перенос chat-логики из chatgui/transport в src/chat/

- Создан src/chat/ с chat_event.h/c — система нотификаций
- Перенесены chat_core, chat_msg, chat_channel, chat_profile, chat_status
- Перенесены chat_sync, member_sync, merkle_sync, db_sync
- gui_bridge_post() заменён на chat_event_post() с возможностью регистрации обработчика
- chat_event_set_handler() — chatgui регистрирует мост в gui_bridge
- Без обработчика: события логируются через DEBUG_DEBUG + log_dump
- chat интегрирован в utun_instance.c (при db_path в конфиге)
- Обновлены include-пути, Makefile.am, CMakeLists.txt
- Из chatgui удалены старые transport/ файлы, обновлён utun_node.cpp
This commit is contained in:
Evgeny committed 2026-07-27 12:58:33 +03:00
1 parent 4595052b5b
commit ed6132fcda
25 files changed
+262 -165

No files matched your search

+1 -9
View File
@@ -85,14 +85,6 @@ add_executable(chatgui
transport/node_config.cpp
transport/config_updater.cpp
transport/gui_bridge_impl.cpp
transport/chat_core.c
transport/chat_msg.c
transport/chat_channel.c
transport/chat_profile.c
transport/chat_status.c
transport/chat_sync.c
transport/member_sync.c
transport/merkle_sync.c
transport/miniaudio_impl.c
db/db_manager.cpp
../../lib/sqlite3.c
@@ -101,7 +93,7 @@ add_executable(chatgui
target_include_directories(chatgui PRIVATE ${CMAKE_SOURCE_DIR}/../../lib ${CMAKE_SOURCE_DIR}/../../src ${CMAKE_SOURCE_DIR}/db ${CMAKE_SOURCE_DIR}/../../lib/libopus/include)
target_compile_definitions(chatgui PRIVATE SQLITE_THREADSAFE=1 USE_SQLITE)
set_source_files_properties(../../lib/sqlite3.c transport/chat_core.c transport/chat_msg.c transport/chat_channel.c transport/chat_profile.c transport/chat_status.c transport/chat_sync.c transport/member_sync.c transport/merkle_sync.c transport/miniaudio_impl.c PROPERTIES LANGUAGE C)
set_source_files_properties(../../lib/sqlite3.c transport/miniaudio_impl.c PROPERTIES LANGUAGE C)
if(WIN32)
target_link_libraries(chatgui PRIVATE ${QT_LIBS} rlottie::rlottie ZLIB::ZLIB OpenSSL::Crypto ZXing::ZXing utun pthread)
else()
-1
View File
@@ -68,7 +68,6 @@ target_include_directories(utun PUBLIC
${SRC_DIR}/transport_layer
${SRC_DIR}/routing_layer
${LIB_DIR}
${TRANSPORT_DIR}
${SRC_DIR}/uip
${CMAKE_SOURCE_DIR}/db
)
-257
View File
@@ -1,257 +0,0 @@
/*
* chat_channel.c — управление каналами: ensure_ready, create, connect
*
* Вынесено из chat_core.c для уменьшения размера модуля.
*/
#include "chat_core_priv.h"
#include "gui_bridge.h"
#include "../../../src/utun_instance.h"
#include "../../../src/ntp_time.h"
#include "topo_node_sqlite.h"
#include "topo_group.h"
#include "topo_group_connect.h"
#include "member_sync.h"
#include "secure_channel.h"
#include "../../../lib/mem.h"
#include "../../../lib/ll_queue.h"
#include "../../../lib/platform_compat.h"
#include "etcp.h"
#include <openssl/sha.h>
#include <openssl/evp.h>
#include <openssl/rand.h>
/* ─── Ed25519 подпись join-сообщения ─── */
static int channel_sign_join(const uint8_t* ch_x25519, const uint8_t* ch_ed25519,
uint64_t node_id, const uint8_t* node_pubkey,
uint64_t join_ts, const uint8_t* ed25519_privkey,
uint8_t sig_out[64])
{
uint8_t msg[256]; size_t mlen = 0;
memcpy(msg + mlen, ch_x25519, 32); mlen += 32;
memcpy(msg + mlen, ch_ed25519, 32); mlen += 32;
memcpy(msg + mlen, &node_id, 8); mlen += 8;
memcpy(msg + mlen, node_pubkey, 32); mlen += 32;
memcpy(msg + mlen, &join_ts, 8); mlen += 8;
EVP_PKEY* pkey = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL,
ed25519_privkey, 32);
if (!pkey) { memset(sig_out, 0, 64); return -1; }
EVP_MD_CTX* mdctx = EVP_MD_CTX_new();
if (!mdctx) { EVP_PKEY_free(pkey); memset(sig_out, 0, 64); return -1; }
int ok = EVP_DigestSignInit(mdctx, NULL, NULL, NULL, pkey) == 1
&& EVP_DigestSign(mdctx, sig_out, &(size_t){64}, msg, mlen) == 1;
EVP_MD_CTX_free(mdctx);
EVP_PKEY_free(pkey);
if (!ok) memset(sig_out, 0, 64);
return ok ? 0 : -1;
}
/* ─── сбор адресов текущего узла (для member_sync_put) ─── */
static int collect_my_addrs(uint8_t* buf, int max_addrs) {
int count = 0;
struct ETCP_SOCKET* s = g_cc.inst->etcp_sockets;
while (s && count < max_addrs) {
struct sockaddr_storage* sa = s->interface_addr.ss_family ? &s->interface_addr : NULL;
if (!sa) sa = s->local_addr.ss_family ? &s->local_addr : NULL;
if (sa && sa->ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)sa;
buf[count * 8] = 4;
buf[count * 8 + 1] = s->sock_id;
memcpy(buf + count * 8 + 2, &sin->sin_addr, 4);
uint16_t port = ntohs(sin->sin_port);
buf[count * 8 + 6] = (uint8_t)(port >> 8);
buf[count * 8 + 7] = (uint8_t)(port & 0xFF);
count++;
}
s = s->next;
}
return count;
}
/* ─── подготовка инфраструктуры канала (db_sync instance) ─── */
void chat_core_ensure_channel_ready(const char* ch_id) {
if (!g_cc.initialized || !ch_id || !ch_id[0]) return;
if (si_find(ch_id)) return;
char tbl_msg[80]; msg_table_name(ch_id, tbl_msg, sizeof(tbl_msg));
uint64_t ch_hash = 0;
{ const uint8_t* chd = (const uint8_t*)ch_id; size_t chl = strlen(ch_id); uint8_t sh[32]; SHA256(chd, chl, sh); memcpy(&ch_hash, sh, 8); }
struct DB_SYNC_INSTANCE* si = db_sync_instance_add(g_cc.inst, tbl_msg, ch_hash, 1);
uint64_t gid = strtoull(ch_id, NULL, 10);
if (gid != 0 && g_cc.inst->topo_groups && !topo_groups_find(g_cc.inst->topo_groups, gid))
topo_groups_create_group(g_cc.inst->topo_groups, gid, TOPO_GROUP_TYPE_CHAT, ch_id);
if (si) {
si_register(si, ch_id);
db_sync_set_insert_cb(si, on_msg_inserted, u_strdup(ch_id));
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: channel ready ch=%s tbl=%s hash=0x%016llx",
CC_ID, ch_id, tbl_msg, (unsigned long long)ch_hash);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: db_sync_instance_add failed for ch=%s",
CC_ID, ch_id);
}
}
/* ─── создание канала ─── */
void chat_core_create_channel(struct chat_channel_create* req) {
if (!g_cc.initialized) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: create_channel NOT INITIALIZED ch=%s",
CC_ID, req ? req->channel_id : "(null)");
return;
}
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: create_channel req=NULL", CC_ID); return; }
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: create_channel BEGIN ch=%s name=%s",
CC_ID, req->channel_id, req->name);
int rc = topo_node_sqlite_channel_put(g_cc.db,
req->channel_id, req->name, req->owner_node_id,
req->x25519_pubkey, req->x25519_privkey,
req->ed25519_pubkey, req->ed25519_privkey,
req->signature);
if (rc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: channel_put FAILED ch=%s rc=%d",
CC_ID, req->channel_id, rc);
} else {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: channel_put OK ch=%s name=%s owner=0x%016llx",
CC_ID, req->channel_id, req->name, (unsigned long long)req->owner_node_id);
chat_core_ensure_channel_ready(req->channel_id);
uint8_t ch_id_len = (uint8_t)strlen(req->channel_id);
uint8_t data[65];
data[0] = ch_id_len;
memcpy(data + 1, req->channel_id, ch_id_len);
{
uint64_t myid = g_cc.inst->node_id;
uint64_t join_ts = (uint64_t)ntp_time_get_seconds(g_cc.inst);
uint8_t join_sig[64];
channel_sign_join(req->x25519_pubkey, req->ed25519_pubkey,
myid, g_cc.inst->my_keys.public_key,
join_ts, g_cc.inst->my_ed25519_privkey, join_sig);
uint8_t my_addrs[256]; int my_addr_cnt = collect_my_addrs(my_addrs, 16);
char juser3[256]; snprintf(juser3, sizeof(juser3), "{\"name\":\"%s\"}",
g_cc.inst->name[0] ? g_cc.inst->name : "");
int mrc = member_sync_put(g_cc.inst, req->channel_id, myid,
g_cc.inst->my_keys.public_key, g_cc.inst->my_ed25519_pubkey,
join_sig, join_ts, NULL, 0, juser3, my_addrs, my_addr_cnt);
if (mrc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_sync_put(self) FAILED ch=%s rc=%d",
CC_ID, req->channel_id, mrc);
}
}
gui_bridge_post(GUI_EVT_CHANNEL_UPDATED, data, 1 + ch_id_len);
}
}
void chat_core_create_channel_trampoline(void* arg) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "chat_core: TRAMPOLINE invoked arg=%p", arg);
struct chat_channel_create* req = (struct chat_channel_create*)arg;
chat_core_create_channel(req);
u_free(req);
}
void chat_core_create_channel_auto_trampoline(void* arg) {
chat_core_create_channel_auto((const char*)arg);
u_free(arg);
}
void chat_core_create_channel_auto(const char* name) {
if (!g_cc.initialized || !name) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: create_channel_auto — not initialized or NULL name", CC_ID);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: create_channel_auto BEGIN name=%s", CC_ID, name);
uint8_t x25519_pub[32] = {0}, x25519_priv[32] = {0};
{ EVP_PKEY* pkey = NULL; EVP_PKEY_CTX* ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_X25519, NULL);
if (ctx) { EVP_PKEY_keygen_init(ctx); EVP_PKEY_generate(ctx, &pkey); EVP_PKEY_CTX_free(ctx); }
if (pkey) { size_t pkLen = 32, skLen = 32; EVP_PKEY_get_raw_public_key(pkey, x25519_pub, &pkLen); EVP_PKEY_get_raw_private_key(pkey, x25519_priv, &skLen); EVP_PKEY_free(pkey); } }
uint8_t ed25519_pub[32] = {0}, ed25519_priv[32] = {0};
EVP_PKEY* ed_pkey = NULL;
{ EVP_PKEY_CTX* ctx = EVP_PKEY_CTX_new_id(EVP_PKEY_ED25519, NULL);
if (ctx) { EVP_PKEY_keygen_init(ctx); EVP_PKEY_generate(ctx, &ed_pkey); EVP_PKEY_CTX_free(ctx); }
if (ed_pkey) { size_t pkLen = 32, skLen = 32; EVP_PKEY_get_raw_public_key(ed_pkey, ed25519_pub, &pkLen); EVP_PKEY_get_raw_private_key(ed_pkey, ed25519_priv, &skLen); } }
uint64_t chId = 0; RAND_bytes((uint8_t*)&chId, sizeof(chId));
char channelId[64]; snprintf(channelId, sizeof(channelId), "%llu", (unsigned long long)chId);
uint8_t sig[64] = {0};
if (ed_pkey) {
size_t nameLen = strlen(name);
size_t chIdLen = strlen(channelId);
size_t msgLen = chIdLen + 1 + nameLen + 1 + 8 + 32 + 32;
uint8_t* msg = u_malloc(msgLen);
if (msg) {
uint8_t* p = msg;
memcpy(p, channelId, chIdLen); p += chIdLen; *p++ = '\0';
memcpy(p, name, nameLen); p += nameLen; *p++ = '\0';
memcpy(p, &g_cc.my_node_id, 8); p += 8;
memcpy(p, x25519_pub, 32); p += 32;
memcpy(p, ed25519_pub, 32);
EVP_MD_CTX* mdctx = EVP_MD_CTX_new();
if (mdctx) { EVP_DigestSignInit(mdctx, NULL, NULL, NULL, ed_pkey); size_t sl = 64; EVP_DigestSign(mdctx, sig, &sl, msg, msgLen); EVP_MD_CTX_free(mdctx); }
u_free(msg);
}
EVP_PKEY_free(ed_pkey);
}
struct chat_channel_create* req = u_calloc(1, sizeof(*req));
if (!req) return;
snprintf(req->channel_id, sizeof(req->channel_id), "%s", channelId);
snprintf(req->name, sizeof(req->name), "%s", name);
req->owner_node_id = g_cc.my_node_id;
memcpy(req->x25519_pubkey, x25519_pub, 32); memcpy(req->x25519_privkey, x25519_priv, 32);
memcpy(req->ed25519_pubkey, ed25519_pub, 32); memcpy(req->ed25519_privkey, ed25519_priv, 32);
memcpy(req->signature, sig, 64);
chat_core_create_channel(req);
u_free(req);
}
/* ─── управление подключением к каналу (выбор в GUI) ─── */
void chat_core_connect_channel(const char* ch_id) {
DEBUG_TRACE(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel called ch_id=%s initialized=%d", CC_ID, ch_id ? ch_id : "(null)", g_cc.initialized);
if (!g_cc.initialized || !ch_id || !ch_id[0]) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel — skip (not ready)", CC_ID); return; }
if (!g_cc.inst->topo_groups) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel — no topo_groups", CC_ID); return; }
uint64_t gid = strtoull(ch_id, NULL, 10);
if (gid == 0) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel — invalid ch_id=%s", CC_ID, ch_id); return; }
chat_core_ensure_channel_ready(ch_id);
struct TOPO_GROUP* group = topo_groups_find(g_cc.inst->topo_groups, gid);
if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel — group not found ch=%s gid=%016llx total_groups=%d", CC_ID, ch_id, (unsigned long long)gid, queue_entry_count(g_cc.inst->topo_groups->group_list)); return; }
if (group->connect) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel — already in progress ch=%s active=%d",
CC_ID, ch_id, topo_group_connect_active_count(group));
return;
}
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel ch=%s gid=%016llx group_type=%d — starting connect_init", CC_ID, ch_id, (unsigned long long)gid, group->group_type);
topo_group_connect_init(group);
}
void chat_core_connect_channel_trampoline(void* arg) {
char* ch_id = (char*)arg;
DEBUG_TRACE(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect_channel_trampoline ch_id=%s", CC_ID, ch_id ? ch_id : "(null)");
if (!ch_id) return;
chat_core_connect_channel(ch_id);
u_free(ch_id);
}
-236
View File
@@ -1,236 +0,0 @@
/*
* chat_core.c — жизненный цикл чата: init, destroy, глобальное состояние g_cc
*
* Под-модули (каждый в своём .c):
* chat_msg.c — отправка сообщений, DB-операции для chat_sync
* chat_channel.c — создание/подключение каналов, db_sync instance
* chat_profile.c — имя узла, адреса, ui_state
* chat_status.c — сбор статуса NTP+connections
*
* Все DB-операции и сетевые функции выполняются в uasync-потоке.
*/
#include "chat_core_priv.h"
#include "gui_bridge.h"
#include "../../../src/utun_instance.h"
#include "../../../src/ntp_time.h"
#include "topo_node_sqlite.h"
#include "../../../lib/mem.h"
#include <string.h>
#include <stdio.h>
#include "../../../lib/platform_compat.h"
/* ─── глобальное состояние (extern в chat_core_priv.h) ─── */
struct chat_core_ctx g_cc;
/* ─── хелперы ─── */
void sanitize_ch_id(const char* ch_id, char* out, size_t out_sz) {
size_t i = 0;
while (*ch_id && i < out_sz - 1) {
char c = *ch_id++;
if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z')
|| (c >= '0' && c <= '9') || c == '_')
out[i++] = c;
else
out[i++] = '_';
}
out[i] = '\0';
}
void msg_table_name(const char* ch_id, char* buf, size_t sz) {
char san[64]; sanitize_ch_id(ch_id, san, sizeof(san));
snprintf(buf, sz, "msg_%s", san);
}
void peers_table_name(const char* ch_id, char* buf, size_t sz) {
char san[64]; sanitize_ch_id(ch_id, san, sizeof(san));
snprintf(buf, sz, "peers_%s", san);
}
int db_exec(const char* sql) {
char* err = NULL;
int rc = sqlite3_exec(g_cc.db, sql, NULL, NULL, &err);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: sql error: %s", CC_ID, err);
sqlite3_free(err);
}
return rc;
}
/* ─── db_sync registry ─── */
struct DB_SYNC_INSTANCE* si_find(const char* ch_id) {
for (int i = 0; i < g_cc.si_count; i++)
if (strcmp(g_cc.si_ch_id[i], ch_id) == 0) return g_cc.si[i];
return NULL;
}
void si_register(struct DB_SYNC_INSTANCE* si, const char* ch_id) {
if (g_cc.si_count >= g_cc.si_capacity) {
int nc = g_cc.si_capacity ? g_cc.si_capacity * 2 : 8;
g_cc.si = u_realloc(g_cc.si, nc * sizeof(void*));
g_cc.si_ch_id = u_realloc(g_cc.si_ch_id, nc * sizeof(char*));
g_cc.si_capacity = nc;
}
g_cc.si[g_cc.si_count] = si;
g_cc.si_ch_id[g_cc.si_count] = u_strdup(ch_id);
g_cc.si_count++;
uint32_t msg_cnt = db_sync_count(si);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: si_register ch=%s mc=%u", CC_ID, ch_id, msg_cnt);
}
/* ─── жизненный цикл ─── */
int chat_core_init(struct UTUN_INSTANCE* inst, const char* db_path) {
if (!inst || !db_path) return -1;
memset(&g_cc, 0, sizeof(g_cc));
if (inst->topo_sqlite_db) {
g_cc.db = inst->topo_sqlite_db; g_cc.shared_db = 1;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: using shared SQLite db=%p", CC_ID, (void*)g_cc.db);
} else {
int rc = sqlite3_open_v2(db_path, &g_cc.db,
SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX, NULL);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: cannot open DB %s: %s",
CC_ID, db_path, sqlite3_errmsg(g_cc.db));
sqlite3_close(g_cc.db); g_cc.db = NULL; return -1;
}
sqlite3_exec(g_cc.db, "PRAGMA journal_mode=WAL", NULL, NULL, NULL);
sqlite3_exec(g_cc.db, "PRAGMA foreign_keys=ON", NULL, NULL, NULL);
}
g_cc.inst = inst;
g_cc.my_node_id = inst->node_id;
sqlite3_int64 now_sec = (sqlite3_int64)ntp_time_get_seconds(inst);
/* записать себя в nodes + local_identity */
{
const char* my_name = inst->name[0] ? inst->name : "Me";
topo_node_sqlite_node_update_verified(g_cc.db, g_cc.my_node_id,
my_name, inst->my_keys.public_key, inst->my_ed25519_pubkey,
(uint64_t)now_sec, (time_t)now_sec);
topo_node_sqlite_node_set_online(g_cc.db, g_cc.my_node_id, 1);
sqlite3_stmt* st = NULL;
sqlite3_prepare_v2(g_cc.db,
"INSERT OR REPLACE INTO local_identity(id,node_id,name,x25519_pubkey,ed25519_pubkey,created_at,updated_at)"
" VALUES(1,?,?,?,?,?,?)", -1, &st, NULL);
if (st) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)g_cc.my_node_id);
sqlite3_bind_text(st, 2, my_name, -1, SQLITE_STATIC);
sqlite3_bind_blob(st, 3, inst->my_keys.public_key, 32, SQLITE_STATIC);
sqlite3_bind_blob(st, 4, inst->my_ed25519_pubkey, 32, SQLITE_STATIC);
sqlite3_bind_int64(st, 5, now_sec);
sqlite3_bind_int64(st, 6, now_sec);
sqlite3_step(st); sqlite3_finalize(st);
}
}
g_cc.initialized = 1;
chat_core_sync_my_addresses();
db_exec(
"CREATE TABLE IF NOT EXISTS local_identity ("
" id INTEGER PRIMARY KEY CHECK (id = 1),"
" node_id INTEGER NOT NULL UNIQUE,"
" name TEXT NOT NULL,"
" x25519_pubkey BLOB NOT NULL,"
" x25519_privkey BLOB,"
" ed25519_pubkey BLOB,"
" created_at INTEGER DEFAULT (unixepoch()),"
" updated_at INTEGER DEFAULT (unixepoch())"
");"
"CREATE TABLE IF NOT EXISTS accounts ("
" node_id INTEGER PRIMARY KEY REFERENCES nodes(node_id),"
" display_name TEXT NOT NULL,"
" avatar_color TEXT DEFAULT '#4A90E2',"
" avatar_letter TEXT NOT NULL,"
" is_contact INTEGER DEFAULT 1,"
" created_at INTEGER DEFAULT (unixepoch())"
");"
"CREATE TABLE IF NOT EXISTS ui_state ("
" key TEXT PRIMARY KEY,"
" value TEXT"
");"
);
/* seed stub accounts if empty */
{
sqlite3_stmt* cnt_st = NULL;
if (sqlite3_prepare_v2(g_cc.db, "SELECT COUNT(*) FROM accounts",
-1, &cnt_st, NULL) == SQLITE_OK) {
if (sqlite3_step(cnt_st) == SQLITE_ROW && sqlite3_column_int(cnt_st, 0) == 0) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: seeding 5 stub accounts", CC_ID);
struct { int64_t id; const char* name; const char* letter; const char* color; } accs[] = {
{1, "Alice", "A", "#4A90E2"}, {2, "Bob", "B", "#7ED321"},
{3, "Carol", "C", "#F5A623"}, {4, "Dave", "D", "#9013FE"},
{5, "Eve", "E", "#00B894"},
};
for (int i = 0; i < 5; i++) {
sqlite3_stmt* ins = NULL;
sqlite3_prepare_v2(g_cc.db,
"INSERT OR IGNORE INTO accounts(node_id,display_name,avatar_color,avatar_letter,created_at)"
" VALUES(?,?,?,?,?)", -1, &ins, NULL);
if (ins) {
sqlite3_bind_int64(ins, 1, accs[i].id);
sqlite3_bind_text(ins, 2, accs[i].name, -1, SQLITE_STATIC);
sqlite3_bind_text(ins, 3, accs[i].color, -1, SQLITE_STATIC);
sqlite3_bind_text(ins, 4, accs[i].letter, -1, SQLITE_STATIC);
sqlite3_bind_int64(ins, 5, now_sec);
sqlite3_step(ins); sqlite3_finalize(ins);
}
}
}
sqlite3_finalize(cnt_st);
}
}
/* ensure infrastructure for all existing channels (survives restart) */
{
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, "SELECT channel_id FROM channels ORDER BY created_at ASC",
-1, &stmt, NULL) == SQLITE_OK) {
int loaded = 0;
while (sqlite3_step(stmt) == SQLITE_ROW) {
const char* ch = (const char*)sqlite3_column_text(stmt, 0);
if (ch && ch[0]) { chat_core_ensure_channel_ready(ch); loaded++; }
}
sqlite3_finalize(stmt);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: loaded %d channels from DB", CC_ID, loaded);
}
}
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized, db=%s node_id=0x%016llx",
CC_ID, db_path, (unsigned long long)g_cc.my_node_id);
{ uint8_t nid[8]; memcpy(nid, &g_cc.my_node_id, 8); gui_bridge_post(GUI_EVT_MY_NODE_ID, nid, 8); }
gui_bridge_post(GUI_EVT_DB_READY, NULL, 0);
return 0;
}
void chat_core_destroy(struct UTUN_INSTANCE* inst) {
(void)inst;
if (!g_cc.initialized) return;
g_cc.initialized = 0;
for (int i = 0; i < g_cc.si_count; i++) u_free(g_cc.si_ch_id[i]);
u_free(g_cc.si_ch_id); u_free(g_cc.si);
g_cc.si = NULL; g_cc.si_ch_id = NULL; g_cc.si_count = g_cc.si_capacity = 0;
if (g_cc.db && !g_cc.shared_db) { sqlite3_close(g_cc.db); }
g_cc.db = NULL; g_cc.inst = NULL;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: destroyed", CC_ID);
}
sqlite3* chat_core_get_db(void) { return g_cc.db; }
struct UTUN_INSTANCE* chat_core_get_inst(void) { return g_cc.inst; }
int chat_core_is_initialized(void) { return g_cc.initialized; }
-86
View File
@@ -1,86 +0,0 @@
/*
* chat_core.h — центральный API чата в потоке uasync
*
* Все DB-операции и сетевые функции выполняются в uasync-потоке.
* GUI вызывает chat_core_submit_message() через gui_bridge_post_uasync().
*/
#ifndef CHAT_CORE_H
#define CHAT_CORE_H
#include <stdint.h>
#include <stddef.h>
struct sqlite3;
struct UTUN_INSTANCE;
/* ── Жизненный цикл ── */
int chat_core_init(struct UTUN_INSTANCE* inst, const char* db_path);
void chat_core_destroy(struct UTUN_INSTANCE* inst);
struct sqlite3* chat_core_get_db(void);
struct UTUN_INSTANCE* chat_core_get_inst(void);
int chat_core_is_initialized(void);
/* ── Отправка сообщения (GUI → uasync) ── */
struct chat_msg_submit {
char channel_id[64];
char content_type[32];
uint8_t* data;
uint32_t data_len;
uint64_t timestamp;
};
void chat_core_submit_message(struct chat_msg_submit* req);
/* ── DB-операции (оставлены для интроспекции) ── */
uint32_t chat_core_count(const char* ch_id);
int chat_core_chain_hash_at(const char* ch_id, uint32_t pos, uint8_t* hash_out);
int chat_core_list_channels(uint8_t* buf, size_t buf_size, size_t* out_len);
int chat_core_list_peers(const char* ch_id, uint8_t* buf, size_t buf_size, size_t* out_len);
int chat_core_load_nodeinfo(uint64_t node_id, uint8_t* buf, size_t buf_size, size_t* out_len);
/* ── Создание канала (GUI → uasync) ── */
struct chat_channel_create {
char channel_id[64];
char name[128];
uint64_t owner_node_id;
uint8_t x25519_pubkey[32];
uint8_t x25519_privkey[32];
uint8_t ed25519_pubkey[32];
uint8_t ed25519_privkey[32];
uint8_t signature[64];
};
void chat_core_ensure_channel_ready(const char* ch_id);
void chat_core_create_channel(struct chat_channel_create* req);
void chat_core_create_channel_trampoline(void* arg);
void chat_core_create_channel_auto(const char* name);
void chat_core_create_channel_auto_trampoline(void* arg);
/* ── Утилиты ── */
void chat_core_set_my_node_id(uint64_t node_id);
void chat_core_update_my_name(const char* name);
void chat_core_update_my_name_trampoline(void* arg);
void chat_core_sync_my_addresses(void);
/* Сохранение key-value в ui_state (вызывается из uasync-потока) */
void chat_core_save_ui_state(const char* key, const char* value);
void chat_core_save_ui_state_trampoline(void* arg);
/* Запуск подключения к узлам выбранного канала (GUI → uasync) */
void chat_core_connect_channel(const char* ch_id);
void chat_core_connect_channel_trampoline(void* arg);
/* Трамплин для gui_bridge_post_uasync (GUI → uasync) */
void chat_core_submit_trampoline(void* arg);
/* Сбор статуса (NTP + connections) и отправка в GUI */
void chat_core_collect_status_trampoline(void* arg);
#endif /* CHAT_CORE_H */
-53
View File
@@ -1,53 +0,0 @@
/*
* chat_core_priv.h — внутренний заголовок для под-модулей chat_core
*
* Предоставляет доступ к глобальному состоянию g_cc и общим хелперам.
* Не включать извне transport/ — только для chat_core*.c.
*/
#ifndef CHAT_CORE_PRIV_H
#define CHAT_CORE_PRIV_H
#include "chat_core.h"
#include "db_sync.h"
#include "../../../lib/debug_config.h"
#include <sqlite3.h>
#include <string.h>
#include <stdio.h>
#define CC_ID "chat_core"
/* ── Глобальное состояние (определено в chat_core.c) ── */
struct chat_core_ctx {
struct UTUN_INSTANCE* inst;
sqlite3* db;
uint8_t shared_db;
uint64_t my_node_id;
struct DB_SYNC_INSTANCE** si;
char** si_ch_id;
int si_count, si_capacity;
uint8_t initialized;
};
extern struct chat_core_ctx g_cc;
/* ── Хелперы ── */
void sanitize_ch_id(const char* ch_id, char* out, size_t out_sz);
void msg_table_name(const char* ch_id, char* buf, size_t sz);
void peers_table_name(const char* ch_id, char* buf, size_t sz);
int db_exec(const char* sql);
/* ── db_sync registry (chat_core.c) ── */
struct DB_SYNC_INSTANCE* si_find(const char* ch_id);
void si_register(struct DB_SYNC_INSTANCE* si, const char* ch_id);
/* ── db_sync callback (chat_msg.c) ── */
void on_msg_inserted(struct DB_SYNC_INSTANCE* si, uint64_t record_ts, const char* data, size_t len, uint64_t author, void* arg);
#endif /* CHAT_CORE_PRIV_H */
-230
View File
@@ -1,230 +0,0 @@
/*
* chat_msg.c — сообщения: отправка, DB-операции для chat_sync
*
* Вынесено из chat_core.c для уменьшения размера модуля.
*/
#include "chat_core_priv.h"
#include "gui_bridge.h"
#include "../../../src/utun_instance.h"
#include "secure_channel.h"
#include "../../../lib/mem.h"
#include "../../../lib/platform_compat.h"
#include <openssl/sha.h>
/* ─── отправка сообщения (GUI → uasync) ─── */
static int msg_get_prev_chain_hash(const char* tbl, uint8_t out[32]) {
char sql[128]; snprintf(sql, sizeof(sql),
"SELECT chain_hash FROM \"%s\" ORDER BY timestamp DESC LIMIT 1", tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) != SQLITE_OK) { memset(out,0,32); return -1; }
if (sqlite3_step(st) == SQLITE_ROW) {
const void* blob = sqlite3_column_blob(st, 0); int blen = sqlite3_column_bytes(st, 0);
if (blob && blen == 32) memcpy(out, blob, 32); else memset(out, 0, 32);
} else { memset(out, 0, 32); }
sqlite3_finalize(st); return 0;
}
void chat_core_submit_message(struct chat_msg_submit* req) {
if (!g_cc.initialized) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: submit before init", CC_ID); return; }
if (!req) return;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: SUBMIT ch=%s ct=%s len=%u",
CC_ID, req->channel_id, req->content_type, req->data_len);
struct DB_SYNC_INSTANCE* si = si_find(req->channel_id);
if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: no db_sync instance for ch=%s", CC_ID, req->channel_id); return; }
char json[4096];
snprintf(json, sizeof(json),
"{\"n\":%llu,\"ch\":\"%s\",\"ct\":\"%s\",\"d\":\"%.*s\"}",
(unsigned long long)g_cc.my_node_id, req->channel_id,
req->content_type, (int)req->data_len, (const char*)req->data);
uint64_t ts = db_sync_next_timestamp(si);
uint8_t sig_msg[8192]; size_t soff = 0;
memcpy(sig_msg + soff, &ts, 8); soff += 8;
size_t jl = strlen(json); memcpy(sig_msg + soff, json, jl); soff += jl;
uint8_t sig[64];
if (sc_ed25519_sign(g_cc.inst->my_ed25519_privkey, sig_msg, soff, sig) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: Ed25519 sign failed", CC_ID);
return;
}
int ret = db_sync_insert_signed(si, json, strlen(json), sig, 64, ts);
if (ret != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: db_sync_insert_signed failed ret=%d", CC_ID, ret);
return;
}
}
void chat_core_submit_trampoline(void* arg) { chat_core_submit_message((struct chat_msg_submit*)arg); u_free(arg); }
/* ─── DB-операции для chat_sync ─── */
uint32_t chat_core_count(const char* ch_id) {
if (!g_cc.initialized) return 0;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[128]; snprintf(sql, sizeof(sql),
"SELECT COUNT(*) FROM \"%s\"", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
uint32_t cnt = 0;
if (sqlite3_step(stmt) == SQLITE_ROW)
cnt = (uint32_t)sqlite3_column_int64(stmt, 0);
sqlite3_finalize(stmt);
return cnt;
}
int chat_core_chain_hash_at(const char* ch_id, uint32_t pos, uint8_t* hash_out) {
if (!g_cc.initialized || !hash_out) return -1;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[200]; snprintf(sql, sizeof(sql),
"SELECT chain_hash FROM \"%s\" ORDER BY timestamp, node_id ASC LIMIT 1 OFFSET ?",
tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) {
memset(hash_out, 0, 32); return -1;
}
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)pos);
if (sqlite3_step(stmt) == SQLITE_ROW) {
const void* blob = sqlite3_column_blob(stmt, 0);
int len = sqlite3_column_bytes(stmt, 0);
if (blob && len == 32) memcpy(hash_out, blob, 32);
else memset(hash_out, 0, 32);
} else {
memset(hash_out, 0, 32);
}
sqlite3_finalize(stmt);
return 0;
}
int chat_core_list_channels(uint8_t* buf, size_t buf_size, size_t* out_len) {
if (!g_cc.initialized || !buf || !out_len) return -1;
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db,
"SELECT channel_id FROM channels ORDER BY created_at ASC",
-1, &stmt, NULL) != SQLITE_OK) return -1;
uint8_t* out = buf;
uint8_t* start = out;
out += 2;
uint16_t cnt = 0;
while (sqlite3_step(stmt) == SQLITE_ROW) {
const char* ch_id = (const char*)sqlite3_column_text(stmt, 0);
int ch_len = sqlite3_column_bytes(stmt, 0);
if (!ch_id || ch_len <= 0 || ch_len > 63) continue;
size_t need = (size_t)(out - buf) + 1 + (size_t)ch_len;
if (need > buf_size) break;
*out++ = (uint8_t)ch_len;
memcpy(out, ch_id, (size_t)ch_len); out += ch_len;
cnt++;
}
sqlite3_finalize(stmt);
memcpy(start, &cnt, 2);
*out_len = (size_t)(out - buf);
return 0;
}
int chat_core_list_peers(const char* ch_id, uint8_t* buf, size_t buf_size,
size_t* out_len) {
if (!g_cc.initialized || !buf || !out_len) return -1;
char tbl[80]; peers_table_name(ch_id, tbl, sizeof(tbl));
char sql[128]; snprintf(sql, sizeof(sql),
"SELECT node_id FROM \"%s\"", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) {
uint16_t z = 0; memcpy(buf, &z, 2); *out_len = 2; return -1;
}
uint16_t cnt = 0;
uint8_t* out = buf + 2;
while (sqlite3_step(stmt) == SQLITE_ROW) {
if ((size_t)(out - buf) + 8 > buf_size) break;
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
memcpy(out, &nid, 8); out += 8; cnt++;
}
sqlite3_finalize(stmt);
memcpy(buf, &cnt, 2);
*out_len = (size_t)(out - buf);
return 0;
}
int chat_core_load_nodeinfo(uint64_t node_id, uint8_t* buf, size_t buf_size,
size_t* out_len) {
if (!g_cc.initialized || !buf || !out_len) return -1;
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db,
"SELECT x25519_pubkey, ed25519_pubkey FROM nodes WHERE node_id=?",
-1, &stmt, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
if (sqlite3_step(stmt) != SQLITE_ROW) { sqlite3_finalize(stmt); return -1; }
const uint8_t* x25519 = (const uint8_t*)sqlite3_column_blob(stmt, 0);
int x25519_len = sqlite3_column_bytes(stmt, 0);
const uint8_t* ed25519 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
int ed25519_len = sqlite3_column_bytes(stmt, 1);
uint8_t* out = buf;
if (x25519 && x25519_len == 32) memcpy(out, x25519, 32);
else memset(out, 0, 32);
out += 32;
if (ed25519 && ed25519_len == 32) memcpy(out, ed25519, 32);
else memset(out, 0, 32);
out += 32;
sqlite3_finalize(stmt);
if (sqlite3_prepare_v2(g_cc.db,
"SELECT family, protocol, address, port, rtt FROM node_addresses WHERE node_id=?",
-1, &stmt, NULL) != SQLITE_OK) {
*out++ = 0; *out_len = (size_t)(out - buf); return 0;
}
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
uint8_t* cnt_pos = out; *out++ = 0;
uint8_t addr_cnt = 0;
while (sqlite3_step(stmt) == SQLITE_ROW && addr_cnt < 255) {
int family = sqlite3_column_int(stmt, 0);
int proto = sqlite3_column_int(stmt, 1);
const uint8_t* addr = (const uint8_t*)sqlite3_column_blob(stmt, 2);
int addr_len = sqlite3_column_bytes(stmt, 2);
uint16_t port = (uint16_t)sqlite3_column_int(stmt, 3);
int16_t rtt = (int16_t)sqlite3_column_int(stmt, 4);
if ((size_t)(out - buf) + 7 + (size_t)addr_len > buf_size) break;
*out++ = (uint8_t)family;
*out++ = (uint8_t)proto;
*out++ = (uint8_t)addr_len;
if (addr_len > 0) { memcpy(out, addr, (size_t)addr_len); out += addr_len; }
memcpy(out, &port, 2); out += 2;
memcpy(out, &rtt, 2); out += 2;
addr_cnt++;
}
*cnt_pos = addr_cnt;
sqlite3_finalize(stmt);
*out_len = (size_t)(out - buf);
return 0;
}
/* ─── 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; (void)data; (void)len;
const char* ch_id = (const char*)arg;
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);
gui_bridge_post(GUI_EVT_MSG_RECEIVED, evt, 1 + cl + 8);
sqlite3_stmt* st = NULL;
sqlite3_prepare_v2(g_cc.db, "UPDATE channels SET last_msg_at = ? WHERE channel_id = ?", -1, &st, NULL);
if (st) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)record_ts);
sqlite3_bind_text(st, 2, ch_id, -1, SQLITE_STATIC);
sqlite3_step(st); sqlite3_finalize(st);
}
}
-189
View File
@@ -1,189 +0,0 @@
/*
* chat_profile.c — профиль узла: имя, адреса, ui_state
*
* Вынесено из chat_core.c для уменьшения размера модуля.
*/
#include "chat_core_priv.h"
#include "gui_bridge.h"
#include "../../../lib/json_flat.h"
#include "topo_node_sqlite.h"
#include "member_sync.h"
#include "../../../src/utun_instance.h"
#include "etcp.h"
#include "secure_channel.h"
#include "../../../src/ntp_time.h"
#include "../../../lib/u_async.h"
#include "../../../lib/mem.h"
#include "../../../lib/platform_compat.h"
#include <openssl/evp.h>
void chat_core_set_my_node_id(uint64_t node_id) {
g_cc.my_node_id = node_id;
}
void chat_core_update_my_name(const char* name) {
if (!g_cc.initialized || !name) return;
topo_node_sqlite_node_update_verified(g_cc.db, g_cc.my_node_id,
name, g_cc.inst->my_keys.public_key, g_cc.inst->my_ed25519_pubkey,
(uint64_t)ntp_time_get_seconds(g_cc.inst), 0);
snprintf(g_cc.inst->name, sizeof(g_cc.inst->name), "%s", name);
uint64_t myid = g_cc.my_node_id;
sqlite3_stmt* cs = NULL;
sqlite3_prepare_v2(g_cc.db, "SELECT channel_id, x25519_pubkey, ed25519_pubkey FROM channels", -1, &cs, NULL);
if (cs) {
while (sqlite3_step(cs) == SQLITE_ROW) {
const char* ch = (const char*)sqlite3_column_text(cs, 0);
const uint8_t* ch_x25519 = sqlite3_column_blob(cs, 1);
const uint8_t* ch_ed25519 = sqlite3_column_blob(cs, 2);
if (!ch || !ch_x25519 || !ch_ed25519) continue;
char tbl[80]; peers_table_name(ch, tbl, sizeof(tbl));
char buf[256]; snprintf(buf, sizeof(buf), "SELECT 1 FROM \"%s\" WHERE node_id=?", tbl);
sqlite3_stmt* ps = NULL;
if (sqlite3_prepare_v2(g_cc.db, buf, -1, &ps, NULL) == SQLITE_OK) {
sqlite3_bind_int64(ps, 1, (sqlite3_int64)myid);
if (sqlite3_step(ps) == SQLITE_ROW) {
uint8_t join_msg[256]; size_t mlen = 0;
memcpy(join_msg + mlen, ch_x25519, 32); mlen += 32;
memcpy(join_msg + mlen, ch_ed25519, 32); mlen += 32;
memcpy(join_msg + mlen, &myid, 8); mlen += 8;
memcpy(join_msg + mlen, g_cc.inst->my_keys.public_key, 32); mlen += 32;
uint64_t join_ts = (uint64_t)ntp_time_get_seconds(g_cc.inst);
memcpy(join_msg + mlen, &join_ts, 8); mlen += 8;
uint8_t new_sig[64]; memset(new_sig, 0, 64);
EVP_PKEY* pkey = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL,
g_cc.inst->my_ed25519_privkey, 32);
if (pkey) {
EVP_MD_CTX* mdctx = EVP_MD_CTX_new();
if (mdctx) {
if (EVP_DigestSignInit(mdctx, NULL, NULL, NULL, pkey) == 1)
EVP_DigestSign(mdctx, new_sig, &(size_t){64}, join_msg, mlen);
EVP_MD_CTX_free(mdctx);
}
EVP_PKEY_free(pkey);
}
char juser[256]; snprintf(juser, sizeof(juser), "{\"name\":\"%s\"}", name);
int r1 = topo_node_sqlite_member_put(g_cc.db, ch, myid, new_sig, join_ts, NULL, 0,
g_cc.inst->my_keys.public_key, g_cc.inst->my_ed25519_pubkey, juser, NULL);
if (r1 != 0) DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put(my_name) FAILED ch=%s rc=%d", CC_ID, ch, r1);
int r2 = member_sync_put(g_cc.inst, ch, myid, g_cc.inst->my_keys.public_key,
g_cc.inst->my_ed25519_pubkey, new_sig, join_ts, NULL, 0, juser, NULL, 0);
if (r2 != 0) DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_sync_put(my_name) FAILED ch=%s rc=%d", CC_ID, ch, r2);
uint8_t evt[65]; uint8_t cl = (uint8_t)strlen(ch);
evt[0] = cl; memcpy(evt + 1, ch, cl);
gui_bridge_post(GUI_EVT_MEMBERS_CHANGED, evt, 1 + cl);
}
sqlite3_finalize(ps);
}
}
sqlite3_finalize(cs);
}
}
void chat_core_update_my_name_trampoline(void* arg) {
chat_core_update_my_name((const char*)arg);
u_free(arg);
}
void chat_core_sync_my_addresses(void) {
if (!g_cc.initialized || !g_cc.db || !g_cc.inst) return;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] sync_my_addresses START my_node=0x%016llx", CC_ID, (unsigned long long)g_cc.my_node_id);
sqlite3_stmt* del = NULL;
sqlite3_prepare_v2(g_cc.db, "DELETE FROM node_addresses WHERE node_id=? AND addr_type=0", -1, &del, NULL);
if (del) {
sqlite3_bind_int64(del, 1, (sqlite3_int64)g_cc.my_node_id);
sqlite3_step(del);
int deleted = sqlite3_changes(g_cc.db);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] DELETE addr_type=0 for my_node=0x%016llx: %d rows deleted", CC_ID, (unsigned long long)g_cc.my_node_id, deleted);
sqlite3_finalize(del);
}
sqlite3_stmt* ins = NULL;
sqlite3_prepare_v2(g_cc.db,
"INSERT OR REPLACE INTO node_addresses(node_id,family,protocol,address,port,addr_type,socket_id)"
" VALUES(?,?,1,?,?,0,?)", -1, &ins, NULL);
if (!ins) {
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] FAIL prepare INSERT stmt my_node=0x%016llx", CC_ID, (unsigned long long)g_cc.my_node_id);
return;
}
struct ETCP_SOCKET* sock = g_cc.inst->etcp_sockets;
int sock_count = 0, written = 0;
while (sock) {
sock_count++;
struct sockaddr_storage* sa = sock->interface_addr.ss_family ? &sock->interface_addr : NULL;
if (!sa) sa = sock->local_addr.ss_family ? &sock->local_addr : NULL;
if (!sa) { DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] socket %d — no interface/local addr", CC_ID, sock_count); sock = sock->next; continue; }
if (sa->ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)sa;
sqlite3_bind_int64(ins, 1, (sqlite3_int64)g_cc.my_node_id);
sqlite3_bind_int(ins, 2, 4);
sqlite3_bind_blob(ins, 3, &sin->sin_addr, 4, SQLITE_STATIC);
sqlite3_bind_int(ins, 4, (int)ntohs(sin->sin_port));
sqlite3_bind_int(ins, 5, (int)sock->sock_id);
sqlite3_step(ins); sqlite3_reset(ins);
uint8_t* ip = (uint8_t*)&sin->sin_addr;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] INSERT my_node=0x%016llx sock=%d %d.%d.%d.%d:%d",
CC_ID, (unsigned long long)g_cc.my_node_id, sock->sock_id, ip[0], ip[1], ip[2], ip[3], (int)ntohs(sin->sin_port));
written++;
} else if (sa->ss_family == AF_INET6) {
struct sockaddr_in6* sin6 = (struct sockaddr_in6*)sa;
sqlite3_bind_int64(ins, 1, (sqlite3_int64)g_cc.my_node_id);
sqlite3_bind_int(ins, 2, 6);
sqlite3_bind_blob(ins, 3, &sin6->sin6_addr, 16, SQLITE_STATIC);
sqlite3_bind_int(ins, 4, (int)ntohs(sin6->sin6_port));
sqlite3_bind_int(ins, 5, (int)sock->sock_id);
sqlite3_step(ins); sqlite3_reset(ins);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] INSERT v6 my_node=0x%016llx sock=%d port=%d",
CC_ID, (unsigned long long)g_cc.my_node_id, sock->sock_id, (int)ntohs(sin6->sin6_port));
written++;
}
sock = sock->next;
}
sqlite3_finalize(ins);
if (sock_count == 0)
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] etcp_sockets is NULL — NO sockets, addresses NOT written!", CC_ID);
else if (written == 0)
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] %d sockets found but 0 addresses written (all have no interface_addr nor local_addr)", CC_ID, sock_count);
else
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] DONE: %d of %d sockets written, my_node=0x%016llx",
CC_ID, written, sock_count, (unsigned long long)g_cc.my_node_id);
sqlite3_stmt* chk = NULL;
sqlite3_prepare_v2(g_cc.db, "SELECT COUNT(*) FROM node_addresses WHERE node_id=? AND addr_type=0", -1, &chk, NULL);
if (chk) {
sqlite3_bind_int64(chk, 1, (sqlite3_int64)g_cc.my_node_id);
int cnt = sqlite3_step(chk) == SQLITE_ROW ? sqlite3_column_int(chk, 0) : -1;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] VERIFY: total addr_type=0 rows for my_node=0x%016llx = %d",
CC_ID, (unsigned long long)g_cc.my_node_id, cnt);
sqlite3_finalize(chk);
}
}
void chat_core_save_ui_state(const char* key, const char* value) {
if (!g_cc.initialized || !key || !value) return;
sqlite3_stmt* st = NULL;
sqlite3_prepare_v2(g_cc.db,
"INSERT OR REPLACE INTO ui_state(key, value) VALUES(?, ?)", -1, &st, NULL);
if (st) {
sqlite3_bind_text(st, 1, key, -1, SQLITE_STATIC);
sqlite3_bind_text(st, 2, value, -1, SQLITE_STATIC);
sqlite3_step(st); sqlite3_finalize(st);
}
}
struct save_ui_state_arg { char data[256]; };
void chat_core_save_ui_state_trampoline(void* arg) {
struct save_ui_state_arg* a = (struct save_ui_state_arg*)arg;
const char* key = a->data;
const char* value = key + strlen(key) + 1;
chat_core_save_ui_state(key, value);
u_free(arg);
}
-118
View File
@@ -1,118 +0,0 @@
/*
* chat_status.c — сбор статуса (NTP + ETCP connections) и отправка в GUI
*
* Вынесено из chat_core.c для уменьшения размера модуля.
*/
#include "chat_core_priv.h"
#include "gui_bridge.h"
#include "../../../src/utun_instance.h"
#include "../../../src/ntp_time.h"
#include "../../../lib/ll_queue.h"
#include "../../../lib/platform_compat.h"
#include "etcp.h"
#include "etcp_connections.h"
static const char* nat_type_str(uint8_t t) {
switch (t) { case 0: return "UNKNOWN"; case 1: return "EIM"; case 2: return "STRICT"; case 3: return "DIRECT"; default: return "?"; }
}
static void get_node_name(uint64_t node_id, char* out, size_t sz) {
out[0] = '\0';
if (!g_cc.db || node_id == 0) return;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(g_cc.db, "SELECT name FROM nodes WHERE node_id=?", -1, &st, NULL) != SQLITE_OK) return;
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
const char* n = (const char*)sqlite3_column_text(st, 0);
if (n) snprintf(out, sz, "%s", n);
}
sqlite3_finalize(st);
}
static void chat_core_collect_status(void) {
char buf[8192]; int off = 0;
if (!g_cc.initialized || !g_cc.inst) {
off = snprintf(buf, sizeof(buf), "uTun not initialized\n");
gui_bridge_post(GUI_EVT_STATUS_REFRESH, (const uint8_t*)buf, off);
return;
}
struct NTP_TIME* ntp = &g_cc.inst->ntp;
off += snprintf(buf + off, sizeof(buf) - off, "=== NTP ===\n");
off += snprintf(buf + off, sizeof(buf) - off, "Enabled: %s\n", ntp->enabled ? "yes" : "no");
off += snprintf(buf + off, sizeof(buf) - off, "Synced: %s\n", ntp->synced ? "yes" : "no");
off += snprintf(buf + off, sizeof(buf) - off, "Offset: %.1f s\n", ntp->offset_us / 1000000.0);
if (ntp->server_count > 0 && ntp->servers) {
off += snprintf(buf + off, sizeof(buf) - off, "Servers: %d\n", ntp->server_count);
for (int i = 0; i < ntp->server_count && ntp->servers[i]; i++)
off += snprintf(buf + off, sizeof(buf) - off, " [%d] %s%s\n", i, ntp->servers[i],
i == ntp->server_current ? " (current)" : "");
}
if (ntp->synced && ntp->last_sync_tb > 0) {
uint64_t now_tb = get_time_tb();
uint64_t ago_sec = (now_tb - ntp->last_sync_tb) / 10000;
off += snprintf(buf + off, sizeof(buf) - off, "Last sync: %llu sec ago\n", (unsigned long long)ago_sec);
}
int conn_count = 0;
struct ll_entry* entry = g_cc.inst->connections->head;
while (entry) { conn_count++; entry = entry->next; }
off += snprintf(buf + off, sizeof(buf) - off, "=== Connections (%d) ===\n", conn_count);
uint64_t my_id = g_cc.inst->node_id;
entry = g_cc.inst->connections->head;
while (entry) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data;
if (!ce || !ce->conn) { entry = entry->next; continue; }
uint64_t pid = ce->peer_node_id;
struct ETCP_CONN* conn = ce->conn;
int link_count = 0;
struct ETCP_LINK* tl = conn->links;
while (tl) { link_count++; tl = tl->next; }
const char* init_str = conn->initialized ? "" : "(!init)";
char myhex[5], peerhex[5];
snprintf(myhex, sizeof(myhex), "%04llX", (unsigned long long)(my_id & 0xFFFF));
snprintf(peerhex, sizeof(peerhex), "%04llX", (unsigned long long)(pid & 0xFFFF));
char peername[64];
get_node_name(pid, peername, sizeof(peername));
if (peername[0])
off += snprintf(buf + off, sizeof(buf) - off, "[%s]→[%s] %s ETCP:%s%s(%dL)\n",
myhex, peerhex, peername,
conn->links_up ? "UP" : "DOWN", init_str, link_count);
else
off += snprintf(buf + off, sizeof(buf) - off, "[%s]→[%s] ETCP:%s%s(%dL)\n",
myhex, peerhex,
conn->links_up ? "UP" : "DOWN", init_str, link_count);
int link_idx = 0;
struct ETCP_LINK* link = conn->links;
while (link) {
char rtt_str[32]; rtt_str[0] = '\0';
if (link->rtt_last > 0) snprintf(rtt_str, sizeof(rtt_str), " / rtt=%ums", link->rtt_last / 10);
off += snprintf(buf + off, sizeof(buf) - off, " LINK#%d: %s /NAT=%s%s\n",
link_idx,
link->link_status ? "UP" : "DOWN",
nat_type_str(link->nat_type),
rtt_str);
link = link->next; link_idx++;
}
entry = entry->next;
}
gui_bridge_post(GUI_EVT_STATUS_REFRESH, (const uint8_t*)buf, off);
}
void chat_core_collect_status_trampoline(void* arg) {
(void)arg;
chat_core_collect_status();
}
File diff suppressed because it is too large. Load diff
-60
View File
@@ -1,60 +0,0 @@
#ifndef CHAT_SYNC_H
#define CHAT_SYNC_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
#include <stddef.h>
struct UTUN_INSTANCE;
struct UASYNC;
/* etcp service ID (direct P2P) */
#define ETCP_RT_ID_CHAT_SYNC 0x30
#define ETCP_RT_ID_MEMBER_SYNC 0x31
/* Message types */
#define CS_MSG_INIT_SYNC 0x01
#define CS_MSG_INIT_RESP 0x02
#define CS_MSG_SEND_DATA 0x03
#define CS_MSG_PUSH 0x04
#define CS_MSG_ACK_PUSH 0x05
#define CS_MSG_SYNC_DONE 0x06
#define CS_MSG_ERROR 0x07
#define CS_MSG_CHANNEL_INFO_REQ 0x09
#define CS_MSG_CHANNEL_INFO_RESP 0x0A
#define CS_MSG_CHANNEL_JOIN 0x0B
#define CS_MSG_WELCOME 0x0C
#define CS_MSG_PEER_UPSERT 0x0D
#define CS_MSG_PEER_REMOVE 0x0E
/* Protocol constants */
#define CS_SEND_DATA_MAX 32
#define CS_MAX_SPARSE 16
#define CS_PEER_TIMEOUT_MS 5000
/* Sync states per channel */
#define CS_SYNC_NONE 0
#define CS_SYNC_IN_PROGRESS 1
#define CS_SYNC_DONE 2
/* ── Public API ── */
int chat_sync_init(struct UTUN_INSTANCE* inst,
void (*gui_result_cb)(void*, int, const uint8_t*, int));
void chat_sync_destroy(struct UTUN_INSTANCE* inst);
/* Прямое ETCP-подключение к узлу (pubkey+адреса из БД) */
void chat_sync_connect_node(struct UTUN_INSTANCE* inst, uint64_t node_id);
/* Подключиться к пиру по данным invite-ссылки (вызывается из GUI-потока) */
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);
#ifdef __cplusplus
}
#endif
#endif /* CHAT_SYNC_H */
-472
View File
@@ -1,472 +0,0 @@
#include "member_sync.h"
#include "topo_node_sqlite.h"
#include "chat_core.h"
#include "../../../src/utun_instance.h"
#include "../../../lib/debug_config.h"
#include "../../../lib/mem.h"
#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>
#include <openssl/evp.h>
#define MS_ID "member_sync"
struct addr_item { uint8_t family; uint8_t addr[16]; uint16_t port; };
/* ── DB access ── */
static sqlite3* _db(struct UTUN_INSTANCE* inst) {
return inst ? inst->topo_sqlite_db : NULL;
}
static void _peers_table(const char* ch_id, char* buf, size_t sz) {
size_t i = 0;
while (*ch_id && i < sz - 1) {
char c = *ch_id++;
if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_')
buf[i++] = c; else buf[i++] = '_';
}
buf[i] = '\0';
char tbl[128]; snprintf(tbl, sizeof(tbl), "peers_%s", buf);
snprintf(buf, sz, "%s", tbl);
}
/* ── Member hash (identical to old _compute_member_hash) ── */
static int _addr_cmp(const void* a, const void* b) {
const struct addr_item* ia = (const struct addr_item*)a;
const struct addr_item* ib = (const struct addr_item*)b;
if (ia->family != ib->family) return (int)ia->family - (int)ib->family;
int r = memcmp(ia->addr, ib->addr, (size_t)(ia->family == 4 ? 4 : 16));
if (r) return r;
return (int)ia->port - (int)ib->port;
}
static void _compute_member_hash(uint64_t node_id, const uint8_t* x25519,
const uint8_t* ed25519,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const uint8_t* addrs_data, int addr_count,
uint8_t hash_out[MT_HASH_SIZE]) {
EVP_MD_CTX* ctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
EVP_DigestUpdate(ctx, &node_id, 8);
EVP_DigestUpdate(ctx, x25519, 32);
EVP_DigestUpdate(ctx, ed25519, 32);
EVP_DigestUpdate(ctx, join_sig, 64);
EVP_DigestUpdate(ctx, &join_ts, 8);
if (update_sig) EVP_DigestUpdate(ctx, update_sig, 64); else { static const uint8_t z[64]; EVP_DigestUpdate(ctx, z, 64); }
EVP_DigestUpdate(ctx, &update_ts, 8);
uint8_t ac = (uint8_t)addr_count;
EVP_DigestUpdate(ctx, &ac, 1);
if (addr_count > 0 && addrs_data) {
struct addr_item items[256]; int n = 0;
const uint8_t* p = addrs_data;
for (int i = 0; i < addr_count && n < 256; i++) {
uint8_t fam = *p++; items[n].family = fam;
int ip_len = fam == 4 ? 4 : 16;
memcpy(items[n].addr, p, (size_t)ip_len); p += ip_len;
items[n].port = ((uint16_t)p[0] << 8) | p[1]; p += 2;
n++;
}
qsort(items, (size_t)n, sizeof(struct addr_item), _addr_cmp);
for (int i = 0; i < n; i++) {
EVP_DigestUpdate(ctx, &items[i].family, 1);
EVP_DigestUpdate(ctx, items[i].addr, (size_t)(items[i].family == 4 ? 4 : 16));
uint8_t port_be[2] = { (uint8_t)(items[i].port >> 8), (uint8_t)(items[i].port & 0xFF) };
EVP_DigestUpdate(ctx, port_be, 2);
}
}
EVP_DigestFinal_ex(ctx, hash_out, NULL);
EVP_MD_CTX_free(ctx);
}
/* ── merkle_sync_data_ops implementation ── */
static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
uint64_t prefix64, EVP_MD_CTX* sha_ctx) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db) return -1;
if (level == 0) return 0;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: bucket_hash ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix64);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
int mask_shift = 64 - (int)level * 5;
uint64_t mask = (mask_shift >= 0 && mask_shift < 64) ? (~0ULL << mask_shift) : UINT64_MAX;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts"
" FROM \"%s\""
" WHERE (node_id & %lld) == %lld ORDER BY node_id ASC",
peers_tbl, (long long)mask, (long long)prefix64);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;
int count = 0;
while (sqlite3_step(stmt) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
const uint8_t* ed = (const uint8_t*)sqlite3_column_blob(stmt, 2);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
uint64_t jts = (uint64_t)sqlite3_column_int64(stmt, 4);
const uint8_t* usig = (const uint8_t*)sqlite3_column_blob(stmt, 5);
uint64_t uts = (uint64_t)sqlite3_column_int64(stmt, 6);
if (!x25 || !ed) continue;
uint8_t mh[MT_HASH_SIZE];
_compute_member_hash(nid, x25, ed, sig, jts, usig, uts, NULL, 0, mh);
EVP_DigestUpdate(sha_ctx, mh, MT_HASH_SIZE);
count++;
}
sqlite3_finalize(stmt);
return count;
}
static int _member_get_items(void* ctx, const char* ns, uint8_t level,
uint64_t prefix, uint8_t prefix_bytes,
uint8_t* buf, size_t* len) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db || !buf || !len) return -1;
if (level == 0) { *len = 0; return 0; }
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: get_items ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
int mask_shift = 64 - (int)level * 5;
uint64_t mask = (mask_shift >= 0 && mask_shift < 64) ? (~0ULL << mask_shift) : UINT64_MAX;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, userinfo"
" FROM \"%s\""
" WHERE (node_id & %lld) == %lld ORDER BY node_id ASC",
peers_tbl, (long long)mask, (long long)prefix);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;
size_t off = 0;
if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; }
uint16_t* cnt = (uint16_t*)(buf + off); off += 2; *cnt = 0;
while (sqlite3_step(stmt) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
const uint8_t* ed = (const uint8_t*)sqlite3_column_blob(stmt, 2);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
uint64_t jts = (uint64_t)sqlite3_column_int64(stmt, 4);
const uint8_t* usig = (const uint8_t*)sqlite3_column_blob(stmt, 5);
uint64_t uts = (uint64_t)sqlite3_column_int64(stmt, 6);
const char* nm = (const char*)sqlite3_column_text(stmt, 7);
if (!x25 || !ed) continue;
uint8_t nl = nm ? (uint8_t)strnlen(nm, 255) : 0;
sqlite3_stmt* as = NULL;
sqlite3_prepare_v2(db,
"SELECT family, socket_id, address, port FROM node_addresses WHERE node_id=? AND addr_type=0"
" ORDER BY family, address, port", -1, &as, NULL);
uint8_t addrs[2048]; int addr_off = 0; int addr_count = 0;
if (as) {
sqlite3_bind_int64(as, 1, (sqlite3_int64)nid);
while (sqlite3_step(as) == SQLITE_ROW && addr_off < (int)sizeof(addrs) - 8) {
int fam = sqlite3_column_int(as, 0);
int sid = sqlite3_column_int(as, 1);
addrs[addr_off++] = (uint8_t)fam;
addrs[addr_off++] = (uint8_t)sid;
int ip_sz = fam == 4 ? 4 : 16;
memcpy(addrs + addr_off, sqlite3_column_blob(as, 2), (size_t)ip_sz);
addr_off += ip_sz;
uint16_t p = (uint16_t)sqlite3_column_int(as, 3);
addrs[addr_off++] = (uint8_t)(p >> 8);
addrs[addr_off++] = (uint8_t)(p & 0xFF);
addr_count++;
}
sqlite3_finalize(as);
}
uint8_t flags = (sig && jts) ? PEERS_FLAG_HAS_JOIN : 0;
size_t need = 8 + 32 + 32 + 1 + (flags ? 72ULL : 0ULL) + 64 + 8 + 1 + (size_t)nl + 1 + (size_t)addr_off;
if (off + need > *len) { sqlite3_finalize(stmt); return -2; }
memcpy(buf + off, &nid, 8); off += 8;
memcpy(buf + off, x25, 32); off += 32;
memcpy(buf + off, ed, 32); off += 32;
buf[off++] = flags;
if (flags & PEERS_FLAG_HAS_JOIN) {
memcpy(buf + off, sig, 64); off += 64;
memcpy(buf + off, &jts, 8); off += 8;
}
if (usig && uts) {
memcpy(buf + off, usig, 64); off += 64;
memcpy(buf + off, &uts, 8); off += 8;
} else {
memset(buf + off, 0, 64); off += 64;
uint64_t z = 0; memcpy(buf + off, &z, 8); off += 8;
}
buf[off++] = nl;
if (nl) { memcpy(buf + off, nm, nl); off += nl; }
buf[off++] = (uint8_t)addr_count;
memcpy(buf + off, addrs, (size_t)addr_off); off += (size_t)addr_off;
(*cnt)++;
}
sqlite3_finalize(stmt);
*len = off;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: get_items ns=%s L%d/P%016llx mask=%016llx count=%u off=%zu",
MS_ID, ns, level, (unsigned long long)prefix, (unsigned long long)mask, *cnt, off);
return 0;
}
static int _member_apply_items(void* ctx, const char* ns, uint64_t from_peer,
const uint8_t* data, size_t len) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
if (len < 2) return -1;
uint16_t count; memcpy(&count, data, 2);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: apply_items ns=%s count=%u from=%016llx", MS_ID, ns, count, (unsigned long long)from_peer);
const uint8_t* mp = data + 2; size_t mrem = len - 2;
for (uint16_t i = 0; i < count && mrem >= 148; i++) {
uint64_t nid; memcpy(&nid, mp, 8); mp += 8; mrem -= 8;
const uint8_t* x25 = mp; mp += 32; mrem -= 32;
const uint8_t* ed = mp; mp += 32; mrem -= 32;
uint8_t flags = *mp++; mrem--;
const uint8_t* jsig = NULL; uint64_t jts = 0;
if (flags & PEERS_FLAG_HAS_JOIN) {
if (mrem < 72) break; jsig = mp; mp += 64; mrem -= 64; memcpy(&jts, mp, 8); mp += 8; mrem -= 8;
} else {
/* lookup join from local DB */
uint8_t buf_jsig[64]; uint64_t buf_jts;
if (topo_node_sqlite_member_get_join(_db(inst), ns, nid, buf_jsig, &buf_jts) == 0) { jsig = buf_jsig; jts = buf_jts; }
}
if (mrem < 72) break;
const uint8_t* usig = mp; mp += 64; mrem -= 64; uint64_t uts; memcpy(&uts, mp, 8); mp += 8; mrem -= 8;
uint8_t nl = *mp++; mrem--;
char nm[256] = "";
if (nl && mrem >= nl) { memcpy(nm, mp, nl); nm[nl] = '\0'; mp += nl; mrem -= nl; }
uint8_t ac = *mp++; mrem--;
const uint8_t* addrs = mp;
int consumed = 0;
for (int a = 0; a < (int)ac && mrem >= (size_t)(2 + consumed); a++) {
uint8_t fam = mp[consumed]; consumed++;
uint8_t sid = mp[consumed]; consumed++;
int sz = fam == 4 ? 4 : 16;
consumed += sz + 2;
}
/* verify update_sig if available */
if (usig && uts && jsig && ed) {
uint8_t vmsg[256]; size_t vlen = 0;
memcpy(vmsg + vlen, jsig, 64); vlen += 64;
memcpy(vmsg + vlen, &uts, 8); vlen += 8;
size_t nl2 = nm ? strlen(nm) : 0; memcpy(vmsg + vlen, nm ? nm : "", nl2 + 1); vlen += nl2 + 1;
{ EVP_PKEY* pkey = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, ed, 32);
if (pkey) {
EVP_MD_CTX* ver = EVP_MD_CTX_new();
if (ver) {
int ok = (EVP_DigestVerifyInit(ver, NULL, NULL, NULL, pkey) == 1)
&& (EVP_DigestVerify(ver, usig, 64, vmsg, vlen) == 1);
if (!ok)
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_sync invalid update_sig node=0x%016llx ns=%s", MS_ID,
(unsigned long long)nid, ns);
EVP_MD_CTX_free(ver);
}
EVP_PKEY_free(pkey);
}
}
}
member_sync_put(inst, ns, nid, x25, ed, jsig, jts, usig, uts, nm, addrs, (int)ac);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync recv ns=%s nid=0x%016llx ac=%d", MS_ID, ns, (unsigned long long)nid, ac);
mp += consumed; mrem -= (size_t)consumed;
}
if (count > 0)
merkle_sync_broadcast(inst, ns, from_peer, data, len);
return 0;
}
static member_sync_node_updated_fn g_node_updated_cb = NULL;
static int _ms_apply_update(void* ctx, const char* ns, uint64_t key,
uint8_t type, const uint8_t* data, size_t len) {
(void)ns;
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
if (type == 0x01 && len >= 1) {
int online = data[0];
member_sync_set_online(inst, key, online);
if (g_node_updated_cb) g_node_updated_cb(key, online);
}
return 0;
}
static const struct merkle_sync_data_ops g_member_ops = {
.update_bucket_hash = _member_update_bucket_hash,
.get_items = _member_get_items,
.apply_items = _member_apply_items,
.apply_update = _ms_apply_update,
};
/* ── Public API ── */
static void _rebuild_all_trees(struct UTUN_INSTANCE* inst) {
sqlite3* db = _db(inst); if (!db) return;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(db,
"SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'peers_%'",
-1, &st, NULL) != SQLITE_OK) return;
while (sqlite3_step(st) == SQLITE_ROW) {
const char* tbl = (const char*)sqlite3_column_text(st, 0);
if (!tbl || strncmp(tbl, "peers_", 6) != 0) continue;
const char* ch_id = tbl + 6;
sqlite3_stmt* ns = NULL;
char sql[256]; snprintf(sql, sizeof(sql), "SELECT node_id FROM \"%s\"", tbl);
if (sqlite3_prepare_v2(db, sql, -1, &ns, NULL) == SQLITE_OK) {
while (sqlite3_step(ns) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(ns, 0);
merkle_sync_recompute_path(inst, ch_id, nid);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: rebuild tree ns=%s nid=%016llx", MS_ID, ch_id, (unsigned long long)nid);
}
sqlite3_finalize(ns);
}
}
sqlite3_finalize(st);
}
int member_sync_init(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: init", MS_ID);
int rc = merkle_sync_init(inst, 0x31, &g_member_ops, inst);
if (rc != 0) return rc;
_rebuild_all_trees(inst);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: initialized, merkle_rc=%d", MS_ID, rc);
return 0;
}
void member_sync_destroy(struct UTUN_INSTANCE* inst) {
if (!inst) return;
merkle_sync_destroy(inst);
}
int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ch_id, merkle_sync_done_cb done_cb, void* arg) {
return merkle_sync_start(inst, peer, ch_id, done_cb, arg);
}
void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) {
merkle_sync_cancel(inst, peer, ch_id);
}
int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t member_id, const uint8_t* x25519,
const uint8_t* ed25519,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const char* userinfo,
const uint8_t* addrs_data, int addr_count) {
if (!inst || !ch_id) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: put ch=%s nid=%016llx userinfo=%s ac=%d", MS_ID, ch_id, (unsigned long long)member_id, userinfo ? userinfo : "", addr_count);
sqlite3* db = _db(inst); if (!db) return -1;
int rc = topo_node_sqlite_member_put(db, ch_id, member_id,
join_sig, join_ts, update_sig, update_ts,
x25519, ed25519, userinfo, NULL);
if (rc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put FAILED ch=%s id=0x%016llx userinfo=%s rc=%d",
MS_ID, ch_id, (unsigned long long)member_id, userinfo ? userinfo : "", rc);
return -1;
}
if (addrs_data && addr_count > 0) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put node=0x%016llx ch=%s addr_count=%d",
MS_ID, (unsigned long long)member_id, ch_id, addr_count);
sqlite3_exec(db, "BEGIN", NULL, NULL, NULL);
sqlite3_stmt* ds = NULL;
sqlite3_prepare_v2(db, "DELETE FROM node_addresses WHERE node_id=? AND addr_type=0", -1, &ds, NULL);
if (ds) { sqlite3_bind_int64(ds, 1, (sqlite3_int64)member_id); sqlite3_step(ds);
int deleted = sqlite3_changes(db);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put DELETE addr_type=0 node=0x%016llx: %d rows deleted",
MS_ID, (unsigned long long)member_id, deleted);
sqlite3_finalize(ds); }
sqlite3_stmt* as = NULL;
sqlite3_prepare_v2(db,
"INSERT INTO node_addresses(node_id,family,protocol,address,port,addr_type,socket_id)"
" VALUES(?,?,1,?,?,0,?)", -1, &as, NULL);
if (as) {
const uint8_t* p = addrs_data;
int written = 0;
for (int i = 0; i < addr_count; i++) {
uint8_t fam = *p++;
uint8_t sid = *p++;
int ip_sz = fam == 4 ? 4 : 16;
sqlite3_bind_int64(as, 1, (sqlite3_int64)member_id);
sqlite3_bind_int(as, 2, fam);
sqlite3_bind_blob(as, 3, p, ip_sz, SQLITE_STATIC);
p += ip_sz;
uint16_t port = ((uint16_t)p[0] << 8) | p[1]; p += 2;
sqlite3_bind_int(as, 4, port);
sqlite3_bind_int(as, 5, (int)sid);
sqlite3_step(as); sqlite3_reset(as);
if (fam == 4) {
const uint8_t* ip = p - 6; /* p advanced by ip_sz(4) + port(2) */
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put INSERT node=0x%016llx sock=%d %d.%d.%d.%d:%d",
MS_ID, (unsigned long long)member_id, sid, ip[0], ip[1], ip[2], ip[3], port);
} else {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put INSERT v6 node=0x%016llx sock=%d port=%d",
MS_ID, (unsigned long long)member_id, sid, port);
}
written++;
}
sqlite3_finalize(as);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put DONE node=0x%016llx: %d addrs written",
MS_ID, (unsigned long long)member_id, written);
}
sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
} else {
if (!addrs_data)
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put node=0x%016llx ch=%s addrs_data=NULL — NO addresses (skipping)", MS_ID, (unsigned long long)member_id, ch_id);
else
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: [ADDR_SYNC] member_sync_put node=0x%016llx ch=%s addr_count=%d — empty, NO DELETE (skipping)", MS_ID, (unsigned long long)member_id, ch_id, addr_count);
}
merkle_sync_recompute_path(inst, ch_id, member_id);
return 0;
}
int member_sync_del(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id) {
if (!inst || !ch_id) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: del ch=%s nid=%016llx", MS_ID, ch_id, (unsigned long long)member_id);
sqlite3* db = _db(inst); if (!db) return -1;
topo_node_sqlite_member_del(db, ch_id, member_id);
merkle_sync_recompute_path(inst, ch_id, member_id);
return 0;
}
int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id) {
sqlite3* db = _db(inst);
if (!db || !ch_id) return 0;
char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
char sql[256]; snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM \"%s\"", peers_tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
int c = 0;
if (sqlite3_step(stmt) == SQLITE_ROW) c = sqlite3_column_int(stmt, 0);
sqlite3_finalize(stmt);
return c;
}
void member_sync_set_node_updated_cb(member_sync_node_updated_fn cb) {
g_node_updated_cb = cb;
}
void member_sync_set_online(struct UTUN_INSTANCE* inst, uint64_t node_id, int online) {
if (!inst) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: set_online nid=%016llx online=%d", MS_ID, (unsigned long long)node_id, online);
sqlite3* db = _db(inst); if (!db) return;
topo_node_sqlite_node_set_online(db, node_id, online);
}
const uint8_t* member_sync_get_hash(struct UTUN_INSTANCE* inst, const char* ch_id,
uint8_t level, uint64_t prefix64) {
return merkle_sync_get_hash(inst, ch_id, level, prefix64);
}
-133
View File
@@ -1,133 +0,0 @@
/*
* ── member_sync — синхронизация участников чата ──
*
* Тонкая прослойка над merkle_sync, адаптированная под мемберов каналов.
* Данные хранятся в таблицах: nodes, node_addresses, peers_<channel_id>.
* Хеш мембера = SHA256(node_id || x25519 || ed25519 || join_sig || join_ts || addrs).
*
* ── Использование ──
*
* // разово: инициализация (из chat_sync_init)
* member_sync_init(inst);
*
* // асинхронный запуск синхронизации (из cs_on_conn_up)
* member_sync_start(inst, peer, ch_id, on_done, ch);
*
* // коллбэк: синхронизация завершена
* static void on_done(uint64_t peer, const char* ch_id, int result, void* arg) {
* struct channel_cache* ch = (struct channel_cache*)arg;
* if (result == MT_OK) ch->synced = CS_SYNC_DONE;
* }
*
* // добавление/обновление мембера (из chat_core_create_channel, cs_handle_welcome)
* member_sync_put(inst, ch_id, node_id, x25519, ed25519, join_sig, join_ts, addrs, ac);
* // дерево автоматически пересчитано
*
* // онлайн-статус (из cs_on_conn_up / cs_on_conn_down)
* member_sync_set_online(inst, node_id, 1); // online
* member_sync_set_online(inst, node_id, 0); // offline
*
* // удаление мембера
* member_sync_del(inst, ch_id, node_id);
*
* // отмена синхронизации (коллбэк НЕ вызывается)
* member_sync_cancel(inst, peer, ch_id);
*
* // количество мемберов в канале
* int n = member_sync_count(inst, ch_id);
*
* // разово: завершение (из chat_sync_destroy)
* member_sync_destroy(inst);
*/
#ifndef MEMBER_SYNC_H
#define MEMBER_SYNC_H
#include "merkle_sync.h"
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
#include <stddef.h>
struct UTUN_INSTANCE;
/*
* Инициализировать модуль: создаёт merkle_sync с коллбэками для мемберов
* (ETCP сервис 0x31), регистрирует _on_node_updated для пересчёта дерева
* при обновлении node info.
*/
int member_sync_init(struct UTUN_INSTANCE* inst);
/* Завершить модуль. */
void member_sync_destroy(struct UTUN_INSTANCE* inst);
/*
* Запустить синхронизацию мемберов канала ch_id с пиром peer.
* Делегирует в merkle_sync_start().
*/
int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ch_id, merkle_sync_done_cb done_cb, void* arg);
/*
* Отменить синхронизацию. Делегирует в merkle_sync_cancel().
*/
void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id);
/*
* Добавить/обновить мембера и пересчитать дерево.
*
* ch_id — ID канала
* member_id — node_id участника
* x25519 — X25519 публичный ключ (32 байта)
* ed25519 — Ed25519 публичный ключ (32 байта)
* join_sig — Ed25519 подпись join-сообщения (64 байта)
* сообщение: ch_x25519(32) || ch_ed25519(32) || node_id(8LE) || x25519(32) || join_ts(8LE)
* join_ts — NTP-синхронизированное время подписи (unix epoch seconds)
* userinfo — метаданные узла (имя) в канале
* addrs_data — адреса в wire-формате: [family:1][ip:4|16][port:2]...
* addr_count — количество адресов (0 если нет)
*
* Пишет только в таблицу peers_<ch_id> (INSERT OR REPLACE).
* nodes обновляется отдельно через topo_node_sqlite_node_update_verified().
* Вызывает merkle_sync_recompute_path() для пересчёта дерева.
*
* Возвращает 0 при успехе, -1 при ошибке.
*/
int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t member_id, const uint8_t* x25519,
const uint8_t* ed25519,
const uint8_t* join_sig, uint64_t join_ts,
const uint8_t* update_sig, uint64_t update_ts,
const char* userinfo,
const uint8_t* addrs_data, int addr_count);
/* Удалить мембера из канала и пересчитать дерево. */
int member_sync_del(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id);
/* Количество мемберов в канале (SELECT COUNT из peers_<ch_id>). */
int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id);
/*
* Установить онлайн-статус узла (nodes.online = 0/1).
*/
void member_sync_set_online(struct UTUN_INSTANCE* inst, uint64_t node_id, int online);
/* Получить хеш бакета из merkle_tree_hash (для тестов). */
const uint8_t* member_sync_get_hash(struct UTUN_INSTANCE* inst,
const char* ch_id, uint8_t level, uint64_t prefix64);
/*
* Коллбэк: узел изменился по инициативе удалённого пира (через MSG_ITEM_UPDATE).
* Вызывается из uasync-потока. Потребитель (chat_sync) может из него
* постить GUI-события.
*/
typedef void (*member_sync_node_updated_fn)(uint64_t node_id, int online);
void member_sync_set_node_updated_cb(member_sync_node_updated_fn cb);
#ifdef __cplusplus
}
#endif
#endif /* MEMBER_SYNC_H */
-790
View File
@@ -1,790 +0,0 @@
#include "merkle_sync.h"
#include "../../../src/utun_instance.h"
#include "etcp_api.h"
#include "etcp.h"
#include "../../../lib/debug_config.h"
#include "../../../lib/mem.h"
#include "../../../lib/u_async.h"
#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>
#define MS_ID "merkle_sync"
#define MS_BG_INTERVAL_MS 100
#define MS_PENDING_MAX 128
struct bucket_entry { uint8_t level; uint8_t prefix_bytes; uint64_t prefix; };
enum { SESS_SYNCING, SESS_SYNCED };
enum { MS_PEND_WAITING = 0, MS_PEND_RESOLVED = 1 };
struct ms_pending {
uint8_t level;
uint64_t prefix;
int8_t parent;
int sub_count;
uint8_t state;
};
struct ms_session {
struct ms_session* next;
char ns[64];
uint64_t peer;
uint8_t sess_state;
merkle_sync_done_cb done_cb;
void* cb_arg;
struct ms_pending pending[MS_PENDING_MAX];
int pending_count;
};
struct merkle_sync {
struct UTUN_INSTANCE* inst;
uint8_t svc_id;
const struct merkle_sync_data_ops* ops;
void* data_ctx;
struct ms_session* sessions;
uint8_t initialized;
void* bg_timer;
};
/* ── DB access ── */
static sqlite3* _db(struct UTUN_INSTANCE* inst) {
return inst ? inst->topo_sqlite_db : NULL;
}
/* ── Prefix arithmetic ── */
uint64_t merkle_sync_level_prefix(uint64_t key, uint8_t level) {
if (level == 0) return 0;
int shift = 64 - (int)level * 5; if (shift < 0) shift = 0;
return (key >> shift) << shift;
}
uint8_t merkle_sync_prefix_bytes(uint8_t level) { int bits = level * 5; return (uint8_t)((bits + 7) / 8); }
static void _prefix_write(uint8_t* out, uint64_t prefix, uint8_t pb) {
for (int i = 0; i < (int)pb; i++)
out[i] = (uint8_t)(prefix >> (8 * (7 - i)));
}
static uint64_t _prefix_read(const uint8_t* in, uint8_t pb) {
uint64_t v = 0;
for (uint8_t i = 0; i < pb && i < 8; i++) v = (v << 8) | in[i];
return v << (8 * (8 - (int)pb));
}
static int _popcount_u32(uint32_t v) { return __builtin_popcount(v); }
/* ── Ensure merkle_tree_hash table ── */
static void _ensure_table(struct merkle_sync* ms) {
sqlite3* db = _db(ms->inst); if (!db) return;
sqlite3_exec(db,
"CREATE TABLE IF NOT EXISTS merkle_tree_hash ("
" namespace TEXT NOT NULL,"
" level INTEGER NOT NULL CHECK(level BETWEEN 1 AND 5),"
" prefix64 INTEGER NOT NULL,"
" hash BLOB NOT NULL,"
" member_count INTEGER NOT NULL,"
" PRIMARY KEY (namespace, level, prefix64))",
NULL, NULL, NULL);
}
/* ── Tree operations ── */
static int _recompute_bucket(struct merkle_sync* ms, const char* ns,
uint8_t level, uint64_t prefix64) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix64);
sqlite3* db = _db(ms->inst); if (!db) return -1;
EVP_MD_CTX* ctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
int count = ms->ops->update_bucket_hash(ms->data_ctx, ns, level, prefix64, ctx);
if (count < 0) { DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → error", MS_ID); EVP_MD_CTX_free(ctx); return -1; }
if (count == 0) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → empty, deleted", MS_ID);
EVP_MD_CTX_free(ctx);
sqlite3_stmt* ds = NULL;
sqlite3_prepare_v2(db,
"DELETE FROM merkle_tree_hash WHERE namespace=? AND level=? AND prefix64=?",
-1, &ds, NULL);
if (ds) { sqlite3_bind_text(ds, 1, ns, -1, SQLITE_STATIC); sqlite3_bind_int(ds, 2, level);
sqlite3_bind_int64(ds, 3, (sqlite3_int64)prefix64); sqlite3_step(ds); sqlite3_finalize(ds); }
return 0;
}
uint8_t hash[MT_HASH_SIZE];
EVP_DigestFinal_ex(ctx, hash, NULL);
EVP_MD_CTX_free(ctx);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → saved, count=%d", MS_ID, count);
sqlite3_stmt* is = NULL;
sqlite3_prepare_v2(db,
"INSERT OR REPLACE INTO merkle_tree_hash(namespace, level, prefix64, hash, member_count)"
" VALUES(?,?,?,?,?)", -1, &is, NULL);
if (is) {
sqlite3_bind_text(is, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(is, 2, level);
sqlite3_bind_int64(is, 3, (sqlite3_int64)prefix64);
sqlite3_bind_blob(is, 4, hash, MT_HASH_SIZE, SQLITE_STATIC);
sqlite3_bind_int(is, 5, count);
sqlite3_step(is); sqlite3_finalize(is);
}
return 0;
}
void merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint64_t key) {
struct merkle_sync* ms = inst->msync;
if (!ms || !ms->initialized) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: recompute_path called but msync not initialized", MS_ID); return; }
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: ns=%s key=%016llx", MS_ID, ns, (unsigned long long)key);
for (uint8_t level = 1; level <= MT_MAX_LEVEL; level++)
_recompute_bucket(ms, ns, level, merkle_sync_level_prefix(key, level));
}
const uint8_t* merkle_sync_get_hash(struct UTUN_INSTANCE* inst, const char* ns,
uint8_t level, uint64_t prefix64) {
static uint8_t zero[MT_HASH_SIZE];
sqlite3* db = _db(inst); if (!db) { memset(zero, 0, MT_HASH_SIZE); return zero; }
sqlite3_stmt* stmt = NULL;
sqlite3_prepare_v2(db,
"SELECT hash FROM merkle_tree_hash WHERE namespace=? AND level=? AND prefix64=?",
-1, &stmt, NULL);
if (!stmt) { memset(zero, 0, MT_HASH_SIZE); return zero; }
sqlite3_bind_text(stmt, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, level);
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)prefix64);
memset(zero, 0, MT_HASH_SIZE);
if (sqlite3_step(stmt) == SQLITE_ROW) {
const void* h = sqlite3_column_blob(stmt, 0);
if (h) memcpy(zero, h, MT_HASH_SIZE);
}
sqlite3_finalize(stmt);
return zero;
}
static int _get_level_hashes(struct merkle_sync* ms, const char* ns,
uint8_t level, uint64_t prefix, uint8_t prefix_bytes,
uint32_t* bitmap, uint8_t hashes[MT_BUCKETS][MT_HASH_SIZE]) {
(void)prefix_bytes;
memset(hashes, 0, sizeof(uint8_t) * MT_BUCKETS * MT_HASH_SIZE);
*bitmap = 0;
if (level >= MT_MAX_LEVEL) return 0;
sqlite3* db = _db(ms->inst); if (!db) return -1;
int next_shift = 64 - ((int)level + 1) * 5; if (next_shift < 0) next_shift = 0;
sqlite3_stmt* stmt = NULL;
int query_level = (int)(level + 1);
const char* sql = (level == 0)
? "SELECT prefix64, hash FROM merkle_tree_hash"
" WHERE namespace=? AND level=? ORDER BY prefix64"
: "SELECT prefix64, hash FROM merkle_tree_hash"
" WHERE namespace=? AND level=? AND prefix64>=? AND prefix64<=?"
" ORDER BY prefix64";
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK || !stmt) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: _get_level_hashes SQL error ns=%s L%d — %s", MS_ID, ns, level, sqlite3_errmsg(db));
return -1;
}
sqlite3_bind_text(stmt, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, query_level);
if (level > 0) {
uint64_t range_end = prefix | (0x1FULL << (next_shift > 0 ? next_shift : 0));
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)prefix);
sqlite3_bind_int64(stmt, 4, (sqlite3_int64)range_end);
}
while (sqlite3_step(stmt) == SQLITE_ROW) {
sqlite3_int64 row_pfx = sqlite3_column_int64(stmt, 0);
const void* h = sqlite3_column_blob(stmt, 1);
if (!h || sqlite3_column_bytes(stmt, 1) < MT_HASH_SIZE) continue;
int bucket = (int)(((uint64_t)row_pfx >> next_shift) & 0x1F);
*bitmap |= (1u << bucket);
memcpy(hashes[bucket], h, MT_HASH_SIZE);
}
sqlite3_finalize(stmt);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: _get_level_hashes ns=%s L%d/P%016llx bitmap=%08x",
MS_ID, ns, level, (unsigned long long)prefix, *bitmap);
return 0;
}
/* ── Send helpers ── */
static struct ETCP_CONN* ms_find_conn_for_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (!inst->connections) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — inst->connections is NULL", MS_ID); return NULL; }
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (!e) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx NOT FOUND in connections", MS_ID, (unsigned long long)node_id); return NULL; }
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx found=%d initialized=%d links_up=%d", MS_ID, (unsigned long long)node_id, 1, ce->conn->initialized, ce->conn->links_up);
if (ce->conn->initialized && ce->conn->links_up) return ce->conn;
return NULL;
}
static int _send_msg(struct merkle_sync* ms, uint64_t peer,
const uint8_t* payload, size_t len) {
if (!ms->inst || len < 1) return -1;
uint8_t* buf = u_malloc(1 + len);
if (!buf) return -1;
buf[0] = ms->svc_id;
memcpy(buf + 1, payload, len);
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { u_free(buf); return -1; }
entry->dgram = buf; entry->len = 1 + len;
struct ETCP_CONN* conn = ms_find_conn_for_node(ms->inst, peer);
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: no conn for node %016llx", MS_ID, (unsigned long long)peer); u_free(buf); queue_entry_free(entry); return -1; }
int r = etcp_send(conn, entry);
if (r != 0) { u_free(buf); queue_entry_free(entry); }
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: send OK peer=%016llx len=%zu rc=%d", MS_ID, (unsigned long long)peer, len, r);
return r;
}
static int _send_broadcast_data(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* item_data, size_t item_len) {
uint8_t ch_len = (uint8_t)strlen(ns);
size_t sz = 1 + 1 + ch_len + 1 + 1 + 1 + 2 + item_len;
uint8_t* buf = u_malloc(sz); if (!buf) return -1;
uint8_t* p = buf;
*p++ = ch_len; memcpy(p, ns, ch_len); p += ch_len;
*p++ = 0x01; /* MSG_HASHES */
*p++ = 0; /* level=0 */
*p++ = 0; /* prefix_bytes=0 */
*p++ = 1; /* is_data=1 */
uint16_t dlen = (uint16_t)item_len; memcpy(p, &dlen, 2); p += 2;
memcpy(p, item_data, item_len); p += item_len;
int r = _send_msg(ms, peer, buf, (size_t)(p - buf));
u_free(buf);
return r;
}
static int _send_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
uint8_t level, uint64_t prefix, uint8_t prefix_bytes, int is_data) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: send_hashes peer=%016llx ns=%s L%d P%016llx is_data=%d", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, is_data);
uint8_t ch_len = (uint8_t)strlen(ns);
size_t max_sz = 1 + 1 + ch_len + 1 + 1 + prefix_bytes + 1 + 4 + MT_BUCKETS * MT_HASH_SIZE + 2 + 65536;
uint8_t* buf = u_malloc(max_sz);
if (!buf) return -1;
uint8_t* p = buf;
*p++ = ch_len; memcpy(p, ns, ch_len); p += ch_len;
*p++ = 0x01; /* MS_MSG_HASHES */
*p++ = level;
*p++ = prefix_bytes; _prefix_write(p, prefix, prefix_bytes); p += prefix_bytes;
*p++ = (uint8_t)(is_data ? 1 : 0);
if (is_data) {
size_t mlen = 65536; uint8_t* mbuf = u_malloc(mlen);
if (mbuf) {
if (ms->ops->get_items(ms->data_ctx, ns, level, prefix, prefix_bytes, mbuf, &mlen) == 0) {
uint16_t dlen = (uint16_t)mlen; memcpy(p, &dlen, 2); p += 2;
memcpy(p, mbuf, mlen); p += mlen;
} else { uint16_t zero = 0; memcpy(p, &zero, 2); p += 2; }
u_free(mbuf);
}
} else {
uint32_t bitmap; uint8_t hashes[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, level, prefix, prefix_bytes, &bitmap, hashes);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: send_hashes peer=%016llx ns=%s level=%d prefix=%016llx bitmap=%08x", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, bitmap);
memcpy(p, &bitmap, 4); p += 4;
for (int i = 0; i < MT_BUCKETS; i++)
if (bitmap & (1u << i)) { memcpy(p, hashes[i], MT_HASH_SIZE); p += MT_HASH_SIZE; }
}
int r = _send_msg(ms, peer, buf, (size_t)(p - buf));
u_free(buf);
return r;
}
static int _send_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
struct bucket_entry* buckets, int count) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: send_batch peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
uint8_t ch_len = (uint8_t)strlen(ns);
size_t max_sz = 1 + 1 + ch_len + 1 + (size_t)count * (1 + 1 + 8 + 1 + 4 + MT_BUCKETS * MT_HASH_SIZE + 65536);
uint8_t* buf = u_malloc(max_sz);
if (!buf) return -1;
uint8_t* p = buf;
*p++ = ch_len; memcpy(p, ns, ch_len); p += ch_len;
*p++ = 0x03; /* MS_MSG_BATCH */
*p++ = (uint8_t)count;
for (int i = 0; i < count; i++) {
*p++ = buckets[i].level;
*p++ = buckets[i].prefix_bytes;
_prefix_write(p, buckets[i].prefix, buckets[i].prefix_bytes); p += buckets[i].prefix_bytes;
EVP_MD_CTX* tctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(tctx, EVP_sha256(), NULL);
int bcount = ms->ops->update_bucket_hash(ms->data_ctx, ns,
buckets[i].level, buckets[i].prefix, tctx);
if (bcount < 0) bcount = 0;
EVP_MD_CTX_free(tctx);
int is_terminal = (buckets[i].level >= MT_MAX_LEVEL || bcount < 8);
*p++ = (uint8_t)(is_terminal ? 1 : 0);
if (is_terminal) {
size_t mlen = 65536; uint8_t* mbuf = u_malloc(mlen);
if (mbuf) {
if (ms->ops->get_items(ms->data_ctx, ns, buckets[i].level,
buckets[i].prefix, buckets[i].prefix_bytes, mbuf, &mlen) == 0) {
uint16_t dlen = (uint16_t)mlen; memcpy(p, &dlen, 2); p += 2;
memcpy(p, mbuf, mlen); p += mlen;
} else { uint16_t z = 0; memcpy(p, &z, 2); p += 2; }
u_free(mbuf);
}
} else {
uint32_t bm; uint8_t hs[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, buckets[i].level,
buckets[i].prefix, buckets[i].prefix_bytes, &bm, hs);
memcpy(p, &bm, 4); p += 4;
for (int j = 0; j < MT_BUCKETS; j++)
if (bm & (1u << j)) { memcpy(p, hs[j], MT_HASH_SIZE); p += MT_HASH_SIZE; }
}
}
int r = _send_msg(ms, peer, buf, (size_t)(p - buf));
u_free(buf);
return r;
}
/* ── Sessions ── */
static struct ms_session* _session_find(struct merkle_sync* ms, uint64_t peer, const char* ns) {
for (struct ms_session* s = ms->sessions; s; s = s->next)
if (s->peer == peer && strcmp(s->ns, ns) == 0) return s;
return NULL;
}
static int _find_pending(struct ms_session* s, uint8_t level, uint64_t prefix) {
for (int i = 0; i < s->pending_count; i++)
if (s->pending[i].level == level && s->pending[i].prefix == prefix)
return i;
return -1;
}
static void _session_done(struct ms_session* s, int result) {
if (s->sess_state == SESS_SYNCED) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: session ALREADY DONE peer=%016llx ns=%s", MS_ID, (unsigned long long)s->peer, s->ns); return; }
merkle_sync_done_cb cb = s->done_cb; void* arg = s->cb_arg;
s->done_cb = NULL; s->cb_arg = NULL;
if (result == MT_OK) { s->sess_state = SESS_SYNCED; DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: session SYNCED peer=%016llx ns=%s", MS_ID, (unsigned long long)s->peer, s->ns); }
else { s->sess_state = SESS_SYNCING; DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: session FAILED peer=%016llx ns=%s result=%d", MS_ID, (unsigned long long)s->peer, s->ns, result); }
if (cb) cb(s->peer, s->ns, result, arg);
}
/* ── Recv handlers ── */
static void _handle_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen, int parent_idx) {
if (plen < 3) return;
uint8_t level = pl[0]; uint8_t pb = pl[1];
uint64_t prefix = _prefix_read(pl + 2, pb);
uint8_t is_data = pl[2 + pb];
const uint8_t* payload = pl + 3 + pb;
size_t paylen = plen - 3 - pb;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: handle_hashes peer=%016llx ns=%s L%d/P%016llx is_data=%d", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, is_data);
struct ms_session* s = _session_find(ms, peer, ns);
if (is_data) {
if (paylen < 2) return;
uint16_t dlen; memcpy(&dlen, payload, 2);
const uint8_t* pd = payload + 2;
if (paylen - 2 < dlen) return;
ms->ops->apply_items(ms->data_ctx, ns, peer, pd, dlen);
if (s && s->sess_state == SESS_SYNCING) _session_done(s, MT_OK);
return;
}
if (!s) {
s = u_calloc(1, sizeof(*s)); if (!s) return;
snprintf(s->ns, sizeof(s->ns), "%s", ns); s->peer = peer;
s->next = ms->sessions; ms->sessions = s;
} else {
s->pending_count = 0;
}
s->sess_state = SESS_SYNCING;
if (paylen < 4) return;
uint32_t remote_bm; memcpy(&remote_bm, payload, 4);
const uint8_t* rh = payload + 4;
uint32_t local_bm; uint8_t lh[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, level, prefix, pb, &local_bm, lh);
uint32_t differs = remote_bm ^ local_bm;
for (int i = 0; i < MT_BUCKETS; i++) {
if (!(local_bm & (1u << i)) && !(remote_bm & (1u << i))) continue;
if ((local_bm & (1u << i)) && (remote_bm & (1u << i))) {
int rh_idx = 0;
for (int j = 0; j < i; j++) if (remote_bm & (1u << j)) rh_idx++;
if (memcmp(rh + (size_t)rh_idx * MT_HASH_SIZE, lh[i], MT_HASH_SIZE) != 0)
differs |= (1u << i);
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: handle_hashes peer=%016llx ns=%s level=%d remote_bm=%08x local_bm=%08x differs=%08x is_data=%d",
MS_ID, (unsigned long long)peer, ns, level, remote_bm, local_bm, differs, is_data);
if (differs == 0 && level == 0) {
_send_hashes(ms, peer, ns, 0, 0, 0, 1);
_session_done(s, MT_OK);
return;
}
if (differs == 0) return;
int next_shift = 64 - ((int)level + 1) * 5;
struct bucket_entry requests[MT_BUCKETS]; int rcount = 0;
for (int i = 0; i < MT_BUCKETS && rcount < 32; i++) {
if (!(differs & (1u << i))) continue;
uint8_t nl = (uint8_t)(level + 1); if (nl > MT_MAX_LEVEL) nl = MT_MAX_LEVEL;
requests[rcount].level = nl;
requests[rcount].prefix_bytes = merkle_sync_prefix_bytes(nl);
requests[rcount].prefix = prefix | ((uint64_t)i << (next_shift > 0 ? next_shift : 0));
rcount++;
}
if (rcount > 0) {
if (s) {
for (int i = 0; i < rcount && s->pending_count < MS_PENDING_MAX; i++) {
int idx = s->pending_count++;
s->pending[idx].level = requests[i].level;
s->pending[idx].prefix = requests[i].prefix;
s->pending[idx].parent = (int8_t)parent_idx;
s->pending[idx].sub_count = 0;
s->pending[idx].state = MS_PEND_WAITING;
}
if (parent_idx >= 0 && parent_idx < MS_PENDING_MAX)
s->pending[parent_idx].sub_count++;
}
uint8_t ch_len = (uint8_t)strlen(ns);
size_t rs = 1 + 1 + ch_len + 1;
for (int i = 0; i < rcount; i++) rs += 1 + 1 + requests[i].prefix_bytes + 1;
uint8_t* rbuf = u_malloc(rs);
if (rbuf) {
uint8_t* wr = rbuf;
*wr++ = ch_len;
memcpy(wr, ns, ch_len); wr += ch_len;
*wr++ = 0x02; /* MS_MSG_REQUEST */
*wr++ = (uint8_t)rcount;
for (int i = 0; i < rcount; i++) {
*wr++ = requests[i].level;
*wr++ = requests[i].prefix_bytes;
_prefix_write(wr, requests[i].prefix, requests[i].prefix_bytes);
wr += requests[i].prefix_bytes;
EVP_MD_CTX* tctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(tctx, EVP_sha256(), NULL);
int cnt = ms->ops->update_bucket_hash(ms->data_ctx, ns,
requests[i].level, requests[i].prefix, tctx);
if (cnt < 0) cnt = 0;
EVP_MD_CTX_free(tctx);
*wr++ = (uint8_t)(cnt < 8 || requests[i].level >= MT_MAX_LEVEL ? 1 : 0);
}
_send_msg(ms, peer, rbuf, (size_t)(wr - rbuf));
u_free(rbuf);
}
} else {
if (s) _session_done(s, MT_OK);
}
}
static void _handle_request(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen) {
if (plen < 1) return;
uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t off = 0;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: handle_request peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
struct bucket_entry buckets[32]; int bc = 0;
for (uint8_t i = 0; i < count && bc < 32; i++) {
if (off + 2 > plen - 1) break;
uint8_t lvl = bp[off++]; uint8_t pb_i = bp[off++];
if (off + pb_i > plen - 1) break;
uint64_t pr = _prefix_read(bp + off, pb_i); off += pb_i;
buckets[bc].level = lvl; buckets[bc].prefix_bytes = pb_i; buckets[bc].prefix = pr; bc++;
off++; /* skip is_terminal byte */
}
if (bc > 0) {
_send_batch(ms, peer, ns, buckets, bc);
}
}
static void _handle_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen) {
if (plen < 1) return;
uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t rem = plen - 1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: handle_batch peer=%016llx ns=%s count=%d len=%zu", MS_ID, (unsigned long long)peer, ns, count, plen - 1);
for (uint8_t i = 0; i < count && rem >= 3; i++) {
uint8_t lvl = bp[0]; uint8_t pb_i = bp[1]; rem -= 2; bp += 2;
if (rem < pb_i + 1) break;
uint64_t pr = _prefix_read(bp, pb_i); bp += pb_i; rem -= pb_i;
uint8_t is_data = *bp++; rem--;
if (is_data && rem >= 4) {
uint16_t dlen; memcpy(&dlen, bp, 2); bp += 2; rem -= 2;
if (rem < dlen) break;
ms->ops->apply_items(ms->data_ctx, ns, peer, bp, dlen);
bp += dlen; rem -= dlen;
struct ms_session* sb = _session_find(ms, peer, ns);
if (sb) { int pidx = _find_pending(sb, lvl, pr); if (pidx >= 0) sb->pending[pidx].state = MS_PEND_RESOLVED; }
} else if (!is_data && rem >= 4) {
struct ms_session* sc = _session_find(ms, peer, ns);
int pidx = sc ? _find_pending(sc, lvl, pr) : -1;
if (pidx >= 0) sc->pending[pidx].state = MS_PEND_RESOLVED;
uint8_t sub_pl[4096]; size_t sub_len = 0;
sub_pl[sub_len++] = lvl;
sub_pl[sub_len++] = merkle_sync_prefix_bytes(lvl);
_prefix_write(sub_pl + sub_len, pr, merkle_sync_prefix_bytes(lvl));
sub_len += merkle_sync_prefix_bytes(lvl);
sub_pl[sub_len++] = 0; /* is_data=0 */
uint32_t bm; memcpy(&bm, bp, 4); bp += 4; rem -= 4;
memcpy(sub_pl + sub_len, &bm, 4); sub_len += 4;
int nh = _popcount_u32(bm);
if (rem >= (size_t)nh * MT_HASH_SIZE) {
memcpy(sub_pl + sub_len, bp, (size_t)nh * MT_HASH_SIZE);
sub_len += (size_t)nh * MT_HASH_SIZE;
bp += (size_t)nh * MT_HASH_SIZE; rem -= (size_t)nh * MT_HASH_SIZE;
}
_handle_hashes(ms, peer, ns, sub_pl, sub_len, pidx);
}
}
{
struct ms_session* s = _session_find(ms, peer, ns);
if (!s) return;
int all_done = (s->pending_count > 0);
for (int i = 0; i < s->pending_count; i++) {
if (s->pending[i].state != MS_PEND_RESOLVED) { all_done = 0; break; }
}
if (all_done && s->sess_state == SESS_SYNCING) {
_send_hashes(ms, peer, ns, 0, 0, 0, 1);
_session_done(s, MT_OK);
}
}
}
static void _recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || entry->len < 4) {
if (entry) { if (entry->dgram) u_free(entry->dgram); queue_entry_free(entry); }
return;
}
struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL;
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized) { u_free(entry->dgram); queue_entry_free(entry); return; }
uint64_t peer = conn ? conn->peer_node_id : 0;
const uint8_t* d = entry->dgram;
size_t dlen = entry->len;
uint8_t ch_len = d[1];
if (dlen < (size_t)(2 + ch_len + 1)) { u_free(entry->dgram); queue_entry_free(entry); return; }
char ns[64]; memcpy(ns, d + 2, ch_len); ns[ch_len] = '\0';
uint8_t type = d[2 + ch_len];
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: recv type=%02x from=%016llx ns=%s len=%zu", MS_ID, type, (unsigned long long)peer, ns, dlen);
const uint8_t* pl = d + 3 + ch_len;
size_t plen = dlen - 3 - ch_len;
switch (type) {
case 0x01: _handle_hashes(ms, peer, ns, pl, plen, -1); break;
case 0x02: _handle_request(ms, peer, ns, pl, plen); break;
case 0x03: _handle_batch(ms, peer, ns, pl, plen); break;
case 0x04:
if (plen >= 9 && ms->ops->apply_update) {
uint64_t key; memcpy(&key, pl, 8);
uint8_t utype = pl[8];
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: recv ITEM_UPDATE key=%016llx type=%02x from=%016llx ns=%s",
MS_ID, (unsigned long long)key, utype, (unsigned long long)peer, ns);
ms->ops->apply_update(ms->data_ctx, ns, key, utype, pl + 9, plen - 9);
}
break;
}
u_free(entry->dgram); queue_entry_free(entry);
}
/* ── Push lightweight update to synced peers ── */
void merkle_sync_push_update(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t key, uint8_t type, const uint8_t* data, size_t len) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized || !ns || !data) return;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: push_update ns=%s key=%016llx type=%02x len=%zu",
MS_ID, ns, (unsigned long long)key, type, len);
uint8_t ch_len = (uint8_t)strlen(ns);
if (ch_len > 63) return;
size_t pkt = 1 + 1 + ch_len + 1 + 8 + 1 + len;
uint8_t* buf = u_malloc(pkt);
if (!buf) return;
uint8_t* p = buf;
*p++ = ms->svc_id;
*p++ = ch_len; memcpy(p, ns, ch_len); p += ch_len;
*p++ = 0x04; /* MSG_ITEM_UPDATE */
memcpy(p, &key, 8); p += 8;
*p++ = type;
memcpy(p, data, len);
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (s->sess_state != SESS_SYNCED || strcmp(s->ns, ns) != 0) continue;
struct ll_entry* entry = queue_entry_new(0);
if (!entry) continue;
uint8_t* dcopy = u_malloc(pkt);
if (!dcopy) { queue_entry_free(entry); continue; }
memcpy(dcopy, buf, pkt);
entry->dgram = dcopy; entry->len = pkt;
struct ETCP_CONN* conn = ms_find_conn_for_node(inst, s->peer);
if (!conn) { u_free(dcopy); queue_entry_free(entry); continue; }
int r = etcp_send(conn, entry);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: push_update sent type=%02x to=%016llx rc=%d",
MS_ID, type, (unsigned long long)s->peer, r);
if (r != 0) { u_free(dcopy); queue_entry_free(entry); }
}
u_free(buf);
}
/* ── Broadcast to synced peers ── */
void merkle_sync_broadcast(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t from_peer, const uint8_t* data, size_t len) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized || !ns || !data || len < 2) return;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: broadcast ns=%s from=%016llx len=%zu",
MS_ID, ns, (unsigned long long)from_peer, len);
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (s->sess_state != SESS_SYNCED || s->peer == from_peer || strcmp(s->ns, ns) != 0) continue;
_send_broadcast_data(ms, s->peer, ns, data, len);
}
}
/* ── Background consistency check ── */
int merkle_sync_bg_check(struct UTUN_INSTANCE* inst, const char* ns) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ms->initialized) return -1;
sqlite3* db = _db(inst); if (!db || !ns) return -1;
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(db,
"SELECT prefix64 FROM merkle_tree_hash WHERE namespace=? AND level=? LIMIT 1",
-1, &stmt, NULL) != SQLITE_OK) return -1;
sqlite3_bind_text(stmt, 1, ns, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 2, MT_MAX_LEVEL);
if (sqlite3_step(stmt) != SQLITE_ROW) { sqlite3_finalize(stmt); return -1; }
uint64_t pref = (uint64_t)sqlite3_column_int64(stmt, 0);
sqlite3_finalize(stmt);
uint8_t old_hash[MT_HASH_SIZE];
memcpy(old_hash, merkle_sync_get_hash(inst, ns, MT_MAX_LEVEL, pref), MT_HASH_SIZE);
_recompute_bucket(ms, ns, MT_MAX_LEVEL, pref);
const uint8_t* new_hash = merkle_sync_get_hash(inst, ns, MT_MAX_LEVEL, pref);
if (memcmp(old_hash, new_hash, MT_HASH_SIZE) != 0) {
for (uint8_t lv = (uint8_t)(MT_MAX_LEVEL - 1); lv >= 1; lv--)
_recompute_bucket(ms, ns, lv, merkle_sync_level_prefix(pref, lv));
return 1;
}
return 0;
}
static void _bg_timer_cb(void* arg) {
struct merkle_sync* ms = (struct merkle_sync*)arg;
if (!ms || !ms->initialized || !ms->inst) return;
sqlite3* db = _db(ms->inst); if (!db) return;
sqlite3_stmt* cs = NULL;
sqlite3_prepare_v2(db, "SELECT channel_id FROM channels", -1, &cs, NULL);
if (!cs) return;
while (sqlite3_step(cs) == SQLITE_ROW) {
const char* ch_id = (const char*)sqlite3_column_text(cs, 0);
if (!ch_id) continue;
int rc = merkle_sync_bg_check(ms->inst, ch_id);
if (rc < 0) continue;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: bg_check ns=%s → %s", MS_ID, ch_id, rc == 0 ? "consistent" : "RECALCULATED");
break;
}
sqlite3_finalize(cs);
ms->bg_timer = uasync_set_timeout(ms->inst->ua, (uint32_t)(MS_BG_INTERVAL_MS * 10),
ms, _bg_timer_cb, "ms_bg");
}
/* ── Public lifecycle ── */
int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id,
const struct merkle_sync_data_ops* ops, void* data_ctx) {
if (!inst || !ops) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: init svc=%02x", MS_ID, svc_id);
struct merkle_sync* ms = u_calloc(1, sizeof(*ms));
if (!ms) return -1;
ms->inst = inst; ms->svc_id = svc_id; ms->ops = ops; ms->data_ctx = data_ctx;
ms->initialized = 1;
inst->msync = ms;
_ensure_table(ms);
etcp_bind(inst, svc_id, _recv_cb);
ms->bg_timer = uasync_set_timeout(inst->ua, (uint32_t)(MS_BG_INTERVAL_MS * 10),
ms, _bg_timer_cb, "ms_bg");
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: initialized svc=%02x", MS_ID, svc_id);
return 0;
}
void merkle_sync_destroy(struct UTUN_INSTANCE* inst) {
if (!inst) return;
struct merkle_sync* ms = inst->msync;
if (!ms) return;
ms->initialized = 0; inst->msync = NULL;
etcp_unbind(inst, ms->svc_id);
if (ms->bg_timer) { uasync_cancel_timeout(inst->ua, ms->bg_timer); ms->bg_timer = NULL; }
struct ms_session* s = ms->sessions;
while (s) { struct ms_session* next = s->next;
u_free(s); s = next; }
u_free(ms);
}
/* ── Public async sync ── */
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ns, merkle_sync_done_cb done_cb, void* arg) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!inst || !ns || !ms || !ms->initialized) return -1;
_ensure_table(ms);
struct ms_session* s = _session_find(ms, peer, ns);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: start peer=%016llx ns=%s new=%d", MS_ID, (unsigned long long)peer, ns, s ? 0 : 1);
if (!s) {
s = u_calloc(1, sizeof(*s));
if (!s) return -1;
snprintf(s->ns, sizeof(s->ns), "%s", ns);
s->peer = peer;
s->next = ms->sessions;
ms->sessions = s;
} else {
s->pending_count = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: start OVERWRITE peer=%016llx ns=%s", MS_ID, (unsigned long long)peer, ns);
}
s->sess_state = SESS_SYNCING;
s->done_cb = done_cb; s->cb_arg = arg;
_send_hashes(ms, peer, ns, 0, 0, 0, 0);
return 0;
}
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns) {
struct merkle_sync* ms = inst ? inst->msync : NULL;
if (!ms || !ns) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: cancel peer=%016llx ns=%s", MS_ID, (unsigned long long)peer, ns);
struct ms_session** p = &ms->sessions;
while (*p) {
struct ms_session* s = *p;
if (s->peer == peer && strcmp(s->ns, ns) == 0) {
*p = s->next; u_free(s);
return;
}
p = &(*p)->next;
}
}
-286
View File
@@ -1,286 +0,0 @@
#ifndef MERKLE_SYNC_H
#define MERKLE_SYNC_H
#include <stdint.h>
#include <stddef.h>
#include <openssl/evp.h>
struct UTUN_INSTANCE;
#define MT_MAX_LEVEL 5
#define MT_BUCKETS 32
#define MT_HASH_SIZE 32
/*
* ── Архитектура ──
*
* merkle_sync — универсальный протокол синхронизации на Merkle-деревьях.
* Группирует элементы (items) в префиксное дерево: 5 уровней × 32 бакета,
* SHA256-хеши. Два пира обмениваются хешами уровней, находят различающиеся
* бакеты и передают только их содержимое.
*
* merkle_sync не знает, что такое "элемент" — потребитель (member_sync)
* предоставляет три коллбэка, описывающих модель данных:
*
* update_bucket_hash — перечислить элементы в диапазоне ключей,
* вычислить хеш каждого и подать в SHA256-контекст
* get_items — сериализовать элементы бакета в wire-формат
* apply_items — десериализовать и сохранить полученные элементы
*
* Потребитель (chat_sync/chat_core) работает только с member_sync
* и не видит merkle_sync напрямую:
*
* // ── разово ──
* member_sync_init(inst);
*
* // ── запустил синхронизацию — забыл ──
* member_sync_start(inst, peer, ch_id, on_done, my_ctx);
*
* // ── получил результат ──
* static void on_done(uint64_t peer, const char* ch_id, int result, void* arg) {
* if (result == MT_OK) printf("sync ok\n");
* if (result == MT_ERR_TIMEOUT) printf("timeout\n");
* }
*
* // ── изменил данные — дерево само пересчиталось ──
* member_sync_put(inst, ch_id, node_id, x25519, ed25519, join_sig, addrs, ac);
* member_sync_set_online(inst, node_id, 1);
*
* // ── отменил — коллбэк не вызовется ──
* member_sync_cancel(inst, peer, ch_id);
*
* // ── разово ──
* member_sync_destroy(inst);
*
* ── Формат дерева ──
*
* 5-уровневое префиксное дерево над 64-битными ключами (node_id).
* Каждый уровень берёт 5 старших бит ключа: уровень 1 — биты 59-63,
* уровень 5 — биты 39-63. Каждый узел дерева разбивается на 32 бакета
* (по 5 бит = 32 комбинации).
*
* level 1 [0..31] корень: 32 бакета по 5 бит
* level 2 [0..31]...[0..31] каждый — ещё 32 бакета
* ...
* level 5 [0..31].........[0..31] листья: 32^5 = 33M бакетов макс
*
* Хеш бакета = SHA256(хеш_элемента_1 || ... || хеш_элемента_N).
* Бакет считается терминальным (leaf) на уровне 5 или если в нём < 8 элементов.
*
* ── Wire-протокол (сервис ETCP, id задаётся при init) ──
*
* Каждое сообщение: [svc_id:1][ns_len:1][ns:var][type:1][payload:var]
*
* MSG_HASHES (0x01): level,prefix,is_data, [bitmap+hashes | member_data]
* MSG_REQUEST (0x02): count, [level,prefix,is_terminal]*
* MSG_BATCH (0x03): count, [level,prefix,is_terminal,[member_data|bitmap+hashes]]*
*
* Алгоритм:
* A → MSG_HASHES(level=1, bitmap+hashes всех 32 бакетов уровня 2) → B
* B сравнивает со своим деревом, находит различающиеся бакеты
* B → MSG_REQUEST(level=2, prefix=X, is_terminal) → A
* A → MSG_BATCH(данные бакета) → B
* B сохраняет, пересчитывает хеши
* Рекурсивно для подбакетов, пока хеши не совпадут.
*
* ── Сессии ──
*
* Протокол работает поверх надёжного транспорта (ETCP).
* Таймаутов и ретраев нет — при ошибке сессия завершается.
*/
/* ── Data model callbacks ── */
struct merkle_sync_data_ops {
/*
* Подать хеши всех элементов в префиксном диапазоне в sha_ctx.
*
* ctx — data_ctx, переданный в merkle_sync_init
* ns — namespace (например channel_id)
* level — уровень дерева (1..5)
* prefix64 — префикс ключа (старшие level*5 бит, остальные нули)
* sha_ctx — уже инициализирован (EVP_DigestInit_ex), потребитель
* делает только EVP_DigestUpdate(sha_ctx, item_hash, 32)
* для каждого элемента
*
* Возвращает количество элементов (0 = бакет пуст, -1 = ошибка).
*
* Пример реализации для мемберов:
* SELECT ... FROM peers_<ns> JOIN nodes
* WHERE (node_id & mask) == prefix64 ORDER BY node_id
* для каждой строки: _compute_member_hash() → EVP_DigestUpdate()
*/
int (*update_bucket_hash)(void* ctx, const char* ns, uint8_t level,
uint64_t prefix64, EVP_MD_CTX* sha_ctx);
/*
* Сериализовать элементы бакета в wire-формат.
*
* buf — буфер для записи (выделяет merkle_sync, мин. 64K)
* *len — [in] размер буфера, [out] записанный размер
*
* Wire-формат для мемберов:
* [count:2][node_id:8][x25519:32][ed25519:32][flags:1]([join_sig:64][join_ts:8])[update_sig:64][update_ts:8][userinfo_len:1][userinfo:var][addr_cnt:1][addrs:var]...
*
* Возвращает 0 при успехе, <0 при ошибке, -2 если буфер мал.
*/
int (*get_items)(void* ctx, const char* ns, uint8_t level,
uint64_t prefix, uint8_t pbytes, uint8_t* buf, size_t* len);
/*
* Десериализовать и сохранить элементы из wire-формата.
*
* data, len — данные в том же формате, что выдаёт get_items
*
* Вызывается при приёме MSG_HASHES(is_data=1) или MSG_BATCH(is_terminal=1).
* Должна сохранить элементы в БД и вызвать merkle_sync_recompute_path
* для каждого изменённого ключа (через member_sync_put).
*
* Возвращает 0 при успехе, <0 при ошибке.
*/
int (*apply_items)(void* ctx, const char* ns, uint64_t from_peer,
const uint8_t* data, size_t len);
/*
* Применить лёгкое обновление (не влияющее на Merkle-хеш).
* Вызывается при приёме MSG_ITEM_UPDATE(0x04) — online-статус и т.п.
*
* ns — namespace (channel_id)
* key — node_id изменившегося узла
* type — тип обновления (0x01 = online-статус)
* data,len — данные обновления (type=0x01: [online:1])
*
* Возвращает 0 при успехе, <0 при ошибке.
*/
int (*apply_update)(void* ctx, const char* ns, uint64_t key,
uint8_t type, const uint8_t* data, size_t len);
};
/*
* Коллбэк завершения синхронизации.
*
* peer — node_id пира
* ns — namespace (channel_id)
* result — MT_OK (0) = успех, MT_ERR_TIMEOUT (-1) = исчерпаны ретраи
* arg — пользовательский контекст
*
* При отмене через merkle_sync_cancel() коллбэк НЕ вызывается.
*/
#define MT_OK 0
#define MT_ERR_TIMEOUT -1
typedef void (*merkle_sync_done_cb)(uint64_t peer, const char* ns, int result, void* arg);
/* ── Lifecycle ── */
/*
* Инициализировать модуль. Создаёт таблицу merkle_tree_hash, биндит
* ETCP-сервис svc_id (напр. 0x31), запускает фоновую проверку (bg_timer).
* Вызывается один раз, обычно из member_sync_init().
*
* inst — экземпляр uTun (нужен для etcp_send, uasync, sqlite3)
* svc_id — ID сервиса на ETCP-роутере
* ops — коллбэки модели данных (update_bucket_hash, get_items, apply_items)
* data_ctx — прозрачный контекст, передаваемый в коллбэки первым аргументом
*
* Возвращает 0 при успехе, -1 при ошибке.
*/
int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id,
const struct merkle_sync_data_ops* ops, void* data_ctx);
/*
* Завершить модуль. Отвязывает ETCP-сервис, останавливает bg_timer,
* удаляет все активные сессии (коллбэки НЕ вызываются).
*/
void merkle_sync_destroy(struct UTUN_INSTANCE* inst);
/* ── Async sync ── */
/*
* Запустить синхронизацию namespace ns с пиром peer.
* Отправляет MSG_HASHES(level=1) и запускает таймаут 10s.
*
* Если сессия для (peer, ns) уже существует — перезапускает её
* с новым коллбэком (старый коллбэк теряется без вызова).
*
* Возвращает 0 при успехе, -1 при ошибке.
*/
int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
const char* ns, merkle_sync_done_cb done_cb, void* arg);
/*
* Отменить синхронизацию. Удаляет сессию, done_cb НЕ вызывается.
* Безопасно вызывать, если сессия не существует.
*/
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns);
/* ── Recompute tree path after data change ── */
/*
* Пересчитать Merkle-дерево для ключа key — все уровни от 1 до 5.
* Вызывается после изменения данных: merkle_sync сам не знает,
* когда данные изменились — потребитель должен вызвать явно.
* member_sync делает это внутри put/del.
*/
void merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint64_t key);
/*
* Отправить лёгкое обновление всем synced-пирам в namespace ns.
* Не влияет на Merkle-дерево — используется для online-статуса и т.п.
*
* ns — namespace (channel_id)
* key — node_id изменившегося узла
* type — тип обновления (0x01 = online-статус)
* data,len — данные обновления
*/
void merkle_sync_push_update(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t key, uint8_t type, const uint8_t* data, size_t len);
/*
* Переслать данные всем SYNCED-сессиям (кроме from_peer) в namespace ns.
* Используется для relay: когда apply_items обнаружил реальные изменения,
* потребитель вызывает эту функцию чтобы разослать дельту остальным пирам.
*
* data, len — wire-формат [count:2][key:8][val:4]... как от get_items.
*/
void merkle_sync_broadcast(struct UTUN_INSTANCE* inst, const char* ns,
uint64_t from_peer, const uint8_t* data, size_t len);
/* ── Background consistency check ── */
/*
* Проверить консистентность одного случайного level-5 бакета в ns.
* Пересчитывает хеш из данных, сравнивает с сохранённым в merkle_tree_hash.
* При расхождении — пересчитывает весь путь вверх до корня.
* Возвращает 1 если были изменения, 0 если совпало, -1 ошибка.
*/
int merkle_sync_bg_check(struct UTUN_INSTANCE* inst, const char* ns);
/* ── Test helper ── */
/*
* Получить сохранённый хеш бакета из merkle_tree_hash.
* Возвращает указатель на статический буфер (32 байта), перезаписывается
* при следующем вызове. Если бакет не найден — возвращает нулевой хеш.
*/
const uint8_t* merkle_sync_get_hash(struct UTUN_INSTANCE* inst,
const char* ns, uint8_t level, uint64_t prefix64);
/* ── Prefix arithmetic (pure, exported for convenience) ── */
/*
* Выделить level*5 старших бит из 64-битного ключа.
* Уровень 1: биты 59-63 (сдвиг 59)
* Уровень 5: биты 39-63 (сдвиг 39)
* Пример: merkle_sync_level_prefix(0x1234567890ABCDEF, 1) = 0x1000000000000000
*/
uint64_t merkle_sync_level_prefix(uint64_t key, uint8_t level);
/*
* Количество байт, необходимое для хранения префикса уровня level.
* Уровень 1: 1 байт (5 бит), уровень 4: 3 байта (20 бит), уровень 5: 4 байта (25 бит).
*/
uint8_t merkle_sync_prefix_bytes(uint8_t level);
#endif /* MERKLE_SYNC_H */
+11 -5
View File
@@ -22,10 +22,11 @@ extern "C" {
#include "../lib/memory_pool.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "chat_sync.h"
#include "chat_core.h"
#include "db_sync.h"
#include "gui_bridge.h"
#include "chat/chat_sync.h"
#include "chat/chat_core.h"
#include "chat/chat_event.h"
#include "db_sync.h"
#include "gui_bridge.h"
#include "topo_node_sqlite.h"
#include "topo_node.h"
#include "topo_group.h"
@@ -342,9 +343,14 @@ void UtunNode::runLoop() {
etcp_router_bind(m_instance, ETCP_RT_ID_CHAT, recvCallback);
/* Bridge chat events to GUI via gui_bridge */
chat_event_set_handler([](int type, const uint8_t* data, int len) {
gui_bridge_post(type, data, len);
});
/* Initialize chat_core (DB) and chat_sync (channel/message P2P sync) */
chat_core_init(m_instance, QString(m_dbPath + "/chats.db").toUtf8().constData());
chat_sync_init(m_instance, nullptr);
chat_sync_init(m_instance);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "chat_core + chat_sync initialized");
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "utun_node: entering poll loop");