feat: media_async + media_index modules, media_files table, tests
- media_async: thread-per-task async engine with primitives (sha256, sign, copy, uuid, file_size) - media_index: media_files DB table with register_async for indexing chunks - chat_msg: async media submission via register_async, simplified chat_msg_submit fields - voice_recorder/attachment_sender: write single file (no chunk splitting on caller side) - inputbar/messagelist: updated to new simplified submit interface - test_media_async: 4-thread stress test (200K ops / 2 sec, 0 errors) - test_media_index: 9 cases covering commit edge cases + async flow - build: Makefile.am, CMakeLists.txt, utun_sources.cmake updated
This commit is contained in:
evgeny committed
1 parent
ce5a84da24
commit
65dbfa6813
43 files changed
+1983
-391
No files matched your search
@@ -16,6 +16,7 @@
|
||||
#include "../utun_instance.h"
|
||||
#include "../ntp_time.h"
|
||||
#include "../routing_layer/topo_node_sqlite.h"
|
||||
#include "../media_delivery/media_index.h"
|
||||
#include "../../lib/mem.h"
|
||||
|
||||
#include <string.h>
|
||||
@@ -135,6 +136,9 @@ int chat_core_init(struct UTUN_INSTANCE* inst, const char* db_path) {
|
||||
|
||||
chat_core_sync_my_addresses();
|
||||
|
||||
/* create media_files table */
|
||||
media_index_init(g_cc.db);
|
||||
|
||||
db_exec(
|
||||
"CREATE TABLE IF NOT EXISTS local_identity ("
|
||||
" id INTEGER PRIMARY KEY CHECK (id = 1),"
|
||||
|
||||
@@ -28,11 +28,9 @@ int chat_core_is_initialized(void);
|
||||
struct chat_msg_submit {
|
||||
char channel_id[64];
|
||||
char content_type[32];
|
||||
char media_dt[32];
|
||||
char media_basename[256];
|
||||
char media_suffix[8];
|
||||
char media_ext[16];
|
||||
uint32_t media_num_blocks;
|
||||
char media_src[1024];
|
||||
char media_dest[1024];
|
||||
uint8_t media_copy;
|
||||
uint8_t* data;
|
||||
uint32_t data_len;
|
||||
uint64_t timestamp;
|
||||
|
||||
+100
-139
@@ -9,16 +9,11 @@
|
||||
|
||||
#include "../utun_instance.h"
|
||||
#include "../transport_layer/secure_channel.h"
|
||||
#include "../media_delivery/media_index.h"
|
||||
#include "../../lib/mem.h"
|
||||
#include "../../lib/platform_compat.h"
|
||||
|
||||
#include <openssl/sha.h>
|
||||
#include <unistd.h>
|
||||
#include <sys/stat.h>
|
||||
|
||||
#define MEDIA_BLOCK_MIN (10 * 1024 * 1024)
|
||||
#define MEDIA_BLOCK_MAX (25 * 1024 * 1024)
|
||||
#define MEDIA_BLOCK_TARGET 30
|
||||
|
||||
/* forward decl */
|
||||
static void chat_core_submit_media_message(struct chat_msg_submit* req);
|
||||
@@ -72,174 +67,140 @@ void chat_core_submit_message(struct chat_msg_submit* req) {
|
||||
|
||||
void chat_core_submit_trampoline(void* arg) {
|
||||
struct chat_msg_submit* req = (struct chat_msg_submit*)arg;
|
||||
if (req->media_num_blocks > 0)
|
||||
if (req->media_src[0] != '\0')
|
||||
chat_core_submit_media_message(req);
|
||||
else
|
||||
chat_core_submit_message(req);
|
||||
u_free(req);
|
||||
}
|
||||
|
||||
/* ─── Медиа-сообщения: подпись блоков + сборка в единый файл ─── */
|
||||
/* ─── Медиа-сообщения: async регистрация в media_files + отправка ─── */
|
||||
|
||||
static uint32_t media_calc_block_size(uint64_t file_size) {
|
||||
if (file_size == 0) return 0;
|
||||
uint64_t target = (file_size + MEDIA_BLOCK_TARGET - 1) / MEDIA_BLOCK_TARGET;
|
||||
if (target < MEDIA_BLOCK_MIN) target = MEDIA_BLOCK_MIN;
|
||||
if (target > MEDIA_BLOCK_MAX) target = MEDIA_BLOCK_MAX;
|
||||
return (uint32_t)target;
|
||||
}
|
||||
struct media_submit_ctx {
|
||||
char channel_id[64];
|
||||
char content_type[32];
|
||||
char media_base[512];
|
||||
uint8_t msg_data[4096];
|
||||
uint32_t msg_data_len;
|
||||
uint64_t timestamp;
|
||||
};
|
||||
|
||||
static void chat_core_submit_media_message(struct chat_msg_submit* req) {
|
||||
if (!g_cc.initialized) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media submit before init", CC_ID); return; }
|
||||
if (!req || req->media_num_blocks == 0) return;
|
||||
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;
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: MEDIA SUBMIT ch=%s ct=%s dt=%s stem=%s ext=%s blocks=%u",
|
||||
CC_ID, req->channel_id, req->content_type,
|
||||
req->media_dt, req->media_basename, req->media_ext, req->media_num_blocks);
|
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: db_path='%s' initialized=%d",
|
||||
CC_ID, g_cc.db_path, g_cc.initialized);
|
||||
|
||||
struct DB_SYNC_INSTANCE* si = si_find(req->channel_id);
|
||||
if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: no db_sync instance for ch=%s", CC_ID, req->channel_id); return; }
|
||||
|
||||
char media_dir[1024];
|
||||
{
|
||||
/* g_cc.db_path is a file path (e.g. /data/.../chats.db).
|
||||
Extract directory portion for media path. */
|
||||
const char* last_slash = strrchr(g_cc.db_path, '/');
|
||||
if (last_slash)
|
||||
snprintf(media_dir, sizeof(media_dir), "%.*s/media/%s",
|
||||
(int)(last_slash - g_cc.db_path), g_cc.db_path, req->channel_id);
|
||||
else
|
||||
snprintf(media_dir, sizeof(media_dir), "%s/media/%s", g_cc.db_path, req->channel_id);
|
||||
if (err || !result) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media registration failed ch=%s err=%d",
|
||||
CC_ID, mctx->channel_id, err);
|
||||
u_free(mctx);
|
||||
return;
|
||||
}
|
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: media_dir='%s' dt='%s' stem='%s' ext='%s'",
|
||||
CC_ID, media_dir, req->media_dt, req->media_basename, req->media_ext);
|
||||
|
||||
uint32_t num_blocks = req->media_num_blocks;
|
||||
uint64_t block_size = 0;
|
||||
uint64_t file_size = 0;
|
||||
|
||||
/* allocate sig array */
|
||||
uint8_t (*sigs)[64] = (uint8_t(*)[64])u_malloc(num_blocks * 64);
|
||||
if (!sigs) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: sig alloc failed blocks=%u", CC_ID, num_blocks); return; }
|
||||
|
||||
/* read blocks, sign, compute sizes */
|
||||
for (uint32_t n = 0; n < num_blocks; n++) {
|
||||
char path[1280];
|
||||
snprintf(path, sizeof(path), "%s/%s_%u_%s_%s.%s",
|
||||
media_dir, req->media_dt, n, req->media_basename, req->media_suffix, req->media_ext);
|
||||
|
||||
FILE* f = fopen(path, "rb");
|
||||
if (!f) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: cannot open block %u: %s", CC_ID, n, path);
|
||||
u_free(sigs); return;
|
||||
}
|
||||
fseek(f, 0, SEEK_END); long fsz = ftell(f); fseek(f, 0, SEEK_SET);
|
||||
if (fsz <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: empty block %u: %s", CC_ID, n, path); fclose(f); u_free(sigs); return; }
|
||||
|
||||
uint8_t* buf = (uint8_t*)u_malloc((size_t)fsz);
|
||||
if (!buf || fread(buf, 1, (size_t)fsz, f) != (size_t)fsz) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: read fail block %u", CC_ID, n);
|
||||
if (buf) u_free(buf); fclose(f); u_free(sigs); return;
|
||||
}
|
||||
fclose(f);
|
||||
|
||||
if (sc_ed25519_sign(g_cc.inst->my_ed25519_privkey, buf, (size_t)fsz, sigs[n]) != SC_OK) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: Ed25519 sign fail block %u", CC_ID, n);
|
||||
u_free(buf); u_free(sigs); return;
|
||||
}
|
||||
u_free(buf);
|
||||
|
||||
file_size += (uint64_t)fsz;
|
||||
if (n == 0) block_size = (uint64_t)fsz;
|
||||
struct DB_SYNC_INSTANCE* si = si_find(mctx->channel_id);
|
||||
if (!si) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: no db_sync instance for ch=%s", CC_ID, mctx->channel_id);
|
||||
u_free(mctx);
|
||||
return;
|
||||
}
|
||||
|
||||
if (file_size == 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: zero file size", CC_ID); u_free(sigs); return; }
|
||||
|
||||
/* build final data: <base>|<file_size>|<block_size>|<num_blocks>|<sig_hex>,... */
|
||||
size_t base_len = req->data_len;
|
||||
size_t sigs_hex_len = num_blocks * 129; /* 128 hex + comma */
|
||||
size_t full_data_cap = base_len + 128 + sigs_hex_len;
|
||||
char* full_data = (char*)u_malloc(full_data_cap + 1);
|
||||
if (!full_data) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: data alloc fail", CC_ID); u_free(sigs); 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_DB_SYNC, "%s: data alloc fail", CC_ID); u_free(mctx); return; }
|
||||
|
||||
size_t off = 0;
|
||||
if (base_len > 0) { memcpy(full_data + off, req->data, base_len); off += base_len; }
|
||||
off += snprintf(full_data + off, full_data_cap - off, "|%llu|%llu|%u|",
|
||||
(unsigned long long)file_size, (unsigned long long)block_size, num_blocks);
|
||||
for (uint32_t n = 0; n < num_blocks; n++) {
|
||||
if (base_len > 0) { memcpy(full_data + off, mctx->msg_data, base_len); off += base_len; }
|
||||
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 b = 0; b < 64; b++)
|
||||
off += snprintf(full_data + off, full_data_cap - off, "%02x", sigs[n][b]);
|
||||
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]);
|
||||
}
|
||||
size_t full_data_len = off;
|
||||
|
||||
/* build JSON with dynamic buffer */
|
||||
size_t json_cap = full_data_len + 256;
|
||||
char* json = (char*)u_malloc(json_cap);
|
||||
if (!json) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: json alloc fail", CC_ID); u_free(full_data); u_free(sigs); return; }
|
||||
char* json = u_malloc(json_cap);
|
||||
if (!json) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: json alloc fail", CC_ID); u_free(full_data); u_free(mctx); return; }
|
||||
snprintf(json, json_cap,
|
||||
"{\"n\":%llu,\"ch\":\"%s\",\"ct\":\"%s\",\"d\":\"%.*s\"}",
|
||||
(unsigned long long)g_cc.my_node_id, req->channel_id,
|
||||
req->content_type, (int)full_data_len, full_data);
|
||||
(unsigned long long)g_cc.my_node_id, mctx->channel_id,
|
||||
mctx->content_type, (int)full_data_len, full_data);
|
||||
|
||||
uint64_t ts = db_sync_next_timestamp(si);
|
||||
size_t sig_msg_len = 8 + strlen(json);
|
||||
uint8_t* sig_msg = (uint8_t*)u_malloc(sig_msg_len);
|
||||
if (!sig_msg) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: sig_msg alloc fail", CC_ID); u_free(json); u_free(full_data); u_free(sigs); return; }
|
||||
uint8_t* sig_msg = u_malloc(sig_msg_len);
|
||||
if (!sig_msg) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: sig_msg alloc fail", CC_ID); u_free(json); u_free(full_data); u_free(mctx); return; }
|
||||
memcpy(sig_msg, &ts, 8);
|
||||
memcpy(sig_msg + 8, json, sig_msg_len - 8);
|
||||
uint8_t json_sig[64];
|
||||
if (sc_ed25519_sign(g_cc.inst->my_ed25519_privkey, sig_msg, sig_msg_len, json_sig) != SC_OK) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: Ed25519 sign JSON failed", CC_ID);
|
||||
u_free(sig_msg); u_free(json); u_free(full_data); u_free(sigs); return;
|
||||
u_free(sig_msg); u_free(json); u_free(full_data); u_free(mctx); return;
|
||||
}
|
||||
|
||||
/* assemble single file from blocks, then delete blocks */
|
||||
{
|
||||
char assembled_path[1280];
|
||||
snprintf(assembled_path, sizeof(assembled_path), "%s/%s_%s_%s.%s",
|
||||
media_dir, req->media_dt, req->media_basename, req->media_suffix, req->media_ext);
|
||||
FILE* af = fopen(assembled_path, "wb");
|
||||
if (af) {
|
||||
for (uint32_t n = 0; n < num_blocks; n++) {
|
||||
char block_path[1280];
|
||||
snprintf(block_path, sizeof(block_path), "%s/%s_%u_%s_%s.%s",
|
||||
media_dir, req->media_dt, n, req->media_basename, req->media_suffix, req->media_ext);
|
||||
FILE* bf = fopen(block_path, "rb");
|
||||
if (bf) {
|
||||
uint8_t buf[65536]; size_t rd;
|
||||
while ((rd = fread(buf, 1, sizeof(buf), bf)) > 0) fwrite(buf, 1, rd, af);
|
||||
fclose(bf);
|
||||
unlink(block_path);
|
||||
}
|
||||
}
|
||||
fclose(af);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: assembled %s (%llu bytes)", CC_ID, assembled_path, (unsigned long long)file_size);
|
||||
} else {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: cannot create assembled file %s", CC_ID, assembled_path);
|
||||
}
|
||||
}
|
||||
|
||||
/* build local_attrs JSON */
|
||||
char local_attrs_json[512];
|
||||
{
|
||||
char bl_list[256] = "0"; size_t bl_off = 1;
|
||||
for (uint32_t n = 1; n < num_blocks; n++)
|
||||
bl_off += snprintf(bl_list + bl_off, sizeof(bl_list) - bl_off, ",%u", n);
|
||||
snprintf(local_attrs_json, sizeof(local_attrs_json),
|
||||
"{\"st\":\"fl\",\"fp\":\"%s_%s_%s.%s\",\"bl\":[%s]}",
|
||||
req->media_dt, req->media_basename, req->media_suffix, req->media_ext, bl_list);
|
||||
}
|
||||
|
||||
int ret = db_sync_insert_signed(si, json, strlen(json), json_sig, 64, ts, local_attrs_json);
|
||||
int ret = db_sync_insert_signed(si, json, strlen(json), json_sig, 64, ts, NULL);
|
||||
if (ret != 0)
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media insert failed ret=%d", CC_ID, ret);
|
||||
else
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: media message inserted ch=%s ts=%llu",
|
||||
CC_ID, mctx->channel_id, (unsigned long long)ts);
|
||||
|
||||
u_free(sig_msg); u_free(json); u_free(full_data); u_free(sigs);
|
||||
u_free(sig_msg); u_free(json); u_free(full_data); u_free(mctx);
|
||||
}
|
||||
|
||||
static void chat_core_submit_media_message(struct chat_msg_submit* req) {
|
||||
if (!g_cc.initialized) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media submit before init", CC_ID); return; }
|
||||
if (!req || req->media_src[0] == '\0') return;
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: MEDIA SUBMIT ch=%s ct=%s src=%s dst=%s copy=%d",
|
||||
CC_ID, req->channel_id, req->content_type,
|
||||
req->media_src, req->media_dest, req->media_copy);
|
||||
|
||||
if (!g_cc.inst->media_async) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: media_async not initialized", CC_ID);
|
||||
return;
|
||||
}
|
||||
|
||||
/* build media_base from db_path: dirname(db_path) e.g. /path/to/data */
|
||||
char media_base[1024];
|
||||
{
|
||||
const char* last_slash = strrchr(g_cc.db_path, '/');
|
||||
if (last_slash)
|
||||
snprintf(media_base, sizeof(media_base), "%.*s", (int)(last_slash - g_cc.db_path), g_cc.db_path);
|
||||
else
|
||||
snprintf(media_base, sizeof(media_base), "%s", g_cc.db_path);
|
||||
}
|
||||
|
||||
/* copy fields for async callback (req is freed by trampoline immediately) */
|
||||
struct media_submit_ctx* mctx = u_calloc(1, sizeof(*mctx));
|
||||
if (!mctx) return;
|
||||
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_base, sizeof(mctx->media_base), "%s", media_base);
|
||||
mctx->timestamp = req->timestamp;
|
||||
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);
|
||||
}
|
||||
|
||||
media_index_register_async(
|
||||
g_cc.inst->media_async, g_cc.inst->ua, g_cc.db,
|
||||
g_cc.my_node_id, g_cc.inst->my_ed25519_privkey,
|
||||
req->channel_id,
|
||||
req->media_src, req->media_dest, req->media_copy ? 1 : 0,
|
||||
media_base,
|
||||
on_media_registered, mctx);
|
||||
}
|
||||
|
||||
/* ─── Обновление local_attrs (для будущей приёмной стороны) ─── */
|
||||
|
||||
Reference in new issue
Block a user