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

dm: прямой обмен сообщениями (DM) + soundtouch + архитектура звонков

Browse Source 
- src/dm/: dm_core/dm_crypto/dm_mailbox — прямой обмен сообщениями (DM)
- lib/soundtouch-master/: vendored SoundTouch (изменение темпа/высоты аудио)
- doc/dm_arch.md, doc/chat_call_arch.md: архитектура DM и звонков
- tests: test_dm.c, test_dm_e2e.c
- chat: рефакторинг chat_* модулей, member_sync, invite_build
- chatgui/chatgui-android: синхронизация с новым API чата
- lib/u_async, debug_config, utun_instance: сопутствующие изменения
This commit is contained in:
evgeny committed 2026-09-15 21:09:17 +03:00
1 parent 761f64cf76
commit 73fe99b33f
123 files changed
+14095 -1270
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dm/dm_core.c
+785
View File
@@ -0,0 +1,785 @@
/*
* dm_core.c — прямой p2p чат между двумя пользователями (DM)
*
* Отдельная подсистема (не канал): беседы хранятся в dm_conversations,
* сообщения в dm_messages, синхронизация идёт напрямую через etcp_router
* (сервис 0x34), при недоступности пира — через dm_mailbox (0x35).
*
* Два независимых направленных потока (я→пир, пир→я) с монотонным seq.
* Каноническое тело сообщения — см. dm_core.h (общее с dm_mailbox).
*/
#include "dm_core.h"
#include "dm_crypto.h"
#include "dm_mailbox.h"
#include "../chat/chat_core.h"
#include "../chat/chat_event.h"
#include "../utun_instance.h"
#include "../ntp_time.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../routing_layer/topo_group.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/secure_channel.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
#include "../../lib/debug_config.h"
#include "../../lib/platform_compat.h"
#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>
#define DM_ID "dm_core"
#define DM_DATA_MAX 1024 /* макс размер открытого текста сообщения (v1 — текст) */
/* Подкоманды протокола DM (первый байт payload после ROUTER_SVC-заголовка). */
#define DM_SUBCMD_MSG 0x01 /* [msg] — сообщение */
#define DM_SUBCMD_ACK 0x02 /* [conv_id:8][seq:8] — подтверждение приёма */
#define DM_SUBCMD_HELLO 0x03 /* [conv_id:8][last_out_seq:8][last_in_seq:8] — catch-up при connect */
/* Per-instance состояние модуля. */
struct dm_state {
struct UTUN_INSTANCE* inst;
sqlite3* db;
int initialized;
};
/* ── вспомогательные ── */
static struct dm_state* dm_of(struct UTUN_INSTANCE* inst) {
return inst ? inst->dm : NULL;
}
/* Выполнить SQL без результата. Возвращает 0/ошибку. */
static int dm_exec(struct dm_state* dm, const char* sql) {
char* err = NULL;
if (sqlite3_exec(dm->db, sql, NULL, NULL, &err) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: sql failed: %s (%s)", DM_ID, sql, err ? err : "?");
if (err) sqlite3_free(err);
return -1;
}
return 0;
}
/* Создать таблицы бесед и сообщений (общие для всех узлов). */
static void dm_create_tables(struct dm_state* dm) {
dm_exec(dm, "CREATE TABLE IF NOT EXISTS dm_conversations ("
" conv_id TEXT PRIMARY KEY,"
" peer_node_id INTEGER NOT NULL,"
" peer_x25519 BLOB NOT NULL,"
" peer_ed25519 BLOB NOT NULL,"
" peer_name TEXT DEFAULT '',"
" group_id INTEGER DEFAULT 0,"
" last_out_seq INTEGER DEFAULT 0,"
" last_in_seq INTEGER DEFAULT 0,"
" created_at INTEGER DEFAULT 0,"
" last_ts INTEGER DEFAULT 0)");
dm_exec(dm, "CREATE TABLE IF NOT EXISTS dm_messages ("
" conv_id TEXT NOT NULL,"
" dir INTEGER NOT NULL," /* 0=входящее, 1=исходящее */
" seq INTEGER NOT NULL,"
" ts INTEGER NOT NULL,"
" author INTEGER NOT NULL,"
" ct TEXT DEFAULT '',"
" data BLOB," /* зашифрованное тело (ciphertext+tag) */
" sig BLOB," /* Ed25519 автора */
" PRIMARY KEY (conv_id, dir, seq))");
}
/* Прочитать оба pubkey узла (x25519 + ed25519) из таблицы nodes. 0=ок, <0=нет. */
static int dm_node_pubkeys(struct dm_state* dm, uint64_t node_id, uint8_t x25519[32], uint8_t ed25519[32]) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT x25519_pubkey, ed25519_pubkey FROM nodes WHERE node_id=?",
-1, &st, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
int rc = -1;
if (sqlite3_step(st) == SQLITE_ROW) {
const uint8_t* x = (const uint8_t*)sqlite3_column_blob(st, 0);
const uint8_t* e = (const uint8_t*)sqlite3_column_blob(st, 1);
if (x && e && sqlite3_column_bytes(st, 0) >= 32 && sqlite3_column_bytes(st, 1) >= 32) {
memcpy(x25519, x, 32);
memcpy(ed25519, e, 32);
rc = 0;
}
}
sqlite3_finalize(st);
return rc;
}
/* Есть ли общая группа с пиром (anti-spam: принимаем DM только от знакомых из общих каналов). */
static int dm_share_group(struct dm_state* dm, uint64_t peer_id) {
uint64_t* ch_ids = NULL;
int ch_count = 0;
if (topo_node_sqlite_get_member_channels(dm->db, dm->inst->node_id, &ch_ids, &ch_count) != 0)
return 0;
int shared = 0;
for (int i = 0; i < ch_count; i++) {
char ch[64]; snprintf(ch, sizeof(ch), "%llu", (unsigned long long)ch_ids[i]);
if (topo_node_sqlite_member_in_channel(dm->db, ch, peer_id)) { shared = 1; break; }
}
if (ch_ids) u_free(ch_ids);
return shared;
}
/* Доступен ли пир для прямой доставки: живой ETCP-линк или присутствие в BGP-группе. */
static int dm_peer_reachable(struct dm_state* dm, uint64_t peer_id) {
struct ETCP_CONN* conn = instance_find_conn(dm->inst, peer_id);
if (conn && conn->links_up && conn->initialized) return 1;
if (dm->inst->topo_groups && dm->inst->topo_groups->group_list) {
struct ll_entry* e = dm->inst->topo_groups->group_list->head;
while (e) {
struct TOPO_GROUP* g = (struct TOPO_GROUP*)e;
if (g->group_type == TOPO_GROUP_TYPE_CHAT && topo_node_find_by_id(g, peer_id)) return 1;
e = e->next;
}
}
return 0;
}
/* Отправить payload подкомандой subcmd в dst через etcp_router (mode=0, своё E2E). */
static int dm_route_send(struct dm_state* dm, uint64_t group_id, uint64_t dst, uint8_t subcmd,
const uint8_t* body, size_t body_len) {
struct ll_entry* entry = queue_entry_new(0);
if (!entry) return -1;
entry->dgram = u_malloc(1 + 1 + body_len);
if (!entry->dgram) { queue_entry_free(entry); return -1; }
entry->dgram[0] = ETCP_RT_ID_DM;
entry->dgram[1] = subcmd;
if (body_len) memcpy(entry->dgram + 2, body, body_len);
entry->len = (uint16_t)(2 + body_len);
int rc = etcp_route_send(dm->inst, group_id, dst, entry, 1, 0);
if (rc != 0) { queue_dgram_free(entry); queue_entry_free(entry); return -1; }
return 0;
}
/* ── работа с беседой ── */
/* Загрузить беседу из БД в структуру. Возвращает 0/ошибку. */
struct dm_conv {
char conv_id[64];
uint64_t peer_node_id;
uint8_t peer_x25519[32];
uint8_t peer_ed25519[32];
char peer_name[64];
uint64_t group_id;
uint8_t content_key[DM_CONTENT_KEY_SIZE];
uint64_t last_out_seq;
uint64_t last_in_seq;
};
static int dm_conv_load(struct dm_state* dm, const char* conv_id, struct dm_conv* c) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT peer_node_id, peer_x25519, peer_ed25519, COALESCE(peer_name,''),"
" COALESCE(group_id,0), last_out_seq, last_in_seq FROM dm_conversations WHERE conv_id=?",
-1, &st, NULL) != SQLITE_OK) return -1;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
int rc = -1;
if (sqlite3_step(st) == SQLITE_ROW) {
memset(c, 0, sizeof(*c));
snprintf(c->conv_id, sizeof(c->conv_id), "%s", conv_id);
c->peer_node_id = (uint64_t)sqlite3_column_int64(st, 0);
const uint8_t* x = (const uint8_t*)sqlite3_column_blob(st, 1);
const uint8_t* e = (const uint8_t*)sqlite3_column_blob(st, 2);
if (x && sqlite3_column_bytes(st, 1) >= 32) memcpy(c->peer_x25519, x, 32);
if (e && sqlite3_column_bytes(st, 2) >= 32) memcpy(c->peer_ed25519, e, 32);
const char* name = (const char*)sqlite3_column_text(st, 3);
if (name) snprintf(c->peer_name, sizeof(c->peer_name), "%s", name);
c->group_id = (uint64_t)sqlite3_column_int64(st, 4);
c->last_out_seq = (uint64_t)sqlite3_column_int64(st, 5);
c->last_in_seq = (uint64_t)sqlite3_column_int64(st, 6);
dm_derive_content_key(dm->inst->my_keys.private_key, c->peer_x25519, c->content_key);
rc = 0;
}
sqlite3_finalize(st);
return rc;
}
/* Вставить/обновить беседу в БД. Возвращает 0/ошибку. */
static int dm_conv_save(struct dm_state* dm, const struct dm_conv* c) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"INSERT INTO dm_conversations(conv_id,peer_node_id,peer_x25519,peer_ed25519,"
" peer_name,group_id,last_out_seq,last_in_seq,created_at,last_ts)"
" VALUES(?,?,?,?,?,?,?,?,?,0)"
" ON CONFLICT(conv_id) DO UPDATE SET"
" peer_x25519=excluded.peer_x25519, peer_ed25519=excluded.peer_ed25519,"
" peer_name=excluded.peer_name, group_id=excluded.group_id,"
" last_out_seq=excluded.last_out_seq, last_in_seq=excluded.last_in_seq",
-1, &st, NULL) != SQLITE_OK) return -1;
sqlite3_bind_text(st, 1, c->conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)c->peer_node_id);
sqlite3_bind_blob(st, 3, c->peer_x25519, 32, SQLITE_STATIC);
sqlite3_bind_blob(st, 4, c->peer_ed25519, 32, SQLITE_STATIC);
sqlite3_bind_text(st, 5, c->peer_name, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 6, (sqlite3_int64)c->group_id);
sqlite3_bind_int64(st, 7, (sqlite3_int64)c->last_out_seq);
sqlite3_bind_int64(st, 8, (sqlite3_int64)c->last_in_seq);
sqlite3_bind_int64(st, 9, (sqlite3_int64)ntp_time_get_seconds(dm->inst));
int rc = sqlite3_step(st) == SQLITE_DONE ? 0 : -1;
sqlite3_finalize(st);
return rc;
}
/* ── сборка/разбор канонического тела ── */
/* Собрать каноническое тело сообщения (зашифровать + подписать).
* out — буфер (u_malloc), *out_len — длина. dir=1 для исходящего. */
static int dm_build_body(struct dm_state* dm, const struct dm_conv* c, uint64_t seq, uint64_t ts,
const char* ct, const uint8_t* data, uint32_t data_len,
uint8_t** out, size_t* out_len) {
if (!c || !ct || (!data && data_len) || data_len > DM_DATA_MAX) return -1;
uint8_t ct_len = (uint8_t)strnlen(ct, 255);
uint64_t conv_num = (uint64_t)strtoull(c->conv_id, NULL, 10);
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, dm->inst->node_id, seq, nonce);
/* шифротекст = data_len + tag(16) */
uint8_t* enc = u_malloc(data_len + DM_TAG_SIZE);
if (!enc) return -1;
size_t enc_len = 0;
if (dm_encrypt(c->content_key, nonce, data, data_len, enc, &enc_len) != 0) {
u_free(enc);
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: encrypt failed conv=%s", DM_ID, c->conv_id);
return -1;
}
uint64_t author = dm->inst->node_id;
size_t body_len = DM_MSG_FIXED_HDR + ct_len + 2 + enc_len + DM_MSG_SIG_SIZE;
uint8_t* body = u_malloc(body_len);
if (!body) { u_free(enc); return -1; }
size_t off = 0;
memcpy(body + off, &conv_num, 8); off += 8;
memcpy(body + off, &seq, 8); off += 8;
memcpy(body + off, &ts, 8); off += 8;
memcpy(body + off, &author, 8); off += 8;
body[off++] = ct_len;
memcpy(body + off, ct, ct_len); off += ct_len;
uint16_t el = (uint16_t)enc_len;
memcpy(body + off, &el, 2); off += 2;
memcpy(body + off, enc, enc_len); off += enc_len;
/* подпись по всем полям до sig */
sc_ed25519_sign(dm->inst->my_ed25519_privkey, body, off, body + off);
u_free(enc);
*out = body;
*out_len = body_len;
return 0;
}
/* ── обработка входящего сообщения ── */
/* Обработать каноническое тело входящего сообщения (из live-доставки или mailbox):
* проверить подпись, расшифровать, dedup по seq, вставить, отослать ACK, событие GUI. */
static void dm_on_msg(struct dm_state* dm, const uint8_t* msg, size_t mlen) {
if (!msg || mlen < DM_MSG_FIXED_HDR + DM_MSG_SIG_SIZE) return;
uint64_t conv_num; memcpy(&conv_num, msg, 8);
uint64_t seq; memcpy(&seq, msg + 8, 8);
uint64_t ts; memcpy(&ts, msg + 16, 8);
uint64_t author; memcpy(&author, msg + 24, 8);
uint8_t ct_len = msg[32];
const uint8_t* ct = msg + 33;
uint16_t enc_len; memcpy(&enc_len, msg + 33 + ct_len, 2);
const uint8_t* enc = msg + 35 + ct_len;
const uint8_t* sig = msg + mlen - DM_MSG_SIG_SIZE;
if (author == dm->inst->node_id) return; /* своё (эхо) — игнор */
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv_num);
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) {
/* первое сообщение: автоприём при общей группе + совпадение conv_id */
if (dm_derive_conv_id(dm->inst->node_id, author) != conv_num) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: conv_id mismatch from 0x%016llx", DM_ID, (unsigned long long)author);
return;
}
if (!dm_share_group(dm, author)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: no shared group with 0x%016llx — reject DM", DM_ID, (unsigned long long)author);
return;
}
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.peer_node_id = author;
if (dm_node_pubkeys(dm, author, c.peer_x25519, c.peer_ed25519) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: no pubkeys for 0x%016llx", DM_ID, (unsigned long long)author);
return;
}
dm_derive_content_key(dm->inst->my_keys.private_key, c.peer_x25519, c.content_key);
c.group_id = 0;
dm_conv_save(dm, &c);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: auto-created conversation conv=%s peer=0x%016llx",
DM_ID, conv_id, (unsigned long long)author);
}
/* проверка подписи автора по всем полям до sig */
if (sc_ed25519_verify(c.peer_ed25519, msg, mlen - DM_MSG_SIG_SIZE, sig) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid signature conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
return;
}
/* dedup: вставляем только новое */
if (seq <= c.last_in_seq) {
/* уже есть — только подтвердить */
uint8_t ack[16]; memcpy(ack, &conv_num, 8); memcpy(ack + 8, &seq, 8);
dm_route_send(dm, c.group_id, author, DM_SUBCMD_ACK, ack, sizeof(ack));
return;
}
/* расшифровать (nonce детерминирован по conv_id+author+seq) */
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, author, seq, nonce);
uint8_t* plain = u_malloc(enc_len > DM_TAG_SIZE ? enc_len - DM_TAG_SIZE : 0);
if (!plain) return;
size_t plain_len = 0;
if (dm_decrypt(c.content_key, nonce, enc, enc_len, plain, &plain_len) != 0) {
u_free(plain);
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: decrypt failed conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
return;
}
/* вставить в dm_messages (dir=0, входящее) */
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"INSERT OR IGNORE INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig)"
" VALUES(?,0,?,?,?,?,?,?)",
-1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 3, (sqlite3_int64)ts);
sqlite3_bind_int64(st, 4, (sqlite3_int64)author);
sqlite3_bind_text(st, 5, (const char*)ct, ct_len, SQLITE_STATIC);
sqlite3_bind_blob(st, 6, enc, enc_len, SQLITE_STATIC);
sqlite3_bind_blob(st, 7, sig, DM_MSG_SIG_SIZE, SQLITE_STATIC);
sqlite3_step(st);
}
if (st) sqlite3_finalize(st);
c.last_in_seq = seq;
dm_conv_save(dm, &c);
/* подтвердить приём (нужно и для mailbox-доставки → storage удалит по ACK) */
uint8_t ack[16]; memcpy(ack, &conv_num, 8); memcpy(ack + 8, &seq, 8);
dm_route_send(dm, c.group_id, author, DM_SUBCMD_ACK, ack, sizeof(ack));
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: recv conv=%s seq=%llu author=0x%016llx ct=%.*s len=%zu",
DM_ID, conv_id, (unsigned long long)seq, (unsigned long long)author, ct_len, ct, plain_len);
/* событие GUI: [conv_id_len:1][conv_id][author_node_id:8] */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
memcpy(evt + 1 + cl, &author, 8);
chat_event_post(dm->inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
}
u_free(plain);
}
/* ── обработка HELLO (catch-up) ── */
/* Пересобрать каноническое тело из сохранённых полей (для catch-up resend). */
static int dm_rebuild_body(uint64_t conv_num, uint64_t seq, uint64_t ts, uint64_t author,
const char* ct, const uint8_t* enc, int enc_len, const uint8_t sig[64],
uint8_t** out, size_t* out_len) {
uint8_t ctl = (uint8_t)strnlen(ct, 255);
size_t body_len = DM_MSG_FIXED_HDR + ctl + 2 + (size_t)enc_len + DM_MSG_SIG_SIZE;
uint8_t* body = u_malloc(body_len);
if (!body) return -1;
size_t off = 0;
memcpy(body + off, &conv_num, 8); off += 8;
memcpy(body + off, &seq, 8); off += 8;
memcpy(body + off, &ts, 8); off += 8;
memcpy(body + off, &author, 8); off += 8;
body[off++] = ctl;
memcpy(body + off, ct, ctl); off += ctl;
uint16_t el = (uint16_t)enc_len;
memcpy(body + off, &el, 2); off += 2;
memcpy(body + off, enc, enc_len); off += enc_len;
memcpy(body + off, sig, DM_MSG_SIG_SIZE);
*out = body;
*out_len = body_len;
return 0;
}
/* На connect обе стороны шлют HELLO со своими счётчиками; получив HELLO, досылаем
* пиру недостающие ему наши сообщения (его last_in_seq+1 .. наш last_out_seq). */
static void dm_on_hello(struct dm_state* dm, uint64_t conv_num, uint64_t from_peer, uint64_t peer_out_seq, uint64_t peer_in_seq) {
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv_num);
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) return;
if (peer_in_seq < c.last_out_seq) {
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT seq, ts, author, ct, data, sig FROM dm_messages"
" WHERE conv_id=? AND dir=1 AND seq>? ORDER BY seq",
-1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)peer_in_seq);
while (sqlite3_step(st) == SQLITE_ROW) {
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 0);
uint64_t ts = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 2);
const char* ct = (const char*)sqlite3_column_text(st, 3);
const uint8_t* enc = (const uint8_t*)sqlite3_column_blob(st, 4);
int enc_len = sqlite3_column_bytes(st, 4);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(st, 5);
uint8_t* body = NULL; size_t body_len = 0;
if (dm_rebuild_body(conv_num, seq, ts, author, ct, enc, enc_len, sig, &body, &body_len) == 0) {
dm_route_send(dm, c.group_id, from_peer, DM_SUBCMD_MSG, body, body_len);
u_free(body);
}
}
}
if (st) sqlite3_finalize(st);
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: hello from 0x%016llx conv=%s out=%llu in=%llu",
DM_ID, (unsigned long long)from_peer, conv_id,
(unsigned long long)peer_out_seq, (unsigned long long)peer_in_seq);
}
/* ── обработка ACK ── */
/* Получено подтверждение доставки от пира — логируем (надёжность гарантирует etcp_router). */
static void dm_on_ack(uint64_t conv_num, uint64_t seq) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: ack conv=%llu seq=%llu", DM_ID,
(unsigned long long)conv_num, (unsigned long long)seq);
}
/* ── etcp_router recv (0x34) ── */
static void dm_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || !conn || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF + 1) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
struct dm_state* dm = dm_of(conn->instance);
if (!dm || !dm->initialized) { queue_dgram_free(entry); queue_entry_free(entry); return; }
const uint8_t* d = entry->dgram;
uint64_t from_node; memcpy(&from_node, d + ROUTER_SVC_SRC_OFF, 8);
const uint8_t* p = d + ROUTER_SVC_PAYLOAD_OFF;
size_t plen = entry->len - ROUTER_SVC_PAYLOAD_OFF;
uint8_t subcmd = p[0];
if (subcmd == DM_SUBCMD_MSG) {
size_t mlen = 0;
if (dm_msg_len(p + 1, plen - 1, &mlen))
dm_on_msg(dm, p + 1, mlen);
} else if (subcmd == DM_SUBCMD_ACK) {
if (plen >= 1 + 16) {
uint64_t conv_num; memcpy(&conv_num, p + 1, 8);
uint64_t seq; memcpy(&seq, p + 9, 8);
dm_on_ack(conv_num, seq);
}
} else if (subcmd == DM_SUBCMD_HELLO) {
if (plen >= 1 + 24) {
uint64_t conv_num; memcpy(&conv_num, p + 1, 8);
uint64_t out_seq; memcpy(&out_seq, p + 9, 8);
uint64_t in_seq; memcpy(&in_seq, p + 17, 8);
dm_on_hello(dm, conv_num, from_node, out_seq, in_seq);
}
} else {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unknown subcmd=%02x", DM_ID, subcmd);
}
queue_dgram_free(entry); queue_entry_free(entry);
}
/* ── connection status ── */
/* При поднятии соединения с пиром: HELLO (catch-up) + вытянуть mailbox. */
static void dm_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
struct dm_state* dm = dm_of(inst);
if (!conn || status != ETCP_CONN_STATUS_UP || !dm) return;
uint64_t peer = conn->peer_node_id;
if (!peer) return;
/* На подъём ЛЮБОГО соединения: HELLO для беседы с подключившимся пиром +
* PULL mailbox для бесед, чей пир не подключён напрямую (offline-доставка). */
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT conv_id, peer_node_id, last_out_seq, last_in_seq FROM dm_conversations",
-1, &st, NULL) != SQLITE_OK) return;
while (sqlite3_step(st) == SQLITE_ROW) {
const char* conv_id = (const char*)sqlite3_column_text(st, 0);
uint64_t conv_peer = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t out_seq = (uint64_t)sqlite3_column_int64(st, 2);
uint64_t in_seq = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t conv_num = strtoull(conv_id, NULL, 10);
if (conv_peer == peer) {
uint8_t hello[24];
memcpy(hello, &conv_num, 8);
memcpy(hello + 8, &out_seq, 8);
memcpy(hello + 16, &in_seq, 8);
/* group_id берём из строки беседы — для простоты HELLO по 0, маршрутизация глобальная */
dm_route_send(dm, 0, peer, DM_SUBCMD_HELLO, hello, sizeof(hello));
}
/* пир не подключён напрямую — тянем mailbox со storage-узлов */
struct ETCP_CONN* pc = instance_find_conn(inst, conv_peer);
if (!pc || !pc->links_up || !pc->initialized) {
dm_mailbox_pull(inst, conv_peer, in_seq);
}
}
sqlite3_finalize(st);
}
/* ── публичный API ── */
/* Начать беседу с пользователем. Создаёт запись если её нет, выводит ключи,
* обеспечивает прямое соединение (если есть общая группа). */
int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
const uint8_t peer_x25519[32], const uint8_t peer_ed25519[32],
const char* peer_name, const char* source_ch_id) {
if (!inst || !peer_node_id || !peer_x25519 || !peer_ed25519) return -1;
if (peer_node_id == inst->node_id) return -1;
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized) return -1;
uint64_t conv_num = dm_derive_conv_id(inst->node_id, peer_node_id);
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv_num);
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) == 0) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: conversation already exists conv=%s", DM_ID, conv_id);
return 0;
}
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.peer_node_id = peer_node_id;
memcpy(c.peer_x25519, peer_x25519, 32);
memcpy(c.peer_ed25519, peer_ed25519, 32);
if (peer_name) snprintf(c.peer_name, sizeof(c.peer_name), "%s", peer_name);
c.group_id = source_ch_id ? strtoull(source_ch_id, NULL, 10) : 0;
dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key);
if (dm_conv_save(dm, &c) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conv_save failed conv=%s", DM_ID, conv_id);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: started conv=%s peer=0x%016llx name=%s group=%llu",
DM_ID, conv_id, (unsigned long long)peer_node_id, c.peer_name,
(unsigned long long)c.group_id);
/* событие GUI */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl);
}
return 0;
}
/* Отправить сообщение в беседу: зашифровать, подписать, сохранить, доставить. */
int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content_type,
const uint8_t* data, uint32_t data_len) {
if (!inst || !conv_id || !content_type || (!data && data_len)) return -1;
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized) return -1;
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: no conversation %s", DM_ID, conv_id);
return -1;
}
uint64_t seq = c.last_out_seq + 1;
uint64_t ts = (uint64_t)(ntp_time_get_us(inst) / 1000);
uint8_t* body = NULL;
size_t body_len = 0;
if (dm_build_body(dm, &c, seq, ts, content_type, data, data_len, &body, &body_len) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: build body failed", DM_ID);
return -1;
}
/* сохранить локально (dir=1, исходящее) */
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"INSERT INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig)"
" VALUES(?,1,?,?,?,?,?,?)",
-1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 3, (sqlite3_int64)ts);
sqlite3_bind_int64(st, 4, (sqlite3_int64)inst->node_id);
sqlite3_bind_text(st, 5, content_type, -1, SQLITE_STATIC);
/* data/sig уже внутри body: data_enc по смещению 35+ct_len, sig в конце */
uint8_t ctl = (uint8_t)strnlen(content_type, 255);
const uint8_t* enc = body + 35 + ctl;
uint16_t enc_len; memcpy(&enc_len, body + 33 + ctl, 2);
sqlite3_bind_blob(st, 6, enc, enc_len, SQLITE_STATIC);
sqlite3_bind_blob(st, 7, body + body_len - DM_MSG_SIG_SIZE, DM_MSG_SIG_SIZE, SQLITE_STATIC);
sqlite3_step(st);
}
if (st) sqlite3_finalize(st);
c.last_out_seq = seq;
dm_conv_save(dm, &c);
/* доставка: доступен → прямой send, иначе → mailbox */
if (dm_peer_reachable(dm, c.peer_node_id)) {
dm_route_send(dm, c.group_id, c.peer_node_id, DM_SUBCMD_MSG, body, body_len);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: sent direct conv=%s seq=%llu → 0x%016llx",
DM_ID, conv_id, (unsigned long long)seq, (unsigned long long)c.peer_node_id);
} else {
dm_mailbox_put(inst, c.peer_node_id, inst->node_id, body, body_len);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: peer offline conv=%s seq=%llu → mailbox",
DM_ID, conv_id, (unsigned long long)seq);
}
u_free(body);
return 0;
}
/* Вставка сообщения, полученного через mailbox (вызывается dm_mailbox по PULL_RESP).
* Обрабатывается тем же путём, что и live-сообщение (verify/decrypt/dedup/insert). */
void dm_core_inject_message(const uint8_t* msg, size_t len, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
struct dm_state* dm = dm_of(inst);
if (!dm) return;
dm_on_msg(dm, msg, len);
}
/* ── жизненный цикл ── */
int dm_core_init(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
if (inst->dm && inst->dm->initialized) return 0;
struct dm_state* dm = u_calloc(1, sizeof(*dm));
if (!dm) return -1;
inst->dm = dm;
dm->inst = inst;
dm->db = chat_core_get_db(inst);
if (!dm->db) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: no chat db", DM_ID); u_free(dm); inst->dm = NULL; return -1; }
dm_create_tables(dm);
if (etcp_router_bind(inst, ETCP_RT_ID_DM, dm_recv_cb) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: etcp_router_bind failed", DM_ID);
u_free(dm); inst->dm = NULL; return -1;
}
etcp_add_conn_status_cbk(inst, dm_on_conn_status, inst);
dm_mailbox_set_deliver_cb(inst, dm_core_inject_message, inst);
dm->initialized = 1;
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: initialized", DM_ID);
return 0;
}
void dm_core_destroy(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->dm || !inst->dm->initialized) return;
etcp_remove_conn_status_cbk(inst, dm_on_conn_status, inst);
etcp_router_unbind(inst, ETCP_RT_ID_DM);
u_free(inst->dm);
inst->dm = NULL;
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: destroyed", DM_ID);
}
/* ── JSON для GUI/headless ── */
/* Список бесед в JSON. */
int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf_size, size_t* out_len) {
struct dm_state* dm = dm_of(inst);
if (!buf || !out_len || !dm || !dm->db) return -1;
int off = 0;
off += snprintf(buf + off, buf_size - (size_t)off, "[");
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT conv_id, peer_node_id, COALESCE(peer_name,''), last_ts, last_out_seq, last_in_seq"
" FROM dm_conversations ORDER BY last_ts DESC",
-1, &st, NULL) == SQLITE_OK) {
int first = 1;
while (sqlite3_step(st) == SQLITE_ROW) {
const char* conv_id = (const char*)sqlite3_column_text(st, 0);
uint64_t peer = (uint64_t)sqlite3_column_int64(st, 1);
const char* name = (const char*)sqlite3_column_text(st, 2);
uint64_t last_ts = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t out_seq = (uint64_t)sqlite3_column_int64(st, 4);
uint64_t in_seq = (uint64_t)sqlite3_column_int64(st, 5);
off += snprintf(buf + off, buf_size - (size_t)off,
"%s{\"conv_id\":\"%s\",\"peer_node_id\":%llu,\"peer_name\":\"%s\","
"\"last_ts\":%llu,\"out_seq\":%llu,\"in_seq\":%llu}",
first ? "" : ",", conv_id, (unsigned long long)peer,
name ? name : "", (unsigned long long)last_ts,
(unsigned long long)out_seq, (unsigned long long)in_seq);
first = 0;
}
}
if (st) sqlite3_finalize(st);
off += snprintf(buf + off, buf_size - (size_t)off, "]");
*out_len = (size_t)off;
return 0;
}
/* Сообщения беседы (расшифрованные) в JSON. */
int dm_list_messages_json(struct UTUN_INSTANCE* inst, const char* conv_id, int count, int offset,
char* buf, size_t buf_size, size_t* out_len) {
struct dm_state* dm = dm_of(inst);
if (!conv_id || !buf || !out_len || !dm || !dm->db) return -1;
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) return -1;
char sql[256];
if (count > 0)
snprintf(sql, sizeof(sql),
"SELECT dir,seq,ts,author,ct,data FROM dm_messages WHERE conv_id=? ORDER BY seq LIMIT %d OFFSET %d",
count, offset);
else
snprintf(sql, sizeof(sql),
"SELECT dir,seq,ts,author,ct,data FROM dm_messages WHERE conv_id=? ORDER BY seq");
int off = 0;
off += snprintf(buf + off, buf_size - (size_t)off, "[");
uint64_t conv_num = (uint64_t)strtoull(conv_id, NULL, 10);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
int first = 1;
while (sqlite3_step(st) == SQLITE_ROW) {
int dir = sqlite3_column_int(st, 0);
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t ts = (uint64_t)sqlite3_column_int64(st, 2);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 3);
const char* ct = (const char*)sqlite3_column_text(st, 4);
const uint8_t* enc = (const uint8_t*)sqlite3_column_blob(st, 5);
int enc_len = sqlite3_column_bytes(st, 5);
/* расшифровать (nonce детерминирован по conv_id+author+seq) */
char text[DM_DATA_MAX + 1];
text[0] = '\0';
if (enc && enc_len >= (int)DM_TAG_SIZE) {
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, author, seq, nonce);
uint8_t* plain = u_malloc((size_t)enc_len);
size_t plen = 0;
if (dm_decrypt(c.content_key, nonce, enc, (size_t)enc_len, plain, &plen) == 0) {
if (plen >= sizeof(text)) plen = sizeof(text) - 1;
memcpy(text, plain, plen);
text[plen] = '\0';
}
u_free(plain);
}
off += snprintf(buf + off, buf_size - (size_t)off,
"%s{\"dir\":%d,\"seq\":%llu,\"ts\":%llu,\"author\":%llu,\"ct\":\"%s\",\"data\":\"%s\"}",
first ? "" : ",", dir, (unsigned long long)seq, (unsigned long long)ts,
(unsigned long long)author, ct ? ct : "", text);
first = 0;
}
}
if (st) sqlite3_finalize(st);
off += snprintf(buf + off, buf_size - (size_t)off, "]");
*out_len = (size_t)off;
return 0;
}