jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
910 Commits
15 Branches
2 Tags
165 MiB
 
 
 
 
 
 
Tree: de7841f33a
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'de7841f33a'
${ noResults }
utun2/tools/chatgui/transport/chat_core.c

724 lines
26 KiB
Raw Blame History

/*
* chat_core.c — центральный API чата в потоке uasync
*
* Все DB-операции — через своё sqlite3-соединение (тот же файл БД, WAL).
* Сетевые вызовы — напрямую в ETCP.
*/
#include "chat_core.h"
#include "chat_sync.h"
#include "gui_bridge.h"
#include "topo_node_sqlite.h"
#include "../../../src/utun_instance.h"
#include "../../../src/etcp_router.h"
#include "../../../src/etcp_api.h"
#include "../../../src/topo_group.h"
#include "../../../src/topo_node.h"
#include "../../../src/conn_mgr.h"
#include "../../../src/secure_channel.h"
#include "../../../lib/u_async.h"
#include "../../../lib/ll_queue.h"
#include "../../../lib/mem.h"
#include "../../../lib/debug_config.h"
#include <sqlite3.h>
#include <openssl/sha.h>
#include <string.h>
#include <stdio.h>
#define CC_ID "chat_core"
/* ─── глобальное состояние ─── */
static struct chat_core_ctx {
struct UTUN_INSTANCE* inst;
sqlite3* db;
uint64_t my_node_id;
/* курсоры (макс 16) */
sqlite3_stmt* cursors[16];
uint32_t next_cursor_id;
uint8_t initialized;
} g_cc;
/* ─── утилиты ─── */
static 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';
}
static 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);
}
static 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);
}
static void compute_datahash(const uint8_t* data, size_t len, uint64_t* dh) {
uint8_t hash[32]; SHA256(data, len, hash);
memcpy(dh, hash, 8);
}
static void compute_chain_hash(const uint8_t* prev_chain, int64_t ts,
uint64_t dh, uint8_t* out) {
uint8_t buf[32 + 8 + 8];
memcpy(buf, prev_chain, 32);
memcpy(buf + 32, &ts, 8);
memcpy(buf + 40, &dh, 8);
SHA256(buf, 48, out);
}
static int get_last_chain_hash(const char* ch_id, uint8_t* out) {
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"SELECT chain_hash FROM \"%s\" ORDER BY timestamp, datahash DESC LIMIT 1", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) {
memset(out, 0, 32); return -1;
}
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(out, blob, 32);
else memset(out, 0, 32);
} else {
memset(out, 0, 32);
}
sqlite3_finalize(stmt);
return 0;
}
static 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_DEBUG, "%s: sql error: %s", CC_ID, err);
sqlite3_free(err);
}
return rc;
}
/* ─── жизненный цикл ─── */
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));
int rc = sqlite3_open(db_path, &g_cc.db);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%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;
topo_node_sqlite_init(g_cc.db);
topo_groups_set_sqlite_db(inst->topo_groups, g_cc.db);
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"
");"
);
g_cc.initialized = 1;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: initialized, db=%s node_id=0x%016llx",
CC_ID, db_path, (unsigned long long)g_cc.my_node_id);
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 < 16; i++)
if (g_cc.cursors[i]) sqlite3_finalize(g_cc.cursors[i]);
if (g_cc.db) { sqlite3_close(g_cc.db); g_cc.db = NULL; }
g_cc.inst = NULL;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: destroyed", CC_ID);
}
void chat_core_set_my_node_id(uint64_t node_id) {
g_cc.my_node_id = node_id;
}
/* ─── отправка сообщения (GUI → uasync) ─── */
void chat_core_submit_message(struct chat_msg_submit* req) {
if (!g_cc.initialized) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: submit_message called before init", CC_ID);
return;
}
if (!req) return;
const char* ch_id = req->channel_id;
const uint8_t* data = req->data;
uint32_t data_len = req->data_len;
uint64_t ts = req->timestamp;
uint64_t dh;
compute_datahash(data, data_len, &dh);
uint8_t prev_ch[32]; get_last_chain_hash(ch_id, prev_ch);
uint8_t chain_hash[32];
compute_chain_hash(prev_ch, (int64_t)ts, dh, chain_hash);
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"INSERT OR IGNORE INTO \"%s\""
" (node_id, content_type, data, timestamp, datahash, chain_hash,"
" signature, is_outgoing, is_read)"
" VALUES(?,?,?,?,?,?,?,1,1)", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: prepare insert failed: %s",
CC_ID, sqlite3_errmsg(g_cc.db));
return;
}
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)g_cc.my_node_id);
sqlite3_bind_text(stmt, 2, req->content_type, -1, SQLITE_STATIC);
sqlite3_bind_blob(stmt, 3, data, (int)data_len, SQLITE_STATIC);
sqlite3_bind_int64(stmt, 4, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 5, (sqlite3_int64)dh);
sqlite3_bind_blob(stmt, 6, chain_hash, 32, SQLITE_STATIC);
{
static const char zero64[64] = {0};
sqlite3_bind_blob(stmt, 7, zero64, 64, SQLITE_STATIC);
}
int rc = sqlite3_step(stmt);
sqlite3_finalize(stmt);
if (rc == SQLITE_DONE) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: msg inserted ch=%s ts=%llu dh=0x%016llx",
CC_ID, ch_id, (unsigned long long)ts, (unsigned long long)dh);
/* push пирам */
chat_sync_push(g_cc.inst, ch_id, g_cc.my_node_id,
req->content_type, data, data_len, ts, dh);
} else if (rc == SQLITE_CONSTRAINT) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: msg duplicate ch=%s ts=%llu",
CC_ID, ch_id, (unsigned long long)ts);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: insert msg failed: %s",
CC_ID, sqlite3_errmsg(g_cc.db));
}
}
void chat_core_submit_trampoline(void* arg) {
chat_core_submit_message((struct chat_msg_submit*)arg);
u_free(arg);
}
void chat_core_push_message(struct chat_msg_submit* req) {
if (!g_cc.initialized) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: push_message called before init", CC_ID);
return;
}
if (!req) return;
uint64_t dh;
compute_datahash(req->data, req->data_len, &dh);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: pushing msg ch=%s ts=%llu dh=0x%016llx",
CC_ID, req->channel_id, (unsigned long long)req->timestamp, (unsigned long long)dh);
chat_sync_push(g_cc.inst, req->channel_id, g_cc.my_node_id,
req->content_type, req->data, req->data_len,
req->timestamp, dh);
}
void chat_core_push_trampoline(void* arg) {
chat_core_push_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, datahash 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_insert_record(const char* ch_id, const uint8_t* rec, size_t len) {
if (!g_cc.initialized || !rec || len < 29) return -1;
const uint8_t* p = rec;
size_t rem = len;
int64_t ts; uint64_t dh, nid;
memcpy(&ts, p, 8); p += 8; rem -= 8;
memcpy(&dh, p, 8); p += 8; rem -= 8;
memcpy(&nid, p, 8); p += 8; rem -= 8;
if (rem < 1) return -1;
uint8_t ct_len = *p; p++; rem--;
if (rem < ct_len) return -1;
char ct_buf[64]; memcpy(ct_buf, p, ct_len); ct_buf[ct_len] = '\0';
p += ct_len; rem -= ct_len;
if (rem < 4) return -1;
uint32_t dlen; memcpy(&dlen, p, 4); p += 4; rem -= 4;
if (rem < dlen) return -1;
const uint8_t* data = p; p += dlen; rem -= dlen;
if (rem < 32) return -1;
uint8_t peer_ch[32]; memcpy(peer_ch, p, 32);
/* verify chain_hash */
uint8_t prev[32]; get_last_chain_hash(ch_id, prev);
uint8_t expected[32];
compute_chain_hash(prev, ts, dh, expected);
if (memcmp(expected, peer_ch, 32) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: chain_hash mismatch ch=%s ts=%lld dh=0x%016llx",
CC_ID, ch_id, (long long)ts, (unsigned long long)dh);
return -1;
}
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"INSERT OR IGNORE INTO \"%s\""
" (node_id, content_type, data, timestamp, datahash, chain_hash,"
" signature, is_outgoing, is_read)"
" VALUES(?,?,?,?,?,?,?,0,1)", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)nid);
sqlite3_bind_text(stmt, 2, ct_buf, ct_len, SQLITE_STATIC);
sqlite3_bind_blob(stmt, 3, data, (int)dlen, SQLITE_STATIC);
sqlite3_bind_int64(stmt, 4, ts);
sqlite3_bind_int64(stmt, 5, (sqlite3_int64)dh);
sqlite3_bind_blob(stmt, 6, peer_ch, 32, SQLITE_STATIC);
{
static const char zero64[64] = {0};
sqlite3_bind_blob(stmt, 7, zero64, 64, SQLITE_STATIC);
}
int rc = sqlite3_step(stmt);
sqlite3_finalize(stmt);
if (rc == SQLITE_DONE) return 0;
if (rc == SQLITE_CONSTRAINT) return 1;
return -1;
}
uint32_t chat_core_cursor_open(const char* ch_id) {
if (!g_cc.initialized) return 0;
for (int i = 0; i < 16; i++) {
if (g_cc.cursors[i]) continue;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"SELECT timestamp, datahash, node_id, content_type, data"
" FROM \"%s\" ORDER BY timestamp, datahash ASC", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
uint32_t id = ++g_cc.next_cursor_id;
g_cc.cursors[i] = stmt;
return id;
}
return 0;
}
int chat_core_cursor_next(uint32_t cursor_id, uint8_t* buf, size_t buf_size,
size_t* out_len) {
if (!g_cc.initialized || !buf || !out_len) return -1;
for (int i = 0; i < 16; i++) {
if (!g_cc.cursors[i]) continue;
/* проверяем cursor_id = i+1 */
if ((uint32_t)(i + 1) != cursor_id) continue;
sqlite3_stmt* stmt = g_cc.cursors[i];
if (sqlite3_step(stmt) != SQLITE_ROW) { *out_len = 0; return 0; }
int64_t ts = sqlite3_column_int64(stmt, 0);
uint64_t dh = (uint64_t)sqlite3_column_int64(stmt, 1);
uint64_t nid= (uint64_t)sqlite3_column_int64(stmt, 2);
const char* ct = (const char*)sqlite3_column_text(stmt, 3);
int ct_len = sqlite3_column_bytes(stmt, 3);
const uint8_t* data = (const uint8_t*)sqlite3_column_blob(stmt, 4);
int dlen = sqlite3_column_bytes(stmt, 4);
size_t need = 8 + 8 + 8 + 1 + (size_t)ct_len + 4 + (size_t)dlen;
if (need > buf_size) return -1;
uint8_t* out = buf;
memcpy(out, &ts, 8); out += 8;
memcpy(out, &dh, 8); out += 8;
memcpy(out, &nid, 8); out += 8;
*out++ = (uint8_t)ct_len;
if (ct_len) { memcpy(out, ct, (size_t)ct_len); out += ct_len; }
uint32_t dl32 = (uint32_t)dlen;
memcpy(out, &dl32, 4); out += 4;
if (dlen) { memcpy(out, data, (size_t)dlen); out += dlen; }
*out_len = (size_t)(out - buf);
return 0;
}
return -1;
}
void chat_core_cursor_close(uint32_t cursor_id) {
if (!g_cc.initialized) return;
for (int i = 0; i < 16; i++) {
if (g_cc.cursors[i] && (uint32_t)(i + 1) == cursor_id) {
sqlite3_finalize(g_cc.cursors[i]);
g_cc.cursors[i] = NULL;
return;
}
}
}
void chat_core_mark_sent(const char* ch_id, uint64_t ts, uint64_t dh,
uint64_t node_id) {
if (!g_cc.initialized) return;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"UPDATE \"%s\" SET sync_flags=sync_flags|1"
" WHERE timestamp=? AND datahash=? AND node_id=? AND is_outgoing=1", tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) return;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)ts);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)dh);
sqlite3_bind_int64(stmt, 3, (sqlite3_int64)node_id);
sqlite3_step(stmt); sqlite3_finalize(stmt);
}
void chat_core_ttl_delete(const char* ch_id, uint64_t node_id,
uint64_t cutoff_us) {
if (!g_cc.initialized) return;
char tbl[80]; msg_table_name(ch_id, tbl, sizeof(tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"DELETE FROM \"%s\" WHERE node_id=? AND is_outgoing=1 AND (sync_flags&1)=0 AND timestamp<?",
tbl);
sqlite3_stmt* stmt = NULL;
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) return;
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)node_id);
sqlite3_bind_int64(stmt, 2, (sqlite3_int64)cutoff_us);
sqlite3_step(stmt); sqlite3_finalize(stmt);
}
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; /* placeholder for count */
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;
}
/* ─── подключение к пиру из invite-ссылки ─── */
static void connect_result_cb(int result, uint64_t node_id, void* arg) {
uint8_t data[12];
memcpy(data, &node_id, 8);
memcpy(data + 8, &result, 4);
gui_bridge_post(GUI_EVT_CONNECT_RESULT, data, 12);
}
void chat_core_connect_from_invite(struct chat_invite* inv) {
if (!g_cc.initialized || !g_cc.inst || !inv) return;
struct TOPO_GROUP* group = topo_groups_get_default(g_cc.inst->topo_groups);
if (!group || !g_cc.inst->conn_mgr) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: bgp/conn_mgr not available", CC_ID);
uint8_t err[12]; int r = -7;
memcpy(err, &inv->node_id, 8); memcpy(err + 8, &r, 4);
gui_bridge_post(GUI_EVT_CONNECT_RESULT, err, 12);
return;
}
uint64_t node_id = inv->node_id;
if (topo_node_find_by_id(group, node_id)) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: node 0x%016llx already in BGP, connecting",
CC_ID, (unsigned long long)node_id);
conn_mgr_connect_node(g_cc.inst->conn_mgr, node_id, 30000,
connect_result_cb, NULL);
return;
}
struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_NODEQ));
if (!qe) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: failed to alloc TOPO_NODEQ", CC_ID);
uint8_t err[12]; int r = -7;
memcpy(err, &node_id, 8); memcpy(err + 8, &r, 4);
gui_bridge_post(GUI_EVT_CONNECT_RESULT, err, 12);
return;
}
struct TOPO_NODEQ* nq = (struct TOPO_NODEQ*)qe;
memset((uint8_t*)nq + sizeof(struct ll_entry), 0, sizeof(*nq) - sizeof(struct ll_entry));
struct TOPO_NODE* ni = u_calloc(1, sizeof(struct TOPO_NODE));
if (!ni) { queue_entry_free(qe); return; }
ni->node_id = node_id;
ni->group_id = group->group_id;
ni->ver = 0;
memcpy(ni->public_key, inv->pubkey, SC_PUBKEY_SIZE);
memset(ni->ed25519_public_key, 0, SC_PUBKEY_SIZE);
ni->node_name = u_strdup("");
struct TOPO_ADDR4* addrs_head = NULL;
const uint8_t* src = inv->addrs_data;
for (int i = 0; i < inv->addr_count; i++) {
uint8_t family = *src++;
if (family == 4) {
struct TOPO_ADDR4* a4 = memory_pool_alloc(group->instance->topo_groups->v4_addr_pool);
if (!a4) continue;
memcpy(a4->addr, src, 4); src += 4;
a4->port = ((uint16_t)src[0] << 8) | src[1]; src += 2;
a4->type = TOPO_ADDR_REAL; a4->socket_id = 0; a4->protocol = TOPO_PROTO_UDP;
a4->next = addrs_head; addrs_head = a4;
} else {
src += 18; /* skip v6 */
}
}
ni->v4_addrs = addrs_head;
nq->node = ni; topo_node_ref(ni);
nq->hash_node_id = node_id; nq->alien = 0;
nq->dirty = 0; nq->last_ver = 0;
nq->conn_mgr_type = CONN_TYPE_NONE; nq->conn_mgr_intermediariy_count = 0;
memset(&nq->connectivity, 0, sizeof(nq->connectivity));
nq->best_socket = NULL;
queue_data_put_with_index(group->nodes, &nq->ll);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: created NODEINFO for 0x%016llx, %d addrs, connecting",
CC_ID, (unsigned long long)node_id, inv->addr_count);
conn_mgr_connect_node(g_cc.inst->conn_mgr, node_id, 30000,
connect_result_cb, NULL);
}
/* ─── создание канала ─── */
void chat_core_create_channel(struct chat_channel_create* req) {
if (!g_cc.initialized) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: create_channel NOT INITIALIZED ch=%s",
CC_ID, req ? req->channel_id : "(null)");
return;
}
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "%s: create_channel req=NULL", CC_ID); return; }
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: create_channel BEGIN ch=%s name=%s",
CC_ID, req->channel_id, req->name);
char tbl_msg[80]; msg_table_name(req->channel_id, tbl_msg, sizeof(tbl_msg));
/* создаём таблицу сообщений */
char sql[512]; snprintf(sql, sizeof(sql),
"CREATE TABLE IF NOT EXISTS \"%s\" ("
" id INTEGER PRIMARY KEY AUTOINCREMENT,"
" node_id INTEGER NOT NULL,"
" content_type TEXT NOT NULL,"
" data BLOB NOT NULL,"
" timestamp INTEGER NOT NULL,"
" datahash INTEGER NOT NULL,"
" chain_hash BLOB NOT NULL,"
" signature BLOB,"
" is_outgoing INTEGER DEFAULT 0,"
" is_read INTEGER DEFAULT 0,"
" sync_flags INTEGER DEFAULT 0,"
" UNIQUE(timestamp, datahash))", tbl_msg);
db_exec(sql);
snprintf(sql, sizeof(sql),
"CREATE INDEX IF NOT EXISTS \"idx_%s_ts_dh\" ON \"%s\"(timestamp, datahash)",
tbl_msg, tbl_msg);
db_exec(sql);
/* записываем канал в БД */
int rc = topo_node_sqlite_channel_put(g_cc.db,
req->channel_id, req->name, req->is_dm, 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_DEBUG, "%s: channel_put FAILED ch=%s rc=%d",
CC_ID, req->channel_id, rc);
} else {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%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);
/* уведомляем GUI */
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);
gui_bridge_post(GUI_EVT_CHANNEL_UPDATED, data, 1 + ch_id_len);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: gui_bridge_post sent GUI_EVT_CHANNEL_UPDATED ch=%s",
CC_ID, req->channel_id);
}
}
void chat_core_create_channel_trampoline(void* arg) {
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "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);
}