/*
 * 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_schema.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>
#include <openssl/crypto.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;
}

/* Прочитать оба 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;
}

/* ── работа с беседой ── */

/* Полный локальный снимок беседы с выведенным content_key. */
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[DM_PEER_NAME_MAX + 1];
    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;
    enum dm_mode mode;
    struct dm_party parties[2]; /* Локальная сторона, собеседник; неизменны до удаления беседы. */
    uint8_t stored_key[DM_KEY_WRAP_SIZE]; /* Content key, защищённый ключом конкретного устройства. */
    int stored_key_valid, closed;
};

enum dm_conv_result { DM_CONV_ERROR = -1, DM_CONV_FOUND = 0, DM_CONV_NOT_FOUND = 1 };

/* Привязка локальной упаковки к полной беседе защищает от перестановки строк/ключей. */
static size_t dm_local_key_info(const struct dm_conv* c, uint8_t info[192]) {
    const char domain[] = "utun-dm-local-history-v1"; size_t pos = sizeof(domain);
    uint64_t conv = strtoull(c->conv_id, NULL, 10);
    memcpy(info, domain, pos);
    for (unsigned i = 0; i < 8; i++) info[pos++] = (uint8_t)(conv >> (8 * i));
    if (dm_party_encode(c->parties, info + pos) || dm_party_encode(c->parties + 1, info + pos + DM_PARTY_WIRE_SIZE)) return 0;
    return pos + 2 * DM_PARTY_WIRE_SIZE;
}

/* Архив читаем, но новые операции требуют текущей принадлежности закреплённой стороне. */
static int dm_conv_active(struct dm_state* dm, const struct dm_conv* c) {
    if (!c->closed && (c->mode == DM_MODE_DEVICE || dm_party_member(dm->inst, c->parties, dm->inst->node_id) == 1)) return 1;
    DEBUG_WARN(DEBUG_CATEGORY_DM, "DM conversation is archived/inactive conv=%s mode=%d", c->conv_id, c->mode);
    return 0;
}

/* Копировать целый UTF-8 префикс; NUL внутри SQL-строки и битый суффикс не попадают в имя/JSON. */
static void dm_copy_peer_name(char out[DM_PEER_NAME_MAX + 1], const char* name, size_t len, const char* conv) {
    size_t size = 0;
    while (size < len && size < DM_PEER_NAME_MAX) {
        unsigned c = (unsigned char)name[size];
        if (c && c < 0x80) { size++; continue; }
        unsigned count, scalar, minimum;
        if (c >= 0xc2 && c <= 0xdf) { count = 1; scalar = c & 31; minimum = 0x80; }
        else if (c >= 0xe0 && c <= 0xef) { count = 2; scalar = c & 15; minimum = 0x800; }
        else if (c >= 0xf0 && c <= 0xf4) { count = 3; scalar = c & 7; minimum = 0x10000; }
        else goto invalid;
        if (count > len - size - 1) goto invalid;
        for (unsigned i = 1; i <= count; i++) {
            c = (unsigned char)name[size + i];
            if ((c & 0xc0) != 0x80) goto invalid;
            scalar = (scalar << 6) | (c & 63);
        }
        if (scalar < minimum || scalar > 0x10ffff || (scalar >= 0xd800 && scalar <= 0xdfff)) goto invalid;
        if (count + 1 > DM_PEER_NAME_MAX - size) break;
        size += count + 1;
    }
    if (size < len) DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: peer name shortened conv=%s bytes=%zu->%zu", DM_ID, conv, len, size);
    goto done;
invalid:
    DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: invalid peer name encoding conv=%s offset=%zu bytes=%zu; suffix omitted",
               DM_ID, conv, size, len);
done:
    if (size) memcpy(out, name, size);
    out[size] = '\0';
}

