1281 lines
63 KiB
C
1281 lines
63 KiB
C
/*
|
||
* 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_history.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 (media_ccm_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;
|
||
int deleted = dm_history_deleted(inst, msg, len, 0, NULL);
|
||
if (deleted < 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);
|
||
if (deleted) goto receipt;
|
||
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 (media_ccm_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 ||
|
||
(inserted && dm_history_show(inst, conv_id)) || dm_exec(dm, "COMMIT") != 0) {
|
||
dm_exec(dm, "ROLLBACK");
|
||
return -1;
|
||
}
|
||
receipt:
|
||
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;
|
||
}
|
||
if (deleted) return 0;
|
||
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) {
|
||
uint64_t generation = dm_history_generation(inst, conv_id);
|
||
memcpy(evt + 1 + cl, &generation, 8);
|
||
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl + 8);
|
||
memcpy(evt + 1 + cl, &author, 8);
|
||
}
|
||
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);
|
||
}
|
||
|
||
if (dm_history_show(inst, conv_id)) return -1;
|
||
|
||
/* событие GUI (всегда, чтобы GUI мог открыть/выделить беседу) */
|
||
{
|
||
size_t cl = strlen(conv_id);
|
||
uint8_t evt[1 + 64 + 8];
|
||
evt[0] = (uint8_t)cl;
|
||
memcpy(evt + 1, conv_id, cl);
|
||
uint64_t generation = dm_history_generation(inst, conv_id);
|
||
memcpy(evt + 1 + cl, &generation, 8);
|
||
chat_event_post(inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl + 8);
|
||
}
|
||
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;
|
||
}
|
||
if (!dm_history_current(inst, conv_id, dm_history_generation(inst, conv_id))) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM send: hidden conversation conv=%s", 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 || dm_history_init(inst)) 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 c.conv_id,peer_node_id,COALESCE(peer_name,''),last_ts,last_out_seq,last_in_seq,COALESCE(h.generation,1)"
|
||
" FROM dm_conversations c LEFT JOIN dm_history h USING(conv_id) WHERE COALESCE(h.hidden,0)=0"
|
||
" 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,\"generation\":%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,
|
||
(unsigned long long)sqlite3_column_int64(st, 6)) < 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 ||
|
||
media_ccm_decrypt(c.content_key, nonce, enc, (size_t)enc_len, (uint8_t*)text, &plen)) goto fail;
|
||
text[plen] = '\0';
|
||
const char* group_name = "";
|
||
int invite = ct && !strcmp(ct, "invite");
|
||
if (invite) {
|
||
char* separator = memchr(text, '\0', plen);
|
||
if (!separator || (size_t)(separator - text) + 8 > plen ||
|
||
strncmp(separator + 1, "utun://", 7) || memchr(separator + 1, '\0', plen - (separator + 1 - text))) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: malformed payload conv=%s seq=%llu bytes=%zu", conv_id,
|
||
(unsigned long long)seq, plen);
|
||
goto fail;
|
||
}
|
||
group_name = text;
|
||
}
|
||
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_group[DM_DATA_MAX * 2 + 8], esc_path[2048];
|
||
json_escape(group_name, esc_group, sizeof(esc_group));
|
||
const char* display_data = invite ? text + strlen(group_name) + 1 : text;
|
||
json_escape(ct ? ct : "", esc_ct, sizeof(esc_ct)); json_escape(display_data, 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,"
|
||
"\"tags\":{\"group_name\":\"%s\"},"
|
||
"\"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), esc_group,
|
||
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';
|
||
uint64_t generation = req->generation ? req->generation : dm_history_generation(inst, conv);
|
||
int rc = dm_history_current(inst, conv, generation) ? dm_send(inst, conv, "text", req->data, req->data_len) : -1;
|
||
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[36];
|
||
memcpy(event, &req->request_id, 8); memcpy(event + 8, &status, 4);
|
||
memcpy(event + 12, &req->peer_node_id, 8); memcpy(event + 20, &conv, 8);
|
||
char id[64]; snprintf(id, sizeof(id), "%llu", (unsigned long long)conv);
|
||
uint64_t generation = dm_history_generation(req->inst, id); memcpy(event + 28, &generation, 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));
|
||
if (!dm_history_current(req->inst, conv, req->generation)) { dm_invite_done(req, DM_INVITE_CANCELLED); return; }
|
||
size_t bytes = strlen(link);
|
||
sqlite3_stmt* st = NULL;
|
||
char channel[64], payload[DM_DATA_MAX];
|
||
snprintf(channel, sizeof(channel), "%llu", (unsigned long long)req->channel_id);
|
||
struct dm_state* dm = dm_of(req->inst);
|
||
int rc = sqlite3_prepare_v2(dm->db, "SELECT name FROM channels WHERE channel_id=?", -1, &st, NULL);
|
||
if (rc == SQLITE_OK) { sqlite3_bind_text(st, 1, channel, -1, SQLITE_STATIC); rc = sqlite3_step(st); }
|
||
size_t name_bytes = rc == SQLITE_ROW ? (size_t)sqlite3_column_bytes(st, 0) : 0;
|
||
if (rc != SQLITE_ROW || name_bytes + 1 + bytes > sizeof(payload)) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_DM, "invite PM: group metadata failed ch=%s rc=%d name_bytes=%zu link_bytes=%zu sql=%s",
|
||
channel, rc, name_bytes, bytes, sqlite3_errmsg(dm->db));
|
||
sqlite3_finalize(st); dm_invite_done(req, DM_INVITE_SEND_ERROR); return;
|
||
}
|
||
memcpy(payload, sqlite3_column_text(st, 0), name_bytes); payload[name_bytes] = '\0';
|
||
memcpy(payload + name_bytes + 1, link, bytes);
|
||
sqlite3_finalize(st);
|
||
DEBUG_INFO(DEBUG_CATEGORY_DM, "invite PM: registered request=%llu peer=%llu ch=%llu name_bytes=%zu link_bytes=%zu ttl=%ds",
|
||
(unsigned long long)req->request_id, (unsigned long long)req->peer_node_id,
|
||
(unsigned long long)req->channel_id, name_bytes, bytes, JOIN_KEY_TTL_SECONDS);
|
||
rc = dm_send(req->inst, conv, "invite", (const uint8_t*)payload, (uint32_t)(name_bytes + 1 + 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;
|
||
}
|
||
}
|
||
if (dm_history_show(req->inst, conv)) { dm_invite_done(req, DM_INVITE_START_ERROR); return; }
|
||
req->generation = dm_history_generation(req->inst, conv);
|
||
if (!dm_history_current(req->inst, conv, req->generation)) { dm_invite_done(req, DM_INVITE_CANCELLED); 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);
|
||
}
|
||
|
||
void dm_cancel_pending_history(struct UTUN_INSTANCE* inst, const char* conv) {
|
||
struct dm_state* dm = dm_of(inst);
|
||
if (dm && dm->invite && dm_derive_conv_id(inst->node_id, dm->invite->peer_node_id) == strtoull(conv, NULL, 10))
|
||
chat_invite_build_cancel(dm->invite->pending);
|
||
}
|
||
|
||
void dm_delete_trampoline(void* arg) {
|
||
struct dm_delete_req* req = arg;
|
||
if (!req) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM delete: missing request"); return; }
|
||
int success = dm_history_delete(req->inst, req->conv_id, req->generation) == 0;
|
||
uint64_t conv = strtoull(req->conv_id, NULL, 10);
|
||
uint8_t event[25];
|
||
memcpy(event, &req->request_id, 8); memcpy(event + 8, &conv, 8); memcpy(event + 16, &req->generation, 8); event[24] = success;
|
||
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM delete result: request=%llu conv=%s generation=%llu success=%d",
|
||
(unsigned long long)req->request_id, req->conv_id, (unsigned long long)req->generation, success);
|
||
chat_event_post(req->inst, CHAT_EVT_DM_DELETE_RESULT, event, sizeof(event));
|
||
u_free(req);
|
||
}
|
||
|
||
/* Единая 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_TRACE(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);
|
||
}
|