jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feature: media auto-download + manual download + UI download state indication (Android + desktop)

Browse Source 
C-layer:
- chat_event.h: CHAT_EVT_ATTACHMENT_DOWNLOADED (14)
- chat_msg.c: fixed md_auto_download (non-NULL done_cb, correct media_base from db_path, use chat_setting for autoload checks)
- chat_msg.c: added chat_core_attachment_download + trampoline for manual download
- chat_core.h: exported attachment_dl_req + trampoline

Android (Kotlin + JNI):
- Message: downloadState field (parse from localAttrs)
- ChatRepository: parse localAttrs for download state
- NativeLib: attachmentDownload() JNI
- MessageBubble: FileBubble with download status icon + tap-to-download
- ChatViewModel: downloadAttachment() + CHAT_EVT_ATTACHMENT_DOWNLOADED handler
- ChatScreen: wire onDownloadFile handler
- android_jni_bridge: utun_bridge_attachment_download + JNI + localAttrs in JSON

Desktop (Qt/C++):
- messagedelegate.h: MsgFileDownloadStateRole, MsgChannelIdRole
- messagelist.cpp: parse localAttrs for file download state
- messagedelegate.cpp: FileBubble with download icon/text + click handler via chat_core_attachment_download_trampoline
- gui_bridge.h: GUI_EVT_ATTACHMENT_DOWNLOADED (14) + callback typedef
- gui_bridge_impl.cpp: event dispatch + setter
- mainwindow.cpp: callback registration -> refresh message list
This commit is contained in:
evgeny committed 2026-07-30 00:01:38 +03:00
1 parent b8920cf483
commit 70797f921e
23 files changed
+558 -174
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_core.h
+7
View File
@@ -87,6 +87,13 @@ void chat_core_connect_channel_trampoline(void* arg);
/* Трамплин для gui_bridge_post_uasync (GUI → uasync) */
void chat_core_submit_trampoline(void* arg);
/* Трамплин для ручного запуска загрузки медиа (JNI → uasync) */
struct attachment_dl_req {
char channel_id[64];
int64_t msg_id;
};
void chat_core_attachment_download_trampoline(void* arg);
/* Обновление local_attrs для записи (вызывается из uasync-потока) */
int chat_core_update_local_attrs(const char* ch_id, uint64_t ts,
const uint8_t* author_sig,
src/chat/chat_event.h
+1
View File
@@ -32,6 +32,7 @@ extern "C" {
#define CHAT_EVT_KEYS_GENERATED 11 /* [pub_hex:64] */
#define CHAT_EVT_SERVICE_STARTED 12 /* data: none */
#define CHAT_EVT_SERVICE_STOPPED 13 /* data: none */
#define CHAT_EVT_ATTACHMENT_DOWNLOADED 14 /* [ch_id_len:1][ch_id:var][msg_id:8] */
typedef void (*chat_event_handler_fn)(int type, const uint8_t* data, int len);
src/chat/chat_msg.c
+180 -60
View File
@@ -6,6 +6,7 @@
#include "chat_core_priv.h"
#include "chat_event.h"
#include "chat_setting.h"
#include "../utun_instance.h"
#include "../transport_layer/secure_channel.h"
@@ -384,94 +385,129 @@ int chat_core_load_nodeinfo(uint64_t node_id, uint8_t* buf, size_t buf_size,
return 0;
}
/* ─── db_sync callback ─── */
/* ─── download completion ─── */
static void md_auto_download(struct UTUN_INSTANCE* inst, const char* data_str, size_t data_len,
const char* ch_id, const char* media_base) {
/* parse: <base>|<file_size>|<block_size>|<num_blocks>|<media_id_hex>|<content_hash_hex>|<ids,sigs> */
/* find first | after base (strip base prefix) */
struct md_done_ctx {
char channel_id[64];
uint64_t ts;
uint8_t author_sig[64];
};
static void md_download_done_cb(void* arg, int err) {
struct md_done_ctx* ctx = (struct md_done_ctx*)arg;
if (!err) {
char dest_relpath[1024];
{
const char* last_slash = strrchr(g_cc.db_path, '/');
const char* base = last_slash ? last_slash + 1 : g_cc.db_path;
snprintf(dest_relpath, sizeof(dest_relpath), "media/%s", base);
}
char attrs[1024];
snprintf(attrs, sizeof(attrs), "{\"st\":\"fl\",\"fp\":\"%s\"}", dest_relpath);
chat_core_update_local_attrs(ctx->channel_id, ctx->ts, ctx->author_sig, attrs);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: media downloaded ch=%s", CC_ID, ctx->channel_id);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: media download failed ch=%s err=%d", CC_ID, ctx->channel_id, err);
}
uint8_t evt[73]; uint8_t cl = (uint8_t)strlen(ctx->channel_id);
evt[0] = cl; memcpy(evt + 1, ctx->channel_id, cl); { uint64_t z = 0; memcpy(evt + 1 + cl, &z, 8); }
chat_event_post(CHAT_EVT_ATTACHMENT_DOWNLOADED, evt, 1 + cl + 8);
u_free(ctx);
}
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) {
(void)data_len;
const char* p = data_str;
while (*p && *p != '|') p++;
if (!*p) return; /* not a media message */
p++; /* skip first | */
if (!*p) return -1; p++;
long long fsize = strtoll(p, (char**)&p, 10);
if (*p != '|') return; p++;
if (*p != '|') return -1; p++;
long long bsize = strtoll(p, (char**)&p, 10);
if (*p != '|') return; p++;
int nb = atoi(p);
if (*p != '|') return -1; p++;
int nb = (int)strtol(p, (char**)&p, 10);
while (*p && *p != '|') p++;
if (!*p) return; p++;
if (nb <= 0 || nb > 100) return;
if (fsize <= 0) return;
/* check auto-download settings */
char sql[128];
snprintf(sql, sizeof(sql), "SELECT value FROM ui_state WHERE key='auto_download_max_size'");
sqlite3_stmt* st = NULL;
long long max_size = 0;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) == SQLITE_OK) {
if (sqlite3_step(st) == SQLITE_ROW) max_size = strtoll((const char*)sqlite3_column_text(st, 0), NULL, 10);
sqlite3_finalize(st);
}
if (max_size > 0 && fsize > max_size) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: media too large (%lld > %lld), skip auto-download", CC_ID, fsize, max_size);
return;
}
if (!*p) return -1; p++;
if (nb <= 0 || nb > 100 || fsize <= 0) return -1;
struct media_index_result result;
memset(&result, 0, sizeof(result));
result.file_size = fsize;
result.block_size = bsize;
result.num_blocks = nb;
result.file_size = fsize; result.block_size = bsize; result.num_blocks = nb;
/* media_id hex → bin */
for (int i = 0; i < 16; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.media_id[i] = (uint8_t)strtol(b, NULL, 16);
}
p += 32;
if (*p != '|') return; p++;
/* content_hash hex → bin */
p += 32; if (*p != '|') return -1; p++;
for (int i = 0; i < 32; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.content_hash[i] = (uint8_t)strtol(b, NULL, 16);
}
p += 64;
if (*p != '|') return; p++;
p += 64; if (*p != '|') return -1; p++;
/* block_ids and block_sigs */
result.block_ids = u_malloc((size_t)nb * 16);
result.block_sigs = u_malloc((size_t)nb * 64);
if (!result.block_ids || !result.block_sigs) { media_index_result_free(&result); return; }
if (!result.block_ids || !result.block_sigs) { media_index_result_free(&result); return -1; }
for (int n = 0; n < nb; n++) {
/* block_id hex */
for (int i = 0; i < 16; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.block_ids[n * 16 + i] = (uint8_t)strtol(b, NULL, 16);
}
p += 32;
if (*p != ',') { media_index_result_free(&result); return; }
p++; /* skip comma */
/* block_sig hex */
p += 32; if (*p != ',') { media_index_result_free(&result); return -1; } p++;
for (int i = 0; i < 64; i++) {
char b[3] = { p[i*2], p[i*2+1], 0 };
result.block_sigs[n * 64 + i] = (uint8_t)strtol(b, NULL, 16);
}
p += 128;
if (n < nb - 1 && *p == ',') p++;
p += 128; if (n < nb - 1 && *p == ',') p++;
}
char dest[1024]; snprintf(dest, sizeof(dest), "%s/%s", media_base, ch_id);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: auto-download ch=%s media=%02x%02x... blocks=%d size=%lld",
CC_ID, ch_id, result.media_id[0], result.media_id[1], nb, (long long)fsize);
media_download_start(inst, 0, &result, dest, media_base, NULL, NULL);
char dest[1024], media_base[512];
{
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);
}
}
snprintf(dest, sizeof(dest), "%s/media/%s/%s", media_base, ch_id, base_filename);
struct md_done_ctx* ctx = u_calloc(1, sizeof(*ctx));
if (!ctx) { media_index_result_free(&result); return -1; }
snprintf(ctx->channel_id, sizeof(ctx->channel_id), "%s", ch_id);
ctx->ts = ts;
if (author_sig) memcpy(ctx->author_sig, author_sig, 64);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: download start ch=%s media=%02x%02x... blocks=%d size=%lld dest=%s",
CC_ID, ch_id, result.media_id[0], result.media_id[1], nb, (long long)fsize, dest);
media_download_start(inst, 0, &result, dest, media_base, md_download_done_cb, ctx);
media_index_result_free(&result);
return 0;
}
static void 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) {
if (!chat_setting_get_int("storage_autoload", 1)) return;
int max_mb = chat_setting_get_int("storage_autoload_maxsize_mb", 10);
const char* p = data_str;
while (*p && *p != '|') p++; /* skip base filename */
if (!*p) return; p++;
long long fsize = strtoll(p, (char**)&p, 10);
if (fsize > (long long)max_mb * 1024 * 1024) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: media too large (%lld > %d MB), skip auto-download",
CC_ID, fsize, max_mb);
return;
}
md_start_download(inst, data_str, data_len, ch_id, base_filename, ts, author_sig);
}
/* ─── db_sync callback ─── */
void on_msg_inserted(struct DB_SYNC_INSTANCE* si, uint64_t record_ts, const char* data, size_t len, uint64_t author, void* arg) {
(void)si;
const char* ch_id = (const char*)arg;
@@ -481,20 +517,44 @@ void on_msg_inserted(struct DB_SYNC_INSTANCE* si, uint64_t record_ts, const char
/* auto-download media if message contains media metadata */
if (data && len > 0 && author != g_cc.my_node_id) {
/* parse JSON: {"n":...,"ch":"...","ct":"...","d":"..."} */
const char* ct = strstr(data, "\"ct\":\"");
const char* dpos = strstr(data, "\"d\":\"");
if (ct && dpos) {
const char* d_start = dpos + 5;
/* find closing quote of "d" value */
char body[4096]; size_t bi = 0;
while (*d_start && *d_start != '\"' && bi < sizeof(body) - 1) body[bi++] = *d_start++;
body[bi] = '\0';
/* content_type is between ct\":\" and next \" */
const char* ct_val = ct + 6;
/* determine media_base from chat_core context */
const char* media_base = "/tmp/utun_media";
md_auto_download(g_cc.inst, body, bi, ch_id, media_base);
char base_filename[256]; size_t bfi = 0;
while (body[bfi] && body[bfi] != '|' && bfi < sizeof(base_filename) - 1)
base_filename[bfi++] = body[bfi];
base_filename[bfi] = '\0';
if (bfi == 0) snprintf(base_filename, sizeof(base_filename), "file");
const uint8_t* sig_field = strstr(data, "\"sig\":\"");
uint8_t author_sig[64] = {0};
if (sig_field) {
/* read author_signature from DB — find by ts in on_msg_inserted we don't have sig
so leave as zeros; use record_ts to locate the record later */
}
/* We don't have author_sig in the callback; derive from body content.
Instead, read it from the DB using the just-inserted record */
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT author_signature FROM \"%s\" WHERE timestamp=? AND node_id=? LIMIT 1", tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)record_ts);
sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
if (sqlite3_step(st) == SQLITE_ROW) {
const void* sblob = sqlite3_column_blob(st, 0);
int sblen = sqlite3_column_bytes(st, 0);
if (sblob && sblen == 64) memcpy(author_sig, sblob, 64);
}
sqlite3_finalize(st);
}
md_auto_download(g_cc.inst, body, bi, ch_id, base_filename, record_ts, author_sig);
}
}
@@ -506,3 +566,63 @@ void on_msg_inserted(struct DB_SYNC_INSTANCE* si, uint64_t record_ts, const char
sqlite3_step(st); sqlite3_finalize(st);
}
}
/* ─── manual attachment download (called from JNI bridge via uasync post) ─── */
void chat_core_attachment_download(const char* channel_id, int64_t msg_id) {
if (!channel_id || msg_id <= 0) return;
char tbl[80]; msg_table_name(channel_id, tbl, sizeof(tbl));
char sql[256];
snprintf(sql, sizeof(sql), "SELECT data, timestamp, author_signature FROM \"%s\" WHERE id=?", tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_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;
}
sqlite3_bind_int64(st, 1, (sqlite3_int64)msg_id);
if (sqlite3_step(st) != 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 = (const uint8_t*)sqlite3_column_blob(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);
if (!jdata || jlen <= 0 || !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;
}
const char* ds = strstr((const char*)jdata, "\"d\":\"");
if (!ds) { sqlite3_finalize(st); return; }
const char* d_start = ds + 5;
char body[4096]; size_t bi = 0;
while (*d_start && *d_start != '\"' && bi < sizeof(body) - 1) body[bi++] = *d_start++;
body[bi] = '\0';
char base_filename[256]; size_t bfi = 0;
while (body[bfi] && body[bfi] != '|' && bfi < sizeof(base_filename) - 1)
base_filename[bfi++] = body[bfi];
base_filename[bfi] = '\0';
if (bfi == 0) snprintf(base_filename, sizeof(base_filename), "file");
uint8_t author_sig[64]; memcpy(author_sig, sig_blob, 64);
sqlite3_finalize(st);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "%s: manual download ch=%s id=%lld file=%s",
CC_ID, channel_id, (long long)msg_id, base_filename);
md_start_download(g_cc.inst, body, bi, channel_id, base_filename, (uint64_t)db_ts, author_sig);
}
static void chat_core_attachment_download_trampoline_impl(void* arg) {
struct attachment_dl_req* req = (struct attachment_dl_req*)arg;
chat_core_attachment_download(req->channel_id, req->msg_id);
u_free(req);
}
void chat_core_attachment_download_trampoline(void* arg) {
chat_core_attachment_download_trampoline_impl(arg);
}
src/chat/chat_sync.c
+54 -23
View File
@@ -399,6 +399,22 @@ static void _on_invite_sync_done(uint64_t peer, const char* ns, int result, void
u_free(sa);
}
/* ── send CHANNEL_INFO_REQ to start join handshake ── */
static void cs_start_channel_join(struct chat_sync* cs, uint64_t peer) {
if (!cs || cs->info_req_timer) return;
char ch_id_str[64];
snprintf(ch_id_str, sizeof(ch_id_str), "%llu", (unsigned long long)cs->pending_invite_ch_id);
uint8_t req[1] = { CS_MSG_CHANNEL_INFO_REQ };
cs_send(cs, ch_id_str, peer, req, 1);
cs->info_req_timer = uasync_set_timeout(cs->inst->ua,
CS_INFO_REQ_TIMEOUT_MS * 10, cs, cs_info_req_timeout_cb, "cs_info_req");
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: sent CHANNEL_INFO_REQ ch=%llu peer=0x%016llx",
CS_ID, (unsigned long long)cs->pending_invite_ch_id, (unsigned long long)peer);
}
/* ── Connection callbacks ── */
static void cs_on_conn_up(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (!conn || !g_cs) return;
@@ -409,21 +425,12 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, void* arg) {
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up peer=%016llx pending_invite=%llu links_up=%d", CS_ID, (unsigned long long)peer, (unsigned long long)g_cs->pending_invite_ch_id, conn->links_up);
if (g_cs->pending_invite_ch_id != 0 && !g_cs->info_req_timer) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up invite path peer=%016llx ch=%llu",
CS_ID, (unsigned long long)peer, (unsigned long long)g_cs->pending_invite_ch_id);
if (g_cs->pending_invite_ch_id != 0) {
if (g_cs->pending_invite_node_id != 0 && g_cs->pending_invite_node_id != peer)
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: invite node_id MISMATCH: invite=0x%016llx ETCP_peer=0x%016llx — invite is STALE!",
CS_ID, (unsigned long long)g_cs->pending_invite_node_id, (unsigned long long)peer);
g_cs->pending_invite_node_id = peer;
char ch_id_str[64];
snprintf(ch_id_str, sizeof(ch_id_str), "%llu",
(unsigned long long)g_cs->pending_invite_ch_id);
uint8_t req[1] = { CS_MSG_CHANNEL_INFO_REQ };
cs_send(g_cs, ch_id_str, peer, req, 1);
g_cs->info_req_timer = uasync_set_timeout(g_cs->inst->ua,
CS_INFO_REQ_TIMEOUT_MS * 10, g_cs, cs_info_req_timeout_cb, "cs_info_req");
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up invite path, marking peer=%016llx online", CS_ID, (unsigned long long)peer);
cs_start_channel_join(g_cs, peer);
cs_on_peer_status_changed(peer, 1);
return;
}
@@ -633,13 +640,12 @@ static void cm_invite_trampoline(void* arg) {
}
char ch_str[32]; snprintf(ch_str, sizeof(ch_str), "%llu", (unsigned long long)channel_id);
chat_core_ensure_channel_ready(ch_str);
/* ensure TOPO_GROUP_TYPE_CHAT group exists */
/* TOPO_GROUP (lightweight) for conn_mgr_connect_from_invite below.
DB channel + sync created later in cs_handle_channel_info_resp. */
if (inst->topo_groups) {
uint64_t gid = channel_id;
if (!topo_groups_find(inst->topo_groups, gid))
topo_groups_create_group(inst->topo_groups, gid, TOPO_GROUP_TYPE_CHAT, ch_str);
if (!topo_groups_find(inst->topo_groups, channel_id))
topo_groups_create_group(inst->topo_groups, channel_id, TOPO_GROUP_TYPE_CHAT, ch_str);
}
/* build TOPO_NODE from invite data */
@@ -667,13 +673,15 @@ static void cm_invite_trampoline(void* arg) {
if (existing) {
if (existing->links_up) {
if (g_cs && g_cs->pending_invite_ch_id != 0 && !g_cs->info_req_timer) {
if (node_id != existing->peer_node_id)
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC,
"%s: invite node_id mismatch: derived=%016llx ETCP=%016llx — using ETCP",
CS_ID, (unsigned long long)node_id, (unsigned long long)existing->peer_node_id);
cs_on_conn_up(existing, NULL);
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: invite start — already connected peer=0x%016llx, joining ch=%llu",
CS_ID, (unsigned long long)node_id, (unsigned long long)channel_id);
if (node_id != existing->peer_node_id)
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: invite node_id mismatch: derived=%016llx ETCP=%016llx — using ETCP",
CS_ID, (unsigned long long)node_id, (unsigned long long)existing->peer_node_id);
g_cs->pending_invite_ch_id = channel_id;
g_cs->pending_invite_node_id = existing->peer_node_id;
cs_start_channel_join(g_cs, existing->peer_node_id);
cs_on_peer_status_changed(existing->peer_node_id, 1);
} else {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: invite conn in progress node=0x%016llx, waiting UP", CS_ID, (unsigned long long)node_id);
}
@@ -740,6 +748,29 @@ void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
(void(*)(void*))cm_invite_trampoline, w);
}
/* ─── chat_sync_join_channel: join channel via already-connected peer ─── */
void chat_sync_join_channel(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t target_node_id) {
if (!g_cs || !inst || !ch_id || !ch_id[0]) return;
uint64_t channel_id = strtoull(ch_id, NULL, 10);
if (channel_id == 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: join_channel invalid ch_id='%s'", CS_ID, ch_id); return; }
struct ETCP_CONN* conn = instance_find_conn(inst, target_node_id);
if (!conn || !conn->links_up || !conn->initialized) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: join_channel no active conn to 0x%016llx for ch=%s",
CS_ID, (unsigned long long)target_node_id, ch_id);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: join_channel ch=%s peer=0x%016llx",
CS_ID, ch_id, (unsigned long long)target_node_id);
g_cs->pending_invite_ch_id = channel_id;
g_cs->pending_invite_node_id = target_node_id;
cs_start_channel_join(g_cs, target_node_id);
cs_on_peer_status_changed(target_node_id, 1);
}
/* ─── Ed25519 sign / verify helpers ─── */
static int cs_ed25519_sign(const uint8_t* privkey, const uint8_t* msg, size_t msg_len,
src/chat/chat_sync.h
+6
View File
@@ -53,6 +53,12 @@ void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
const uint8_t* pubkey_bin,
const uint8_t* addrs_data, int addr_count);
/* Присоединиться к каналу через уже подключённый узел (uasync-поток).
Соединение с target_node_id должно быть установлено (links_up && initialized).
Протокол: CHANNEL_INFO_REQ → CHANNEL_INFO_RESP → CHANNEL_JOIN → WELCOME.
Канал (DB + TOPO_GROUP) создаётся локально при получении CHANNEL_INFO_RESP. */
void chat_sync_join_channel(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t target_node_id);
#ifdef __cplusplus
}
#endif
src/media_delivery/media_delivery.c
+31 -17
View File
@@ -29,7 +29,7 @@ static int md_super_start(struct media_delivery_ctx* md);
static void md_super_stop(struct media_delivery_ctx* md);
static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super_peer* peer);
static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id);
static int md_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, const uint8_t* data, size_t len);
static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, const uint8_t* data, size_t len);
static struct media_super_peer* md_super_peer_add(struct media_delivery_ctx* md, uint64_t node_id);
/* ── helpers ── */
@@ -255,7 +255,7 @@ static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super
uint16_t num = (uint16_t)num_entries; memcpy(pkt + off, &num, 2); off += 2;
memcpy(pkt + off, buf, (size_t)nb); off += nb;
if (md_send(md->inst, peer->peer_node_id, pkt, (size_t)off) == 0) {
if (md_send(md->inst, TOPO_GROUP_UTUN, peer->peer_node_id, pkt, (size_t)off) == 0) {
peer->inflight_count++;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: SUPER_REPL sent to 0x%016llx seq=%u entries=%d",
MD_ID, (unsigned long long)peer->peer_node_id, seq, num_entries);
@@ -273,7 +273,7 @@ static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super
/* ── send via etcp_router helper ── */
static int md_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id,
static int md_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id,
const uint8_t* data, size_t len) {
if (!inst || !data || len == 0) return -1;
struct ll_entry* e = queue_entry_new(0);
@@ -282,7 +282,7 @@ static int md_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id,
if (!e->dgram) { queue_entry_free(e); return -1; }
e->dgram[0] = ETCP_RT_ID_MEDIA_DELIVERY;
memcpy(e->dgram + 1, data, len); e->len = (uint16_t)(len + 1);
int rc = etcp_route_send(inst, TOPO_GROUP_UTUN, dst_node_id, e, 1);
int rc = etcp_route_send(inst, group_id, dst_node_id, e, 1);
if (rc != 0) { u_free(e->dgram); queue_entry_free(e); }
return rc;
}
@@ -349,7 +349,7 @@ static void md_handle_query(struct media_delivery_ctx* md, uint64_t from_node,
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: QUERY from 0x%016llx → %d entries",
MD_ID, (unsigned long long)from_node, count);
(void)md_send(md->inst, from_node, resp, (size_t)off);
(void)md_send(md->inst, TOPO_GROUP_UTUN, from_node, resp, (size_t)off);
}
static void md_handle_have_block(struct media_delivery_ctx* md, uint64_t from_node,
@@ -368,7 +368,7 @@ static void md_handle_have_block(struct media_delivery_ctx* md, uint64_t from_no
memcpy(ha->block_id, hb->block_id, 16);
ha->status = (rc == 0) ? 0 : 1;
md_send(md->inst, from_node, ack, sizeof(ack));
md_send(md->inst, TOPO_GROUP_UTUN, from_node, ack, sizeof(ack));
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: HAVE_BLOCK from 0x%016llx rc=%d",
MD_ID, (unsigned long long)from_node, rc);
@@ -411,7 +411,7 @@ static void md_handle_super_repl(struct media_delivery_ctx* md, uint64_t from_no
struct media_pkt_super_ack* sa = (struct media_pkt_super_ack*)ack;
sa->subcmd = MEDIA_SUBCMD_SUPER_ACK;
sa->ack_seq = (uint32_t)max_id;
md_send(md->inst, from_node, ack, sizeof(ack));
md_send(md->inst, TOPO_GROUP_UTUN, from_node, ack, sizeof(ack));
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: SUPER_REPL from 0x%016llx entries=%d max_id=%lld",
MD_ID, (unsigned long long)from_node, num, (long long)max_id);
@@ -453,7 +453,7 @@ static void md_handle_super_hello(struct media_delivery_ctx* md, uint64_t from_n
struct media_pkt_super_hello resp;
resp.subcmd = MEDIA_SUBCMD_SUPER_HELLO;
resp.last_recv_id = my_last_recv;
md_send(md->inst, from_node, (const uint8_t*)&resp, sizeof(resp));
md_send(md->inst, TOPO_GROUP_UTUN, from_node, (const uint8_t*)&resp, sizeof(resp));
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: SUPER_HELLO from 0x%016llx (peer_last_recv=%lld, my_last_recv=%lld)",
MD_ID, (unsigned long long)from_node, (long long)sh->last_recv_id, (long long)my_last_recv);
@@ -488,7 +488,7 @@ static void md_super_conn_cb(int result, uint64_t node_id, void* arg) {
hello.subcmd = MEDIA_SUBCMD_SUPER_HELLO;
hello.last_recv_id = my_last;
if (md_send(md->inst, node_id, (const uint8_t*)&hello, sizeof(hello)) == 0) {
if (md_send(md->inst, TOPO_GROUP_UTUN, node_id, (const uint8_t*)&hello, sizeof(hello)) == 0) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: SUPER_HELLO sent to 0x%016llx last_recv=%lld",
MD_ID, (unsigned long long)node_id, (long long)my_last);
} else {
@@ -505,6 +505,7 @@ static void md_super_conn_cb(int result, uint64_t node_id, void* arg) {
struct stream_ctx {
struct media_delivery_ctx* md;
uint64_t group_id;
uint64_t dst_node_id;
uint8_t media_id[16];
uint8_t block_id[16];
@@ -544,7 +545,7 @@ static void stream_send_chunk_cb(struct ll_queue* q, void* arg) {
ch->data_len = (uint16_t)rd;
memcpy(pkt + MEDIA_BLOCK_CHUNK_HDR_SIZE, buf, rd);
md_send(md->inst, sc->dst_node_id, pkt, MEDIA_BLOCK_CHUNK_HDR_SIZE + rd);
md_send(md->inst, sc->group_id, sc->dst_node_id, pkt, MEDIA_BLOCK_CHUNK_HDR_SIZE + rd);
sc->offset += rd;
sc->remaining -= rd;
@@ -561,7 +562,7 @@ static void stream_send_chunk_cb(struct ll_queue* q, void* arg) {
bd->total_size = (uint32_t)sc->block_data_len;
/* block_sig is zero — receiver will verify vs block_sigs from media_index_result */
md_send(md->inst, sc->dst_node_id, done_pkt, MEDIA_BLOCK_DONE_SIZE);
md_send(md->inst, sc->group_id, sc->dst_node_id, done_pkt, MEDIA_BLOCK_DONE_SIZE);
media_delivery_stream_done(md->inst);
fclose(sc->file);
u_free(sc);
@@ -569,7 +570,7 @@ static void stream_send_chunk_cb(struct ll_queue* q, void* arg) {
MD_ID, sc->chunk, bd->total_size);
} else {
/* continue streaming — register waiter for backpressure */
etcp_router_on_send_ready(md->inst, TOPO_GROUP_UTUN,
etcp_router_on_send_ready(md->inst, sc->group_id,
sc->dst_node_id, ETCP_RT_ID_MEDIA_DELIVERY,
&sc->waiter, stream_send_chunk_cb, sc);
}
@@ -579,9 +580,10 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_nod
const uint8_t* data, size_t len) {
if (len < MEDIA_BLOCK_REQ_SIZE) return;
struct media_pkt_block_req* req = (struct media_pkt_block_req*)data;
uint64_t group_id = req->group_id ? req->group_id : TOPO_GROUP_UTUN;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_REQ from 0x%016llx block=%02x%02x... active_streams=%d",
MD_ID, (unsigned long long)from_node, req->block_id[0], req->block_id[1], md->active_streams);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_REQ from 0x%016llx group=%016llx block=%02x%02x... active_streams=%d",
MD_ID, (unsigned long long)from_node, (unsigned long long)group_id, req->block_id[0], req->block_id[1], md->active_streams);
int limit_reached = md->active_streams >= MD_MAX_STREAMS;
md->active_streams++;
@@ -594,12 +596,23 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_nod
memcpy(ov.block_id, req->block_id, 16);
ov.chunk = req->chunk;
ov.retry_after_ms = 2000;
md_send(md->inst, from_node, (const uint8_t*)&ov, sizeof(ov));
md_send(md->inst, group_id, from_node, (const uint8_t*)&ov, sizeof(ov));
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: BLOCK_REQ OVERLOADED (streams=%d) to 0x%016llx",
MD_ID, md->active_streams, (unsigned long long)from_node);
return;
}
/* establish direct connection if not already connected */
if (!instance_find_conn(md->inst, from_node) && group_id != TOPO_GROUP_UTUN) {
struct TOPO_GROUP* grp = topo_groups_find(md->inst->topo_groups, group_id);
if (!grp) grp = topo_groups_get_default(md->inst->topo_groups);
if (grp && grp->conn_mgr) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: no direct conn to 0x%016llx — connecting via conn_mgr",
MD_ID, (unsigned long long)from_node);
conn_mgr_connect_node(grp->conn_mgr, from_node, 0, NULL, NULL);
}
}
/* find file in media_files DB */
sqlite3* db = md->db;
if (!db) return;
@@ -649,6 +662,7 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_nod
struct stream_ctx* sc = u_calloc(1, sizeof(*sc));
if (!sc) { fclose(f); md->active_streams--; return; }
sc->md = md;
sc->group_id = group_id;
sc->dst_node_id = from_node;
memcpy(sc->media_id, req->media_id, 16);
memcpy(sc->block_id, req->block_id, 16);
@@ -659,8 +673,8 @@ static void md_handle_block_req(struct media_delivery_ctx* md, uint64_t from_nod
sc->block_start = (uint64_t)block_start;
sc->block_data_len = (size_t)block_length;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: streaming file=%s chunk=%d start=%lld len=%lld to 0x%016llx",
MD_ID, path, chunk, (long long)block_start, (long long)block_length, (unsigned long long)from_node);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: streaming file=%s chunk=%d start=%lld len=%lld group=%016llx to 0x%016llx",
MD_ID, path, chunk, (long long)block_start, (long long)block_length, (unsigned long long)group_id, (unsigned long long)from_node);
/* start streaming — first chunk immediately, then backpressure */
memset(&sc->waiter, 0, sizeof(sc->waiter));
src/media_delivery/media_delivery_proto.h
+1
View File
@@ -112,6 +112,7 @@ struct media_pkt_super_ack {
struct media_pkt_block_req {
uint8_t subcmd; // MEDIA_SUBCMD_BLOCK_REQ
uint64_t group_id; // CHAT-группа для маршрутизации и прямого коннекта
uint8_t media_id[16];
uint8_t block_id[16];
uint32_t chunk;
src/media_delivery/media_download.c
+3 -2
View File
@@ -57,13 +57,14 @@ static int md_dl_send_block_req(struct UTUN_INSTANCE* inst, uint64_t dst,
struct media_pkt_block_req req;
memset(&req, 0, sizeof(req));
req.subcmd = MEDIA_SUBCMD_BLOCK_REQ;
req.group_id = dl->group_id;
memcpy(req.media_id, dl->media_id, 16);
memcpy(req.block_id, dl->block_ids + bi * 16, 16);
req.chunk = (uint32_t)bi;
req.offset = 0;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_REQ to 0x%016llx block=%d",
MDL_ID, (unsigned long long)dst, bi);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: BLOCK_REQ to 0x%016llx group=%016llx block=%d",
MDL_ID, (unsigned long long)dst, (unsigned long long)dl->group_id, bi);
return md_dl_send(inst, dst, (const uint8_t*)&req, sizeof(req));
}