/* FOUND выдаёт полную беседу; NOT_FOUND — только SQLITE_DONE; ERROR не меняет *c и БД. */
static enum dm_conv_result dm_conv_load(struct dm_state* dm, const char* conv_id, struct dm_conv* c) {
    sqlite3_stmt* st = NULL;
    int rc = 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,mode,local_party,peer_party,local_key,closed"
            " FROM dm_conversations WHERE conv_id=?",
            -1, &st, NULL);
    if (rc != SQLITE_OK) goto fail;
    rc = sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
    if (rc != SQLITE_OK) goto fail;
    rc = sqlite3_step(st);
    if (rc == SQLITE_DONE) {
        rc = sqlite3_finalize(st);
        st = NULL;
        if (rc != SQLITE_OK) goto fail;
        DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: conversation not found conv=%s", DM_ID, conv_id);
        return DM_CONV_NOT_FOUND;
    }
    if (rc != SQLITE_ROW) goto fail;
    struct dm_conv loaded = {0};
    snprintf(loaded.conv_id, sizeof(loaded.conv_id), "%s", conv_id);
    loaded.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 DM_CONV_ERROR;
    }
    memcpy(loaded.peer_x25519, x, 32);
    memcpy(loaded.peer_ed25519, e, 32);
    const char* name = (const char*)sqlite3_column_text(st, 3);
    dm_copy_peer_name(loaded.peer_name, name, (size_t)sqlite3_column_bytes(st, 3), conv_id);
    loaded.group_id = (uint64_t)sqlite3_column_int64(st, 4);
    loaded.last_out_seq = (uint64_t)sqlite3_column_int64(st, 5);
    loaded.last_in_seq = (uint64_t)sqlite3_column_int64(st, 6);
    loaded.last_ts = (uint64_t)sqlite3_column_int64(st, 7);
    loaded.mode = sqlite3_column_int(st, 8); loaded.closed = sqlite3_column_int(st, 12);
    if (loaded.mode == DM_MODE_GROUPS) {
        if (sqlite3_column_bytes(st, 9) != DM_PARTY_WIRE_SIZE || sqlite3_column_bytes(st, 10) != DM_PARTY_WIRE_SIZE ||
            sqlite3_column_bytes(st, 11) != DM_KEY_WRAP_SIZE || dm_party_decode(sqlite3_column_blob(st, 9), loaded.parties) ||
            dm_party_decode(sqlite3_column_blob(st, 10), loaded.parties + 1) ||
            dm_identity_conv_id(loaded.mode, loaded.parties) != strtoull(conv_id, NULL, 10)) {
            DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM conversation: invalid pinned identity conv=%s", conv_id);
            sqlite3_finalize(st); return DM_CONV_ERROR;
        }
        memcpy(loaded.stored_key, sqlite3_column_blob(st, 11), DM_KEY_WRAP_SIZE); loaded.stored_key_valid = 1;
    } else if (loaded.mode == DM_MODE_DEVICE) {
        loaded.parties[0].node = dm->inst->node_id; loaded.parties[1].node = loaded.peer_node_id;
        memcpy(loaded.parties[0].x, dm->inst->my_keys.public_key, 32); memcpy(loaded.parties[0].ed, dm->inst->my_ed25519_pubkey, 32);
        memcpy(loaded.parties[1].x, loaded.peer_x25519, 32); memcpy(loaded.parties[1].ed, loaded.peer_ed25519, 32);
    } else {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM conversation: unknown mode=%d conv=%s", loaded.mode, conv_id);
        sqlite3_finalize(st); return DM_CONV_ERROR;
    }
    /* Если область ещё не выбрана, используем известную общую группу. */
    if (loaded.group_id == 0) {
        uint64_t gid = 0;
        if (dm_shared_group_id(dm, loaded.peer_node_id, &gid)) loaded.group_id = gid;
    }
    rc = sqlite3_finalize(st);
    st = NULL;
    if (rc != SQLITE_OK) goto fail;
    int key_error;
    if (loaded.mode == DM_MODE_GROUPS) {
        uint8_t info[192]; size_t info_size = dm_local_key_info(&loaded, info);
        key_error = !info_size || dm_key_open(dm->inst->my_keys.private_key, info, info_size, loaded.stored_key, loaded.content_key);
    } else key_error = dm_derive_content_key(dm->inst->my_keys.private_key, loaded.peer_x25519, loaded.content_key);
    if (key_error) {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation key derivation failed conv=%s", DM_ID, conv_id);
        return DM_CONV_ERROR;
    }
    *c = loaded;
    DEBUG_TRACE(DEBUG_CATEGORY_DM, "%s: conversation loaded conv=%s out=%llu in=%llu", DM_ID, conv_id,
                (unsigned long long)c->last_out_seq, (unsigned long long)c->last_in_seq);
    return DM_CONV_FOUND;
fail:
    DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation load failed conv=%s rc=%d sql=%s", DM_ID, conv_id, rc, sqlite3_errmsg(dm->db));
    sqlite3_finalize(st);
    return DM_CONV_ERROR;
}

