jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
3069d1c781a80e19bbf1f553d75dbde7565265dc
utun2/src/chat/chat_msg.c
T

2044 lines
110 KiB
C
Raw Blame History

/*
* chat_msg.c — сообщения: отправка, DB-операции для chat_sync
*
* Вынесено из chat_core.c для уменьшения размера модуля.
*/
#include "chat_core_priv.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "chat_attachment.h"
#include "chat_event.h"
#include "chat_feedback.h"
#include "chat_poll.h"
#include "chat_setting.h"
#include "chat_whisper.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "../transport_layer/secure_channel.h"
#include "../media_delivery/media_index.h"
#include "../media_delivery/media_download.h"
#include "../media_delivery/media_delivery.h"
#include "../media_async/media_async.h"
#include "../video/video.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../../lib/platform_compat.h"
#include "../../lib/file_io.h"
#include <openssl/sha.h>
#include <openssl/rand.h>
#include <stdlib.h>
static void media_queue_channel_backfill(struct UTUN_INSTANCE* inst, const char* channel);
static void media_startup_backfill_timer_cb(void* arg);
static void media_queue_full_backfill(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->media_backfill_inst || cc->media_backfill_timer) return;
cc->media_backfill_timer = uasync_set_timeout(inst->ua, 20000, inst, media_startup_backfill_timer_cb, "chat_media_bl");
if (!cc->media_backfill_timer) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media backfill: complete scan timer allocation failed");
}
static void wh_transcribe_done_cb(void* arg, const char* ch_id, const char* text, int err,
const uint8_t reply_id[32]);
static void chat_media_cache_cleanup_orphans(struct UTUN_INSTANCE* inst);
static void chat_media_cache_reconcile(struct UTUN_INSTANCE* inst);
/* ─── счётчик объёма медиакеша (сумма файлов, соответствующих сообщениям) ─── */
static void cache_bytes_add(struct chat_core_ctx* cc, int64_t sz) {
if (sz > 0) cc->media_cache_bytes += (uint64_t)sz;
}
static void cache_bytes_sub(struct chat_core_ctx* cc, int64_t sz) {
uint64_t v = sz > 0 ? (uint64_t)sz : 0;
cc->media_cache_bytes = cc->media_cache_bytes >= v ? cc->media_cache_bytes - v : 0;
}
/* Единый разбор JSON для live/manual/startup/cache:полные строки без скрытого NUL и усечения.
* 0 — body принадлежит caller/u_free,1 — обычный текст без manifest,-1 — SQL/формат,-2 — память. */
static int msg_media_fields(sqlite3* db, const char* data, size_t bytes, const char* attributes,
char** body, char content_type[32], char filename[256]) {
*body = NULL; content_type[0] = filename[0] = 0;
if (!data || !bytes || bytes > DB_SYNC_MESSAGE_MAX || memchr(data, 0, bytes)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media fields: invalid JSON bytes=%zu", bytes); return -1;
}
sqlite3_stmt* st = NULL; int result = -1;
int rc = sqlite3_prepare_v2(db, "SELECT json_extract(?1,'$.d'),json_extract(?1,'$.ct'),"
"CASE WHEN ?2 IS NULL OR ?2='' THEN NULL ELSE json_extract(?2,'$.fp') END", -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, data, (int)bytes, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = attributes ? sqlite3_bind_text(st, 2, attributes, -1, SQLITE_STATIC) : sqlite3_bind_null(st, 2);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
const size_t limits[] = {DB_SYNC_MESSAGE_MAX, 31, 255};
char* values[3] = {NULL, content_type, filename};
for (int i = 0; i < 3; i++) {
int type = sqlite3_column_type(st, i); const char* value = (const char*)sqlite3_column_text(st, i);
int length = sqlite3_column_bytes(st, i);
if (type == SQLITE_NULL) continue;
if (type != SQLITE_TEXT || !value || length < 0 || (size_t)length > limits[i] || strlen(value) != (size_t)length) {
rc = SQLITE_CORRUPT; break;
}
if (!i) values[0] = (char*)value;
else memcpy(values[i], value, (size_t)length + 1);
}
if (rc == SQLITE_ROW) {
if (!values[0] || !strcmp(content_type, "text") || !strcmp(content_type, "text/plain") ||
!strcmp(content_type, "voice_transcription") || !strchr(values[0], '|')) result = 1;
else {
*body = u_strdup(values[0]); result = *body ? 0 : -2;
}
}
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_ROW || final != SQLITE_OK || result < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media fields: rc=%d finalize=%d result=%d: %s", rc, final, result, sqlite3_errmsg(db));
u_free(*body); *body = NULL; return result == -2 ? -2 : -1;
}
return result;
}
/* ─── отправка сообщения (GUI → uasync) ─── */
/* Локальный ответ ссылается только на существующий оригинал своего канала; wire хранит общий ID. */
static int msg_reply_hex(struct chat_core_ctx* cc, const char* channel, const uint8_t reply[32], char hex[65]) {
static const uint8_t empty[32] = {0};
hex[0] = 0;
if (!memcmp(reply, empty, 32)) return 0;
for (unsigned i = 0; i < 32; i++) snprintf(hex + i * 2, 3, "%02x", reply[i]);
char table[80], sql[160]; msg_table_name(channel, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT 1 FROM \"%s\" WHERE message_id=?", table);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, reply, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int final = sqlite3_finalize(st);
if (rc != SQLITE_ROW || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "reply: original unavailable ch=%s id=%s rc=%d: %s",
channel, hex, rc, sqlite3_errmsg(cc->db)); return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "reply: original resolved ch=%s id=%s", channel, hex);
return 0;
}
int chat_core_submit_message(struct UTUN_INSTANCE* inst, struct chat_msg_submit* req) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !req || (req->data_len && !req->data) || req->data_len > 4000 ||
!memchr(req->channel_id, 0, sizeof(req->channel_id)) || !chat_channel_id(req->channel_id) ||
!memchr(req->content_type, 0, sizeof(req->content_type)) || (req->data_len && memchr(req->data, 0, req->data_len))) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: invalid message submission", CC_ID); return -1;
}
memset(req->message_id, 0, sizeof(req->message_id));
if (!topo_node_sqlite_member_allowed(cc->db, req->channel_id, inst->node_id)) {
DEBUG_WARN(DEBUG_CATEGORY_CHAT, "message denied: local member excluded ch=%s", req->channel_id); return -1;
}
struct DB_SYNC_INSTANCE* si = si_find(inst, req->channel_id);
if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: no history for ch=%s", CC_ID, req->channel_id); return -1; }
/* SQLite формирует JSON без обрезки и с экранированием текста/Unicode. */
char reply_hex[65];
if (msg_reply_hex(cc, req->channel_id, req->reply_to, reply_hex) < 0) return -1;
const char* query = reply_hex[0]
? "SELECT json_object('n',?1,'ch',?2,'ct',?3,'d',?4,'reply_to',?5)"
: "SELECT json_object('n',?1,'ch',?2,'ct',?3,'d',?4)";
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(cc->db, query, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: message JSON: %s", CC_ID, sqlite3_errmsg(cc->db)); return -1;
}
int rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)cc->my_node_id);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 2, req->channel_id, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 3, req->content_type, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 4, req->data ? (const char*)req->data : "", req->data_len, SQLITE_STATIC);
if (rc == SQLITE_OK && reply_hex[0]) rc = sqlite3_bind_text(st, 5, reply_hex, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int bytes = rc == SQLITE_ROW ? sqlite3_column_bytes(st, 0) : 0;
const char* encoded = rc == SQLITE_ROW ? (const char*)sqlite3_column_text(st, 0) : NULL;
char json[4001];
if (rc != SQLITE_ROW || !encoded || bytes <= 0 || bytes >= (int)sizeof(json)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: message JSON rejected ch=%s bytes=%d rc=%d: %s",
CC_ID, req->channel_id, bytes, rc, sqlite3_errmsg(cc->db)); sqlite3_finalize(st); return -1;
}
memcpy(json, encoded, bytes); json[bytes] = 0;
rc = sqlite3_finalize(st);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "message JSON finalize failed rc=%d", rc); return -1; }
uint64_t ts = db_sync_next_timestamp(si);
uint8_t signature[64], message_id[32]; struct db_message_revision revision = {.ver = 1};
if (RAND_bytes(revision.nonce, sizeof(revision.nonce)) != 1 ||
db_sync_sign_message(db_sync_instance_group_id(si), ts, cc->my_node_id, &revision, json, bytes,
inst->my_ed25519_privkey, signature) < 0 ||
chat_message_id(db_sync_instance_group_id(si), ts, cc->my_node_id, revision.nonce, message_id) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: message signing failed ch=%s", CC_ID, req->channel_id); return -1;
}
int result = db_sync_insert_signed(si, json, bytes, signature, 64, ts, NULL, &revision);
if (result != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: message insert failed ch=%s result=%d", CC_ID, req->channel_id, result); return -1;
}
req->timestamp = ts; memcpy(req->message_id, message_id, sizeof(message_id));
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: message saved ch=%s ct=%s ts=%llu bytes=%d",
CC_ID, req->channel_id, req->content_type, (unsigned long long)ts, bytes);
return 0;
}
/* Правим только d существующего text; reply_to и остальные поля остаются в подписанном JSON. */
int chat_core_edit_message(struct UTUN_INSTANCE* inst, const char* channel, const uint8_t message_id[32],
uint64_t expected_ver, const char* text, size_t len) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
struct DB_SYNC_INSTANCE* si = cc && channel ? si_find(inst, channel) : NULL;
if (!cc || !cc->initialized || !si || !message_id || !text || !len || len > DB_SYNC_MESSAGE_MAX || !expected_ver) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "edit: invalid request channel=%s bytes=%zu", channel ? channel : "", len); return -1;
}
char table[80], sql[768], json[DB_SYNC_MESSAGE_MAX + 1]; msg_table_name(channel, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT timestamp,node_id,nonce,ver,edited_at,"
"json_extract(CAST(data AS TEXT),'$.ct'),json_extract(CAST(data AS TEXT),'$.d'),"
"json_set(CAST(data AS TEXT),'$.d',?2) FROM \"%s\" WHERE message_id=?1", table);
sqlite3_stmt* st = NULL; struct db_message_revision revision = {0}; uint64_t ts = 0, observed_ver = 0; int result = -1;
const char* stage = "lookup";
int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 2, text, (int)len, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_ROW) goto done;
ts = (uint64_t)sqlite3_column_int64(st, 0);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 1), ver = (uint64_t)sqlite3_column_int64(st, 3);
observed_ver = ver;
const char* ct = (const char*)sqlite3_column_text(st, 5);
if (author != cc->my_node_id) { stage = "foreign author"; goto done; }
if (!ct || (strcmp(ct, "text") && strcmp(ct, "text/plain"))) { stage = "unsupported content type"; goto done; }
if (ver != expected_ver) { stage = "stale editor version"; result = -2; goto done; }
revision.ver = ver; revision.edited_at = (uint64_t)sqlite3_column_int64(st, 4);
const char* old = (const char*)sqlite3_column_text(st, 6);
if (old && (size_t)sqlite3_column_bytes(st, 6) == len && !memcmp(old, text, len)) { stage = "unchanged"; result = 0; goto done; }
const void* nonce = sqlite3_column_blob(st, 2); int bytes = sqlite3_column_bytes(st, 7);
const char* body = (const char*)sqlite3_column_text(st, 7);
if (!nonce || sqlite3_column_bytes(st, 2) != 16) { stage = "invalid stored nonce"; goto done; }
if (!body || bytes <= 0 || bytes > DB_SYNC_MESSAGE_MAX) { stage = "encoded message exceeds limit"; goto done; }
if (ver == INT64_MAX) { stage = "version exhausted"; goto done; }
memcpy(revision.nonce, nonce, 16); revision.ver = ver + 1;
uint64_t edited = (uint64_t)sqlite3_column_int64(st, 4);
revision.edited_at = db_sync_next_timestamp(si);
if (revision.edited_at <= ts) revision.edited_at = ts + 1;
if (revision.edited_at <= edited) revision.edited_at = edited + 1;
memcpy(json, body, bytes); json[bytes] = 0;
rc = sqlite3_finalize(st); st = NULL;
uint8_t signature[64];
stage = "sign/store";
if (rc == SQLITE_OK && db_sync_sign_message(db_sync_instance_group_id(si), ts, author, &revision, json, bytes,
inst->my_ed25519_privkey, signature) == 0 &&
db_sync_insert_signed(si, json, bytes, signature, 64, ts, NULL, &revision) == 0) result = 0;
done:
{
int final = sqlite3_finalize(st);
if (final != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "edit finalize failed rc=%d", final); result = -1; }
}
if (result) DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "edit: rejected ch=%s stage=%s expected=%llu stored=%llu result=%d sql=%d: %s",
channel, stage, (unsigned long long)expected_ver, (unsigned long long)observed_ver, result, rc, sqlite3_errmsg(cc->db));
else DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "edit: %s ch=%s id=%02x%02x%02x%02x ver=%llu edited=%llu",
!strcmp(stage, "unchanged") ? "unchanged" : "saved", channel, message_id[0], message_id[1], message_id[2], message_id[3],
(unsigned long long)(revision.ver ? revision.ver : expected_ver), (unsigned long long)revision.edited_at);
return result;
}
void chat_core_submit_trampoline(void* arg) {
struct chat_msg_submit* req = (struct chat_msg_submit*)arg;
if (!req) return;
if (!memchr(req->channel_id, 0, sizeof(req->channel_id))) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "message trampoline: unterminated channel");
chat_event_submitted(req->inst, req->request_id, 0); u_free(req); return;
}
if (req->expected_ver) {
int result = chat_core_edit_message(req->inst, req->channel_id, req->message_id,
req->expected_ver, (const char*)req->data, req->data_len);
chat_event_submitted(req->inst, req->request_id, result == 0);
} else if (req->media_src[0] != '\0')
chat_core_submit_media_message(req->inst, req, NULL, NULL);
else {
int result = chat_core_submit_message(req->inst, req);
chat_event_submitted(req->inst, req->request_id, result == 0);
}
u_free(req);
}
/* ─── Медиа-сообщения: async регистрация в media_files + отправка ─── */
struct media_submit_ctx {
struct media_submit_ctx* next;
struct media_destination* destination; // Общая с загрузчиком резервация пути до завершения workers.
int delete_pending;
struct UTUN_INSTANCE* inst;
char channel_id[64];
char content_type[32];
char media_dest[1024];
uint8_t msg_data[4096];
uint32_t msg_data_len;
uint64_t timestamp;
int64_t duration_ms; /* видео: длительность после транскода */
int width; /* видео: размеры после транскода */
int height;
char reply_hex[65];
uint64_t request_id;
chat_media_submitted_fn complete;
void* complete_arg;
};
/* Завершение владеющей асинхронной отправки, включая все ошибки подготовки/регистрации. */
static void media_submit_finish(struct media_submit_ctx* ctx, int success) {
struct UTUN_INSTANCE* inst = ctx->inst;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
chat_media_submitted_fn complete = ctx->complete; void* complete_arg = ctx->complete_arg;
char channel[64]; snprintf(channel, sizeof(channel), "%s", ctx->channel_id);
/* Синхронный обработчик события может запросить удаление; ctx пока остаётся в списке владельца. */
chat_event_submitted(ctx->inst, ctx->request_id, success);
int delete_pending = ctx->delete_pending;
struct media_submit_ctx** slot = &cc->media_submissions;
while (*slot && *slot != ctx) slot = &(*slot)->next;
if (*slot) *slot = ctx->next;
else DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media submission missing from owner ch=%s", channel);
media_delivery_release(inst, ctx->destination);
u_free(ctx);
if (delete_pending) {
int waiting = 0;
for (struct media_submit_ctx* pending = cc->media_submissions; pending; pending = pending->next)
if (!strcmp(pending->channel_id, channel)) { waiting = 1; break; }
if (!waiting) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "media submissions drained; continuing channel deletion ch=%s", channel);
chat_core_delete_channel(inst, channel);
}
}
if (complete) complete(complete_arg, success);
}
int chat_media_defer_channel_delete(struct UTUN_INSTANCE* inst, const char* channel) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL; int pending = 0;
for (struct media_submit_ctx* ctx = cc->media_submissions; ctx; ctx = ctx->next) {
if (strcmp(ctx->channel_id, channel)) continue;
ctx->delete_pending = 1; pending++;
}
if (pending) DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "channel deletion waiting for outgoing media ch=%s tasks=%d", channel, pending);
return pending != 0;
}
static void on_media_registered(void* arg, int err, const struct media_index_result* result) {
struct media_submit_ctx* mctx = (struct media_submit_ctx*)arg;
struct chat_core_ctx* cc = CC(mctx->inst);
if (err || !result) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: media registration failed ch=%s err=%d",
CC_ID, mctx->channel_id, err);
media_submit_finish(mctx, 0);
return;
}
if (mctx->delete_pending) {
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "prepared media retained for pending deletion ch=%s", mctx->channel_id);
media_submit_finish(mctx, 0); return;
}
struct DB_SYNC_INSTANCE* si = si_find(mctx->inst, mctx->channel_id);
if (!si) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: no db_sync instance for ch=%s", CC_ID, mctx->channel_id);
media_submit_finish(mctx, 0);
return;
}
/* build full_data: <base>|<file_size>|<block_size>|<num_blocks>|<media_id_hex>|<content_hash_hex>|<id0,sig0,...> */
size_t base_len = mctx->msg_data_len;
int nb = result->num_blocks;
size_t pairs_len = (size_t)nb * (32 + 1 + 128 + 1); /* id_hex, sig_hex, commas */
size_t extra = 128 + 32 + 64 + pairs_len + 256;
size_t fdcap = base_len + extra;
char* full_data = u_malloc(fdcap);
if (!full_data) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: data alloc fail", CC_ID); media_submit_finish(mctx, 0); return; }
size_t off = 0;
if (base_len > 0) {
char b64[2048]; size_t b64len = b64_encode(mctx->msg_data, base_len, b64, sizeof(b64));
full_data[off++] = '*';
memcpy(full_data + off, b64, b64len); off += b64len;
}
off += snprintf(full_data + off, fdcap - off, "|%lld|%lld|%d|",
(long long)result->file_size, (long long)result->block_size, nb);
for (int i = 0; i < 16; i++) off += snprintf(full_data + off, fdcap - off, "%02x", result->media_id[i]);
off += snprintf(full_data + off, fdcap - off, "|");
for (int i = 0; i < 32; i++) off += snprintf(full_data + off, fdcap - off, "%02x", result->content_hash[i]);
off += snprintf(full_data + off, fdcap - off, "|");
for (int n = 0; n < nb; n++) {
if (n > 0) full_data[off++] = ',';
for (int i = 0; i < 16; i++)
off += snprintf(full_data + off, fdcap - off, "%02x", result->block_ids[n * 16 + i]);
full_data[off++] = ',';
for (int i = 0; i < 64; i++)
off += snprintf(full_data + off, fdcap - off, "%02x", result->block_sigs[n * 64 + i]);
}
/* Длительность/размеры входят в подписанное тело после списка блоков. */
if (mctx->content_type[0] && strcmp(mctx->content_type, "video/mp4") == 0 && mctx->duration_ms > 0) {
off += snprintf(full_data + off, fdcap - off, "|dur=%lld|w=%d|h=%d",
(long long)mctx->duration_ms, mctx->width, mctx->height);
}
/* для фото дописываем разрешение в конец */
else if (mctx->content_type[0] && strncmp(mctx->content_type, "image/", 6) == 0
&& (mctx->width > 0 || mctx->height > 0)) {
off += snprintf(full_data + off, fdcap - off, "|w=%d|h=%d", mctx->width, mctx->height);
}
size_t full_data_len = off;
size_t json_cap = full_data_len + 384;
char* json = u_malloc(json_cap);
if (!json) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: json alloc fail", CC_ID); u_free(full_data); media_submit_finish(mctx, 0); return; }
snprintf(json, json_cap,
"{\"n\":%llu,\"ch\":\"%s\",\"ct\":\"%s\",\"d\":\"%.*s\"%s%s%s}",
(unsigned long long)cc->my_node_id, mctx->channel_id,
mctx->content_type, (int)full_data_len, full_data,
mctx->reply_hex[0] ? ",\"reply_to\":\"" : "", mctx->reply_hex, mctx->reply_hex[0] ? "\"" : "");
size_t json_len = strlen(json);
if (json_len > DB_SYNC_MESSAGE_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media manifest exceeds history limit ch=%s blocks=%d bytes=%zu limit=%u",
mctx->channel_id, nb, json_len, (unsigned)DB_SYNC_MESSAGE_MAX);
u_free(json); u_free(full_data); media_submit_finish(mctx, 0); return;
}
uint64_t ts = db_sync_next_timestamp(si);
struct db_message_revision revision = {.ver = 1}; uint8_t json_sig[64], message_id[32];
if (RAND_bytes(revision.nonce, sizeof(revision.nonce)) != 1 ||
db_sync_sign_message(db_sync_instance_group_id(si), ts, cc->my_node_id, &revision, json, json_len,
mctx->inst->my_ed25519_privkey, json_sig) < 0 ||
chat_message_id(db_sync_instance_group_id(si), ts, cc->my_node_id, revision.nonce, message_id) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: Ed25519 sign JSON failed", CC_ID);
u_free(json); u_free(full_data); media_submit_finish(mctx, 0); return;
}
/* local_attrs готовим заранее и пишем атомарно при вставке — иначе GUI по
MSG_RECEIVED может прочитать сообщение раньше отдельного UPDATE (гонка),
и своё голосовое покажется «без медиа» (серый media bar + попытка скачать) */
const char* base = strrchr(mctx->media_dest, '/');
const char* filename = base ? base + 1 : mctx->media_dest;
char attrs[1100]; snprintf(attrs, sizeof(attrs), "{\"st\":\"fl\",\"fp\":\"%s\"}", filename);
if (!strcmp(mctx->content_type, ENCRYPTED_ATTACHMENT_TYPE)) {
uint8_t crypto_id[16];
if (chat_attachment_identity(full_data, full_data_len, crypto_id) ||
chat_attachment_bind(mctx->inst, mctx->channel_id, crypto_id, message_id, result->content_hash)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "encrypted media: cannot publish without bound author key ch=%s", mctx->channel_id);
u_free(json); u_free(full_data); media_submit_finish(mctx, 0); return;
}
}
int ret = db_sync_insert_signed(si, json, json_len, json_sig, 64, ts, attrs, &revision);
if (ret != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: media insert failed ret=%d", CC_ID, ret);
} else {
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: media message inserted ch=%s ts=%llu",
CC_ID, mctx->channel_id, (unsigned long long)ts);
/* анонсируем блоки суперузлам — иначе первый скачивающий не найдёт держателя */
media_delivery_announce_media(mctx->inst, strtoull(mctx->channel_id, NULL, 10),
result->media_id, result->block_ids, result->num_blocks);
/* учесть локально отправленный файл в счётчике и применить лимит медиакеша */
cache_bytes_add(cc, result->file_size);
chat_media_cache_enforce(mctx->inst);
/* whisper транскрипция для своих голосовых сообщений */
if (mctx->content_type[0] && strncmp(mctx->content_type, "voice", 5) == 0
&& chat_setting_get_int(mctx->inst, "whisper_enabled", 0)) {
chat_whisper_transcribe(mctx->inst, mctx->media_dest, mctx->channel_id,
message_id,
wh_transcribe_done_cb, mctx->inst);
}
}
u_free(json); u_free(full_data); media_submit_finish(mctx, ret == 0);
}
static void on_video_prepared(void* arg, int err, const struct video_info* vi) {
struct media_submit_ctx* mctx = (struct media_submit_ctx*)arg;
struct chat_core_ctx* cc = CC(mctx->inst);
if (err) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: video prepare failed ch=%s err=%d",
CC_ID, mctx->channel_id, err);
media_submit_finish(mctx, 0);
return;
}
/* после транскода знаем итоговую длительность и размеры — для тела сообщения */
if (vi && vi->is_video) {
mctx->duration_ms = vi->duration_ms;
mctx->width = vi->width;
mctx->height = vi->height;
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: video prepared %dx%d dur=%lldms",
CC_ID, vi->width, vi->height, (long long)vi->duration_ms);
}
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) { media_submit_finish(mctx, 0); return; }
media_index_register_async(
mctx->inst->media_async, mctx->inst->ua, cc->db,
cc->my_node_id, mctx->inst->my_ed25519_privkey,
mctx->channel_id,
mctx->media_dest, mctx->media_dest, 0,
media_base,
on_media_registered, mctx);
}
void chat_core_submit_media_message(struct UTUN_INSTANCE* inst, struct chat_msg_submit* req,
chat_media_submitted_fn complete, void* arg) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !req || !req->media_src[0] || !memchr(req->channel_id, 0, sizeof(req->channel_id)) ||
!memchr(req->content_type, 0, sizeof(req->content_type)) || !memchr(req->media_src, 0, sizeof(req->media_src)) ||
!memchr(req->media_dest, 0, sizeof(req->media_dest)) || !chat_channel_id(req->channel_id) || req->data_len >= 4096 ||
(req->data_len && !req->data)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: invalid media submit/stopped core", CC_ID);
if (req) chat_event_submitted(inst, req->request_id, 0);
if (complete) complete(arg, 0);
return;
}
char media_base[MEDIA_BASE_MAX], location[MEDIA_PATH_MAX], destination[MEDIA_PATH_MAX];
if (media_index_get_base(cc->db, media_base) || media_index_location(req->media_dest, media_base, location)
|| media_index_path(media_base, location, destination)) {
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return;
}
for (struct media_submit_ctx* pending = cc->media_submissions; pending; pending = pending->next) {
if ((!strcmp(pending->channel_id, req->channel_id) && pending->delete_pending)
|| !strcmp(pending->media_dest, destination)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media submit rejected: channel deleting/destination busy ch=%s path=%s",
req->channel_id, req->media_dest);
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return;
}
}
if (!si_find(inst, req->channel_id)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media submit without channel ch=%s", req->channel_id);
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return;
}
char reply_hex[65];
if (msg_reply_hex(cc, req->channel_id, req->reply_to, reply_hex) < 0) {
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: MEDIA SUBMIT ch=%s ct=%s src=%s dst=%s copy=%d video=%d",
CC_ID, req->channel_id, req->content_type,
req->media_src, req->media_dest, req->media_copy, req->media_video);
if (!inst->media_async) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: media_async not initialized", CC_ID);
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0);
return;
}
/* copy fields for async callback (req is freed by trampoline immediately) */
struct media_submit_ctx* mctx = u_calloc(1, sizeof(*mctx));
if (!mctx) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: media context allocation failed", CC_ID);
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return;
}
mctx->destination = media_delivery_reserve(inst, destination);
if (!mctx->destination) { u_free(mctx); chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return; }
if (media_index_track_preparation(cc->db, req->channel_id, destination, media_base)) {
media_delivery_release(inst, mctx->destination); u_free(mctx);
chat_event_submitted(inst, req->request_id, 0); if (complete) complete(arg, 0); return;
}
mctx->inst = inst;
mctx->request_id = req->request_id;
mctx->complete = complete; mctx->complete_arg = arg;
mctx->next = cc->media_submissions; cc->media_submissions = mctx;
memcpy(mctx->reply_hex, reply_hex, strlen(reply_hex) + 1);
snprintf(mctx->channel_id, sizeof(mctx->channel_id), "%s", req->channel_id);
snprintf(mctx->content_type, sizeof(mctx->content_type), "%s", req->content_type);
snprintf(mctx->media_dest, sizeof(mctx->media_dest), "%s", destination);
mctx->timestamp = req->timestamp;
mctx->duration_ms = req->duration_ms;
mctx->width = req->width;
mctx->height = req->height;
if (req->data && req->data_len > 0) {
mctx->msg_data_len = req->data_len < sizeof(mctx->msg_data) ? req->data_len : sizeof(mctx->msg_data) - 1;
memcpy(mctx->msg_data, req->data, mctx->msg_data_len);
}
if (req->media_video) {
if (video_transcode_start(inst->media_async, inst->ua, req->media_src, destination, on_video_prepared, mctx) < 0)
on_video_prepared(mctx, -1, NULL);
return;
}
media_index_register_async(
inst->media_async, inst->ua, cc->db,
cc->my_node_id, inst->my_ed25519_privkey,
req->channel_id,
req->media_src, destination, req->media_copy ? 1 : 0,
media_base,
on_media_registered, mctx);
}
/* Локальные атрибуты не подписаны: отдельный commit, JSON меняется целиком либо через json_set. */
static int msg_local_attrs_write(struct UTUN_INSTANCE* inst, const char* channel, uint64_t timestamp,
const uint8_t* signature, uint64_t author, const char* attributes) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
const char* action = signature ? "replace" : "voice played";
if (!cc || !cc->initialized || !chat_channel_id(channel) || !timestamp || timestamp > INT64_MAX ||
(signature ? !attributes : !author) || !sqlite3_get_autocommit(cc->db)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "local attrs: invalid state/arguments/transaction action=%s", action); return -1;
}
sqlite3* db = cc->db;
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE /* local attrs */", NULL, NULL, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "local attrs: begin rc=%d: %s", rc, sqlite3_errmsg(db)); return -1;
}
char table[80], sql[768]; msg_table_name(channel, table, sizeof(table));
if (signature) snprintf(sql, sizeof(sql), "UPDATE \"%s\" SET local_attrs=?3 WHERE timestamp=?1 AND author_signature=?2"
" AND CASE WHEN json_valid(?3) THEN json_type(?3)='object' ELSE 0 END RETURNING id", table);
else snprintf(sql, sizeof(sql), "UPDATE \"%s\" SET local_attrs=json_set(CASE WHEN local_attrs IS NULL OR local_attrs=''"
" THEN '{}' ELSE local_attrs END,'$.pl',1) WHERE id=(SELECT id FROM \"%s\" WHERE timestamp=?1 AND node_id=?2"
" ORDER BY id LIMIT 1) AND CASE WHEN local_attrs IS NULL OR local_attrs='' THEN 1 WHEN json_valid(local_attrs)"
" THEN json_type(local_attrs)='object' ELSE 0 END RETURNING id", table, table);
sqlite3_stmt* st = NULL; int rows = 0;
rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)timestamp);
if (rc == SQLITE_OK) rc = signature ? sqlite3_bind_blob(st, 2, signature, DB_SIG_SIZE, SQLITE_STATIC)
: sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
if (rc == SQLITE_OK && signature) rc = sqlite3_bind_text(st, 3, attributes, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) rows++;
int final = sqlite3_finalize(st);
if (rc == SQLITE_DONE && final == SQLITE_OK && rows > 0) rc = sqlite3_exec(db, "COMMIT /* local attrs */", NULL, NULL, NULL);
else if (rc == SQLITE_DONE) rc = final != SQLITE_OK ? final : SQLITE_NOTFOUND;
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "local attrs: action=%s channel=%s rows=%d rc=%d finalize=%d: %s",
action, channel, rows, rc, final, sqlite3_errmsg(db));
if (!sqlite3_get_autocommit(db)) {
int rollback = sqlite3_exec(db, "ROLLBACK /* local attrs */", NULL, NULL, NULL);
if (rollback != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "local attrs: rollback rc=%d", rollback);
if (!sqlite3_get_autocommit(db)) utun_instance_stop(inst);
}
return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "local attrs: committed action=%s channel=%s timestamp=%llu rows=%d",
action, channel, (unsigned long long)timestamp, rows); return 0;
}
int chat_core_update_local_attrs(struct UTUN_INSTANCE* inst, const char* channel, uint64_t timestamp,
const uint8_t* signature, const char* attributes) {
if (!signature || !attributes) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "local attrs: signature/JSON required"); return -1; }
return msg_local_attrs_write(inst, channel, timestamp, signature, 0, attributes);
}
void chat_core_mark_voice_played(struct UTUN_INSTANCE* inst, const char* channel, uint64_t timestamp, uint64_t author) {
msg_local_attrs_write(inst, channel, timestamp, NULL, author, NULL);
}
void chat_core_mark_voice_played_trampoline(void* arg) {
struct voice_played_req* req = (struct voice_played_req*)arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "voice played: null request"); return; }
if (!memchr(req->channel_id, 0, sizeof(req->channel_id))) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "voice played: unterminated channel");
else chat_core_mark_voice_played(req->inst, req->channel_id, req->ts, req->author_node_id);
u_free(req);
}
/* ─── DB-операции для chat_sync ─── */
uint32_t chat_core_count(struct UTUN_INSTANCE* inst, const char* channel) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !chat_channel_id(channel)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "message count: invalid context/channel"); return 0;
}
char table[80], sql[128]; msg_table_name(channel, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM \"%s\"", table);
sqlite3_stmt* st = NULL; sqlite3_int64 count = -1;
int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW) count = sqlite3_column_int64(st, 0);
int final = sqlite3_finalize(st);
if (rc != SQLITE_ROW || final != SQLITE_OK || count < 0 || count > UINT32_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "message count: channel=%s rc=%d finalize=%d count=%lld: %s",
channel, rc, final, (long long)count, sqlite3_errmsg(cc->db)); return 0;
}
return (uint32_t)count;
}
/* Ограниченный буфер даёт только целые первые записи;SQL всё равно читается до конца для проверки ошибок. */
static int msg_list_ids(struct UTUN_INSTANCE* inst, const char* channel, uint8_t* buf, size_t capacity, size_t* length) {
if (length) *length = 0;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !buf || !length || capacity < 2 || (channel && !chat_channel_id(channel))) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "ID list: invalid context/channel/buffer capacity=%zu", capacity); return -1;
}
size_t maximum = 2 + (size_t)UINT16_MAX * (channel ? 8 : 21);
if (capacity > maximum) capacity = maximum;
uint8_t* temporary = u_malloc(capacity);
if (!temporary) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "ID list: allocation bytes=%zu failed", capacity); return -1; }
char sql[160];
if (channel) snprintf(sql, sizeof(sql), "SELECT node_id FROM peers_%s WHERE deleted=0 AND admitted=1 ORDER BY node_id<0,node_id", channel);
else strcpy(sql, "SELECT channel_id FROM channels ORDER BY created_at,channel_id");
sqlite3_stmt* st = NULL; size_t used = 2; uint16_t count = 0; int full = 0;
int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
uint64_t node = 0; const char* id = NULL; size_t bytes = 8;
if (channel) {
if (sqlite3_column_type(st, 0) != SQLITE_INTEGER || !(node = (uint64_t)sqlite3_column_int64(st, 0))) {
rc = SQLITE_CORRUPT; break;
}
} else {
int type = sqlite3_column_type(st, 0); id = (const char*)sqlite3_column_text(st, 0);
int size = sqlite3_column_bytes(st, 0);
if (type != SQLITE_TEXT || !id || size < 1 || size > 19 || strlen(id) != (size_t)size || !chat_channel_id(id)) {
rc = SQLITE_CORRUPT; break;
}
bytes = (size_t)size + 1;
}
if (full || count == UINT16_MAX || bytes > capacity - used) { full = 1; continue; }
if (channel) memcpy(temporary + used, &node, 8);
else { temporary[used] = (uint8_t)(bytes - 1); memcpy(temporary + used + 1, id, bytes - 1); }
used += bytes; count++;
}
int final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "ID list: channel=%s rc=%d finalize=%d: %s",
channel ? channel : "all", rc, final, sqlite3_errmsg(cc->db)); u_free(temporary); return -1;
}
memcpy(temporary, &count, 2); memcpy(buf, temporary, used); *length = used; u_free(temporary);
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "ID list: channel=%s count=%u bytes=%zu limited=%d", channel ? channel : "all", count, used, full);
return 0;
}
int chat_core_list_channels(struct UTUN_INSTANCE* inst, uint8_t* buf, size_t capacity, size_t* length) {
return msg_list_ids(inst, NULL, buf, capacity, length);
}
int chat_core_list_peers(struct UTUN_INSTANCE* inst, const char* channel, uint8_t* buf, size_t capacity, size_t* length) {
if (!channel) { if (length) *length = 0; DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "peer list: channel required"); return -1; }
return msg_list_ids(inst, channel, buf, capacity, length);
}
int chat_core_load_nodeinfo(struct UTUN_INSTANCE* inst, uint64_t node, uint8_t* buf, size_t capacity, size_t* length) {
if (length) *length = 0;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !node || !buf || !length || capacity < 65) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "node info: invalid arguments/capacity=%zu", capacity); return -1;
}
uint8_t temporary[65 + 255 * 23] = {0}; size_t used = 65; unsigned count = 0; int full = 0;
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(cc->db, "SELECT x25519_pubkey,ed25519_pubkey FROM nodes WHERE node_id=?", -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
for (int i = 0; i < 2; i++) {
int type = sqlite3_column_type(st, i); const void* key = sqlite3_column_blob(st, i);
int bytes = sqlite3_column_bytes(st, i);
if (type == SQLITE_NULL) continue;
if (type != SQLITE_BLOB || !key || bytes != 32) { rc = SQLITE_CORRUPT; break; }
memcpy(temporary + i * 32, key, 32);
}
}
int final = sqlite3_finalize(st); st = NULL;
if (rc != SQLITE_ROW || final != SQLITE_OK) goto fail;
rc = sqlite3_prepare_v2(cc->db, "SELECT family,protocol,address,port,rtt FROM node_addresses WHERE node_id=?"
" ORDER BY family,protocol,socket_id,address_kind,address,port", -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node);
if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
int family_type = sqlite3_column_type(st, 0), proto_type = sqlite3_column_type(st, 1);
int address_type = sqlite3_column_type(st, 2), port_type = sqlite3_column_type(st, 3), rtt_type = sqlite3_column_type(st, 4);
int family = sqlite3_column_int(st, 0), protocol = sqlite3_column_int(st, 1), port = sqlite3_column_int(st, 3);
int rtt = sqlite3_column_int(st, 4), bytes = sqlite3_column_bytes(st, 2);
const void* address = sqlite3_column_blob(st, 2);
if (family_type != SQLITE_INTEGER || proto_type != SQLITE_INTEGER || address_type != SQLITE_BLOB ||
port_type != SQLITE_INTEGER || (rtt_type != SQLITE_INTEGER && rtt_type != SQLITE_NULL) ||
(family != 4 && family != 6) || bytes != (family == 4 ? 4 : 16) || !address || port < 0 || port > UINT16_MAX ||
(protocol != 1 && protocol != 2 && protocol != 6)) { rc = SQLITE_CORRUPT; break; }
if (full || count == 255 || (size_t)bytes + 7 > capacity - used) { full = 1; continue; }
temporary[used++] = (uint8_t)family; temporary[used++] = (uint8_t)protocol; temporary[used++] = (uint8_t)bytes;
memcpy(temporary + used, address, (size_t)bytes); used += (size_t)bytes;
uint16_t wire_port = (uint16_t)port; int16_t wire_rtt = rtt < INT16_MIN ? INT16_MIN : rtt > INT16_MAX ? INT16_MAX : (int16_t)rtt;
memcpy(temporary + used, &wire_port, 2); used += 2; memcpy(temporary + used, &wire_rtt, 2); used += 2; count++;
}
final = sqlite3_finalize(st);
if (rc != SQLITE_DONE || final != SQLITE_OK) goto fail;
temporary[64] = (uint8_t)count; memcpy(buf, temporary, used); *length = used;
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "node info: node=%016llx addresses=%u bytes=%zu limited=%d", (unsigned long long)node, count, used, full);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "node info: node=%016llx rc=%d finalize=%d: %s",
(unsigned long long)node, rc, final, sqlite3_errmsg(cc->db)); return -1;
}
/* ─── download completion ─── */
struct md_done_ctx {
struct UTUN_INSTANCE* inst;
char channel_id[64];
uint64_t ts;
int64_t msg_id;
uint8_t author_sig[64];
uint8_t message_id[32];
char dest_relpath[512];
char dest_abs_path[1024];
char content_type[32];
uint64_t author_node_id;
};
/* Локальный номер не является сетевым ID; асинхронная загрузка сохраняет постоянный ключ из БД. */
static int msg_read_id(struct chat_core_ctx* cc, const char* channel, int64_t id, uint8_t message_id[32]) {
char table[80], sql[160]; msg_table_name(channel, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT message_id FROM \"%s\" WHERE id=?", table);
sqlite3_stmt* st = NULL; int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL), result = -1;
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, id);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 32) {
const void* key = sqlite3_column_blob(st, 0);
if (key) { memcpy(message_id, key, 32); result = 0; }
}
int final_rc = sqlite3_finalize(st);
if (final_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media identity finalize failed ch=%s id=%lld rc=%d", channel, (long long)id, final_rc);
result = -1;
}
if (result) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media identity missing ch=%s id=%lld rc=%d: %s",
channel, (long long)id, rc, sqlite3_errmsg(cc->db));
return result;
}
static void md_download_progress_cb(void* arg, int blocks_done, int num_blocks) {
struct md_done_ctx* ctx = (struct md_done_ctx*)arg;
uint8_t evt[256];
uint8_t cl = (uint8_t)strlen(ctx->channel_id);
evt[0] = cl; memcpy(evt + 1, ctx->channel_id, cl);
memcpy(evt + 1 + cl, &ctx->msg_id, 8);
uint32_t bd = (uint32_t)blocks_done, nb = (uint32_t)num_blocks;
memcpy(evt + 1 + cl + 8, &bd, 4); memcpy(evt + 1 + cl + 12, &nb, 4);
chat_event_post(ctx->inst, CHAT_EVT_DOWNLOAD_PROGRESS, evt, 1 + cl + 16);
}
static void md_download_done_cb(void* arg, int err) {
struct md_done_ctx* ctx = (struct md_done_ctx*)arg;
struct chat_core_ctx* cc = CC(ctx->inst);
if (!err) {
if (!strcmp(ctx->content_type, ENCRYPTED_ATTACHMENT_TYPE)) {
if (chat_attachment_downloaded(ctx->inst, ctx->channel_id, ctx->message_id, 1))
media_queue_channel_backfill(ctx->inst, ctx->channel_id);
}
char attrs[1024];
snprintf(attrs, sizeof(attrs), "{\"st\":\"fl\",\"fp\":\"%s\"}", ctx->dest_relpath);
chat_core_update_local_attrs(ctx->inst, ctx->channel_id, ctx->ts, ctx->author_sig, attrs);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: media downloaded ch=%s", CC_ID, ctx->channel_id);
/* whisper транскрипция для голосовых сообщений */
if (ctx->content_type[0] && strncmp(ctx->content_type, "voice", 5) == 0
&& chat_setting_get_int(ctx->inst, "whisper_enabled", 0)) {
chat_whisper_transcribe(ctx->inst, ctx->dest_abs_path, ctx->channel_id,
ctx->message_id,
wh_transcribe_done_cb, ctx->inst);
}
/* после докачки — учесть файл в счётчике и применить лимит медиакеша */
cache_bytes_add(cc, ma_file_size(ctx->dest_abs_path));
chat_media_cache_enforce(ctx->inst);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: media download failed ch=%s err=%d", CC_ID, ctx->channel_id, err);
/* пометить сообщение ошибкой, чтобы UI показал «повторить» */
char attrs[64];
snprintf(attrs, sizeof(attrs), "{\"st\":\"er\"}");
chat_core_update_local_attrs(ctx->inst, ctx->channel_id, ctx->ts, ctx->author_sig, attrs);
}
uint8_t evt[80]; uint8_t cl = (uint8_t)strlen(ctx->channel_id);
evt[0] = cl; memcpy(evt + 1, ctx->channel_id, cl);
memcpy(evt + 1 + cl, &ctx->msg_id, 8);
chat_event_post(ctx->inst, CHAT_EVT_ATTACHMENT_DOWNLOADED, evt, 1 + cl + 8);
u_free(ctx);
}
static void wh_transcribe_done_cb(void* arg, const char* ch_id, const char* text, int err,
const uint8_t reply_id[32]) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
if (err || !text || !text[0]) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: transcription failed ch=%s err=%d", CC_ID, ch_id, err);
return;
}
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !chat_channel_id(ch_id) || !reply_id) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "transcription: inactive core/invalid identity"); return;
}
/* Не публикуем результат после удаления исходного сообщения/канала. */
char table[80], sql[192]; msg_table_name(ch_id, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT timestamp,node_id FROM \"%s\" WHERE message_id=?", table);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, reply_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc != SQLITE_ROW) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "transcription: original unavailable ch=%s rc=%d", ch_id, rc);
int final = sqlite3_finalize(st);
if (final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "transcription: cleanup finalize rc=%d", final);
return;
}
uint64_t ts = (uint64_t)sqlite3_column_int64(st, 0), author = (uint64_t)sqlite3_column_int64(st, 1);
rc = sqlite3_finalize(st);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "transcription: read finalize rc=%d", rc); return; }
struct chat_msg_submit req = {.inst = inst, .data = (uint8_t*)text, .data_len = (uint32_t)strlen(text)};
memcpy(req.channel_id, ch_id, strlen(ch_id) + 1); strcpy(req.content_type, "voice_transcription"); memcpy(req.reply_to, reply_id, 32);
if (chat_core_submit_message(inst, &req)) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "transcription: message save failed ch=%s bytes=%u", ch_id, req.data_len); return;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT, "transcription: saved ch=%s bytes=%u", ch_id, req.data_len);
chat_core_mark_voice_played(inst, ch_id, ts, author);
}
static void sanitize_filename(char* buf, size_t size) {
char orig[256]; snprintf(orig, sizeof(orig), "%s", buf);
size_t w = 0;
for (size_t r = 0; buf[r]; r++) {
char c = buf[r];
if ((unsigned char)c < 0x20) continue;
if (c == '/' || c == '\\' || c == ':' || c == '*' || c == '?'
|| c == '"' || c == '<' || c == '>' || c == '|') continue;
if (w < size - 1) buf[w++] = c;
}
buf[w] = '\0';
if (w == 0) snprintf(buf, size, "file");
if (strcmp(orig, buf) != 0)
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: filename sanitized [%s] -> [%s]", CC_ID, orig, buf);
}
/* Извлекает имя файла из тела "d"-поля медиасообщения: берёт сегмент до '|',
декодирует base64url-префикс '*', если он есть, и санитизирует. */
static void extract_base_filename(const char* body, char* out, size_t out_sz) {
size_t bfi = 0;
while (body[bfi] && body[bfi] != '|' && bfi < out_sz - 1)
out[bfi++] = body[bfi];
out[bfi] = '\0';
if (out[0] == '*' && bfi > 1) {
uint8_t dec[256];
int dlen = (int)b64_decode(out + 1, bfi - 1, dec, sizeof(dec));
if (dlen > 0 && dlen < (int)out_sz) { memcpy(out, dec, (size_t)dlen); out[dlen] = '\0'; }
}
if (bfi == 0) snprintf(out, out_sz, "file");
sanitize_filename(out, out_sz);
}
/* Беззнаковое десятичное число до delimiter; delimiter=0 требует точного конца строки. */
static int msg_media_number(const char** cursor, const char* end, uint64_t* value, char delimiter) {
const char* p = *cursor; uint64_t number = 0;
if (p == end || *p < '0' || *p > '9') return -1;
while (p < end && *p >= '0' && *p <= '9') {
unsigned digit = (unsigned)(*p++ - '0');
if (number > ((uint64_t)INT64_MAX - digit) / 10) return -1;
number = number * 10 + digit;
}
if (delimiter ? (p == end || *p++ != delimiter) : p != end) return -1;
*cursor = p; *value = number; return 0;
}
static int msg_media_hex(const char** cursor, const char* end, uint8_t* out, size_t bytes, char delimiter) {
const char* p = *cursor;
if ((size_t)(end - p) < bytes * 2 + (delimiter ? 1 : 0)) return -1;
for (size_t i = 0; i < bytes; i++) {
unsigned value = 0;
for (unsigned j = 0; j < 2; j++) {
unsigned c = (unsigned char)*p++, digit;
if (c >= '0' && c <= '9') digit = c - '0';
else if (c >= 'a' && c <= 'f') digit = c - 'a' + 10;
else if (c >= 'A' && c <= 'F') digit = c - 'A' + 10;
else return -1;
value = value * 16 + digit;
}
out[i] = (uint8_t)value;
}
if (delimiter && *p++ != delimiter) return -1;
*cursor = p; return 0;
}
/* Проверяет блоки и необязательный хвост |[dur=<ms>|]w=<width>|h=<height>; памяти хвоста не выделяет. */
int chat_msg_parse_media_body(const char* body, struct media_index_result* output) {
if (!body || !output) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media manifest: body/result required"); return -1; }
memset(output, 0, sizeof(*output));
const char* stage = "length/name";
size_t length = strnlen(body, DB_SYNC_MESSAGE_MAX + 1);
if (length > DB_SYNC_MESSAGE_MAX) goto invalid;
const char* end = body + length; const char* p = memchr(body, '|', length);
if (!p) return -1; /* Обычный текст также проходит через scans. */
if (p == body) goto invalid;
p++;
uint64_t size, block, count;
stage = "block geometry";
if (msg_media_number(&p, end, &size, '|') || msg_media_number(&p, end, &block, '|') ||
msg_media_number(&p, end, &count, '|') || !size || !block || !count ||
count > 100 || (size - 1) / block + 1 != count) goto invalid;
struct media_index_result result = {.file_size = (int64_t)size, .block_size = (int64_t)block, .num_blocks = (int)count};
stage = "media identity";
if (msg_media_hex(&p, end, result.media_id, 16, '|') || msg_media_hex(&p, end, result.content_hash, 32, '|')) goto invalid;
const char* pairs_end = memchr(p, '|', (size_t)(end - p));
if (!pairs_end) pairs_end = end;
stage = "block signature length";
if ((size_t)(pairs_end - p) != (size_t)count * 162 - 1) goto invalid;
if (pairs_end < end) {
const char* metadata = pairs_end + 1; uint64_t duration, width, height;
stage = "media dimensions/duration";
if (end - metadata >= 4 && !memcmp(metadata, "dur=", 4)) {
metadata += 4;
if (msg_media_number(&metadata, end, &duration, '|') || !duration) goto invalid;
}
if (end - metadata < 2 || memcmp(metadata, "w=", 2)) goto invalid;
metadata += 2;
if (msg_media_number(&metadata, end, &width, '|') || width > INT_MAX || end - metadata < 2 ||
memcmp(metadata, "h=", 2)) goto invalid;
metadata += 2;
if (msg_media_number(&metadata, end, &height, 0) || height > INT_MAX || (!width && !height)) goto invalid;
}
result.block_ids = u_malloc((size_t)count * 16); result.block_sigs = u_malloc((size_t)count * 64);
if (!result.block_ids || !result.block_sigs) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media manifest: allocation failed blocks=%llu", (unsigned long long)count);
media_index_result_free(&result); return -2;
}
stage = "block signature encoding";
for (unsigned i = 0; i < count; i++) {
if (msg_media_hex(&p, pairs_end, result.block_ids + i * 16, 16, ',') ||
msg_media_hex(&p, pairs_end, result.block_sigs + i * 64, 64, i + 1 < count ? ',' : 0)) {
media_index_result_free(&result); goto invalid;
}
}
*output = result; return 0;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media manifest: invalid %s bytes=%zu", stage, length); return -1;
}
/* Имя файла на диске: голосовое — voice_<media_id>.opus, иначе — базовое имя из "d". */
static void media_final_name(const struct media_index_result* result, const char* content_type,
const char* base_filename, char* out, size_t out_sz) {
if (content_type && strcmp(content_type, "audio/opus") == 0) {
char hex[33];
for (int i = 0; i < 16; i++) snprintf(hex + i * 2, 3, "%02x", result->media_id[i]);
snprintf(out, out_sz, "voice_%s.opus", hex);
} else {
snprintf(out, out_sz, "%s", base_filename);
}
}
static int md_start_download(struct UTUN_INSTANCE* inst,
const char* data_str, size_t data_len,
const char* ch_id, const char* base_filename,
uint64_t ts, const uint8_t* author_sig,
uint64_t author_node_id, int64_t msg_id,
const char* content_type) {
(void)data_len;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
struct media_index_result result;
if (chat_msg_parse_media_body(data_str, &result) != 0) return -1;
int nb = result.num_blocks;
long long fsize = result.file_size;
/* для голосовых имя генерируем из media_id: msg_data — это wf-строка, а не имя файла */
char final_name[256];
media_final_name(&result, content_type, base_filename, final_name, sizeof(final_name));
char dest[MEDIA_PATH_MAX], media_base[MEDIA_BASE_MAX], relative[MEDIA_PATH_MAX];
int length = snprintf(relative, sizeof(relative), "media/%s/%s", ch_id, final_name);
if (length < 0 || length >= (int)sizeof(relative) || media_index_get_base(cc->db, media_base)
|| media_index_path(media_base, relative, dest)) { media_index_result_free(&result); return -1; }
struct md_done_ctx* ctx = u_calloc(1, sizeof(*ctx));
if (!ctx) { media_index_result_free(&result); return -1; }
ctx->inst = inst;
snprintf(ctx->channel_id, sizeof(ctx->channel_id), "%s", ch_id);
ctx->ts = ts; ctx->msg_id = msg_id;
if (msg_read_id(cc, ch_id, msg_id, ctx->message_id) < 0) {
media_index_result_free(&result); u_free(ctx); return -1;
}
snprintf(ctx->dest_relpath, sizeof(ctx->dest_relpath), "%s", final_name);
snprintf(ctx->dest_abs_path, sizeof(ctx->dest_abs_path), "%s", dest);
ctx->author_node_id = author_node_id;
if (content_type) snprintf(ctx->content_type, sizeof(ctx->content_type), "%s", content_type);
else ctx->content_type[0] = '\0';
if (author_sig) memcpy(ctx->author_sig, author_sig, 64);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: download start ch=%s ct=%s file=%s blocks=%d size=%lld dest=%s",
CC_ID, ch_id, content_type ? content_type : "?", final_name, nb, (long long)fsize, dest);
uint64_t gid = strtoull(ch_id, NULL, 10);
int started = media_download_start(inst, gid, &result, dest, media_base, author_node_id,
md_download_done_cb, ctx, md_download_progress_cb, ctx);
media_index_result_free(&result);
if (started < 0) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media download request declined ch=%s path=%s; see download reason", ch_id, dest); u_free(ctx);
}
return started;
}
static int md_auto_download(struct UTUN_INSTANCE* inst,
const char* data_str, size_t data_len,
const char* ch_id, const char* base_filename,
uint64_t ts, const uint8_t* author_sig,
uint64_t author, int64_t msg_id,
const char* content_type) {
if (!chat_setting_get_int(inst, "storage_autoload", 1)) return 0;
uint64_t max_size = inst->config->global.chatserver_storage_unit_size;
const char* p = data_str;
while (*p && *p != '|') p++;
if (!*p) return 0; p++;
long long fsize = strtoll(p, (char**)&p, 10);
if (max_size > 0 && fsize > (long long)max_size) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: media too large (%lld > %llu bytes), skip auto-download ch=%s file=%s",
CC_ID, fsize, (unsigned long long)max_size, ch_id, base_filename);
return 0;
}
return md_start_download(inst, data_str, data_len, ch_id, base_filename, ts, author_sig, author, msg_id, content_type) == 0 ? 1 : 0;
}
/* ─── db_sync callback ─── */
void on_msg_inserted(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t record_ts,
const char* data, size_t len, uint64_t author, int initial_sync, int updated, void* arg) {
struct UTUN_INSTANCE* inst = arg;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc) return;
char ch_id[32]; snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)db_sync_instance_group_id(si));
if (updated) {
uint8_t evt[65]; uint8_t size = (uint8_t)strlen(ch_id); evt[0] = size; memcpy(evt + 1, ch_id, size);
memcpy(evt + 1 + size, message_id, 32);
chat_event_post(inst, CHAT_EVT_MSG_UPDATED, evt, 1 + size + 32);
return;
}
int attachment_result = chat_attachment_observe(inst, ch_id, message_id, author, data, len);
if (attachment_result) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "encrypted media: manifest indexing deferred/rejected ch=%s result=%d", ch_id, attachment_result);
if (attachment_result == -1) media_queue_channel_backfill(inst, ch_id);
}
sqlite3_stmt* touched = NULL;
int touch = sqlite3_prepare_v2(cc->db, "UPDATE channels SET last_msg_at=max(last_msg_at,?1) WHERE channel_id=?2", -1, &touched, NULL);
if (touch == SQLITE_OK) touch = sqlite3_bind_int64(touched, 1, (sqlite3_int64)record_ts);
if (touch == SQLITE_OK) touch = sqlite3_bind_text(touched, 2, ch_id, -1, SQLITE_STATIC);
if (touch == SQLITE_OK) touch = sqlite3_step(touched);
int touch_final = sqlite3_finalize(touched);
if (touch != SQLITE_DONE || touch_final != SQLITE_OK)
DEBUG_ERROR(DEBUG_CATEGORY_CHAT, "message: channel timestamp failed ch=%s rc=%d finalize=%d", ch_id, touch, touch_final);
chat_feedback_check_message(inst, ch_id, message_id);
uint8_t evt[73]; uint8_t cl = (uint8_t)strlen(ch_id); evt[0] = cl; memcpy(evt + 1, ch_id, cl);
memcpy(evt + 1 + cl, &author, 8);
evt[1 + cl + 8] = initial_sync != 0;
chat_event_post(inst, CHAT_EVT_MSG_RECEIVED, evt, 1 + cl + 9);
if (attachment_result) return; // История уже committed; ошибка производного индекса не отменяет её уведомление.
/* Поля JSON и постоянный ID одинаковы для live-PUSH и повторной загрузки. */
if (data && len && author != cc->my_node_id && !initial_sync) {
char* body = NULL; char content_type[32], filename[256];
int fields = msg_media_fields(cc->db, data, len, NULL, &body, content_type, filename);
if (!fields) {
extract_base_filename(body, filename, sizeof(filename));
char table[80], sql[192]; msg_table_name(ch_id, table, sizeof(table));
snprintf(sql, sizeof(sql), "SELECT author_signature,id FROM \"%s\" WHERE message_id=?", table);
sqlite3_stmt* st = NULL; uint8_t signature[64]; int64_t id = 0;
int rc = sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, message_id, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW && sqlite3_column_bytes(st, 0) == 64 && sqlite3_column_blob(st, 0)) {
memcpy(signature, sqlite3_column_blob(st, 0), 64); id = sqlite3_column_int64(st, 1);
}
int final = sqlite3_finalize(st);
if (rc == SQLITE_ROW && id > 0 && final == SQLITE_OK)
md_auto_download(inst, body, strlen(body), ch_id, filename, record_ts, signature, author, id, content_type);
else DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "live media: identity read failed ch=%s rc=%d finalize=%d", ch_id, rc, final);
} else if (fields < 0) media_queue_channel_backfill(inst, ch_id);
u_free(body);
}
}
/* ─── manual attachment download (called from JNI bridge via uasync post) ─── */
void chat_core_attachment_download(struct UTUN_INSTANCE* inst, const char* channel_id, int64_t msg_id) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !chat_channel_id(channel_id) || msg_id <= 0 || !sqlite3_get_autocommit(cc->db)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "manual media download: invalid state/arguments or external transaction"); return;
}
char tbl[80]; msg_table_name(channel_id, tbl, sizeof(tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT data, timestamp, author_signature, node_id FROM \"%s\" WHERE id=?", tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: download sql prep failed ch=%s", CC_ID, channel_id);
return;
}
int read = sqlite3_bind_int64(st, 1, (sqlite3_int64)msg_id);
if (read == SQLITE_OK) read = sqlite3_step(st);
if (read != SQLITE_ROW) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: download msg not found ch=%s id=%lld", CC_ID, channel_id, (long long)msg_id);
sqlite3_finalize(st);
return;
}
const uint8_t* jdata = sqlite3_column_text(st, 0);
int jlen = sqlite3_column_bytes(st, 0);
int64_t db_ts = sqlite3_column_int64(st, 1);
const uint8_t* sig_blob = (const uint8_t*)sqlite3_column_blob(st, 2);
int sig_len = sqlite3_column_bytes(st, 2);
int64_t author_node_id = sqlite3_column_int64(st, 3);
if (!jdata || jlen <= 0 || jlen > DB_SYNC_MESSAGE_MAX || !sig_blob || sig_len != 64) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: download bad msg data ch=%s id=%lld", CC_ID, channel_id, (long long)msg_id);
sqlite3_finalize(st);
return;
}
char* body = NULL; char content_type[32], ignored_filename[256];
int fields = msg_media_fields(cc->db, (const char*)jdata, (size_t)jlen, NULL, &body, content_type, ignored_filename);
if (fields) { sqlite3_finalize(st); return; }
size_t bi = strlen(body); char body_storage[DB_SYNC_MESSAGE_MAX + 1];
memcpy(body_storage, body, bi + 1); u_free(body); body = body_storage;
char base_filename[256];
extract_base_filename(body, base_filename, sizeof(base_filename));
uint8_t author_sig[64]; memcpy(author_sig, sig_blob, 64);
uint8_t message[32]; int encrypted_attachment = !strcmp(content_type, ENCRYPTED_ATTACHMENT_TYPE);
if (encrypted_attachment) {
int observed = msg_read_id(cc, channel_id, msg_id, message) ? -1 :
chat_attachment_observe(inst, channel_id, message, (uint64_t)author_node_id, (const char*)jdata, (size_t)jlen);
if (observed) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "manual encrypted download: cannot index manifest ch=%s result=%d", channel_id, observed);
if (observed == -1) media_queue_channel_backfill(inst, channel_id);
sqlite3_finalize(st); return;
}
}
int final_rc = sqlite3_finalize(st);
if (final_rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "manual media download: history finalize failed ch=%s rc=%d", channel_id, final_rc); return;
}
if (encrypted_attachment) {
if (chat_attachment_want(inst, channel_id, message)) return;
sqlite3_stmt* encrypted = NULL;
int rc = sqlite3_prepare_v2(cc->db, "SELECT cipher_path FROM attachment_keys WHERE channel=? AND message_id=? AND downloaded=1",
-1, &encrypted, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_text(encrypted, 1, channel_id, -1, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_bind_blob(encrypted, 2, message, 32, SQLITE_STATIC);
if (rc == SQLITE_OK) rc = sqlite3_step(encrypted);
const char* path = rc == SQLITE_ROW ? (const char*)sqlite3_column_text(encrypted, 0) : NULL;
int present = path && ma_file_size(path) > 0;
int finish = sqlite3_finalize(encrypted);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || finish != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "encrypted media: local file lookup failed ch=%s step=%d finalize=%d db=%s",
channel_id, rc, finish, sqlite3_errmsg(cc->db)); return;
}
if (present) {
if (chat_attachment_downloaded(inst, channel_id, message, 0)) media_queue_channel_backfill(inst, channel_id);
return;
}
}
/* сбрасываем пометку «файл есть»: файл мог быть вычищен кешем, а local_attrs ещё "fl".
по завершении md_download_done_cb снова запишет st=fl (успех) или st=er (ошибка). */
chat_core_update_local_attrs(inst, channel_id, (uint64_t)db_ts, author_sig, "{}");
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: manual download ch=%s id=%lld ct=%s file=%s",
CC_ID, channel_id, (long long)msg_id, content_type, base_filename);
md_start_download(inst, body, bi, channel_id, base_filename, (uint64_t)db_ts, author_sig, (uint64_t)author_node_id, msg_id, content_type);
}
static void chat_core_attachment_download_trampoline_impl(void* arg) {
struct attachment_dl_req* req = (struct attachment_dl_req*)arg;
if (!req) return;
chat_core_attachment_download(req->inst, req->channel_id, req->msg_id);
u_free(req);
}
void chat_core_attachment_download_trampoline(void* arg) {
chat_core_attachment_download_trampoline_impl(arg);
}
/* ─── Backfill: регистрация уже скачанных медиафайлов в media_files ─── */
void chat_media_backfill(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->db) return;
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) return;
int total = 0;
sqlite3_stmt* cs = NULL; int rc, failed = 0;
rc = sqlite3_prepare_v2(cc->db, "SELECT channel_id FROM channels WHERE EXISTS(SELECT 1 FROM sqlite_master"
" WHERE type='table' AND name='msg_'||channel_id)", -1, &cs, NULL);
if (rc != SQLITE_OK) goto finish;
while ((rc = sqlite3_step(cs)) == SQLITE_ROW) {
const char* ch_id = (const char*)sqlite3_column_text(cs, 0);
if (!chat_channel_id(ch_id) || strlen(ch_id) != (size_t)sqlite3_column_bytes(cs, 0)) { failed = 1; break; }
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT data,message_id,node_id,local_attrs FROM \"%s\"", tbl);
sqlite3_stmt* ms = NULL;
int mrc = sqlite3_prepare_v2(cc->db, sql, -1, &ms, NULL);
if (mrc != SQLITE_OK) { sqlite3_finalize(ms); media_queue_channel_backfill(inst, ch_id); failed = 1; break; }
while ((mrc = sqlite3_step(ms)) == SQLITE_ROW) {
const char* jdata = (const char*)sqlite3_column_text(ms, 0);
if (!jdata) continue;
char* body = NULL; char content_type[32], fp[256];
const char* attributes = (const char*)sqlite3_column_text(ms, 3);
int fields = msg_media_fields(cc->db, jdata, (size_t)sqlite3_column_bytes(ms, 0), attributes, &body, content_type, fp);
if (fields == 1) continue;
if (fields < 0) { failed = 1; break; }
struct media_index_result result;
int parsed = chat_msg_parse_media_body(body, &result);
if (parsed) { u_free(body); if (parsed == -2) { failed = 1; break; } continue; }
int encrypted_attachment = !strcmp(content_type, ENCRYPTED_ATTACHMENT_TYPE);
uint8_t message[32];
if (encrypted_attachment) {
if (sqlite3_column_bytes(ms, 1) != 32 || !sqlite3_column_blob(ms, 1)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media startup: missing encrypted message identity ch=%s", ch_id);
media_index_result_free(&result); u_free(body); continue;
}
memcpy(message, sqlite3_column_blob(ms, 1), 32);
int observed = chat_attachment_observe(inst, ch_id, message, (uint64_t)sqlite3_column_int64(ms, 2), jdata,
(size_t)sqlite3_column_bytes(ms, 0));
if (observed) {
if (observed == -1) media_queue_channel_backfill(inst, ch_id);
media_index_result_free(&result); u_free(body); continue;
}
}
char base_filename[256];
extract_base_filename(body, base_filename, sizeof(base_filename));
char final_name[256];
if (fp[0]) strcpy(final_name, fp);
else media_final_name(&result, content_type, base_filename, final_name, sizeof(final_name));
char path[MEDIA_PATH_MAX], relative[MEDIA_PATH_MAX];
snprintf(relative, sizeof(relative), "media/%s/%s", ch_id, final_name);
if (media_index_path(media_base, relative, path)) { u_free(body); media_index_result_free(&result); failed = 1; break; }
int64_t fsz = ma_file_size(path);
if (fsz <= 0 || fsz != result.file_size) {
u_free(body); media_index_result_free(&result);
continue;
}
int registered = media_index_commit(cc->db, &result, cc->my_node_id, NULL, ch_id, path, media_base);
if (!registered) total += result.num_blocks;
else {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media backfill registration failed ch=%s file=%s", ch_id, path);
media_queue_channel_backfill(inst, ch_id);
}
if (encrypted_attachment && !registered && chat_attachment_downloaded(inst, ch_id, message, 0))
media_queue_channel_backfill(inst, ch_id);
u_free(body);
media_index_result_free(&result);
}
int final = sqlite3_finalize(ms);
if (mrc != SQLITE_DONE || final != SQLITE_OK || failed) {
media_queue_channel_backfill(inst, ch_id); failed = 1; break;
}
}
finish:
{
int final = sqlite3_finalize(cs);
if (rc != SQLITE_DONE || final != SQLITE_OK || failed) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media startup: incomplete snapshot rc=%d finalize=%d: %s", rc, final, sqlite3_errmsg(cc->db));
media_queue_full_backfill(inst);
}
}
if (total > 0)
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: media backfill done, %d blocks registered", CC_ID, total);
}
/* ─── Докачка недостающего медиа + вычистка старых файлов (единый проход) ─── */
struct bf_entry {
uint64_t ts_ms;
char full_path[1536];
int64_t claimed_size; /* размер из тела сообщения */
int64_t actual_size; /* >0 — файл на диске */
uint64_t author;
int64_t msg_id;
char ch_id[64];
char content_type[32];
char base_filename[256];
uint8_t author_sig[64];
char* body; /* "d"-поле (NULL для present) */
size_t body_len;
};
struct bf_list { struct bf_entry* items; int count, cap; };
static int bf_cmp_ts_desc(const void* a, const void* b) {
const struct bf_entry* x = (const struct bf_entry*)a;
const struct bf_entry* y = (const struct bf_entry*)b;
if (x->ts_ms > y->ts_ms) return -1;
if (x->ts_ms < y->ts_ms) return 1;
return 0;
}
/* Собирает медиасообщения (presence + размеры + контекст докачки). only_ch_id != NULL — только этот канал. */
static int bf_collect(struct chat_core_ctx* cc, const char* media_base, const char* only_ch_id, struct bf_list* l) {
sqlite3_stmt* cs = NULL; int rc, failed = 0, incomplete = 0;
rc = sqlite3_prepare_v2(cc->db, "SELECT channel_id FROM channels WHERE EXISTS(SELECT 1 FROM sqlite_master"
" WHERE type='table' AND name='msg_'||channel_id)", -1, &cs, NULL);
if (rc != SQLITE_OK) goto finish;
while ((rc = sqlite3_step(cs)) == SQLITE_ROW) {
const char* ch_id = (const char*)sqlite3_column_text(cs, 0);
if (!chat_channel_id(ch_id) || strlen(ch_id) != (size_t)sqlite3_column_bytes(cs, 0)) { failed = 1; break; }
if (only_ch_id && strcmp(only_ch_id, ch_id) != 0) continue;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256];
snprintf(sql, sizeof(sql),
"SELECT id, timestamp, node_id, data, author_signature, local_attrs, message_id FROM \"%s\"", tbl);
sqlite3_stmt* ms = NULL;
int mrc = sqlite3_prepare_v2(cc->db, sql, -1, &ms, NULL);
if (mrc != SQLITE_OK) { sqlite3_finalize(ms); failed = 1; break; }
while ((mrc = sqlite3_step(ms)) == SQLITE_ROW) {
int64_t msg_id = sqlite3_column_int64(ms, 0);
uint64_t ts = (uint64_t)sqlite3_column_int64(ms, 1);
uint64_t author = (uint64_t)sqlite3_column_int64(ms, 2);
const char* jdata = (const char*)sqlite3_column_text(ms, 3);
const uint8_t* sig_blob = (const uint8_t*)sqlite3_column_blob(ms, 4);
int sig_len = sqlite3_column_bytes(ms, 4);
const char* la = (const char*)sqlite3_column_text(ms, 5);
if (!jdata || !sig_blob || sig_len != 64) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "backfill snapshot: invalid stored message ch=%s id=%lld", ch_id, (long long)msg_id);
failed = 1; break;
}
char* body = NULL; char content_type[32], fp[256];
int fields = msg_media_fields(cc->db, jdata, (size_t)sqlite3_column_bytes(ms, 3), la, &body, content_type, fp);
if (fields == 1) continue;
if (fields < 0) { failed = 1; break; }
size_t dlen = strlen(body); struct media_index_result result;
int parsed = chat_msg_parse_media_body(body, &result);
if (parsed) { u_free(body); failed = 1; break; }
int encrypted_attachment = !strcmp(content_type, ENCRYPTED_ATTACHMENT_TYPE);
uint8_t message[32];
if (encrypted_attachment) {
if (sqlite3_column_bytes(ms, 6) != 32 || !sqlite3_column_blob(ms, 6)) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media backfill: missing encrypted message identity ch=%s", ch_id);
media_index_result_free(&result); u_free(body); continue;
}
memcpy(message, sqlite3_column_blob(ms, 6), 32);
int observed = chat_attachment_observe(cc->inst, ch_id, message, author, jdata, (size_t)sqlite3_column_bytes(ms, 3));
if (observed) {
if (observed == -1) media_queue_channel_backfill(cc->inst, ch_id);
incomplete = 1;
media_index_result_free(&result); u_free(body); continue;
}
}
char base_filename[256];
extract_base_filename(body, base_filename, sizeof(base_filename));
char final_name[256];
if (fp[0]) snprintf(final_name, sizeof(final_name), "%s", fp);
else media_final_name(&result, content_type, base_filename, final_name, sizeof(final_name));
char full_path[MEDIA_PATH_MAX], relative[MEDIA_PATH_MAX];
snprintf(relative, sizeof(relative), "media/%s/%s", ch_id, final_name);
if (media_index_path(media_base, relative, full_path)) { u_free(body); media_index_result_free(&result); failed = 1; break; }
int64_t actual = ma_file_size(full_path);
if (actual < 0) actual = 0;
if (encrypted_attachment && actual > 0 && chat_attachment_downloaded(cc->inst, ch_id, message, 0))
media_queue_channel_backfill(cc->inst, ch_id);
int slot = -1;
for (int i = 0; i < l->count; i++) if (!strcmp(l->items[i].full_path, full_path)) { slot = i; break; }
if (slot >= 0 && l->items[slot].ts_ms >= ts) { media_index_result_free(&result); u_free(body); continue; }
if (slot < 0 && l->count == l->cap) {
if ((size_t)l->cap > UINT32_MAX / (2 * sizeof(struct bf_entry))) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "backfill snapshot exceeds allocator capacity");
media_index_result_free(&result); u_free(body); failed = 1; break;
}
int nc = l->cap ? l->cap * 2 : 64;
struct bf_entry* ni = u_realloc(l->items, (size_t)nc * sizeof(struct bf_entry));
if (!ni) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "backfill snapshot allocation failed");
media_index_result_free(&result); u_free(body); failed = 1; break; }
l->items = ni; l->cap = nc;
}
struct bf_entry* e = &l->items[slot < 0 ? l->count++ : slot];
if (slot >= 0) u_free(e->body);
memset(e, 0, sizeof(*e));
e->ts_ms = ts;
snprintf(e->full_path, sizeof(e->full_path), "%s", full_path);
e->claimed_size = result.file_size;
e->actual_size = actual;
e->author = author;
e->msg_id = msg_id;
snprintf(e->ch_id, sizeof(e->ch_id), "%s", ch_id);
snprintf(e->content_type, sizeof(e->content_type), "%s", content_type);
snprintf(e->base_filename, sizeof(e->base_filename), "%s", base_filename);
memcpy(e->author_sig, sig_blob, 64);
if (actual <= 0) { e->body = body; e->body_len = dlen; }
else u_free(body);
media_index_result_free(&result);
}
int final = sqlite3_finalize(ms);
if (mrc != SQLITE_DONE || final != SQLITE_OK || failed) { failed = 1; break; }
}
finish:
{
int final = sqlite3_finalize(cs);
if (rc != SQLITE_DONE || final != SQLITE_OK || failed || incomplete) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "backfill snapshot failed channel=%s rc=%d finalize=%d: %s",
only_ch_id ? only_ch_id : "all", rc, final, sqlite3_errmsg(cc->db)); return -1;
}
}
return 0;
}
void chat_media_autodownload_backfill(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->db || !inst) return;
uint64_t limit = inst->config->global.chatserver_storage_total_size;
int autoload = chat_setting_get_int(inst, "storage_autoload", 1);
int days = chat_setting_get_int(inst, "storage_backfill_days", 7);
uint64_t now_ms = (uint64_t)ntp_time_get_seconds(inst) * 1000ULL;
uint64_t age_ms = (uint64_t)days * 86400ULL * 1000ULL;
uint64_t cutoff_ms = now_ms > age_ms ? now_ms - age_ms : 0;
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) return;
struct bf_list list = {0};
int collected = bf_collect(cc, media_base, NULL, &list);
if (collected) { media_queue_full_backfill(inst); goto release; }
if (list.count > 1) qsort(list.items, (size_t)list.count, sizeof(struct bf_entry), bf_cmp_ts_desc);
/* инициализация счётчика: сумма всех present-файлов */
cc->media_cache_bytes = 0;
for (int i = 0; i < list.count; i++)
if (list.items[i].actual_size > 0) cc->media_cache_bytes += (uint64_t)list.items[i].actual_size;
uint64_t used = 0;
int present = 0, downloaded = 0, evicted = 0, skipped = 0;
uint64_t evicted_bytes = 0;
for (int i = 0; i < list.count; i++) {
struct bf_entry* e = &list.items[i];
if (e->actual_size > 0) {
if (limit == 0 || used + (uint64_t)e->actual_size <= limit) {
used += (uint64_t)e->actual_size;
present++;
} else {
if (remove(e->full_path) == 0) {
cache_bytes_sub(cc, e->actual_size);
evicted++; evicted_bytes += (uint64_t)e->actual_size;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache evict: %s ts=%llu size=%lld",
CC_ID, e->full_path, (unsigned long long)e->ts_ms, (long long)e->actual_size);
} else {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cache evict failed: %s", CC_ID, e->full_path);
}
}
} else {
if (!autoload || e->author == cc->my_node_id) { skipped++; continue; }
if (e->ts_ms < cutoff_ms) { skipped++; continue; }
if (limit > 0 && used + (uint64_t)e->claimed_size > limit) { skipped++; continue; }
if (md_auto_download(inst, e->body, e->body_len, e->ch_id, e->base_filename,
e->ts_ms, e->author_sig, e->author, e->msg_id, e->content_type)) {
downloaded++;
used += (uint64_t)e->claimed_size;
} else {
skipped++;
}
}
}
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache backfill done present=%d downloaded=%d skipped=%d evicted=%d evicted_bytes=%llu bytes=%llu limit=%llu",
CC_ID, present, downloaded, skipped, evicted, (unsigned long long)evicted_bytes,
(unsigned long long)cc->media_cache_bytes, (unsigned long long)limit);
release:
for (int i = 0; i < list.count; i++) if (list.items[i].body) u_free(list.items[i].body);
u_free(list.items);
chat_media_cache_reconcile(inst);
}
/* Докачка медиа одного канала после завершения его chat_sync.
Порядок — от последних сообщений назад (список сортируется desc по ts). */
void chat_media_autodownload_backfill_channel(struct UTUN_INSTANCE* inst, const char* ch_id) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->db || !inst || !ch_id || !ch_id[0]) return;
uint64_t limit = inst->config->global.chatserver_storage_total_size;
int autoload = chat_setting_get_int(inst, "storage_autoload", 1);
int days = chat_setting_get_int(inst, "storage_backfill_days", 7);
uint64_t now_ms = (uint64_t)ntp_time_get_seconds(inst) * 1000ULL;
uint64_t age_ms = (uint64_t)days * 86400ULL * 1000ULL;
uint64_t cutoff_ms = now_ms > age_ms ? now_ms - age_ms : 0;
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) return;
struct bf_list list = {0};
int collected = bf_collect(cc, media_base, ch_id, &list);
if (collected) { media_queue_channel_backfill(inst, ch_id); goto release; }
if (list.count > 1) qsort(list.items, (size_t)list.count, sizeof(struct bf_entry), bf_cmp_ts_desc);
uint64_t used = cc->media_cache_bytes;
int downloaded = 0, skipped = 0;
for (int i = 0; i < list.count; i++) {
struct bf_entry* e = &list.items[i];
if (e->actual_size > 0) continue;
if (!autoload || e->author == cc->my_node_id) { skipped++; continue; }
if (e->ts_ms < cutoff_ms) { skipped++; continue; }
if (limit > 0 && used + (uint64_t)e->claimed_size > limit) { skipped++; continue; }
if (md_auto_download(inst, e->body, e->body_len, e->ch_id, e->base_filename,
e->ts_ms, e->author_sig, e->author, e->msg_id, e->content_type)) {
downloaded++;
used += (uint64_t)e->claimed_size;
} else {
skipped++;
}
}
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: channel backfill done ch=%s downloaded=%d skipped=%d",
CC_ID, ch_id, downloaded, skipped);
release:
for (int i = 0; i < list.count; i++) if (list.items[i].body) u_free(list.items[i].body);
u_free(list.items);
}
/* ─── Кеш медиафайлов: очистка сирот + лимит storage_total_size ─── */
struct media_msg_file {
char rel_path[1536]; // media/<ch_id>/<final_name>
char full_path[1536]; // <media_base>/media/<ch_id>/<final_name>
uint64_t ts_ms;
};
typedef int (*media_msg_file_cb)(const struct media_msg_file* f, void* arg);
/* Итератор по всем медиасообщениям всех каналов (без фильтра по node_id/времени). */
static int chat_media_foreach_media_message(struct chat_core_ctx* cc, const char* media_base, media_msg_file_cb cb, void* arg) {
if (!cc->db || !cb) return -1;
sqlite3_stmt* cs = NULL; int rc, failed = 0;
rc = sqlite3_prepare_v2(cc->db, "SELECT channel_id FROM channels WHERE EXISTS(SELECT 1 FROM sqlite_master"
" WHERE type='table' AND name='msg_'||channel_id)", -1, &cs, NULL);
if (rc != SQLITE_OK) goto finish;
while ((rc = sqlite3_step(cs)) == SQLITE_ROW) {
const char* ch_id = (const char*)sqlite3_column_text(cs, 0);
if (!chat_channel_id(ch_id) || strlen(ch_id) != (size_t)sqlite3_column_bytes(cs, 0)) { failed = 1; break; }
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT timestamp, data, local_attrs FROM \"%s\"", tbl);
sqlite3_stmt* ms = NULL;
int mrc = sqlite3_prepare_v2(cc->db, sql, -1, &ms, NULL);
if (mrc != SQLITE_OK) { sqlite3_finalize(ms); failed = 1; break; }
while ((mrc = sqlite3_step(ms)) == SQLITE_ROW) {
uint64_t ts = (uint64_t)sqlite3_column_int64(ms, 0);
const char* jdata = (const char*)sqlite3_column_text(ms, 1);
const char* la = (const char*)sqlite3_column_text(ms, 2);
if (!jdata) continue;
char* body = NULL; char content_type[32], fp[256];
int fields = msg_media_fields(cc->db, jdata, (size_t)sqlite3_column_bytes(ms, 1), la, &body, content_type, fp);
if (fields == 1) continue;
if (fields < 0) { failed = 1; break; }
size_t dlen = strlen(body); struct media_index_result result;
int parsed = chat_msg_parse_media_body(body, &result);
if (parsed) { u_free(body); if (parsed == -2 || (strcmp(content_type, "text") && strcmp(content_type, "text/plain"))) {
failed = 1; break;
} continue;
}
char final_name[256];
if (fp[0]) {
snprintf(final_name, sizeof(final_name), "%s", fp);
} else {
char base_filename[256];
extract_base_filename(body, base_filename, sizeof(base_filename));
media_final_name(&result, content_type, base_filename, final_name, sizeof(final_name));
}
struct media_msg_file f;
snprintf(f.rel_path, sizeof(f.rel_path), "media/%s/%s", ch_id, final_name);
if (media_index_path(media_base, f.rel_path, f.full_path)) {
u_free(body); media_index_result_free(&result); failed = 1; break;
}
f.ts_ms = ts;
if (cb(&f, arg)) failed = 1;
if (cc->attachment_initialized && !strcmp(content_type, ENCRYPTED_ATTACHMENT_TYPE)) {
uint8_t crypto_id[16];
sqlite3_stmt* local = NULL; int lrc = SQLITE_ERROR;
if (!chat_attachment_identity(body, dlen, crypto_id)) lrc = sqlite3_prepare_v2(cc->db,
"SELECT plain_path FROM attachment_keys WHERE channel=? AND crypto_id=? AND ready=1 AND decrypting=0",
-1, &local, NULL);
if (lrc == SQLITE_OK) lrc = sqlite3_bind_text(local, 1, ch_id, -1, SQLITE_STATIC);
if (lrc == SQLITE_OK) lrc = sqlite3_bind_blob(local, 2, crypto_id, 16, SQLITE_STATIC);
if (lrc == SQLITE_OK) lrc = sqlite3_step(local);
if (lrc == SQLITE_ROW) {
const char* path = (const char*)sqlite3_column_text(local, 0);
int bytes = sqlite3_column_bytes(local, 0);
if (!path || bytes <= 0 || bytes >= sizeof(f.full_path) || strlen(path) != (size_t)bytes ||
media_index_location(path, media_base, f.rel_path)) lrc = SQLITE_CORRUPT;
else { strcpy(f.full_path, path); if (cb(&f, arg)) failed = 1; }
}
int final = sqlite3_finalize(local);
if ((lrc != SQLITE_ROW && lrc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "encrypted cache: metadata rc=%d finalize=%d ch=%s", lrc, final, ch_id);
failed = 1;
}
}
u_free(body);
media_index_result_free(&result);
if (failed) break;
}
int final = sqlite3_finalize(ms);
if (mrc != SQLITE_DONE || final != SQLITE_OK || failed) { failed = 1; break; }
}
finish:
{
int final = sqlite3_finalize(cs);
if (rc != SQLITE_DONE || final != SQLITE_OK || failed) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "cache snapshot incomplete rc=%d finalize=%d: %s", rc, final, sqlite3_errmsg(cc->db)); return -1;
}
}
return 0;
}
struct cache_file { char full_path[1536]; uint64_t ts_ms; int64_t size; };
struct cache_rel_list { char** items; int count, cap; };
struct cache_file_list { struct cache_file* files; int count, cap; };
static int cache_str_cmp(const void* a, const void* b) {
return strcmp(*(const char* const*)a, *(const char* const*)b);
}
static int cache_file_cmp_ts(const void* a, const void* b) {
const struct cache_file* x = (const struct cache_file*)a;
const struct cache_file* y = (const struct cache_file*)b;
if (x->ts_ms < y->ts_ms) return -1;
if (x->ts_ms > y->ts_ms) return 1;
return 0;
}
static int cache_rel_add(struct cache_rel_list* l, const char* rel) {
if (l->count == l->cap) {
if ((size_t)l->cap > UINT32_MAX / (2 * sizeof(char*))) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "cache references exceed allocator capacity"); return -1;
}
int nc = l->cap ? l->cap * 2 : 64;
char** ni = u_realloc(l->items, (size_t)nc * sizeof(char*));
if (!ni) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: cache rel list realloc failed (%d)", CC_ID, nc); return -1; }
l->items = ni; l->cap = nc;
}
l->items[l->count] = u_strdup(rel);
if (!l->items[l->count]) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: cache rel strdup failed", CC_ID); return -1; }
l->count++;
return 0;
}
static int cache_file_add(struct cache_file_list* l, const char* full_path, uint64_t ts_ms, int64_t size) {
for (int i = 0; i < l->count; i++) if (!strcmp(l->files[i].full_path, full_path)) {
if (ts_ms > l->files[i].ts_ms) l->files[i].ts_ms = ts_ms;
return 0;
}
if (l->count == l->cap) {
if ((size_t)l->cap > UINT32_MAX / (2 * sizeof(struct cache_file))) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "cache files exceed allocator capacity"); return -1;
}
int nc = l->cap ? l->cap * 2 : 64;
struct cache_file* ni = u_realloc(l->files, (size_t)nc * sizeof(struct cache_file));
if (!ni) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: cache file list realloc failed (%d)", CC_ID, nc); return -1; }
l->files = ni; l->cap = nc;
}
snprintf(l->files[l->count].full_path, sizeof(l->files[l->count].full_path), "%s", full_path);
l->files[l->count].ts_ms = ts_ms;
l->files[l->count].size = size;
l->count++;
return 0;
}
static int cache_rel_collect_cb(const struct media_msg_file* f, void* arg) {
return cache_rel_add((struct cache_rel_list*)arg, f->rel_path);
}
static int cache_file_collect_cb(const struct media_msg_file* f, void* arg) {
int64_t fsz = ma_file_size(f->full_path);
return fsz > 0 ? cache_file_add((struct cache_file_list*)arg, f->full_path, f->ts_ms, fsz) : 0;
}
/* Фактическая сумма размеров файлов, соответствующих сообщениям. */
static int chat_media_cache_scan_sum(struct chat_core_ctx* cc, uint64_t* out) {
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) return -1;
struct cache_file_list files = {0};
if (chat_media_foreach_media_message(cc, media_base, cache_file_collect_cb, &files)) {
u_free(files.files); return -1;
}
uint64_t sum = 0;
for (int i = 0; i < files.count; i++)
if (files.files[i].size > 0) sum += (uint64_t)files.files[i].size;
u_free(files.files);
*out = sum; return 0;
}
/* Сверка счётчика с фактической суммой: при расхождении — WARN + самокоррекция. */
static void chat_media_cache_reconcile(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->db || !inst) return;
uint64_t actual;
if (chat_media_cache_scan_sum(cc, &actual)) return;
if (actual != cc->media_cache_bytes) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cache counter drift: counter=%llu actual=%llu diff=%lld",
CC_ID, (unsigned long long)cc->media_cache_bytes, (unsigned long long)actual,
(long long)((int64_t)actual - (int64_t)cc->media_cache_bytes));
cc->media_cache_bytes = actual;
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "%s: cache reconcile ok bytes=%llu",
CC_ID, (unsigned long long)cc->media_cache_bytes);
}
}
struct orphan_walk_ctx {
const char* media_base;
size_t mb_len;
char** rels; int rel_count;
int removed; uint64_t removed_bytes;
};
static void orphan_walk_cb(void* arg, const char* full_path, int64_t size) {
struct orphan_walk_ctx* ow = (struct orphan_walk_ctx*)arg;
char relative[MEDIA_PATH_MAX];
if (media_index_location(full_path, ow->media_base, relative)) return;
const char* rel = relative;
char* found = ow->rel_count ? bsearch(&rel, ow->rels, (size_t)ow->rel_count, sizeof(char*), cache_str_cmp) : NULL;
if (found) return;
int removed = file_io_remove_orphan(full_path);
if (removed > 0) {
ow->removed++;
ow->removed_bytes += (uint64_t)(size > 0 ? size : 0);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache orphan removed: %s (%lld bytes)",
CC_ID, rel, (long long)size);
} else if (removed < 0) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cache orphan remove failed: %s", CC_ID, full_path);
}
}
static void chat_media_cache_cleanup_orphans(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->db || !inst) return;
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) return;
char media_dir[600]; snprintf(media_dir, sizeof(media_dir), "%s/media", media_base);
struct file_io_info directory;
int directory_status = file_io_inspect(media_dir, 0, &directory);
if (directory_status == 1) return; // До первого вложения каталога может не быть.
if (directory_status < 0) return;
struct cache_rel_list rels = {0};
if (chat_media_foreach_media_message(cc, media_base, cache_rel_collect_cb, &rels)) goto release;
sqlite3_stmt* preparations = NULL;
int prep_rc = sqlite3_prepare_v2(cc->db, "SELECT location FROM media_preparations UNION SELECT location FROM media_files",
-1, &preparations, NULL);
if (prep_rc == SQLITE_OK) {
while ((prep_rc = sqlite3_step(preparations)) == SQLITE_ROW) {
const char* location = (const char*)sqlite3_column_text(preparations, 0);
int bytes = sqlite3_column_bytes(preparations, 0);
if (!location || bytes <= 0 || strlen(location) != (size_t)bytes || cache_rel_add(&rels, location)) {
prep_rc = SQLITE_NOMEM; break;
}
}
}
int prep_finish = preparations ? sqlite3_finalize(preparations) : SQLITE_OK;
if (prep_rc != SQLITE_DONE || prep_finish != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media orphan scan cannot protect indexed/preparing files rc=%d", prep_rc);
goto release;
}
/* Конечные контейнеры принадлежат attachment_keys; активные temp защищены file_io-реестром. */
if (cc->attachment_initialized) {
sqlite3_stmt* active = NULL;
int rc = sqlite3_prepare_v2(cc->db, "SELECT cipher_path FROM attachment_keys", -1, &active, NULL);
if (rc == SQLITE_OK) {
while ((rc = sqlite3_step(active)) == SQLITE_ROW) {
const char* path = (const char*)sqlite3_column_text(active, 0);
char relative[MEDIA_PATH_MAX]; int bytes = sqlite3_column_bytes(active, 0);
if (path && !bytes) continue; /* Ключ мог прийти раньше manifest: контейнеру ещё не назначен путь. */
if (!path || bytes <= 0 || strlen(path) != (size_t)bytes || media_index_location(path, media_base, relative) ||
cache_rel_add(&rels, relative)) {
rc = SQLITE_CORRUPT; break;
}
}
}
int final = sqlite3_finalize(active);
if (rc != SQLITE_DONE || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "encrypted cache: active scan failed rc=%d finalize=%d: %s", rc, final, sqlite3_errmsg(cc->db));
goto release;
}
}
if (rels.count > 1) qsort(rels.items, (size_t)rels.count, sizeof(char*), cache_str_cmp);
struct orphan_walk_ctx ow;
ow.media_base = media_base; ow.mb_len = strlen(media_base);
ow.rels = rels.items; ow.rel_count = rels.count;
ow.removed = 0; ow.removed_bytes = 0;
int walk_result = ma_dir_walk(media_dir, orphan_walk_cb, &ow);
if (walk_result) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "%s: cache orphan cleanup incomplete removed=%d bytes=%llu",
CC_ID, ow.removed, (unsigned long long)ow.removed_bytes);
else DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache orphan cleanup done removed=%d bytes=%llu",
CC_ID, ow.removed, (unsigned long long)ow.removed_bytes);
release:
for (int i = 0; i < rels.count; i++) u_free(rels.items[i]);
u_free(rels.items);
}
void chat_media_cache_enforce(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->db || !inst) return;
uint64_t limit = inst->config->global.chatserver_storage_total_size;
if (cc->attachment_initialized) {
uint64_t actual;
if (chat_media_cache_scan_sum(cc, &actual)) { media_queue_full_backfill(inst); return; }
cc->media_cache_bytes = actual;
}
if (limit == 0 || cc->media_cache_bytes <= limit) {
DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "%s: cache limit ok bytes=%llu limit=%llu",
CC_ID, (unsigned long long)cc->media_cache_bytes, (unsigned long long)limit);
chat_media_cache_reconcile(inst);
return;
}
char media_base[MEDIA_BASE_MAX];
if (media_index_get_base(cc->db, media_base)) return;
struct cache_file_list files = {0};
if (chat_media_foreach_media_message(cc, media_base, cache_file_collect_cb, &files)) { u_free(files.files); return; }
if (files.count > 1) qsort(files.files, (size_t)files.count, sizeof(struct cache_file), cache_file_cmp_ts);
int deleted = 0; uint64_t deleted_bytes = 0;
for (int i = 0; i < files.count && cc->media_cache_bytes > limit; i++) {
if (files.files[i].size <= 0) continue;
if (remove(files.files[i].full_path) == 0) {
cache_bytes_sub(cc, files.files[i].size);
deleted++; deleted_bytes += (uint64_t)files.files[i].size;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache evict: %s ts=%llu size=%lld",
CC_ID, files.files[i].full_path,
(unsigned long long)files.files[i].ts_ms, (long long)files.files[i].size);
} else {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cache evict failed: %s", CC_ID, files.files[i].full_path);
}
}
if (cc->media_cache_bytes > limit)
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: cache still over limit bytes=%llu limit=%llu",
CC_ID, (unsigned long long)cc->media_cache_bytes, (unsigned long long)limit);
else
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: cache limit enforced bytes=%llu limit=%llu deleted=%d bytes=%llu",
CC_ID, (unsigned long long)cc->media_cache_bytes, (unsigned long long)limit,
deleted, (unsigned long long)deleted_bytes);
u_free(files.files);
chat_media_cache_reconcile(inst);
}
/* ─── После первой подтверждённой Merkle-сверки (+ дебаунс) — анонс локальных блоков и докачка ─── */
static void media_startup_backfill_timer_cb(void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
cc->media_backfill_timer = NULL;
cc->media_backfill_ran = 1;
/* 1) переанонс локальных блоков суперузлам (block_availability),
* 2) чистка сирот, 3) докачка недостающего + вычистка старого (единый проход) */
media_delivery_announce_local_blocks(inst);
chat_media_cache_cleanup_orphans(inst);
chat_media_autodownload_backfill(inst);
}
static void media_startup_backfill_done_cb(struct DB_SYNC_INSTANCE* si, uint64_t peer_node_id, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
(void)si; (void)peer_node_id;
if (!cc || cc->media_backfill_ran || !cc->media_backfill_inst) return;
if (cc->media_backfill_timer) uasync_cancel_timeout(inst->ua, cc->media_backfill_timer);
cc->media_backfill_timer = uasync_set_timeout(inst->ua, 20000, inst,
media_startup_backfill_timer_cb, "chat_media_bl");
if (!cc->media_backfill_timer) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media startup: backfill timer allocation failed");
}
/* После подтверждённой Merkle-сверки канала — докачка его медиа (дебаунс 2с, от новых к старым). */
static void media_channel_backfill_timer_cb(void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc) return;
cc->media_channel_backfill_timer = NULL;
char channels[CHAT_MEDIA_BACKFILL_CHANNELS][sizeof(cc->media_channel_backfill[0])]; int count = cc->media_channel_backfill_pending;
memcpy(channels, cc->media_channel_backfill, (size_t)count * sizeof(channels[0]));
cc->media_channel_backfill_pending = 0;
for (int i = 0; i < count; i++) chat_media_autodownload_backfill_channel(inst, channels[i]);
}
/* Один ограниченный batch; отказ индекса во время scan ставит следующий timer, не меняя текущий batch. */
static void media_queue_channel_backfill(struct UTUN_INSTANCE* inst, const char* ch_id) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized || !cc->media_backfill_inst) return;
for (int i = 0; i < cc->media_channel_backfill_pending; i++)
if (strcmp(cc->media_channel_backfill[i], ch_id) == 0) goto schedule;
if (cc->media_channel_backfill_pending == (int)(sizeof(cc->media_channel_backfill) / sizeof(cc->media_channel_backfill[0]))) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "media backfill: channel batch full, scheduling complete scan ch=%s", ch_id);
media_queue_full_backfill(inst);
return;
}
snprintf(cc->media_channel_backfill[cc->media_channel_backfill_pending++], sizeof(cc->media_channel_backfill[0]), "%s", ch_id);
schedule:
if (!cc->media_channel_backfill_timer) cc->media_channel_backfill_timer = uasync_set_timeout(inst->ua, 20000, inst,
media_channel_backfill_timer_cb, "chat_md_chbl");
if (!cc->media_channel_backfill_timer) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media backfill: channel timer allocation failed ch=%s", ch_id);
else DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "media backfill: channel queued ch=%s pending=%d", ch_id, cc->media_channel_backfill_pending);
}
static void media_channel_backfill_done_cb(struct DB_SYNC_INSTANCE* si, uint64_t peer_node_id, void* arg) {
(void)peer_node_id;
if (!si) return;
char ch_id[32]; snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)db_sync_instance_group_id(si));
media_queue_channel_backfill(arg, ch_id);
}
int chat_media_startup_backfill_init(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!cc || !cc->initialized) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media backfill: active chat required"); return -1; }
if (cc->media_backfill_inst) return 0;
if (db_sync_add_done_cbk(inst, media_startup_backfill_done_cb, inst) ||
db_sync_add_done_cbk(inst, media_channel_backfill_done_cb, inst)) {
db_sync_remove_done_cbk(inst, media_startup_backfill_done_cb, inst);
db_sync_remove_done_cbk(inst, media_channel_backfill_done_cb, inst);
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "media backfill: history completion subscription failed"); return -1;
}
cc->media_backfill_inst = inst;
cc->media_backfill_ran = 0;
cc->media_channel_backfill_pending = 0;
cc->media_channel_backfill_timer = NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "media backfill: history completion subscriptions installed"); return 0;
}
void chat_media_startup_backfill_destroy(struct UTUN_INSTANCE* inst) {
struct chat_core_ctx* cc = inst ? CC(inst) : NULL;
if (!inst || !cc) return;
db_sync_remove_done_cbk(inst, media_startup_backfill_done_cb, inst);
db_sync_remove_done_cbk(inst, media_channel_backfill_done_cb, inst);
if (cc->media_backfill_timer) {
uasync_cancel_timeout(inst->ua, cc->media_backfill_timer);
cc->media_backfill_timer = NULL;
}
if (cc->media_channel_backfill_timer) {
uasync_cancel_timeout(inst->ua, cc->media_channel_backfill_timer);
cc->media_channel_backfill_timer = NULL;
}
cc->media_channel_backfill_pending = 0;
cc->media_backfill_inst = NULL;
cc->media_backfill_ran = 0;
}
Reference in new issue View Git Blame Copy Permalink