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

1202 lines
58 KiB
Raw Blame History

/*
* 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 "../chat/invite_build.h"
#include "../chat/chat_join.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 "../../lib/strbuf.h"
#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>
#include <openssl/sha.h>
#define DM_ID "dm_core"
/* Подкоманды протокола DM (первый байт payload после ROUTER_SVC-заголовка). */
#define DM_SUBCMD_MSG 0x01 /* [msg] — сообщение */
#define DM_SUBCMD_ACK 0x02 /* Подписанная квитанция получателя. */
/* Per-instance состояние модуля. */
struct dm_state {
struct UTUN_INSTANCE* inst;
sqlite3* db;
int initialized;
void* timer;
struct dm_invite_req* invite;
};
/* ── вспомогательные ── */
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 int dm_create_tables(struct dm_state* dm) {
if (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)") != 0) return -1;
if (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))") != 0) return -1;
return dm_exec(dm, "CREATE TABLE IF NOT EXISTS dm_outbox (conv_id TEXT NOT NULL, seq INTEGER NOT NULL,"
" body BLOB NOT NULL, retry_at INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(conv_id,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) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: node keys query failed: %s", DM_ID, sqlite3_errmsg(dm->db));
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);
if (rc) DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: missing node keys peer=%llu", DM_ID, (unsigned long long)node_id);
return rc;
}
/* Общий CHAT-канал с пиром. Возвращает 1 и кладёт group_id (== channel_id) в *out_group_id. */
static int dm_shared_group_id(struct dm_state* dm, uint64_t peer_id, uint64_t* out_group_id) {
if (!dm || !out_group_id) return 0;
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)) {
*out_group_id = ch_ids[i];
shared = 1;
break;
}
}
if (ch_ids) u_free(ch_ids);
return shared;
}
/* Выбрать общую область с действительно живым маршрутом, включая BGP-транзит. */
static int dm_find_route_group(struct UTUN_INSTANCE* inst, uint64_t peer, uint64_t preferred,
uint64_t* group_id, int bgp_only) {
if (!inst || !inst->topo_sqlite_db || !group_id) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid route lookup", DM_ID);
return 0;
}
uint64_t* ids = NULL;
int count = 0;
if (topo_node_sqlite_get_member_channels(inst->topo_sqlite_db, inst->node_id, &ids, &count) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: route membership lookup failed", DM_ID);
return 0;
}
int found = 0;
for (int i = -1; i < count; i++) {
uint64_t gid = i < 0 ? preferred : ids[i];
if (!gid) continue;
char ch[64];
snprintf(ch, sizeof(ch), "%llu", (unsigned long long)gid);
if (!topo_node_sqlite_member_in_channel(inst->topo_sqlite_db, ch, inst->node_id) ||
!topo_node_sqlite_member_in_channel(inst->topo_sqlite_db, ch, peer)) continue;
if (bgp_only) {
struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, gid);
if (!topo_group_find_conn_for_node(g, peer)) continue;
} else {
struct ETCP_ROUTER_PATH path = etcp_router_get_path(inst, gid, peer, ETCP_RT_ID_DM);
if (!path.next_hop_node_id) continue;
}
*group_id = gid;
found = 1;
break;
}
u_free(ids);
return found;
}
int dm_route_group(struct UTUN_INSTANCE* inst, uint64_t peer, uint64_t preferred, uint64_t* group_id) {
return dm_find_route_group(inst, peer, preferred, group_id, 0);
}
/* Router владеет entry после вызова, в том числе при ошибке. */
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) {
if (!etcp_router_conn_get(dm->inst, group_id, dst, ETCP_RT_ID_DM)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: router channel allocation failed peer=%llu", DM_ID, (unsigned long long)dst);
return -1;
}
if (!etcp_router_send_q_has_room(dm->inst, group_id, dst, ETCP_RT_ID_DM)) {
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: send deferred by backpressure peer=%llu", DM_ID, (unsigned long long)dst);
return -1;
}
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: packet allocation failed", DM_ID); return -1; }
entry->dgram = u_malloc(2 + body_len);
if (!entry->dgram) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: payload allocation failed", DM_ID);
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, ROUTER_FLAG_SIGNED);
if (rc) DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: route send failed peer=%llu cmd=%u rc=%d",
DM_ID, (unsigned long long)dst, subcmd, rc);
return rc;
}
/* ── работа с беседой ── */
/* Загрузить беседу из БД в структуру. Возвращает 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;
uint64_t last_ts;
};
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, last_ts FROM dm_conversations WHERE conv_id=?",
-1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation query failed: %s", DM_ID, sqlite3_errmsg(dm->db));
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 || !e || sqlite3_column_bytes(st, 1) != 32 || sqlite3_column_bytes(st, 2) != 32) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: damaged conversation keys conv=%s", DM_ID, conv_id);
sqlite3_finalize(st);
return -1;
}
memcpy(c->peer_x25519, x, 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);
c->last_ts = (uint64_t)sqlite3_column_int64(st, 7);
/* Старые беседы могли быть авто-созданы с group_id=0 (маршрутизация только напрямую) —
* подтягиваем общую группу, чтобы DM шёл через неё (indirect/BGP). */
if (c->group_id == 0) {
uint64_t gid = 0;
if (dm_shared_group_id(dm, c->peer_node_id, &gid)) c->group_id = gid;
}
rc = dm_derive_content_key(dm->inst->my_keys.private_key, c->peer_x25519, c->content_key);
}
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(?,?,?,?,?,?,?,?,?,?)"
" 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, last_ts=excluded.last_ts",
-1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation save prepare failed: %s", DM_ID, sqlite3_errmsg(dm->db));
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));
sqlite3_bind_int64(st, 10, (sqlite3_int64)c->last_ts);
int rc = sqlite3_step(st) == SQLITE_DONE ? 0 : -1;
if (rc) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation save failed conv=%s: %s", DM_ID, c->conv_id, sqlite3_errmsg(dm->db));
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 || strlen(ct) > 255) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid message size=%u", DM_ID, data_len);
return -1;
}
uint8_t ct_len = (uint8_t)strnlen(ct, 255);
const uint8_t* media = NULL;
if (!strcmp(ct, "file")) {
if (data_len <= DM_MEDIA_DESC_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: file message lacks descriptor/name", DM_ID);
return -1;
}
media = data;
data += DM_MEDIA_DESC_SIZE;
data_len -= DM_MEDIA_DESC_SIZE;
struct attachment_info info;
if (attachment_decode(data, data_len, &info)) return -1;
}
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 + (media ? DM_MEDIA_DESC_SIZE : 0) + 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;
if (media) { memcpy(body + off, media, DM_MEDIA_DESC_SIZE); off += DM_MEDIA_DESC_SIZE; }
/* подпись по всем полям до sig */
if (sc_ed25519_sign(dm->inst->my_ed25519_privkey, body, off, body + off) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: message signing failed", DM_ID);
u_free(enc);
u_free(body);
return -1;
}
u_free(enc);
*out = body;
*out_len = body_len;
return 0;
}
/* Проверить тело перед любой записью: подпись связывает conv, автора и адресата. */
int dm_verify_message(struct UTUN_INSTANCE* inst, uint64_t recipient, const uint8_t* msg, size_t len) {
struct dm_state view = { .inst = inst, .db = inst ? inst->topo_sqlite_db : NULL };
size_t actual = 0;
uint64_t conv, seq, author;
if (!view.db || !dm_msg_len(msg, len, &actual) || actual != len) goto invalid;
memcpy(&conv, msg, 8);
memcpy(&seq, msg + 8, 8);
memcpy(&author, msg + 24, 8);
if (!recipient || !author || author == recipient || !seq || seq > INT64_MAX ||
conv != dm_derive_conv_id(author, recipient)) goto invalid;
const uint8_t* media = dm_message_media(msg, len);
if (media) {
uint64_t size;
memcpy(&size, media + 16, 8);
if (size > DM_MEDIA_MAX_SIZE) goto invalid;
}
uint8_t x[32], ed[32];
if (dm_node_pubkeys(&view, author, x, ed) != 0 ||
sc_ed25519_verify(ed, msg, len - DM_MSG_SIG_SIZE, msg + len - DM_MSG_SIG_SIZE) != SC_OK) goto invalid;
return 0;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: rejected message recipient=%llu bytes=%zu", DM_ID,
(unsigned long long)recipient, len);
return -1;
}
/* Сохранить тело; точный повтор допустим, другое тело с тем же seq запрещено. Транзакция у вызывающего. */
static int dm_store_message(struct dm_state* dm, const char* conv, int dir, const uint8_t* msg, size_t len) {
uint64_t seq, ts, author;
memcpy(&seq, msg + 8, 8);
memcpy(&ts, msg + 16, 8);
memcpy(&author, msg + 24, 8);
uint16_t enc_len;
uint8_t ctl = msg[32];
memcpy(&enc_len, msg + 33 + ctl, 2);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db,
"INSERT INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig) VALUES(?,?,?,?,?,?,?,?)"
" ON CONFLICT(conv_id,dir,seq) DO NOTHING", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
sqlite3_bind_int(st, 2, dir);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
sqlite3_bind_int64(st, 4, (sqlite3_int64)ts);
sqlite3_bind_int64(st, 5, (sqlite3_int64)author);
sqlite3_bind_text(st, 6, (const char*)msg + 33, ctl, SQLITE_STATIC);
sqlite3_bind_blob(st, 7, msg + 35 + ctl, enc_len, SQLITE_STATIC);
sqlite3_bind_blob(st, 8, msg + len - DM_MSG_SIG_SIZE, DM_MSG_SIG_SIZE, SQLITE_STATIC);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
int inserted = sqlite3_changes(dm->db);
sqlite3_finalize(st);
st = NULL;
if (inserted) return 1;
rc = sqlite3_prepare_v2(dm->db, "SELECT sig FROM dm_messages WHERE conv_id=? AND dir=? AND seq=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
sqlite3_bind_int(st, 2, dir);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
rc = sqlite3_step(st);
if (rc != SQLITE_ROW || sqlite3_column_bytes(st, 0) != DM_MSG_SIG_SIZE) goto fail;
if (memcmp(sqlite3_column_blob(st, 0), msg + len - DM_MSG_SIG_SIZE, DM_MSG_SIG_SIZE)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: conflicting body for saved message conv=%s dir=%d seq=%llu", DM_ID,
conv, dir, (unsigned long long)seq);
sqlite3_finalize(st);
return -1;
}
sqlite3_finalize(st);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: message write/conflict conv=%s dir=%d seq=%llu rc=%d sql=%s",
DM_ID, conv, dir, (unsigned long long)seq, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
return -1;
}
/* Принять сообщение и создать подписанную квитанцию только после успешного commit. */
int dm_accept_message(struct UTUN_INSTANCE* inst, const uint8_t* msg, size_t len, uint8_t receipt[DM_RECEIPT_SIZE]) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !receipt) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receive without initialized owner", DM_ID);
return -1;
}
if (dm_verify_message(inst, inst->node_id, msg, len) != 0) return -1;
uint64_t conv, seq, author;
memcpy(&conv, msg, 8);
memcpy(&seq, msg + 8, 8);
memcpy(&author, msg + 24, 8);
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv);
struct dm_conv c;
int created = dm_conv_load(dm, conv_id, &c) != 0;
if (created) {
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.peer_node_id = author;
if (!dm_shared_group_id(dm, author, &c.group_id)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unknown correspondent peer=%llu", DM_ID, (unsigned long long)author);
return -1;
}
if (dm_node_pubkeys(dm, author, c.peer_x25519, c.peer_ed25519) != 0 ||
dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key) != 0) return -1;
chat_core_get_node_name(inst, author, c.peer_name, sizeof(c.peer_name));
}
if (c.peer_node_id != author) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation author mismatch conv=%s", DM_ID, conv_id);
return -1;
}
uint16_t enc_len;
memcpy(&enc_len, msg + 33 + msg[32], 2);
uint8_t nonce[DM_NONCE_SIZE], plain[DM_DATA_MAX + 1];
size_t plain_len = 0;
dm_build_nonce(conv, author, seq, nonce);
if (dm_decrypt(c.content_key, nonce, msg + 35 + msg[32], enc_len, plain, &plain_len) != 0) return -1;
if (dm_message_media(msg, len)) {
struct attachment_info info;
if (attachment_decode(plain, plain_len, &info)) return -1;
}
if (dm_exec(dm, "BEGIN IMMEDIATE") != 0) return -1;
int inserted = dm_store_message(dm, conv_id, 0, msg, len);
if (seq > c.last_in_seq) c.last_in_seq = seq; /* Только статистика; не курсор dedup. */
uint64_t ts;
memcpy(&ts, msg + 16, 8);
if (ts > c.last_ts) c.last_ts = ts;
if (inserted < 0 || dm_conv_save(dm, &c) != 0 || dm_media_track(inst, 0, msg, len) != 0 || dm_exec(dm, "COMMIT") != 0) {
dm_exec(dm, "ROLLBACK");
return -1;
}
memcpy(receipt, "DMACK001", 8);
memcpy(receipt + 8, msg, 16);
memcpy(receipt + 24, &author, 8);
memcpy(receipt + 32, &inst->node_id, 8);
SHA256(msg, len, receipt + 40);
if (sc_ed25519_sign(inst->my_ed25519_privkey, receipt, 72, receipt + 72) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt signing failed conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: committed receive conv=%s seq=%llu new=%d plain=%zu", DM_ID,
conv_id, (unsigned long long)seq, inserted, plain_len);
if (inserted) {
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);
if (created) chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl);
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: received conv=%s seq=%llu author=%llu", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)author);
}
return 0;
}
/* Квитанцию может пересылать суперузел, но подписывает её только получатель. */
int dm_verify_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE], const uint8_t* msg, size_t len) {
struct dm_state view = { .inst = inst, .db = inst ? inst->topo_sqlite_db : NULL };
uint64_t conv, seq, author, recipient;
if (!view.db || !receipt || memcmp(receipt, "DMACK001", 8)) goto invalid;
memcpy(&conv, receipt + 8, 8);
memcpy(&seq, receipt + 16, 8);
memcpy(&author, receipt + 24, 8);
memcpy(&recipient, receipt + 32, 8);
if (!author || !recipient || author == recipient || !seq || seq > INT64_MAX ||
conv != dm_derive_conv_id(author, recipient)) goto invalid;
uint8_t x[32], ed[32];
if (dm_node_pubkeys(&view, recipient, x, ed) != 0 ||
sc_ed25519_verify(ed, receipt, 72, receipt + 72) != SC_OK) goto invalid;
if (msg) {
uint8_t digest[32];
size_t actual = 0;
if (!dm_msg_len(msg, len, &actual) || actual != len || memcmp(receipt + 8, msg, 16) ||
memcmp(receipt + 24, msg + 24, 8)) goto invalid;
SHA256(msg, len, digest);
if (memcmp(digest, receipt + 40, 32)) goto invalid;
}
return 0;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid delivery receipt bytes=%zu", DM_ID, len);
return -1;
}
/* Удалить исходящую задачу по точной подписанной квитанции; историю беседы оставить. */
int dm_accept_receipt(struct UTUN_INSTANCE* inst, const uint8_t receipt[DM_RECEIPT_SIZE]) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || dm_verify_receipt(inst, receipt, NULL, 0) != 0) return -1;
uint64_t author, conv, seq;
memcpy(&author, receipt + 24, 8);
memcpy(&conv, receipt + 8, 8);
memcpy(&seq, receipt + 16, 8);
if (author != inst->node_id) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receipt for another sender=%llu", DM_ID, (unsigned long long)author);
return -1;
}
char conv_id[64];
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv);
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db, "SELECT body FROM dm_outbox WHERE conv_id=? AND seq=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); return 0; }
if (rc != SQLITE_ROW) goto fail;
if (dm_verify_receipt(inst, receipt, sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0)) != 0) {
sqlite3_finalize(st);
return -1;
}
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(dm->db, "DELETE FROM dm_outbox WHERE conv_id=? AND seq=?", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
rc = sqlite3_step(st);
if (rc != SQLITE_DONE) goto fail;
sqlite3_finalize(st);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: delivery confirmed conv=%s seq=%llu", DM_ID, conv_id, (unsigned long long)seq);
dm_media_notify(inst, conv);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: receipt database error rc=%d sql=%s", DM_ID, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
return -1;
}
/* Ограниченная порция durable outbox; retry_at обеспечивает продвижение всей очереди. */
static void dm_pump(struct dm_state* dm) {
int64_t now = (int64_t)ntp_time_get_seconds(dm->inst);
for (int i = 0; i < 32; i++) {
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(dm->db,
"SELECT o.conv_id,o.seq,o.body,c.peer_node_id,c.group_id FROM dm_outbox o"
" JOIN dm_conversations c USING(conv_id) WHERE o.retry_at<=? ORDER BY o.retry_at,o.seq LIMIT 1", -1, &st, NULL);
if (rc != SQLITE_OK) goto fail;
sqlite3_bind_int64(st, 1, now);
rc = sqlite3_step(st);
if (rc == SQLITE_DONE) { sqlite3_finalize(st); break; }
if (rc != SQLITE_ROW) goto fail;
char conv[64];
snprintf(conv, sizeof(conv), "%s", (const char*)sqlite3_column_text(st, 0));
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 1);
size_t len = (size_t)sqlite3_column_bytes(st, 2);
uint8_t* body = u_malloc(len);
if (!body) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: outbox allocation failed bytes=%zu", DM_ID, len);
sqlite3_finalize(st);
break;
}
memcpy(body, sqlite3_column_blob(st, 2), len);
uint64_t peer = (uint64_t)sqlite3_column_int64(st, 3), gid = (uint64_t)sqlite3_column_int64(st, 4);
sqlite3_finalize(st);
st = NULL;
rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_outbox SET retry_at=? WHERE conv_id=? AND seq=?", -1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_int64(st, 1, now + 5);
sqlite3_bind_text(st, 2, conv, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 3, (sqlite3_int64)seq);
rc = sqlite3_step(st);
}
sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_DONE) { u_free(body); goto fail; }
if (dm_route_group(dm->inst, peer, gid, &gid)) {
if (!dm_route_send(dm, gid, peer, DM_SUBCMD_MSG, body, len))
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: outbox direct conv=%s seq=%llu group=%llu", DM_ID,
conv, (unsigned long long)seq, (unsigned long long)gid);
} else {
dm_mailbox_put(dm->inst, peer, dm->inst->node_id, body, len);
}
u_free(body);
continue;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: outbox scan failed rc=%d sql=%s", DM_ID, rc, sqlite3_errmsg(dm->db));
sqlite3_finalize(st);
break;
}
}
/* Durable-задачи переживают stop; таймер является только локальным триггером. */
static void dm_tick(void* arg) {
struct dm_state* dm = arg;
dm->timer = NULL;
dm_pump(dm);
dm->timer = uasync_set_timeout(dm->inst->ua, 10000, dm, dm_tick, "dm_outbox");
if (!dm->timer) DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: retry timer allocation failed", DM_ID);
}
/* Live-доставка использует тот же commit/receipt, что и доставка с суперузла. */
static void dm_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || !conn || !entry->dgram || entry->len <= ROUTER_SVC_PAYLOAD_OFF) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return; /* Уведомление router о закрытии канала. */
}
struct dm_state* dm = dm_of(conn->instance);
if (!dm || !dm->initialized) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: receive while stopped", DM_ID);
goto done;
}
const uint8_t* p = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF - 1;
uint64_t source, gid;
memcpy(&source, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
memcpy(&gid, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
if (!(entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: unsigned route packet source=%llu", DM_ID, (unsigned long long)source);
goto done;
}
if (p[0] == DM_SUBCMD_MSG) {
uint8_t receipt[DM_RECEIPT_SIZE];
if (dm_accept_message(dm->inst, p + 1, len, receipt) == 0) {
dm_route_send(dm, gid, source, DM_SUBCMD_ACK, receipt, sizeof(receipt));
dm_mailbox_receipt(dm->inst, receipt);
}
} else if (p[0] == DM_SUBCMD_ACK && len == DM_RECEIPT_SIZE) {
if (dm_accept_receipt(dm->inst, p + 1) == 0) dm_mailbox_receipt(dm->inst, p + 1);
} else {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid command=%u bytes=%zu", DM_ID, p[0], len);
}
done:
queue_dgram_free(entry);
queue_entry_free(entry);
}
/* ── публичный 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) {
/* Беседа уже есть, но могла быть авто-создана с group_id=0 — дописываем из source_ch_id,
* чтобы маршрутизация шла через группу (а не только прямое соединение). */
if (c.group_id == 0 && source_ch_id && source_ch_id[0]) {
uint64_t gid = strtoull(source_ch_id, NULL, 10);
if (gid != 0) {
c.group_id = gid;
dm_conv_save(dm, &c);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: conversation exists conv=%s — updated group=%llu",
DM_ID, conv_id, (unsigned long long)gid);
}
} else {
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: conversation already exists conv=%s", DM_ID, conv_id);
}
} else {
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;
if (dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key) != 0) return -1;
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 (всегда, чтобы 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;
}
/* Атомарно сохранить seq, историю и неизменяемое тело исходящей задачи. */
int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content_type,
const uint8_t* data, uint32_t data_len) {
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !conv_id || !content_type || (!data && data_len)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: send invalid arguments/state", DM_ID);
return -1;
}
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0 || c.last_out_seq >= INT64_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: send conversation missing/seq exhausted conv=%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 len = 0;
if (dm_build_body(dm, &c, seq, ts, content_type, data, data_len, &body, &len) != 0) return -1;
sqlite3_stmt* st = NULL;
if (dm_exec(dm, "BEGIN IMMEDIATE") != 0) { u_free(body); return -1; }
c.last_out_seq = seq;
if (ts > c.last_ts) c.last_ts = ts;
if (dm_store_message(dm, conv_id, 1, body, len) != 1 || dm_conv_save(dm, &c) != 0 ||
dm_media_track(inst, 1, body, len) != 0) goto rollback;
int rc = sqlite3_prepare_v2(dm->db, "INSERT INTO dm_outbox(conv_id,seq,body) VALUES(?,?,?)", -1, &st, NULL);
if (rc == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int64(st, 2, (sqlite3_int64)seq);
sqlite3_bind_blob(st, 3, body, (int)len, SQLITE_STATIC);
rc = sqlite3_step(st);
}
sqlite3_finalize(st);
st = NULL;
if (rc != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: outbox insert failed conv=%s seq=%llu rc=%d sql=%s", DM_ID,
conv_id, (unsigned long long)seq, rc, sqlite3_errmsg(dm->db));
goto rollback;
}
if (dm_exec(dm, "COMMIT") != 0) goto rollback;
u_free(body);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: queued conv=%s seq=%llu peer=%llu bytes=%u", DM_ID, conv_id,
(unsigned long long)seq, (unsigned long long)c.peer_node_id, data_len);
dm_pump(dm);
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, &inst->node_id, 8);
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
return 0;
rollback:
dm_exec(dm, "ROLLBACK");
u_free(body);
return -1;
}
/* Таблицы принадлежат DM, соединение с SQLite — общему ядру. */
int dm_core_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: init without instance", DM_ID); return -1; }
if (inst->dm && inst->dm->initialized) return 0;
struct dm_state* dm = u_calloc(1, sizeof(*dm));
if (!dm) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: state allocation failed", DM_ID); return -1; }
inst->dm = dm;
dm->inst = inst;
dm->db = inst->topo_sqlite_db;
if (!dm->db || dm_create_tables(dm) != 0) goto fail;
if (etcp_router_bind(inst, ETCP_RT_ID_DM, dm_recv_cb) != 0) goto fail;
dm->initialized = 1;
dm->timer = uasync_set_timeout(inst->ua, 10000, dm, dm_tick, "dm_outbox");
if (!dm->timer) {
etcp_router_unbind(inst, ETCP_RT_ID_DM);
goto fail;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: initialized persistent outbox", DM_ID);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: initialization failed db=%p", DM_ID, (void*)dm->db);
u_free(dm);
inst->dm = NULL;
return -1;
}
/* Снять триггеры перед освобождением контекста; pending остаётся в БД. */
void dm_core_destroy(struct UTUN_INSTANCE* inst) {
struct dm_state* dm = dm_of(inst);
if (!dm) return;
if (dm->invite) chat_invite_build_cancel(dm->invite->pending);
if (dm->timer) uasync_cancel_timeout(inst->ua, dm->timer);
dm->timer = NULL;
dm->initialized = 0;
etcp_router_unbind(inst, ETCP_RT_ID_DM);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: destroyed, durable outbox preserved", DM_ID);
u_free(dm);
inst->dm = NULL;
}
/* ── JSON для GUI/headless ── */
/* Экранирование строки для вставки в JSON-строку: " \ \n \r \t, прочие <0x20 — drop. */
static void json_escape(const char* src, char* dst, size_t dst_sz) {
size_t i = 0;
if (!src) src = "";
for (; *src && i + 2 < dst_sz; src++) {
unsigned char c = (unsigned char)*src;
switch (c) {
case '"': dst[i++] = '\\'; dst[i++] = '"'; break;
case '\\': dst[i++] = '\\'; dst[i++] = '\\'; break;
case '\n': dst[i++] = '\\'; dst[i++] = 'n'; break;
case '\r': dst[i++] = '\\'; dst[i++] = 'r'; break;
case '\t': dst[i++] = '\\'; dst[i++] = 't'; break;
default:
if (c < 0x20) break;
dst[i++] = (char)c;
}
}
dst[i] = '\0';
}
/* Список бесед в 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;
*out_len = 0;
struct strbuf json = strbuf_new();
int failed = strbuf_addf(&json, "[") < 0;
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 LIMIT 200",
-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);
char resolved[128];
if (!name || !name[0]) {
resolved[0] = '\0';
chat_core_get_node_name(inst, peer, resolved, sizeof(resolved));
name = resolved;
}
char esc_name[512];
json_escape(name ? name : "", esc_name, sizeof(esc_name));
if (strbuf_addf(&json,
"%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,
esc_name, (unsigned long long)last_ts,
(unsigned long long)out_seq, (unsigned long long)in_seq) < 0) { failed = 1; break; }
first = 0;
}
if (sqlite3_errcode(dm->db) != SQLITE_DONE && sqlite3_errcode(dm->db) != SQLITE_OK) failed = 1;
} else failed = 1;
if (st) sqlite3_finalize(st);
if (strbuf_addf(&json, "]") < 0) failed = 1;
if (failed || json.len >= buf_size) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversations snapshot failed bytes=%zu capacity=%zu sql=%s", DM_ID,
json.len, buf_size, sqlite3_errmsg(dm->db));
strbuf_free(&json); return -1;
}
memcpy(buf, strbuf_str(&json), json.len + 1);
*out_len = json.len;
strbuf_free(&json);
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 || offset < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: invalid messages snapshot request", DM_ID);
return -1;
}
*out_len = 0;
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) return -1;
/* Ограниченный снимок не блокирует uasync на длинной истории. */
if (count <= 0 || count > 200) count = 200;
const char* sql = "SELECT * FROM (SELECT m.dir,m.seq,m.ts,m.author,m.ct,m.data,o.seq IS NOT NULL AS pending,"
"COALESCE(f.ready,0) AS ready,COALESCE(f.downloading,0) AS downloading,f.receipt,"
"COALESCE(f.local_path,'') AS path,COALESCE(f.plain_size,0) AS size"
" FROM dm_messages m LEFT JOIN dm_outbox o ON m.dir=1 AND o.conv_id=m.conv_id AND o.seq=m.seq"
" LEFT JOIN dm_files f ON f.conv_id=m.conv_id AND f.dir=m.dir AND f.seq=m.seq"
" WHERE m.conv_id=? ORDER BY m.ts DESC,m.seq DESC,m.dir DESC LIMIT ? OFFSET ?) ORDER BY ts,seq,dir";
struct strbuf sb = strbuf_new();
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 || strbuf_addf(&sb, "[") < 0) goto fail;
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
sqlite3_bind_int(st, 2, count); sqlite3_bind_int(st, 3, offset);
int first = 1, rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
int dir = sqlite3_column_int(st, 0);
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 1), 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 = sqlite3_column_blob(st, 5);
int enc_len = sqlite3_column_bytes(st, 5);
char text[DM_DATA_MAX + 1];
size_t plen = 0;
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, author, seq, nonce);
if (enc_len < DM_TAG_SIZE || enc_len > DM_DATA_MAX + DM_TAG_SIZE ||
dm_decrypt(c.content_key, nonce, enc, (size_t)enc_len, (uint8_t*)text, &plen)) goto fail;
text[plen] = '\0';
struct attachment_info attachment = {0};
int file = ct && !strcmp(ct, "file");
if (file) {
if (attachment_decode((const uint8_t*)text, plen, &attachment)) goto fail;
snprintf(text, sizeof(text), "%s", attachment.name);
}
int ready = sqlite3_column_int(st, 7), downloading = sqlite3_column_int(st, 8);
int delivered = sqlite3_column_bytes(st, 9) == DM_MEDIA_RECEIPT_SIZE;
const char* path = ready == 1 ? (const char*)sqlite3_column_text(st, 10) : "";
const char* status = delivered && dir ? "delivered" : ready == 1 ? "ready" : downloading ? "downloading" :
ready < 0 ? "expired" : "pending";
char esc_ct[512], esc_data[DM_DATA_MAX * 2 + 8], esc_path[2048];
json_escape(ct ? ct : "", esc_ct, sizeof(esc_ct)); json_escape(text, esc_data, sizeof(esc_data));
json_escape(path, esc_path, sizeof(esc_path));
if (strbuf_addf(&sb, "%s{\"dir\":%d,\"seq\":%llu,\"ts\":%llu,\"author\":\"%llu\",\"ct\":\"%s\",\"data\":\"%s\","
"\"pending\":%s,\"media_status\":\"%s\",\"media_path\":\"%s\",\"media_size\":%llu,"
"\"content_type\":\"%s\",\"duration_ms\":%u,\"width\":%u,\"height\":%u,\"waveform\":\"%s\"}", first ? "" : ",", dir,
(unsigned long long)seq, (unsigned long long)ts, (unsigned long long)author, esc_ct, esc_data,
sqlite3_column_int(st, 6) ? "true" : "false", status, esc_path,
(unsigned long long)sqlite3_column_int64(st, 11), file ? attachment_content_type(attachment.kind) : "text/plain",
attachment.duration_ms, attachment.width, attachment.height, attachment.waveform) < 0) goto fail;
first = 0;
}
if (rc != SQLITE_DONE || strbuf_addf(&sb, "]") < 0 || sb.len + 1 > buf_size) goto fail;
memcpy(buf, strbuf_str(&sb), sb.len + 1); *out_len = sb.len;
sqlite3_finalize(st); strbuf_free(&sb);
return 0;
fail:
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: message snapshot failed conv=%s buffer=%zu required=%zu sql=%s", DM_ID,
conv_id, buf_size, sb.len + 1, sqlite3_errmsg(dm->db));
sqlite3_finalize(st); strbuf_free(&sb);
return -1;
}
/* ── трамплины для gui_bridge (GUI → uasync) ── */
/* Начать беседу с узлом из общей группы: находим pubkey пира в nodes, зовём dm_start. */
void dm_start_trampoline(void* arg) {
struct dm_start_req* req = (struct dm_start_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
uint64_t peer = req->peer_node_id;
char src[64]; strncpy(src, req->source_ch_id, sizeof(src) - 1); src[sizeof(src) - 1] = '\0';
char name[128]; strncpy(name, req->peer_name, sizeof(name) - 1); name[sizeof(name) - 1] = '\0';
u_free(arg);
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !inst || !peer) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — not initialized", DM_ID);
return;
}
if (peer == inst->node_id) return;
uint8_t x25519[32], ed25519[32];
if (dm_node_pubkeys(dm, peer, x25519, ed25519) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — no pubkeys for peer=0x%016llx",
DM_ID, (unsigned long long)peer);
return;
}
if (dm_start(inst, peer, x25519, ed25519, name[0] ? name : NULL, src[0] ? src : NULL) != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — dm_start failed peer=0x%016llx",
DM_ID, (unsigned long long)peer);
}
/* Отправить текстовое сообщение в беседу. */
void dm_send_trampoline(void* arg) {
struct dm_send_req* req = (struct dm_send_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
char conv[64]; strncpy(conv, req->conv_id, sizeof(conv) - 1); conv[sizeof(conv) - 1] = '\0';
int rc = dm_send(inst, conv, "text", req->data, req->data_len);
u_free(arg);
if (rc != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_send_trampoline — dm_send failed conv=%s",
DM_ID, conv);
}
/* Завершить операцию и вернуть результат с неизменяемым адресатом. */
static void dm_invite_done(struct dm_invite_req* req, int result) {
struct dm_state* dm = dm_of(req->inst);
if (dm && dm->invite == req) dm->invite = NULL;
uint64_t conv = dm_derive_conv_id(req->inst->node_id, req->peer_node_id);
int32_t status = result;
uint8_t event[28];
memcpy(event, &req->request_id, 8); memcpy(event + 8, &status, 4);
memcpy(event + 12, &req->peer_node_id, 8); memcpy(event + 20, &conv, 8);
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: finished request=%llu peer=%llu ch=%llu result=%d",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id,
(unsigned long long)req->channel_id, result);
chat_event_post(req->inst, CHAT_EVT_DM_INVITE_RESULT, event, sizeof(event));
u_free(req);
}
/* ACK регистрации разрешает отправку ссылки; ошибка/отмена не создаёт сообщение. */
static void dm_invite_ready(void* arg, int result, const char* link) {
struct dm_invite_req* req = arg;
req->pending = NULL;
if (result == CHAT_JOIN_CANCELLED) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
if (result != CHAT_JOIN_OK || !link || !link[0]) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: registration failed request=%llu ch=%llu result=%d",
(unsigned long long)req->request_id, (unsigned long long)req->channel_id, result);
dm_invite_done(req, DM_INVITE_BUILD_ERROR); return;
}
char conv[64];
snprintf(conv, sizeof(conv), "%llu", (unsigned long long)dm_derive_conv_id(req->inst->node_id, req->peer_node_id));
size_t bytes = strlen(link);
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: registered request=%llu peer=%llu ch=%llu bytes=%zu ttl=%ds",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id,
(unsigned long long)req->channel_id, bytes, JOIN_KEY_TTL_SECONDS);
int rc = dm_send(req->inst, conv, "text", (const uint8_t*)link, (uint32_t)bytes);
if (rc) DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: send failed request=%llu conv=%s rc=%d",
(unsigned long long)req->request_id, conv, rc);
dm_invite_done(req, rc ? DM_INVITE_SEND_ERROR : DM_INVITE_OK);
}
/* Создать/найти PM, затем зарегистрировать свежий invite; все данные принадлежат запросу. */
void dm_invite_trampoline(void* arg) {
struct dm_invite_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: missing request"); return; }
struct dm_state* dm = dm_of(req->inst);
if (req->cancel) {
if (dm && dm->invite && dm->invite->request_id == req->request_id) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: cancelling request=%llu", (unsigned long long)req->request_id);
chat_invite_build_cancel(dm->invite->pending);
} else DEBUG_DEBUG(DEBUG_CATEGORY_DM, "invite PM: cancel already completed request=%llu",
(unsigned long long)req->request_id);
u_free(req); return;
}
if (!req->inst) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: missing instance"); u_free(req); return; }
if (!dm || !dm->initialized || !req->request_id || !req->peer_node_id || req->peer_node_id == req->inst->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: invalid state/recipient request=%llu peer=%llu",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id);
dm_invite_done(req, DM_INVITE_START_ERROR); return;
}
if (dm->invite) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "invite PM: busy request=%llu pending=%llu",
(unsigned long long)req->request_id, (unsigned long long)dm->invite->request_id);
dm_invite_done(req, DM_INVITE_BUSY); return;
}
char channel[64], conv[64];
snprintf(channel, sizeof(channel), "%llu", (unsigned long long)req->channel_id);
struct TOPO_GROUP* group = topo_groups_find(req->inst->topo_groups, req->channel_id);
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT ||
!topo_node_sqlite_member_in_channel(dm->db, channel, req->inst->node_id)) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: no active membership ch=%s", channel);
dm_invite_done(req, DM_INVITE_BUILD_ERROR); return;
}
snprintf(conv, sizeof(conv), "%llu", (unsigned long long)dm_derive_conv_id(req->inst->node_id, req->peer_node_id));
struct dm_conv c;
if (dm_conv_load(dm, conv, &c) != 0) {
uint8_t x25519[32], ed25519[32];
if (dm_node_pubkeys(dm, req->peer_node_id, x25519, ed25519) != 0 ||
dm_start(req->inst, req->peer_node_id, x25519, ed25519, req->peer_name, req->source_ch_id) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: conversation failed peer=%llu", (unsigned long long)req->peer_node_id);
dm_invite_done(req, DM_INVITE_START_ERROR); return;
}
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: building request=%llu peer=%llu source=%s ch=%s conv=%s",
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id, req->source_ch_id, channel, conv);
dm->invite = req;
req->pending = chat_invite_build_link(req->inst, req->channel_id, 0, NULL, dm_invite_ready, req);
if (!req->pending) dm_invite_ready(req, CHAT_JOIN_ERROR, NULL);
}
/* Единая BGP-проверка для кнопки PM и запуска звонка; только uasync-поток. */
int dm_call_route(struct UTUN_INSTANCE* inst, const char* conv_id, uint64_t* peer, uint64_t* group_id) {
struct dm_state* dm = dm_of(inst);
if (!peer || !group_id || !conv_id || !dm || !dm->db) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "PM route: invalid request or stopped service"); return 0;
}
*peer = 0; *group_id = 0;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db, "SELECT peer_node_id,group_id FROM dm_conversations WHERE conv_id=?",
-1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "PM route: query failed: %s", sqlite3_errmsg(dm->db)); return 0;
}
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
int rc = sqlite3_step(st);
uint64_t preferred = 0;
if (rc == SQLITE_ROW) {
*peer = (uint64_t)sqlite3_column_int64(st, 0);
preferred = (uint64_t)sqlite3_column_int64(st, 1);
} else if (rc == SQLITE_DONE) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "PM route: conversation missing conv=%s", conv_id);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "PM route: read failed conv=%s: %s", conv_id, sqlite3_errmsg(dm->db));
}
sqlite3_finalize(st);
int available = *peer && *peer != inst->node_id && dm_find_route_group(inst, *peer, preferred, group_id, 1);
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "PM route: conv=%s peer=%llu preferred=%llu group=%llu available=%d",
conv_id, (unsigned long long)*peer, (unsigned long long)preferred, (unsigned long long)*group_id, available);
return available;
}
void dm_call_route_trampoline(void* arg) {
struct dm_messages_req* req = arg;
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "PM route: empty request"); return; }
uint64_t peer = 0, group = 0;
dm_call_route(req->inst, req->conv_id, &peer, &group);
size_t len = strlen(req->conv_id);
uint8_t evt[1 + 63 + 16];
evt[0] = (uint8_t)len;
memcpy(evt + 1, req->conv_id, len);
memcpy(evt + 1 + len, &peer, 8);
memcpy(evt + 9 + len, &group, 8);
chat_event_post(req->inst, CHAT_EVT_DM_CALL_ROUTE, evt, (int)(17 + len));
u_free(req);
}
/* Ограниченный JSON-снимок доставляется GUI асинхронно, без доступа UI к instance/SQLite. */
void dm_messages_trampoline(void* arg) {
struct dm_messages_req* req = arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
char conv[64];
snprintf(conv, sizeof(conv), "%s", req->conv_id);
u_free(req);
if (!dm_of(inst)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: snapshot after stop", DM_ID); return; }
size_t cl = strlen(conv), cap = 1024 * 1024, len = 0;
uint8_t* evt = u_malloc(1 + cl + cap);
if (!evt) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: snapshot allocation failed", DM_ID); return; }
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv, cl);
if (!dm_list_messages_json(inst, conv, 200, 0, (char*)evt + 1 + cl, cap, &len))
chat_event_post(inst, CHAT_EVT_DM_MESSAGES, evt, (int)(1 + cl + len));
u_free(evt);
}
void dm_conversations_trampoline(void* arg) {
struct dm_conversations_req* req = arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
u_free(req);
if (!dm_of(inst)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: conversations snapshot after stop", DM_ID); return; }
size_t cap = 256 * 1024, len = 0;
char* json = u_malloc(cap);
if (!json) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversations allocation failed", DM_ID); return; }
if (!dm_list_conversations_json(inst, json, cap, &len))
chat_event_post(inst, CHAT_EVT_DM_CONVERSATIONS, (const uint8_t*)json, (int)len);
u_free(json);
}