int dm_conversation_access(struct UTUN_INSTANCE* inst, const char* conv, struct dm_party parties[2], uint8_t key[32]) {
    struct dm_state* dm = dm_of(inst);
    struct dm_conv c;
    if (!dm || !conv) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM access: invalid context"); return -1; }
    if (dm_conv_load(dm, conv, &c) != DM_CONV_FOUND) return -1;
    int active = dm_conv_active(dm, &c);
    if (active && parties) memcpy(parties, c.parties, sizeof(c.parties));
    if (active && key) memcpy(key, c.content_key, 32);
    OPENSSL_cleanse(c.content_key, sizeof(c.content_key));
    return active ? 0 : -1;
}

int dm_group_leave_prepare(struct UTUN_INSTANCE* inst, const uint8_t group_ed[32]) {
    struct dm_state* dm = dm_of(inst);
    if (!dm) return 0;
    if (!group_ed || sqlite3_get_autocommit(dm->db)) {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM leave: transaction/group required"); return -1;
    }
    sqlite3_stmt* st = NULL;
    int rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_conversations SET closed=1 WHERE mode=1"
        " AND substr(local_party,1,1)=x'01' AND substr(local_party,10,32)=?", -1, &st, NULL);
    if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 1, group_ed, 32, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    int changed = sqlite3_changes(dm->db), final = sqlite3_finalize(st);
    if (rc != SQLITE_DONE || final != SQLITE_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM leave: close failed rc=%d final=%d: %s", rc, final, sqlite3_errmsg(dm->db)); return -1;
    }
    if (dm_exec(dm, "DELETE FROM dm_outbox WHERE conv_id IN (SELECT conv_id FROM dm_conversations WHERE closed=1)")) return -1;
    DEBUG_INFO(DEBUG_CATEGORY_DM, "DM leave: archived conversations=%d, queued sends stopped", changed);
    return 0;
}

