Browse Source
chat_core.c (236 lines): init/destroy, getters, g_cc, helpers, si-registry chat_msg.c (223 lines): submit_message, count, chain_hash, list_*, load_nodeinfo chat_channel.c (197 lines): ensure_ready, create_channel, connect_channel, sign_join helper chat_profile.c (189 lines): update_my_name, sync_my_addresses, save_ui_state chat_status.c (118 lines): collect_status (NTP + ETCP connections) chat_core_priv.h: shared g_cc extern + helpers chat_conn_mgr.c/h: dead code (never compiled, never called) CMakeLists.txt: add 4 new .c files to buildtopo_upd
Evgeny
2 months ago
10 changed files with 833 additions and 1435 deletions
@ -0,0 +1,197 @@
|
||||
/*
|
||||
* 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> |
||||
|
||||
/* ─── 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); |
||||
|
||||
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); |
||||
} |
||||
|
||||
/* ─── управление подключением к каналу (выбор в 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); |
||||
} |
||||
@ -1,656 +0,0 @@
|
||||
/*
|
||||
* chat_conn_mgr.c — менеджер ETCP-подключений (1 на инстанс) |
||||
* |
||||
* Управляет подключениями к узлам: перебор адресов, случайный выбор, |
||||
* ротация при падении соединения. До CM_MAX_PARALLEL одновременных |
||||
* подключений на узел. Фоновый пинг для измерения RTT. |
||||
*/ |
||||
|
||||
#include "chat_conn_mgr.h" |
||||
#include "chat_core.h" |
||||
#include "gui_bridge.h" |
||||
|
||||
#include "../../../src/utun_instance.h" |
||||
#include "etcp_api.h" |
||||
#include "etcp.h" |
||||
#include "etcp_connections.h" |
||||
#include "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 <string.h> |
||||
#include <stdlib.h> |
||||
#include "../../../lib/platform_compat.h" |
||||
|
||||
#define CM_ID "chat_conn_mgr" |
||||
|
||||
/* ─── константы ─── */ |
||||
#define CM_MAX_PARALLEL 10 |
||||
#define CM_MAX_ADDRS 16 |
||||
#define CM_MAX_NODES 32 |
||||
#define CM_INIT_TIMEOUT_MS 3000 |
||||
#define CM_RETRY_MS 3000 |
||||
#define CM_PING_MS 5000 |
||||
#define CM_PING_TIMEOUT_MS 2000 |
||||
#define CM_INIT_TIMEOUT_TB (CM_INIT_TIMEOUT_MS * 10) |
||||
#define CM_RETRY_TB (CM_RETRY_MS * 10) |
||||
#define CM_PING_TB (CM_PING_MS * 10) |
||||
|
||||
/* ─── состояния слота ─── */ |
||||
#define CM_SLOT_FREE 0 |
||||
#define CM_SLOT_CONNECTING 1 |
||||
#define CM_SLOT_ACTIVE 2 |
||||
#define CM_SLOT_DEAD 3 |
||||
|
||||
/* ─── типы ─── */ |
||||
|
||||
struct cm_cb { |
||||
void (*cb)(int result, uint64_t node_id, void* arg); |
||||
void* arg; |
||||
struct cm_cb* next; |
||||
}; |
||||
|
||||
struct cm_slot { |
||||
uint8_t addr_idx; |
||||
uint8_t state; |
||||
uint16_t rtt; |
||||
struct ETCP_CONN* conn; |
||||
void* init_timer; |
||||
}; |
||||
|
||||
struct cm_addr { |
||||
struct sockaddr_storage sa; |
||||
uint16_t rtt; |
||||
}; |
||||
|
||||
struct cm_node; |
||||
|
||||
struct cm_slot_ctx { |
||||
struct cm_node* node; |
||||
int slot_idx; |
||||
}; |
||||
|
||||
struct cm_node_changed_ctx { |
||||
struct chat_conn_mgr* mgr; |
||||
uint64_t node_id; |
||||
}; |
||||
|
||||
struct cm_node { |
||||
uint64_t node_id; |
||||
uint8_t pubkey[SC_PUBKEY_SIZE]; |
||||
struct cm_addr addrs[CM_MAX_ADDRS]; |
||||
int addr_count; |
||||
int addr_loaded; |
||||
struct cm_slot slots[CM_MAX_PARALLEL]; |
||||
struct cm_cb* cbs; |
||||
int delivered; |
||||
void* retry_timer; |
||||
struct chat_conn_mgr* mgr; /* back-pointer */ |
||||
}; |
||||
|
||||
struct chat_conn_mgr { |
||||
struct UTUN_INSTANCE* inst; |
||||
struct sqlite3* db; |
||||
struct cm_node nodes[CM_MAX_NODES]; |
||||
int node_count; |
||||
void* ping_timer; |
||||
}; |
||||
|
||||
/* ─── forward ─── */ |
||||
static struct cm_node* cm_find_node(struct chat_conn_mgr* mgr, uint64_t node_id); |
||||
static struct cm_node* cm_ensure_node(struct chat_conn_mgr* mgr, uint64_t node_id); |
||||
static void cm_load_addrs(struct cm_node* node); |
||||
static int cm_count_active(struct cm_node* node); |
||||
static int cm_addr_taken(struct cm_node* node, int idx); |
||||
static int cm_external_conn_active(struct cm_node* node); |
||||
static struct cm_slot* cm_find_free(struct cm_node* node); |
||||
static struct cm_slot* cm_find_dead(struct cm_node* node); |
||||
static struct cm_slot* cm_find_slot_by_conn(struct cm_node* node, struct ETCP_CONN* conn); |
||||
static void cm_deliver_all(struct cm_node* node, int result); |
||||
static void cm_rotation(struct cm_node* node); |
||||
static void cm_init_cb(struct ETCP_CONN* conn, void* arg); |
||||
static void cm_init_timeout(void* arg); |
||||
static void cm_conn_down_cb(struct ETCP_CONN* conn, void* arg); |
||||
static void cm_retry_cb(void* arg); |
||||
static void cm_ping_timer_cb(void* arg); |
||||
static void cm_ping_result_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len); |
||||
static struct ETCP_SOCKET* cm_best_socket(struct chat_conn_mgr* mgr); |
||||
static void cm_invite_result_cb(int result, uint64_t node_id, void* arg); |
||||
static void cm_node_changed_cb(struct ETCP_CONN* conn, void* arg); |
||||
static void cm_node_changed_deferred(void* arg); |
||||
|
||||
/* ─── init / destroy ─── */ |
||||
|
||||
struct chat_conn_mgr* chat_conn_mgr_init(struct UTUN_INSTANCE* inst, struct sqlite3* db) { |
||||
if (!inst || !db) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: init bad args", CM_ID); return NULL; } |
||||
struct chat_conn_mgr* mgr = u_calloc(1, sizeof(struct chat_conn_mgr)); |
||||
if (!mgr) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: alloc failed", CM_ID); return NULL; } |
||||
mgr->inst = inst; mgr->db = db; |
||||
mgr->ping_timer = uasync_set_timeout(inst->ua, CM_PING_TB, mgr, cm_ping_timer_cb, "cm_ping"); |
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized", CM_ID); |
||||
return mgr; |
||||
} |
||||
|
||||
void chat_conn_mgr_destroy(struct chat_conn_mgr* mgr) { |
||||
if (!mgr) return; |
||||
if (mgr->ping_timer) { uasync_cancel_timeout(mgr->inst->ua, mgr->ping_timer); mgr->ping_timer = NULL; } |
||||
for (int ni = 0; ni < mgr->node_count; ni++) { |
||||
struct cm_node* node = &mgr->nodes[ni]; |
||||
if (node->retry_timer) { uasync_cancel_timeout(mgr->inst->ua, node->retry_timer); node->retry_timer = NULL; } |
||||
for (int si = 0; si < CM_MAX_PARALLEL; si++) { |
||||
struct cm_slot* slot = &node->slots[si]; |
||||
if (slot->init_timer) { uasync_cancel_timeout(mgr->inst->ua, slot->init_timer); slot->init_timer = NULL; } |
||||
if (slot->conn) { etcp_connection_close(slot->conn); slot->conn = NULL; } |
||||
} |
||||
cm_deliver_all(node, CC_ERR_INTERNAL); |
||||
} |
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: destroyed", CM_ID); |
||||
u_free(mgr); |
||||
} |
||||
|
||||
struct sqlite3* chat_conn_mgr_get_db(struct chat_conn_mgr* mgr) { |
||||
return mgr ? mgr->db : NULL; |
||||
} |
||||
|
||||
/* ─── управление узлами ─── */ |
||||
|
||||
static struct cm_node* cm_find_node(struct chat_conn_mgr* mgr, uint64_t node_id) { |
||||
for (int i = 0; i < mgr->node_count; i++) |
||||
if (mgr->nodes[i].node_id == node_id) return &mgr->nodes[i]; |
||||
return NULL; |
||||
} |
||||
|
||||
static struct cm_node* cm_ensure_node(struct chat_conn_mgr* mgr, uint64_t node_id) { |
||||
struct cm_node* node = cm_find_node(mgr, node_id); |
||||
if (node) return node; |
||||
if (mgr->node_count >= CM_MAX_NODES) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: too many nodes", CM_ID); return NULL; } |
||||
node = &mgr->nodes[mgr->node_count++]; |
||||
memset(node, 0, sizeof(*node)); |
||||
node->node_id = node_id; node->mgr = mgr; |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) node->slots[i].addr_idx = 255; |
||||
return node; |
||||
} |
||||
|
||||
void chat_conn_mgr_add_node(struct chat_conn_mgr* mgr, uint64_t node_id) { |
||||
if (!mgr) return; |
||||
if (cm_find_node(mgr, node_id)) return; |
||||
cm_ensure_node(mgr, node_id); |
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: node 0x%016llx added", CM_ID, (unsigned long long)node_id); |
||||
} |
||||
|
||||
/* ─── загрузка адресов ─── */ |
||||
|
||||
static void cm_load_addrs(struct cm_node* node) { |
||||
if (!node || node->addr_loaded) return; |
||||
struct sqlite3* db = node->mgr->db; |
||||
struct { uint8_t a[4]; uint16_t p; } raw[CM_MAX_ADDRS]; |
||||
int rc = 0; |
||||
sqlite3_stmt* st = NULL; |
||||
if (sqlite3_prepare_v2(db, "SELECT address,port FROM node_addresses WHERE node_id=? AND family=4", |
||||
-1, &st, NULL) == SQLITE_OK) { |
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)node->node_id); |
||||
while (sqlite3_step(st) == SQLITE_ROW && rc < CM_MAX_ADDRS) { |
||||
const void* a = sqlite3_column_blob(st, 0); |
||||
if (a && sqlite3_column_bytes(st, 0) == 4) { memcpy(raw[rc].a, a, 4); raw[rc].p = (uint16_t)sqlite3_column_int(st, 1); rc++; } |
||||
} |
||||
sqlite3_finalize(st); |
||||
} |
||||
int uniq = 0; |
||||
for (int i = 0; i < rc; i++) { |
||||
int dup = 0; |
||||
for (int j = 0; j < uniq; j++) if (memcmp(raw[i].a, raw[j].a, 4) == 0 && raw[i].p == raw[j].p) { dup = 1; break; } |
||||
if (!dup) { if (i != uniq) raw[uniq] = raw[i]; uniq++; } |
||||
} |
||||
for (int i = 0; i < uniq; i++) { |
||||
struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; |
||||
memcpy(&sin.sin_addr.s_addr, raw[i].a, 4); sin.sin_port = htons(raw[i].p); |
||||
memcpy(&node->addrs[i].sa, &sin, sizeof(sin)); node->addrs[i].rtt = 65535; |
||||
} |
||||
node->addr_count = uniq; node->addr_loaded = 1; |
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: node 0x%016llx loaded %d addrs", CM_ID, (unsigned long long)node->node_id, uniq); |
||||
|
||||
uint8_t pubkey[SC_PUBKEY_SIZE] = {0}; |
||||
if (sqlite3_prepare_v2(db, "SELECT x25519_pubkey FROM nodes WHERE node_id=?", |
||||
-1, &st, NULL) == SQLITE_OK) { |
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)node->node_id); |
||||
if (sqlite3_step(st) == SQLITE_ROW) { |
||||
const void* pk = sqlite3_column_blob(st, 0); |
||||
if (pk && sqlite3_column_bytes(st, 0) == SC_PUBKEY_SIZE) memcpy(pubkey, pk, SC_PUBKEY_SIZE); |
||||
} |
||||
sqlite3_finalize(st); |
||||
} |
||||
if (pubkey[0] || pubkey[1]) memcpy(node->pubkey, pubkey, SC_PUBKEY_SIZE); |
||||
else DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: node 0x%016llx no pubkey in DB", CM_ID, (unsigned long long)node->node_id); |
||||
} |
||||
|
||||
/* ─── helpers ─── */ |
||||
|
||||
static int cm_count_active(struct cm_node* node) { |
||||
int a = 0; |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) if (node->slots[i].state == CM_SLOT_ACTIVE) a++; |
||||
return a; |
||||
} |
||||
|
||||
static int cm_addr_taken(struct cm_node* node, int idx) { |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) |
||||
if (node->slots[i].state != CM_SLOT_FREE && node->slots[i].addr_idx == idx) return 1; |
||||
return 0; |
||||
} |
||||
|
||||
static int cm_external_conn_active(struct cm_node* node) { |
||||
struct UTUN_INSTANCE* inst = node->mgr->inst; |
||||
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node->node_id); |
||||
if (!e) return 0; |
||||
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; |
||||
return (ce && ce->conn && ce->conn->links_up && ce->conn->peer_node_id == node->node_id); |
||||
} |
||||
|
||||
static struct cm_slot* cm_find_free(struct cm_node* node) { |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) |
||||
if (node->slots[i].state == CM_SLOT_FREE) return &node->slots[i]; |
||||
return NULL; |
||||
} |
||||
|
||||
static struct cm_slot* cm_find_dead(struct cm_node* node) { |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) |
||||
if (node->slots[i].state == CM_SLOT_DEAD) return &node->slots[i]; |
||||
return NULL; |
||||
} |
||||
|
||||
static struct cm_slot* cm_find_slot_by_conn(struct cm_node* node, struct ETCP_CONN* conn) { |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) |
||||
if (node->slots[i].conn == conn) return &node->slots[i]; |
||||
return NULL; |
||||
} |
||||
|
||||
static void cm_deliver_all(struct cm_node* node, int result) { |
||||
while (node->cbs) { struct cm_cb* cb = node->cbs; node->cbs = cb->next; cb->cb(result, node->node_id, cb->arg); u_free(cb); } |
||||
} |
||||
|
||||
static struct ETCP_SOCKET* cm_best_socket(struct chat_conn_mgr* mgr) { |
||||
struct ETCP_SOCKET* s = mgr->inst->etcp_sockets; |
||||
while (s) { if (s->local_addr.ss_family == AF_INET) return s; s = s->next; } |
||||
return NULL; |
||||
} |
||||
|
||||
/* ─── ротация ─── */ |
||||
|
||||
static void cm_rotation(struct cm_node* node) { |
||||
if (!node) return; |
||||
struct chat_conn_mgr* mgr = node->mgr; |
||||
|
||||
int active = cm_count_active(node); |
||||
|
||||
if (cm_external_conn_active(node)) { |
||||
if (!node->delivered) { node->delivered = 1; cm_deliver_all(node, CC_OK); } |
||||
if (node->retry_timer) { uasync_cancel_timeout(mgr->inst->ua, node->retry_timer); node->retry_timer = NULL; } |
||||
return; |
||||
} |
||||
|
||||
if (active >= CM_MAX_PARALLEL) { |
||||
if (!node->delivered) { node->delivered = 1; cm_deliver_all(node, CC_OK); } |
||||
if (node->retry_timer) { uasync_cancel_timeout(mgr->inst->ua, node->retry_timer); node->retry_timer = NULL; } |
||||
return; |
||||
} |
||||
|
||||
struct cm_slot* slot = cm_find_free(node); |
||||
if (!slot) slot = cm_find_dead(node); |
||||
if (!slot) return; |
||||
|
||||
struct ETCP_SOCKET* sock = cm_best_socket(mgr); |
||||
if (!sock) return; |
||||
|
||||
for (int attempt = 0; attempt < 3; attempt++) { |
||||
int idx = rand() % node->addr_count; |
||||
if (!cm_addr_taken(node, idx)) { |
||||
slot->addr_idx = idx; slot->state = CM_SLOT_CONNECTING; |
||||
struct ETCP_CONN* conn = etcp_connection_create(mgr->inst, NULL); |
||||
if (!conn) { slot->state = CM_SLOT_FREE; return; } |
||||
sc_init_ctx(&conn->crypto_ctx, &mgr->inst->my_keys); |
||||
sc_set_peer_public_key(&conn->crypto_ctx, node->pubkey, 0); |
||||
etcp_conn_set_peer_node_id(conn, node->node_id); |
||||
struct cm_slot_ctx* ctx = u_calloc(1, sizeof(struct cm_slot_ctx)); |
||||
if (!ctx) { etcp_connection_close(conn); slot->state = CM_SLOT_FREE; return; } |
||||
ctx->node = node; ctx->slot_idx = (int)(slot - node->slots); |
||||
etcp_conn_add_init_cbk(conn, cm_init_cb, ctx); |
||||
etcp_conn_add_node_changed_cbk(conn, cm_node_changed_cb, mgr); |
||||
if (!etcp_link_new(conn, sock, &node->addrs[idx].sa, 0)) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: link_new failed node=0x%016llx idx=%d", CM_ID, (unsigned long long)node->node_id, idx); |
||||
etcp_connection_close(conn); u_free(ctx); slot->state = CM_SLOT_DEAD; |
||||
cm_rotation(node); return; |
||||
} |
||||
slot->conn = conn; |
||||
slot->init_timer = uasync_set_timeout(mgr->inst->ua, CM_INIT_TIMEOUT_TB, ctx, cm_init_timeout, "cm_init"); |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: INIT node=0x%016llx slot=%d addr=%d", CM_ID, (unsigned long long)node->node_id, ctx->slot_idx, idx); |
||||
return; |
||||
} |
||||
} |
||||
|
||||
if (slot->addr_idx != 255) { |
||||
slot->state = CM_SLOT_CONNECTING; |
||||
struct ETCP_CONN* conn = etcp_connection_create(mgr->inst, NULL); |
||||
if (!conn) { slot->state = CM_SLOT_FREE; return; } |
||||
sc_init_ctx(&conn->crypto_ctx, &mgr->inst->my_keys); |
||||
sc_set_peer_public_key(&conn->crypto_ctx, node->pubkey, 0); |
||||
etcp_conn_set_peer_node_id(conn, node->node_id); |
||||
struct cm_slot_ctx* ctx = u_calloc(1, sizeof(struct cm_slot_ctx)); |
||||
if (!ctx) { etcp_connection_close(conn); slot->state = CM_SLOT_FREE; return; } |
||||
ctx->node = node; ctx->slot_idx = (int)(slot - node->slots); |
||||
etcp_conn_add_init_cbk(conn, cm_init_cb, ctx); |
||||
etcp_conn_add_node_changed_cbk(conn, cm_node_changed_cb, mgr); |
||||
if (!etcp_link_new(conn, sock, &node->addrs[slot->addr_idx].sa, 0)) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: link_new(retry) failed node=0x%016llx idx=%d", CM_ID, (unsigned long long)node->node_id, slot->addr_idx); |
||||
etcp_connection_close(conn); u_free(ctx); slot->state = CM_SLOT_DEAD; |
||||
cm_rotation(node); return; |
||||
} |
||||
slot->conn = conn; |
||||
slot->init_timer = uasync_set_timeout(mgr->inst->ua, CM_INIT_TIMEOUT_TB, ctx, cm_init_timeout, "cm_init"); |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: INIT(retry) node=0x%016llx slot=%d addr=%d", CM_ID, (unsigned long long)node->node_id, ctx->slot_idx, slot->addr_idx); |
||||
return; |
||||
} |
||||
|
||||
slot->state = CM_SLOT_FREE; |
||||
} |
||||
|
||||
/* ─── коллбэки ─── */ |
||||
|
||||
static void cm_init_cb(struct ETCP_CONN* conn, void* arg) { |
||||
struct cm_slot_ctx* ctx = (struct cm_slot_ctx*)arg; |
||||
struct cm_node* node = ctx->node; |
||||
struct cm_slot* slot = &node->slots[ctx->slot_idx]; |
||||
|
||||
if (slot->init_timer) { uasync_cancel_timeout(node->mgr->inst->ua, slot->init_timer); slot->init_timer = NULL; } |
||||
|
||||
if (slot->conn && slot->conn != conn) { etcp_connection_close(slot->conn); slot->conn = NULL; } |
||||
|
||||
slot->state = CM_SLOT_ACTIVE; slot->conn = conn; slot->rtt = conn->rtt_last; |
||||
etcp_conn_add_down_cbk(conn, cm_conn_down_cb, node); |
||||
etcp_conn_remove_init_cbk(conn, cm_init_cb, ctx); |
||||
u_free(ctx); |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: INIT OK node=0x%016llx slot=%d rtt=%u active=%d", |
||||
CM_ID, (unsigned long long)node->node_id, (int)(slot - node->slots), slot->rtt, cm_count_active(node)); |
||||
|
||||
struct chat_conn_mgr* mgr = node->mgr; |
||||
int active = cm_count_active(node); |
||||
|
||||
if (active > CM_MAX_PARALLEL) { |
||||
int worst_idx = -1; uint16_t worst_rtt = 0; |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) { |
||||
if (node->slots[i].state == CM_SLOT_ACTIVE && node->slots[i].rtt > worst_rtt) { |
||||
worst_rtt = node->slots[i].rtt; worst_idx = i; |
||||
} |
||||
} |
||||
if (worst_idx >= 0) { |
||||
struct cm_slot* w = &node->slots[worst_idx]; |
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: evict slot=%d rtt=%u node=0x%016llx", |
||||
CM_ID, worst_idx, w->rtt, (unsigned long long)node->node_id); |
||||
w->state = CM_SLOT_FREE; w->conn = NULL; w->addr_idx = 255; |
||||
} |
||||
active = cm_count_active(node); |
||||
} |
||||
|
||||
if (active >= CM_MAX_PARALLEL) { |
||||
if (!node->delivered) { node->delivered = 1; cm_deliver_all(node, CC_OK); } |
||||
if (node->retry_timer) { uasync_cancel_timeout(mgr->inst->ua, node->retry_timer); node->retry_timer = NULL; } |
||||
return; |
||||
} |
||||
|
||||
cm_rotation(node); |
||||
} |
||||
|
||||
static void cm_init_timeout(void* arg) { |
||||
struct cm_slot_ctx* ctx = (struct cm_slot_ctx*)arg; |
||||
struct cm_node* node = ctx->node; |
||||
struct cm_slot* slot = &node->slots[ctx->slot_idx]; |
||||
slot->init_timer = NULL; |
||||
|
||||
if (slot->state == CM_SLOT_CONNECTING) slot->state = CM_SLOT_DEAD; |
||||
node->delivered = 0; |
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: INIT TIMEOUT node=0x%016llx slot=%d", |
||||
CM_ID, (unsigned long long)node->node_id, ctx->slot_idx); |
||||
|
||||
cm_rotation(node); |
||||
u_free(ctx); |
||||
} |
||||
|
||||
static void cm_conn_down_cb(struct ETCP_CONN* conn, void* arg) { |
||||
struct cm_node* node = (struct cm_node*)arg; |
||||
struct cm_slot* slot = cm_find_slot_by_conn(node, conn); |
||||
if (!slot) return; |
||||
|
||||
slot->state = CM_SLOT_FREE; slot->conn = NULL; |
||||
node->delivered = 0; |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: DOWN node=0x%016llx slot=%d active=%d", |
||||
CM_ID, (unsigned long long)node->node_id, (int)(slot - node->slots), cm_count_active(node)); |
||||
|
||||
cm_rotation(node); |
||||
} |
||||
|
||||
static void cm_retry_cb(void* arg) { |
||||
struct cm_node* node = (struct cm_node*)arg; |
||||
node->retry_timer = NULL; |
||||
|
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) |
||||
if (node->slots[i].state == CM_SLOT_DEAD) { node->slots[i].state = CM_SLOT_FREE; node->slots[i].addr_idx = 255; } |
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: retry node=0x%016llx active=%d", CM_ID, (unsigned long long)node->node_id, cm_count_active(node)); |
||||
|
||||
cm_rotation(node); |
||||
|
||||
if (cm_count_active(node) < CM_MAX_PARALLEL && !node->delivered) |
||||
node->retry_timer = uasync_set_timeout(node->mgr->inst->ua, CM_RETRY_TB, node, cm_retry_cb, "cm_retry"); |
||||
} |
||||
|
||||
/* ─── node_changed ─── */ |
||||
|
||||
static void cm_node_changed_deferred(void* arg) { |
||||
struct cm_node_changed_ctx* ctx = (struct cm_node_changed_ctx*)arg; |
||||
struct cm_node* node = cm_find_node(ctx->mgr, ctx->node_id); |
||||
if (node) { |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) { |
||||
struct cm_slot* slot = &node->slots[i]; |
||||
if (slot->conn) { etcp_connection_close(slot->conn); slot->conn = NULL; } |
||||
slot->state = CM_SLOT_FREE; slot->addr_idx = 255; |
||||
} |
||||
cm_deliver_all(node, CC_ERR_INTERNAL); |
||||
node->delivered = 0; |
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: NODE_CHANGED deferred close done node=0x%016llx", |
||||
CM_ID, (unsigned long long)ctx->node_id); |
||||
} |
||||
u_free(ctx); |
||||
} |
||||
|
||||
static void cm_node_changed_cb(struct ETCP_CONN* conn, void* arg) { |
||||
struct chat_conn_mgr* mgr = (struct chat_conn_mgr*)arg; |
||||
if (!mgr || !conn) return; |
||||
uint64_t node_id = conn->peer_node_id; |
||||
struct cm_node* node = cm_find_node(mgr, node_id); |
||||
if (!node) return; |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: NODE_CHANGED node=0x%016llx — clearing addresses and closing connections", |
||||
CM_ID, (unsigned long long)node_id); |
||||
|
||||
sqlite3_stmt* st = NULL; |
||||
if (sqlite3_prepare_v2(mgr->db, "DELETE FROM node_addresses WHERE node_id=?", -1, &st, NULL) == SQLITE_OK) { |
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); |
||||
sqlite3_step(st); sqlite3_finalize(st); |
||||
} |
||||
|
||||
node->addr_loaded = 0; |
||||
node->addr_count = 0; |
||||
|
||||
if (node->retry_timer) { uasync_cancel_timeout(mgr->inst->ua, node->retry_timer); node->retry_timer = NULL; } |
||||
|
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) { |
||||
struct cm_slot* slot = &node->slots[i]; |
||||
if (slot->init_timer) { uasync_cancel_timeout(mgr->inst->ua, slot->init_timer); slot->init_timer = NULL; } |
||||
if (slot->conn && slot->conn != conn) { etcp_connection_close(slot->conn); slot->conn = NULL; } |
||||
if (slot->conn != conn) { slot->state = CM_SLOT_FREE; slot->addr_idx = 255; } |
||||
} |
||||
|
||||
struct cm_node_changed_ctx* ctx = u_calloc(1, sizeof(*ctx)); |
||||
if (ctx) { ctx->mgr = mgr; ctx->node_id = node_id; uasync_post(mgr->inst->ua, cm_node_changed_deferred, ctx); } |
||||
|
||||
uint8_t evt[8]; memcpy(evt, &node_id, 8); |
||||
gui_bridge_post(GUI_EVT_NODE_CHANGED, evt, 8); |
||||
} |
||||
|
||||
/* ─── фоновый пинг ─── */ |
||||
|
||||
static void cm_ping_timer_cb(void* arg) { |
||||
struct chat_conn_mgr* mgr = (struct chat_conn_mgr*)arg; |
||||
if (mgr->node_count == 0) { mgr->ping_timer = uasync_set_timeout(mgr->inst->ua, CM_PING_TB, mgr, cm_ping_timer_cb, "cm_ping"); return; } |
||||
|
||||
int ni = rand() % mgr->node_count; |
||||
struct cm_node* node = &mgr->nodes[ni]; |
||||
if (!node->addr_loaded || node->addr_count == 0) { mgr->ping_timer = uasync_set_timeout(mgr->inst->ua, CM_PING_TB, mgr, cm_ping_timer_cb, "cm_ping"); return; } |
||||
|
||||
int ai = rand() % node->addr_count; |
||||
|
||||
if (cm_addr_taken(node, ai)) { mgr->ping_timer = uasync_set_timeout(mgr->inst->ua, CM_PING_TB, mgr, cm_ping_timer_cb, "cm_ping"); return; } |
||||
|
||||
struct ETCP_SOCKET* sock = cm_best_socket(mgr); |
||||
if (sock) etcp_send_ping_to_socket(mgr->inst, sock, node->pubkey, &node->addrs[ai].sa, CM_PING_TIMEOUT_MS, cm_ping_result_cb, &node->addrs[ai], NULL, 0, 0); |
||||
|
||||
mgr->ping_timer = uasync_set_timeout(mgr->inst->ua, CM_PING_TB, mgr, cm_ping_timer_cb, "cm_ping"); |
||||
} |
||||
|
||||
static void cm_ping_result_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len) { |
||||
(void)nonce; (void)resp_data; (void)resp_data_len; |
||||
struct cm_addr* addr = (struct cm_addr*)arg; |
||||
addr->rtt = success ? rtt : 65535; |
||||
} |
||||
|
||||
/* ─── публичное API ─── */ |
||||
|
||||
void chat_conn_mgr_connect(struct chat_conn_mgr* mgr, uint64_t node_id, |
||||
void (*cb)(int result, uint64_t node_id, void* arg), |
||||
void* arg) { |
||||
if (!mgr) { if (cb) cb(CC_ERR_INTERNAL, node_id, arg); return; } |
||||
|
||||
struct cm_node* node = cm_ensure_node(mgr, node_id); |
||||
if (!node) { if (cb) cb(CC_ERR_INTERNAL, node_id, arg); return; } |
||||
|
||||
if (!node->addr_loaded) cm_load_addrs(node); |
||||
|
||||
if (cb) { |
||||
struct cm_cb* cb_node = u_calloc(1, sizeof(struct cm_cb)); |
||||
if (cb_node) { cb_node->cb = cb; cb_node->arg = arg; cb_node->next = node->cbs; node->cbs = cb_node; } |
||||
} |
||||
|
||||
if (node->pubkey[0] == 0 && node->pubkey[1] == 0) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect to 0x%016llx — no pubkey in DB", CM_ID, (unsigned long long)node_id); |
||||
if (!node->delivered) { node->delivered = 1; cm_deliver_all(node, CC_ERR_NOT_FOUND); } |
||||
return; |
||||
} |
||||
if (node->addr_count == 0) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: connect to 0x%016llx — no addresses", CM_ID, (unsigned long long)node_id); |
||||
if (!node->delivered) { node->delivered = 1; cm_deliver_all(node, CC_ERR_NO_ADDRESSES); } |
||||
return; |
||||
} |
||||
|
||||
if (node->delivered) return; |
||||
|
||||
if (cm_external_conn_active(node)) { |
||||
node->delivered = 1; cm_deliver_all(node, CC_OK); |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: node 0x%016llx already connected externally", CM_ID, (unsigned long long)node_id); |
||||
return; |
||||
} |
||||
|
||||
if (cm_count_active(node) >= 1) { |
||||
// at least one active already (might be from a previous rotation that didn't deliver yet)
|
||||
if (cm_count_active(node) >= CM_MAX_PARALLEL && !node->delivered) { node->delivered = 1; cm_deliver_all(node, CC_OK); } |
||||
return; |
||||
} |
||||
|
||||
cm_rotation(node); |
||||
|
||||
if (cm_count_active(node) == 0 && !node->retry_timer && !node->delivered) |
||||
node->retry_timer = uasync_set_timeout(mgr->inst->ua, CM_RETRY_TB, node, cm_retry_cb, "cm_retry"); |
||||
} |
||||
|
||||
void chat_conn_mgr_cancel(struct chat_conn_mgr* mgr, uint64_t node_id) { |
||||
if (!mgr) return; |
||||
struct cm_node* node = cm_find_node(mgr, node_id); |
||||
if (!node) return; |
||||
|
||||
if (node->retry_timer) { uasync_cancel_timeout(mgr->inst->ua, node->retry_timer); node->retry_timer = NULL; } |
||||
for (int i = 0; i < CM_MAX_PARALLEL; i++) { |
||||
struct cm_slot* slot = &node->slots[i]; |
||||
if (slot->init_timer) { uasync_cancel_timeout(mgr->inst->ua, slot->init_timer); slot->init_timer = NULL; } |
||||
if (slot->conn) { etcp_connection_close(slot->conn); slot->conn = NULL; } |
||||
slot->state = CM_SLOT_FREE; slot->addr_idx = 255; |
||||
} |
||||
cm_deliver_all(node, CC_ERR_INTERNAL); |
||||
node->delivered = 0; |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: cancelled node 0x%016llx", CM_ID, (unsigned long long)node_id); |
||||
} |
||||
|
||||
void chat_conn_mgr_connect_from_invite(struct chat_conn_mgr* mgr, struct chat_invite* inv) { |
||||
if (!mgr || !inv) return; |
||||
uint64_t node_id = inv->node_id; |
||||
|
||||
/* save pubkey — minimal node record, ed25519/name придут позже через протокол */ |
||||
{ |
||||
sqlite3_stmt* st = NULL; |
||||
sqlite3_prepare_v2(mgr->db, |
||||
"INSERT INTO nodes(node_id,x25519_pubkey,created_at) VALUES(?,?,?)" |
||||
" ON CONFLICT(node_id) DO UPDATE SET x25519_pubkey=excluded.x25519_pubkey," |
||||
" created_at=COALESCE(nodes.created_at, excluded.created_at)", |
||||
-1, &st, NULL); |
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); |
||||
sqlite3_bind_blob(st, 2, inv->pubkey, 32, SQLITE_STATIC); |
||||
sqlite3_bind_int64(st, 3, (sqlite3_int64)ntp_time_get_seconds(mgr->inst)); |
||||
sqlite3_step(st); sqlite3_finalize(st); |
||||
} |
||||
|
||||
/* save addresses (socket_id=0) */ |
||||
sqlite3_stmt* ds = NULL; |
||||
if (sqlite3_prepare_v2(mgr->db, "DELETE FROM node_addresses WHERE node_id=? AND socket_id=0", |
||||
-1, &ds, NULL) == SQLITE_OK) { |
||||
sqlite3_bind_int64(ds, 1, (sqlite3_int64)node_id); |
||||
sqlite3_step(ds); sqlite3_finalize(ds); |
||||
} |
||||
sqlite3_stmt* is = NULL; |
||||
if (sqlite3_prepare_v2(mgr->db, "INSERT INTO node_addresses(node_id,family,protocol,address,port,addr_type,socket_id)" |
||||
" VALUES(?,?,1,?,?,0,?)", -1, &is, NULL) == SQLITE_OK) { |
||||
const uint8_t* src = inv->addrs_data; |
||||
for (int i = 0; i < inv->addr_count; i++) { |
||||
uint8_t family = *src++; uint8_t sid = *src++; |
||||
if (family == 4) { |
||||
sqlite3_bind_int64(is, 1, (sqlite3_int64)node_id); |
||||
sqlite3_bind_int(is, 2, 4); |
||||
sqlite3_bind_blob(is, 3, src, 4, SQLITE_STATIC); src += 4; |
||||
uint16_t port = ((uint16_t)src[0] << 8) | src[1]; src += 2; |
||||
sqlite3_bind_int(is, 4, (int)port); |
||||
sqlite3_bind_int(is, 5, (int)sid); |
||||
sqlite3_step(is); sqlite3_reset(is); |
||||
} else src += 18; |
||||
} |
||||
sqlite3_finalize(is); |
||||
} |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: invite node=0x%016llx addrs=%d — saved, connecting", |
||||
CM_ID, (unsigned long long)node_id, inv->addr_count); |
||||
|
||||
/* add node + reset addr_loaded to reload from DB */ |
||||
struct cm_node* node = cm_ensure_node(mgr, node_id); |
||||
if (node) { node->addr_loaded = 0; cm_load_addrs(node); } |
||||
|
||||
{ char ch_str[32]; snprintf(ch_str, sizeof(ch_str), "%llu", (unsigned long long)inv->channel_id); |
||||
chat_core_ensure_channel_ready(ch_str); } |
||||
|
||||
chat_conn_mgr_connect(mgr, node_id, cm_invite_result_cb, &inv->channel_id); |
||||
} |
||||
|
||||
static void cm_invite_result_cb(int result, uint64_t node_id, void* arg) { |
||||
uint64_t channel_id = arg ? *(uint64_t*)arg : 0; |
||||
uint8_t data[20]; memcpy(data, &node_id, 8); memcpy(data + 8, &result, 4); memcpy(data + 12, &channel_id, 8); |
||||
gui_bridge_post(GUI_EVT_CONNECT_RESULT, data, 20); |
||||
} |
||||
@ -1,55 +0,0 @@
|
||||
/*
|
||||
* chat_conn_mgr.h — менеджер ETCP-подключений (1 на инстанс) |
||||
* |
||||
* Управляет подключениями к узлам: перебор адресов, случайный выбор, |
||||
* ротация при падении соединения. До CM_MAX_PARALLEL одновременных |
||||
* подключений на узел. Фоновый пинг для измерения RTT. |
||||
*/ |
||||
|
||||
#ifndef CHAT_CONN_MGR_H |
||||
#define CHAT_CONN_MGR_H |
||||
|
||||
#include <stdint.h> |
||||
#include <stddef.h> |
||||
|
||||
struct sqlite3; |
||||
struct UTUN_INSTANCE; |
||||
|
||||
/* ── Коды возврата ── */ |
||||
#define CC_OK 0 |
||||
#define CC_ERR_NOT_FOUND -1 |
||||
#define CC_ERR_NO_ADDRESSES -2 |
||||
#define CC_ERR_TIMEOUT -3 |
||||
#define CC_ERR_UNREACHABLE -4 |
||||
#define CC_ERR_INTERNAL -7 |
||||
|
||||
/* ── Жизненный цикл ── */ |
||||
|
||||
struct chat_conn_mgr* chat_conn_mgr_init(struct UTUN_INSTANCE* inst, struct sqlite3* db); |
||||
void chat_conn_mgr_destroy(struct chat_conn_mgr* mgr); |
||||
struct sqlite3* chat_conn_mgr_get_db(struct chat_conn_mgr* mgr); |
||||
|
||||
/* ── Управление узлами ── */ |
||||
|
||||
void chat_conn_mgr_add_node(struct chat_conn_mgr* mgr, uint64_t node_id); |
||||
|
||||
/* ── Подключение ── */ |
||||
|
||||
void chat_conn_mgr_connect(struct chat_conn_mgr* mgr, uint64_t node_id, |
||||
void (*cb)(int result, uint64_t node_id, void* arg), |
||||
void* arg); |
||||
void chat_conn_mgr_cancel(struct chat_conn_mgr* mgr, uint64_t node_id); |
||||
|
||||
/* ── invite ── */ |
||||
|
||||
struct chat_invite { |
||||
uint64_t channel_id; |
||||
uint64_t node_id; |
||||
uint8_t pubkey[32]; |
||||
uint8_t* addrs_data; |
||||
int addr_count; |
||||
}; |
||||
|
||||
void chat_conn_mgr_connect_from_invite(struct chat_conn_mgr* mgr, struct chat_invite* inv); |
||||
|
||||
#endif /* CHAT_CONN_MGR_H */ |
||||
@ -0,0 +1,53 @@
|
||||
/*
|
||||
* 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 */ |
||||
@ -0,0 +1,223 @@
|
||||
/*
|
||||
* 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; |
||||
} |
||||
{ uint8_t evt[65]; uint8_t cl=(uint8_t)strlen(req->channel_id); evt[0]=cl; memcpy(evt+1,req->channel_id,cl); |
||||
gui_bridge_post(GUI_EVT_MSG_RECEIVED, evt, 1+cl); } |
||||
} |
||||
|
||||
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)record_ts; (void)data; (void)len; (void)author; |
||||
const char* ch_id = (const char*)arg; |
||||
uint8_t evt[65]; uint8_t cl = (uint8_t)strlen(ch_id); evt[0] = cl; memcpy(evt + 1, ch_id, cl); |
||||
gui_bridge_post(GUI_EVT_MSG_RECEIVED, evt, 1 + cl); |
||||
} |
||||
@ -0,0 +1,189 @@
|
||||
/*
|
||||
* 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); |
||||
} |
||||
@ -0,0 +1,118 @@
|
||||
/*
|
||||
* 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); |
||||
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(); |
||||
} |
||||
Loading…
Reference in new issue