/* Вставить/обновить беседу в БД. Возвращает 0/ошибку. */
static int dm_conv_save(struct dm_state* dm, const struct dm_conv* c) {
    sqlite3_stmt* st = NULL;
    uint8_t parties[2][DM_PARTY_WIRE_SIZE], stored[DM_KEY_WRAP_SIZE];
    if (c->mode == DM_MODE_GROUPS) {
        if (dm_party_encode(c->parties, parties[0]) || dm_party_encode(c->parties + 1, parties[1])) return -1;
        if (c->stored_key_valid) memcpy(stored, c->stored_key, sizeof(stored));
        else {
            uint8_t info[192]; size_t size = dm_local_key_info(c, info);
            if (!size || dm_key_seal(dm->inst->my_keys.public_key, info, size, c->content_key, stored)) return -1;
        }
    }
    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,mode,local_party,peer_party,local_key,closed)"
            " 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,"
            " mode=excluded.mode,local_party=excluded.local_party,peer_party=excluded.peer_party,local_key=excluded.local_key,closed=excluded.closed",
            -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);
    sqlite3_bind_int(st, 11, c->mode);
    if (c->mode == DM_MODE_GROUPS) {
        sqlite3_bind_blob(st, 12, parties[0], DM_PARTY_WIRE_SIZE, SQLITE_STATIC);
        sqlite3_bind_blob(st, 13, parties[1], DM_PARTY_WIRE_SIZE, SQLITE_STATIC);
        sqlite3_bind_blob(st, 14, stored, DM_KEY_WRAP_SIZE, SQLITE_STATIC);
    }
    sqlite3_bind_int(st, 15, c->closed);
    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;
    uint8_t context[DM_CONTEXT_MAX]; size_t context_size = 0;
    if (c->mode == DM_MODE_GROUPS && dm_context_encode(dm->inst, c->parties, context, sizeof(context), &context_size)) {
        u_free(enc); return -1;
    }
    size_t body_len = DM_MSG_FIXED_HDR + ct_len + 2 + enc_len + (media ? DM_MEDIA_DESC_SIZE : 0) + DM_MSG_SIG_SIZE;
    body_len += context_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 | (context_size ? DM_CONTEXT_FLAG : 0));
    memcpy(body + off, &el, 2);       off += 2;
    if (context_size) { memcpy(body + off, context, context_size); off += context_size; }
    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 };
    struct dm_message_view message;
    if (!view.db || dm_message_parse(msg, len, &message) || message.size != len) goto invalid;
    uint64_t conv = message.conv, seq = message.seq, author = message.author;
    if (!recipient || !author || author == recipient || !seq || seq > INT64_MAX) goto invalid;
    uint8_t x[32], ed[32];
    if (message.mode == DM_MODE_GROUPS) {
        if (dm_message_verify_targets(&message) || dm_message_target(&message, recipient, NULL, NULL) < 0 ||
            dm_message_target(&message, author, x, ed) != 0) goto invalid;
    } else if (conv != dm_derive_conv_id(author, recipient) || dm_node_pubkeys(&view, author, x, ed)) goto invalid;
    const uint8_t* media = message.media;
    if (media) {
        uint64_t size;
        memcpy(&size, media + 16, 8);
        if (size > DM_MEDIA_MAX_SIZE) goto invalid;
    }
    if (sc_ed25519_verify(ed, msg, len - DM_MSG_SIG_SIZE, message.signature) != 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) {
    struct dm_message_view view;
    if (dm_message_parse(msg, len, &view) || view.size != len) return -1;
    uint64_t seq = view.seq, ts = view.timestamp, author = view.author;
    sqlite3_stmt* st = NULL;
    int rc = sqlite3_prepare_v2(dm->db,
        "INSERT INTO dm_messages(conv_id,dir,seq,ts,author,ct,data,sig,body) VALUES(?,?,?,?,?,?,?,?,?)"
        " ON CONFLICT(conv_id,author,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*)view.content_type, view.content_type_size, SQLITE_STATIC);
    sqlite3_bind_blob(st, 7, view.ciphertext, view.ciphertext_size, SQLITE_STATIC);
    sqlite3_bind_blob(st, 8, view.signature, DM_MSG_SIG_SIZE, SQLITE_STATIC);
    sqlite3_bind_blob(st, 9, msg, (int)len, 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,body FROM dm_messages WHERE conv_id=? AND author=? AND seq=?", -1, &st, NULL);
    if (rc != SQLITE_OK) goto fail;
    sqlite3_bind_text(st, 1, conv, -1, SQLITE_STATIC);
    sqlite3_bind_int64(st, 2, (sqlite3_int64)author);
    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), view.signature, DM_MSG_SIG_SIZE) ||
        (sqlite3_column_type(st, 1) != SQLITE_NULL &&
         (sqlite3_column_bytes(st, 1) != (int)len || memcmp(sqlite3_column_blob(st, 1), msg, len)))) {
        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;
    struct dm_message_view message;
    if (dm_message_parse(msg, len, &message)) return -1;
    int side = message.mode == DM_MODE_GROUPS ? dm_message_target(&message, inst->node_id, NULL, NULL) : 1;
    int dir = side == 0;
    int deleted = dm_history_deleted(inst, msg, len, dir, 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;
    enum dm_conv_result loaded = dm_conv_load(dm, conv_id, &c);
    if (loaded == DM_CONV_ERROR) return -1;
    int created = loaded == DM_CONV_NOT_FOUND;
    if (created) {
        memset(&c, 0, sizeof(c));
        snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
        c.mode = message.mode;
        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 (c.mode == DM_MODE_GROUPS) {
            c.parties[0] = message.parties[side]; c.parties[1] = message.parties[1 - side];
            if (!side) {
                c.peer_node_id = 0;
                for (unsigned i = 0; i < message.target_count; i++) {
                    struct dm_target target;
                    if (dm_message_target_at(&message, i, &target)) return -1;
                    if (target.side == 1) { c.peer_node_id = target.node; break; }
                }
            }
            memcpy(c.peer_x25519, c.parties[1].x, 32); memcpy(c.peer_ed25519, c.parties[1].ed, 32);
            uint8_t private_key[32];
            if (dm_party_private(inst, c.parties, private_key)) return -1;
            int error = dm_derive_content_key(private_key, c.parties[1].x, c.content_key);
            OPENSSL_cleanse(private_key, sizeof(private_key));
            if (error) return -1;
        } else 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;
        char name[128] = {0};
        chat_core_get_node_name(inst, c.peer_node_id, name, sizeof(name));
        dm_copy_peer_name(c.peer_name, name, strlen(name), conv_id);
    }
    if (c.mode != message.mode || (c.mode == DM_MODE_DEVICE && c.peer_node_id != author) ||
        (c.mode == DM_MODE_GROUPS && (!dm_party_equal(c.parties, message.parties + side) ||
                                    !dm_party_equal(c.parties + 1, message.parties + 1 - side)))) {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conversation author mismatch conv=%s", DM_ID, conv_id);
        return -1;
    }
    if (!dm_conv_active(dm, &c)) return -1;
    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, message.ciphertext, message.ciphertext_size, 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, dir, msg, len);
    if (!dir && 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, dir, msg, len) != 0 ||
        (inserted && dm_history_show(inst, conv_id)) || dm_exec(dm, "COMMIT") != 0) {
        dm_exec(dm, "ROLLBACK");
        return -1;
    }
receipt:
    memcpy(receipt, message.mode == DM_MODE_GROUPS ? "DMACKG01" : "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) goto invalid;
    int grouped = !memcmp(receipt, "DMACKG01", 8);
    if (!grouped && 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 || conv > INT64_MAX) goto invalid;
    if (!grouped && 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];
        struct dm_message_view message;
        if (dm_message_parse(msg, len, &message) || message.size != len || memcmp(receipt + 8, msg, 16) ||
            memcmp(receipt + 24, msg + 24, 8)) goto invalid;
        if (grouped != (message.mode == DM_MODE_GROUPS)) goto invalid;
        if (message.mode == DM_MODE_GROUPS) {
            uint8_t target_ed[32];
            if (dm_message_target(&message, recipient, NULL, target_ed) < 0 || memcmp(target_ed, ed, 32)) goto invalid;
        } else if (conv != dm_derive_conv_id(author, recipient)) 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, recipient;
    memcpy(&author, receipt + 24, 8);
    memcpy(&conv, receipt + 8, 8);
    memcpy(&seq, receipt + 16, 8);
    memcpy(&recipient, receipt + 32, 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 transaction = 0;
    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;
    if (dm_exec(dm, "BEGIN IMMEDIATE")) return -1;
    transaction = 1;
    rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_delivery SET receipt=? WHERE conv_id=? AND author=? AND seq=? AND recipient=?", -1, &st, NULL);
    if (rc != SQLITE_OK) goto fail;
    sqlite3_bind_blob(st, 1, receipt, DM_RECEIPT_SIZE, SQLITE_STATIC);
    sqlite3_bind_text(st, 2, conv_id, -1, SQLITE_STATIC);
    sqlite3_bind_int64(st, 3, (sqlite3_int64)author); sqlite3_bind_int64(st, 4, (sqlite3_int64)seq);
    sqlite3_bind_int64(st, 5, (sqlite3_int64)recipient);
    rc = sqlite3_step(st);
    if (rc != SQLITE_DONE || sqlite3_changes(dm->db) != 1) goto fail;
    sqlite3_finalize(st); st = NULL;
    rc = sqlite3_prepare_v2(dm->db, "DELETE FROM dm_outbox WHERE conv_id=? AND seq=? AND NOT EXISTS"
        " (SELECT 1 FROM dm_delivery d WHERE d.conv_id=dm_outbox.conv_id AND d.seq=dm_outbox.seq AND d.author=? AND d.receipt IS NULL)",
        -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);
    sqlite3_bind_int64(st, 3, (sqlite3_int64)author);
    rc = sqlite3_step(st);
    if (rc != SQLITE_DONE) goto fail;
    int complete = sqlite3_changes(dm->db);
    sqlite3_finalize(st); st = NULL;
    if (dm_exec(dm, "COMMIT")) goto fail;
    DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: delivery confirmed conv=%s seq=%llu recipient=%llu complete=%d", DM_ID, conv_id,
        (unsigned long long)seq, (unsigned long long)recipient, complete);
    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);
    if (transaction) dm_exec(dm, "ROLLBACK");
    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,d.recipient,c.group_id,d.direct_until FROM dm_outbox o"
            " JOIN dm_conversations c USING(conv_id) JOIN dm_delivery d ON d.conv_id=o.conv_id AND d.seq=o.seq"
            " WHERE d.retry_at<=? AND d.receipt IS NULL AND c.closed=0 ORDER BY d.retry_at,o.seq,d.recipient 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);
        int64_t deadline = sqlite3_column_int64(st, 5);
        if (!deadline) deadline = now + 5;
        sqlite3_finalize(st);
        st = NULL;
        rc = sqlite3_prepare_v2(dm->db, "UPDATE dm_delivery SET retry_at=?,direct_until=?,"
            " first_attempt=CASE WHEN first_attempt=0 THEN ? ELSE first_attempt END WHERE conv_id=? AND seq=? AND recipient=?",
            -1, &st, NULL);
        if (rc == SQLITE_OK) {
            sqlite3_bind_int64(st, 1, now + 5);
            sqlite3_bind_int64(st, 2, deadline); sqlite3_bind_int64(st, 3, now);
            sqlite3_bind_text(st, 4, conv, -1, SQLITE_STATIC);
            sqlite3_bind_int64(st, 5, (sqlite3_int64)seq); sqlite3_bind_int64(st, 6, (sqlite3_int64)peer);
            rc = sqlite3_step(st);
        }
        sqlite3_finalize(st);
        st = NULL;
        if (rc != SQLITE_DONE) { u_free(body); goto fail; }
        int direct = dm_route_group(dm->inst, peer, gid, &gid);
        if (direct) {
            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);
        }
        if (!direct || now >= deadline) {
            DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM delivery: mailbox fallback conv=%s seq=%llu recipient=%llu reason=%s", conv,
                (unsigned long long)seq, (unsigned long long)peer, direct ? "direct ACK deadline" : "no route");
            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, enum dm_mode mode, uint64_t* result_conv) {
    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);
    struct dm_party parties[2] = {{0}};
    if (mode == DM_MODE_GROUPS) {
        if (dm_party_resolve(inst, inst->node_id, mode, parties) || dm_party_resolve(inst, peer_node_id, mode, parties + 1)) return -1;
        conv_num = dm_identity_conv_id(mode, parties);
        if (!conv_num) return -1;
    } else if (mode != DM_MODE_DEVICE) {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM start: invalid mode=%d", mode); return -1;
    }
    char conv_id[64];
    snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv_num);

    struct dm_conv c;
    enum dm_conv_result loaded = dm_conv_load(dm, conv_id, &c);
    if (loaded == DM_CONV_ERROR) return -1;
    if (loaded == DM_CONV_FOUND) {
        if (c.mode != mode || (mode == DM_MODE_DEVICE && c.peer_node_id != peer_node_id) ||
            (mode == DM_MODE_GROUPS && (!dm_party_equal(c.parties, parties) || !dm_party_equal(c.parties + 1, parties + 1)))) {
            DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM start: conversation identity collision conv=%s", conv_id); return -1;
        }
        /* Беседа уже есть, но могла быть авто-создана с 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;
                if (dm_conv_save(dm, &c) != 0) return -1;
                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.mode = mode;
        c.peer_node_id = peer_node_id;
        memcpy(c.peer_x25519, peer_x25519, 32);
        memcpy(c.peer_ed25519, peer_ed25519, 32);
        dm_copy_peer_name(c.peer_name, peer_name, peer_name ? strlen(peer_name) : 0, conv_id);
        c.group_id = source_ch_id ? strtoull(source_ch_id, NULL, 10) : 0;
        if (mode == DM_MODE_GROUPS) {
            memcpy(c.parties, parties, sizeof(parties));
            memcpy(c.peer_x25519, parties[1].x, 32); memcpy(c.peer_ed25519, parties[1].ed, 32);
            uint8_t private_key[32];
            if (dm_party_private(inst, parties, private_key)) return -1;
            int error = dm_derive_content_key(private_key, parties[1].x, c.content_key);
            OPENSSL_cleanse(private_key, sizeof(private_key));
            if (error) return -1;
        } else 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);
    }
    if (result_conv) *result_conv = conv_num;
    return 0;
}

/* Снимок доставки сохраняется только источником, в транзакции отправки. */
static int dm_queue_targets(struct dm_state* dm, const struct dm_conv* c, const uint8_t* body, size_t length) {
    struct dm_message_view view;
    if (dm_message_parse(body, length, &view) || view.size != length || view.author != dm->inst->node_id) {
        DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM queue: invalid source/body conv=%s", c->conv_id); return -1;
    }
    sqlite3_stmt* st = NULL;
    int rc = sqlite3_prepare_v2(dm->db, "INSERT INTO dm_delivery(conv_id,author,seq,recipient) VALUES(?,?,?,?)", -1, &st, NULL);
    if (rc != SQLITE_OK) goto fail;
    unsigned count = view.mode == DM_MODE_GROUPS ? view.target_count : 1, queued = 0;
    for (unsigned i = 0; i < count; i++) {
        struct dm_target target = {.node = c->peer_node_id, .side = 1};
        if (view.mode == DM_MODE_GROUPS && dm_message_target_at(&view, i, &target)) goto fail;
        if (target.node == dm->inst->node_id) continue;
        rc = sqlite3_reset(st);
        if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, c->conv_id, -1, SQLITE_STATIC);
        if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)view.author);
        if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)view.seq);
        if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)target.node);
        if (rc == SQLITE_OK) rc = sqlite3_step(st);
        if (rc != SQLITE_DONE) goto fail;
        queued++;
    }
    rc = sqlite3_finalize(st); st = NULL;
    if (rc != SQLITE_OK || !queued) goto fail;
    DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM queue: frozen delivery list conv=%s seq=%llu recipients=%u", c->conv_id,
        (unsigned long long)view.seq, queued);
    return 0;
fail:
    DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM queue: recipient insert failed conv=%s rc=%d sql=%s", c->conv_id, rc, sqlite3_errmsg(dm->db));
    sqlite3_finalize(st); return -1;
}

/* Атомарно сохранить 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) != DM_CONV_FOUND || 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_conv_active(dm, &c)) 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;
    if (dm_sequence_next(inst, &seq)) return -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_sequence_take(inst, seq) || dm_store_message(dm, conv_id, 1, body, len) != 1 || dm_conv_save(dm, &c) != 0 ||
        dm_media_track(inst, 1, body, len) != 0 || dm_queue_targets(dm, &c, body, len)) 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_schema_init(inst) != 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),mode,closed"
            " 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);
            size_t name_len = (size_t)sqlite3_column_bytes(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_len) {
                resolved[0] = '\0';
                chat_core_get_node_name(inst, peer, resolved, sizeof(resolved));
                name = resolved;
                name_len = strlen(resolved);
            }
            char safe_name[DM_PEER_NAME_MAX + 1];
            dm_copy_peer_name(safe_name, name, name_len, conv_id);
            char esc_name[512];
            json_escape(safe_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,\"mode\":%d,\"closed\":%s}",
                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), sqlite3_column_int(st, 7),
                sqlite3_column_int(st, 8) ? "true" : "false") < 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) != DM_CONV_FOUND) 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.author=?4 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.author=m.author AND f.seq=m.seq"
                      " WHERE m.conv_id=?1 ORDER BY m.ts DESC,m.seq DESC,m.author DESC LIMIT ?2 OFFSET ?3) ORDER BY ts,seq,author";
    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);
    sqlite3_bind_int64(st, 4, (sqlite3_int64)inst->node_id);
    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;
    enum dm_mode mode = req->mode;
    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, mode, 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);
    uint64_t request_id = req->request_id;
    int rc = dm_history_current(inst, conv, generation) ? dm_send(inst, conv, "text", req->data, req->data_len) : -1;
    chat_event_submitted(inst, request_id, rc == 0);
    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;
    enum dm_conv_result loaded = dm_conv_load(dm, conv, &c);
    if (loaded == DM_CONV_ERROR) { dm_invite_done(req, DM_INVITE_START_ERROR); return; }
    if (loaded == DM_CONV_NOT_FOUND) {
        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, DM_MODE_DEVICE, NULL) != 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) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: missing history request", DM_ID); return; }
    struct UTUN_INSTANCE* inst = req->inst;
    char conv[64];
    snprintf(conv, sizeof(conv), "%s", req->conv_id);
    uint64_t generation = req->generation;
    u_free(req);
    if (!dm_of(inst)) { DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: snapshot after stop", DM_ID); return; }
    if (!generation) generation = dm_history_generation(inst, conv);
    if (!dm_history_current(inst, conv, generation)) {
        DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: skipped stale history conv=%s generation=%llu", DM_ID, conv,
                   (unsigned long long)generation); return;
    }
    size_t cl = strlen(conv), cap = 1024 * 1024, len = 0;
    uint8_t* evt = u_malloc(9 + 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);
    memcpy(evt + 1 + cl, &generation, 8);
    if (!dm_list_messages_json(inst, conv, 200, 0, (char*)evt + 9 + cl, cap, &len)) {
        DEBUG_DEBUG(DEBUG_CATEGORY_DM, "%s: history snapshot conv=%s generation=%llu bytes=%zu", DM_ID, conv,
                    (unsigned long long)generation, len);
        chat_event_post(inst, CHAT_EVT_DM_MESSAGES, evt, (int)(9 + 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);
}