/* db_sync — подписанная история SQLite, модель MESSAGES общего Merkle и адресные PUSH/ACK. */
#include "db_sync.h"
#include "merkle_sync.h"
#include "member_sync.h"
#include "chat_feedback.h"
#include "../utun_instance.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/secure_channel.h"
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../../lib/platform_compat.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
#include <openssl/evp.h>
#include <string.h>

#define DB_BATCH_RECORDS 16
#define DB_WIRE_FIXED (32 + 8 + 8 + 16 + 8 + 8 + 4 + DB_SIG_SIZE)
#define DB_WIRE_MAX (DB_WIRE_FIXED + DB_SYNC_MESSAGE_MAX)
#define SI_DB(si) ((si)->db_sync->db)
#define SI_TBL(si) ((si)->table_name)

struct SI_PEER {
    uint64_t node_id;
    uint8_t sync_state, sound_on, paused;
};
struct DB_SYNC_INSTANCE {
    struct DB_SYNC* db_sync;
    uint64_t group_id, next_id, last_timestamp_ms;
    char table_name[64], ns[21];
    uint8_t enabled, sync_gated;
    unsigned holds;
    void* ttl_timer;
    struct {
        uint8_t pending, restart, phase, after_valid, waiting_key;
        sqlite3_int64 after;
        void* timer;
    } index;
    struct SI_PEER* peers;
    int peer_count, peer_capacity;
    db_sync_insert_cb on_insert;
    void* on_insert_arg;
};
struct db_sync_done_cbk_entry {
    db_sync_done_fn fn;
    void* arg;
    uint8_t removed;
    struct db_sync_done_cbk_entry* next;
};
struct db_notice {
    struct db_notice* next;
    struct DB_SYNC* db;
    uint64_t group_id;
    uint8_t initial;
    unsigned count;
    void* work;
    /* Записи и тела лежат непосредственно после этой структуры. */
};
struct DB_SYNC {
    struct UTUN_INSTANCE* inst;
    sqlite3* db;
    uint8_t enabled;
    struct DB_SYNC_INSTANCE** instances;
    int instance_count, instance_capacity;
    void* peer_check_timer;
#ifdef UTUN_HAVE_STANDBY
    void* peer_check_wait; /* Отменяемое ожидание активной фазы Android. */
#endif
    struct db_sync_done_cbk_entry* done_cbks;
    unsigned dispatch_depth;
    struct db_notice* notices;
};
struct db_record {
    sqlite3_int64 rowid;
    uint64_t id, timestamp, author;
    size_t data_len, sig_len;
    uint8_t signature[64], message_id[32];
    struct db_message_revision revision;
    const char* data;
    int cache_valid, updated;
};
struct db_checked {
    struct db_record record;
    int check;
    char data[DB_SYNC_MESSAGE_MAX];
};

static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry);
static void db_sync_on_peer_ready(struct TOPO_GROUP* group, uint64_t peer, int ready, void* arg);
static void db_sync_peer_check_cb(void* arg);
static void db_sync_resume_peer_check(struct DB_SYNC* db);
static void db_index_tick(void* arg);
static void db_index_request(struct DB_SYNC_INSTANCE* si);
static int db_start_peer(struct DB_SYNC_INSTANCE* si, uint64_t peer);
static int db_sync_peer_throttled(struct DB_SYNC* db, struct SI_PEER* peer);
static int db_message_hash(void*, const char*, uint8_t, uint64_t, EVP_MD_CTX*);
static const struct merkle_sync_data_ops db_message_ops;

static uint64_t db_read64(const uint8_t* p) { uint64_t n; memcpy(&n, p, 8); return be64toh(n); }
static void db_write64(uint8_t* p, uint64_t n) { n = htobe64(n); memcpy(p, &n, 8); }
static uint32_t db_read32(const uint8_t* p) { uint32_t n; memcpy(&n, p, 4); return ntohl(n); }
static void db_write32(uint8_t* p, uint32_t n) { n = htonl(n); memcpy(p, &n, 4); }

/* Канонические подписанные байты: домен, канал, постоянный ID, время, автор, nonce, версия, правка, длина и тело. */
static int db_signed_bytes(uint64_t group, const struct db_record* r, uint8_t* out) {
    static const char domain[] = "utun-chat-record-v2";
    if (!r->data || !r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || !r->timestamp || r->timestamp > INT64_MAX ||
        !r->author || !r->revision.ver || r->revision.ver > INT64_MAX || r->revision.edited_at > INT64_MAX ||
        (r->revision.ver == 1 ? r->revision.edited_at != 0 : r->revision.edited_at <= r->timestamp)) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid envelope group=%llu ts=%llu ver=%llu edited=%llu bytes=%zu",
            (unsigned long long)group, (unsigned long long)r->timestamp, (unsigned long long)r->revision.ver,
            (unsigned long long)r->revision.edited_at, r->data_len); return -1;
    }
    size_t off = sizeof(domain) - 1; memcpy(out, domain, off); db_write64(out + off, group); off += 8;
    memcpy(out + off, r->message_id, 32); off += 32;
    db_write64(out + off, r->timestamp); off += 8; db_write64(out + off, r->author); off += 8;
    memcpy(out + off, r->revision.nonce, 16); off += 16;
    db_write64(out + off, r->revision.ver); off += 8; db_write64(out + off, r->revision.edited_at); off += 8;
    db_write32(out + off, (uint32_t)r->data_len); off += 4; memcpy(out + off, r->data, r->data_len);
    return (int)(off + r->data_len);
}

int db_sync_sign_message(uint64_t group, uint64_t timestamp, uint64_t author,
        const struct db_message_revision* revision, const char* data, size_t len,
        const uint8_t private_key[32], uint8_t signature[64]) {
    if (!revision || !private_key || !signature) {
        DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid signing arguments"); return -1;
    }
    struct db_record r = {.timestamp = timestamp, .author = author, .revision = *revision, .data = data, .data_len = len};
    uint8_t body[DB_WIRE_MAX + 64];
    if (chat_message_id(group, timestamp, author, revision->nonce, r.message_id) < 0) return -1;
    int bytes = db_signed_bytes(group, &r, body);
    if (bytes < 0 || sc_ed25519_sign(private_key, body, (size_t)bytes, signature) != SC_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history signing failed group=%llu ver=%llu",
            (unsigned long long)group, (unsigned long long)revision->ver); return -1;
    }
    return 0;
}

static int db_error(struct DB_SYNC_INSTANCE* si, const char* stage) {
    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history tbl=%s stage=%s: %s", SI_TBL(si), stage, sqlite3_errmsg(SI_DB(si)));
    return -1;
}
static void db_rollback(struct DB_SYNC_INSTANCE* si, const char* stage) {
    db_error(si, stage);
    if (sqlite3_exec(SI_DB(si), "ROLLBACK", NULL, NULL, NULL) != SQLITE_OK) db_error(si, "rollback");
}
// Форматирует SQL-запрос, подставляя имя таблицы из инстанса (через %s)
static void si_sql(char* buf, size_t sz, struct DB_SYNC_INSTANCE* si, const char* fmt)
{
    snprintf(buf, sz, fmt, SI_TBL(si));
}

// Готовит SQLite statement с подстановкой имени таблицы — обёртка над si_sql + sqlite3_prepare_v2
static int si_prep(struct DB_SYNC_INSTANCE* si, sqlite3_stmt** stmt, const char* fmt)
{
    char sql[512];
    si_sql(sql, sizeof(sql), si, fmt);
    return sqlite3_prepare_v2(SI_DB(si), sql, -1, stmt, NULL);
}


static struct DB_SYNC_INSTANCE* db_instance_find(struct DB_SYNC* db, uint64_t group) {
    for (int i = 0; db && i < db->instance_count; i++)
        if (db->instances[i]->enabled && db->instances[i]->group_id == group) return db->instances[i];
    return NULL;
}
struct DB_SYNC_INSTANCE* db_sync_instance_find(struct UTUN_INSTANCE* inst, uint64_t group) {
    return db_instance_find(inst ? inst->db_sync : NULL, group);
}
uint64_t db_sync_instance_group_id(struct DB_SYNC_INSTANCE* si) { return si ? si->group_id : 0; }
static void db_release(struct DB_SYNC_INSTANCE* si) {
    if (--si->holds == 0 && !si->enabled) u_free(si);
}
static int si_name_valid(const char* name) {
    if (!name || !*name || strlen(name) > 48) return 0;
    for (const char* p = name; *p; p++)
        if (!((*p >= 'a' && *p <= 'z') || (*p >= 'A' && *p <= 'Z') || (*p >= '0' && *p <= '9') || *p == '_')) return 0;
    return 1;
}
// Ищет пира в инстансе по node_id
static struct SI_PEER* si_peer_find(struct DB_SYNC_INSTANCE* si, uint64_t node_id)
{
    for (int i = 0; i < si->peer_count; i++) {
        if (si->peers[i].node_id == node_id) return &si->peers[i];
    }
    return NULL;
}

// Добавляет пира в инстанс (если уже существует — возвращает существующего)
static struct SI_PEER* si_peer_add(struct DB_SYNC_INSTANCE* si, uint64_t node_id)
{
    struct SI_PEER* p = si_peer_find(si, node_id);
    if (p) return p;
    if (si->peer_count >= si->peer_capacity) {
        int nc = si->peer_capacity ? si->peer_capacity * 2 : 8;
        struct SI_PEER* np = u_realloc(si->peers, nc * sizeof(*si->peers));
        if (!np) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history peer allocation failed group=%016llx", si->group_id); return NULL; }
        si->peers = np;
        si->peer_capacity = nc;
    }
    p = &si->peers[si->peer_count++];
    memset(p, 0, sizeof(*p));
    p->node_id = node_id;
    p->sound_on = 1;
    return p;
}

// Получает Ed25519 публичный ключ узла: сначала из локального, затем из topo_node БД, затем из ETCP-соединения
static int db_get_ed25519_pubkey(struct DB_SYNC* db, uint64_t node_id, uint8_t out[32])
{
    if (node_id == db->inst->node_id) { memcpy(out, db->inst->my_ed25519_pubkey, 32); return 0; }
    /* из in-memory registry (BGP/topo, self-sig верифицирован) — первичный источник */
    if (db->inst->topo_groups) {
        struct TOPO_NODE* ni = topo_node_registry_find(db->inst->topo_groups, node_id);
        if (ni) {
            uint64_t chk; memcpy(&chk, ni->ed25519_public_key, 8);
            if (chk != 0) { memcpy(out, ni->ed25519_public_key, 32); return 0; }
        }
    }
    if (db->inst->topo_sqlite_db) {
        int rc = topo_node_sqlite_get_ed25519_pubkey(db->inst->topo_sqlite_db, node_id, out);
        if (rc == 0) return 0;
        if (rc == TOPO_SQLITE_LOOKUP_ERROR) return -1;
    }
    struct ETCP_CONN* conn = instance_find_conn(db->inst, node_id);
    if (conn) {
        memcpy(out, conn->peer_ed25519_pubkey, 32);
        uint64_t chk; memcpy(&chk, out, 8);
        if (chk != 0) return 0;
    }
    return 1; /* Ключ ещё не получен; это не доказательство порчи. */
}

enum db_record_check { DB_CHECK_ERROR = -1, DB_CHECK_VALID, DB_CHECK_BAD_SIGNATURE, DB_CHECK_BAD_FORMAT, DB_CHECK_DEFERRED };

/* Проверяем конкретные байты. Только BAD_* доказывает их непригодность. */
static enum db_record_check db_verify_author_sig(struct DB_SYNC_INSTANCE* si, const struct db_record* r)
{
    uint8_t msg[DB_WIRE_MAX + 64];
    int bytes = db_signed_bytes(si->group_id, r, msg);
    if (r->sig_len != DB_SIG_SIZE || bytes < 0) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed record tbl=%s ts=%llu body=%zu signature=%zu",
            SI_TBL(si), (unsigned long long)r->timestamp, r->data_len, r->sig_len); return DB_CHECK_BAD_FORMAT;
    }
    uint8_t pubkey[32];
    int key_rc = db_get_ed25519_pubkey(si->db_sync, r->author, pubkey);
    if (key_rc != 0) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history verification unavailable tbl=%s author=%016llx key_status=%d; preserving data",
            SI_TBL(si), (unsigned long long)r->author, key_rc);
        return key_rc < 0 ? DB_CHECK_ERROR : DB_CHECK_DEFERRED;
    }
    int rc = sc_ed25519_verify(pubkey, msg, (size_t)bytes, r->signature);
    if (rc != SC_OK) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history signature check tbl=%s ts=%llu author=%016llx rc=%d result=%s",
            SI_TBL(si), (unsigned long long)r->timestamp, (unsigned long long)r->author, rc,
            rc == SC_ERR_AUTH_FAILED ? "mismatch" : "operational failure; preserving data");
        return rc == SC_ERR_AUTH_FAILED ? DB_CHECK_BAD_SIGNATURE : DB_CHECK_ERROR;
    }
    return DB_CHECK_VALID;
}

/* Вызывается после завершения SELECT и только внутри собственной транзакции. */
static int db_delete_bad_row(struct DB_SYNC_INSTANCE* si, sqlite3_int64 rowid) {
    sqlite3_stmt* stmt = NULL;
    int rc = si_prep(si, &stmt, "DELETE FROM \"%s\" WHERE rowid=?");
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 1, rowid);
    if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
    int changed = sqlite3_changes(SI_DB(si)), final_rc = sqlite3_finalize(stmt);
    if (rc == SQLITE_DONE && final_rc == SQLITE_OK && changed == 1) return 0;
    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history delete tbl=%s rowid=%lld rc=%d finalize=%d changed=%d: %s",
        SI_TBL(si), (long long)rowid, rc, final_rc, changed, sqlite3_errmsg(SI_DB(si))); return -1;
}

// Формат v2: одна строка на постоянный ID, подпись и версия заменяются атомарно.
static int si_table_create(struct DB_SYNC_INSTANCE* si, const char* name, int if_not_exists) {
    char sql[1024];
    snprintf(sql, sizeof(sql),
        "CREATE TABLE %s\"%s\" ("
        " timestamp INTEGER NOT NULL, node_id INTEGER NOT NULL, id INTEGER NOT NULL,"
        " flags INTEGER NOT NULL DEFAULT 0, data BLOB,"
        " author_signature BLOB NOT NULL, message_id BLOB NOT NULL UNIQUE CHECK(length(message_id)=32),"
        " bucket INTEGER NOT NULL, local_attrs TEXT DEFAULT '', delivered_peers INTEGER NOT NULL DEFAULT 0,"
        " delivery_chain TEXT NOT NULL DEFAULT '', nonce BLOB NOT NULL CHECK(length(nonce)=16),"
        " ver INTEGER NOT NULL CHECK(ver>0), edited_at INTEGER NOT NULL CHECK(edited_at>=0))",
        if_not_exists ? "IF NOT EXISTS " : "", name);
    if (sqlite3_exec(SI_DB(si), sql, NULL, NULL, NULL) == SQLITE_OK) return 0;
    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history schema: create table=%s: %s", name, sqlite3_errmsg(SI_DB(si))); return -1;
}

/* Dev-формат без переноса: удаляем только старую групповую историю и зависимые feedback.
 * PM, участники, ключи и прочие таблицы общей БД сохраняются. DDL и очистка индекса атомарны. */
static int si_table_init(struct DB_SYNC_INSTANCE* si) {
    sqlite3* db = SI_DB(si); sqlite3_stmt* st = NULL;
    if (si_table_create(si, SI_TBL(si), 1) < 0) return -1;
    int rc = si_prep(si, &st, "PRAGMA table_info(\"%s\")"), versioned = 0;
    if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
        const char* name = (const char*)sqlite3_column_text(st, 1);
        if (name && !strcmp(name, "nonce")) versioned = 1;
    }
    int final_rc = sqlite3_finalize(st); st = NULL;
    if (rc != SQLITE_DONE || final_rc != SQLITE_OK) return db_error(si, "inspect history format");
    if (versioned) return 0;
    if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "format reset begin");
    char sql[512];
    snprintf(sql, sizeof(sql), "DROP TABLE \"%s\"", SI_TBL(si));
    if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK || si_table_create(si, SI_TBL(si), 0) < 0) goto rollback;
    /* В тестах произвольные таблицы не имеют feedback; в чате он подготовлен до регистрации истории. */
    snprintf(sql, sizeof(sql), "SELECT 1 FROM sqlite_master WHERE type='table' AND name='feedback_%s'", si->ns);
    if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) goto rollback;
    rc = sqlite3_step(st); final_rc = sqlite3_finalize(st); st = NULL;
    if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final_rc != SQLITE_OK) goto rollback;
    if (rc == SQLITE_ROW) {
        snprintf(sql, sizeof(sql), "DELETE FROM \"feedback_%s\"", si->ns);
        if (sqlite3_exec(db, sql, NULL, NULL, NULL) != SQLITE_OK) goto rollback;
    }
    char messages[25], feedback[25];
    snprintf(messages, sizeof(messages), "%u:%s", MT_DATA_MESSAGES, si->ns);
    snprintf(feedback, sizeof(feedback), "%u:%s", MT_DATA_FEEDBACK, si->ns);
    const char* dependencies[] = {"merkle_tree_hash", "attachment_keys", "attachment_key_requests", "attachment_key_outbox", "channels", "ui_state"};
    const char* cleanup[] = {
        "DELETE FROM merkle_tree_hash WHERE namespace IN (?1,?2)",
        "DELETE FROM attachment_keys WHERE channel=?1", "DELETE FROM attachment_key_requests WHERE channel=?1",
        "DELETE FROM attachment_key_outbox WHERE channel=?1",
        "UPDATE channels SET last_msg_at=0,last_read_msg_id=0,last_pos_msg_id=0 WHERE channel_id=?1",
        "DELETE FROM ui_state WHERE key IN ('ch_'||?1||'_last_read','ch_'||?1||'_last_pos')"
    };
    for (unsigned i = 0; i < sizeof(dependencies) / sizeof(dependencies[0]); i++) {
        if (sqlite3_prepare_v2(db, "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", -1, &st, NULL) != SQLITE_OK) goto rollback;
        rc = sqlite3_bind_text(st, 1, dependencies[i], -1, SQLITE_STATIC);
        if (rc == SQLITE_OK) rc = sqlite3_step(st);
        final_rc = sqlite3_finalize(st); st = NULL;
        if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final_rc != SQLITE_OK) goto rollback;
        if (rc == SQLITE_DONE) continue;
        if (sqlite3_prepare_v2(db, cleanup[i], -1, &st, NULL) != SQLITE_OK) goto rollback;
        rc = sqlite3_bind_text(st, 1, i ? si->ns : messages, -1, SQLITE_STATIC);
        if (rc == SQLITE_OK && !i) rc = sqlite3_bind_text(st, 2, feedback, -1, SQLITE_STATIC);
        if (rc == SQLITE_OK) rc = sqlite3_step(st);
        final_rc = sqlite3_finalize(st); st = NULL;
        if (rc != SQLITE_DONE || final_rc != SQLITE_OK) goto rollback;
    }
    if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history format v2: reset old group history and feedback tbl=%s group=%s",
        SI_TBL(si), si->ns);
    return 0;
rollback:
    sqlite3_finalize(st); db_rollback(si, "history format reset"); return -1;
}

#define DB_COLUMNS "rowid,id,timestamp,node_id,data,author_signature,message_id,bucket,nonce,ver,edited_at"

/* Чтение конкретной строки. cache_valid — производные поля, их можно восстановить без удаления сообщения. */
static enum db_record_check db_read_record(struct DB_SYNC_INSTANCE* si, sqlite3_stmt* st, struct db_record* r) {
    memset(r, 0, sizeof(*r));
    r->rowid = sqlite3_column_int64(st, 0); r->id = (uint64_t)sqlite3_column_int64(st, 1);
    r->timestamp = (uint64_t)sqlite3_column_int64(st, 2); r->author = (uint64_t)sqlite3_column_int64(st, 3);
    r->data = sqlite3_column_blob(st, 4);
    if (!r->data && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
    r->data_len = (size_t)sqlite3_column_bytes(st, 4);
    const uint8_t* sig = sqlite3_column_blob(st, 5);
    if (!sig && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
    r->sig_len = (size_t)sqlite3_column_bytes(st, 5);
    if (!r->data || !r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || !sig || r->sig_len != 64) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed stored row tbl=%s rowid=%lld body=%zu signature=%zu",
            SI_TBL(si), (long long)r->rowid, r->data_len, r->sig_len); return DB_CHECK_BAD_FORMAT;
    }
    memcpy(r->signature, sig, 64);
    const void* nonce = sqlite3_column_blob(st, 8);
    if (!nonce && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
    if (!nonce || sqlite3_column_bytes(st, 8) != 16) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history malformed nonce tbl=%s rowid=%lld", SI_TBL(si), (long long)r->rowid);
        return DB_CHECK_BAD_FORMAT;
    }
    memcpy(r->revision.nonce, nonce, 16);
    r->revision.ver = (uint64_t)sqlite3_column_int64(st, 9);
    r->revision.edited_at = (uint64_t)sqlite3_column_int64(st, 10);
    if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return DB_CHECK_ERROR;
    const uint8_t* key = sqlite3_column_blob(st, 6);
    if (!key && sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) return DB_CHECK_ERROR;
    r->cache_valid = key && sqlite3_column_bytes(st, 6) == 32 && !memcmp(key, r->message_id, 32) &&
        (uint64_t)sqlite3_column_int64(st, 7) == (db_read64(r->message_id) >> 39);
    return DB_CHECK_VALID;
}

static int db_encode_record(const struct db_record* r, uint8_t* buf) {
    memcpy(buf, r->message_id, 32); db_write64(buf + 32, r->timestamp); db_write64(buf + 40, r->author);
    memcpy(buf + 48, r->revision.nonce, 16); db_write64(buf + 64, r->revision.ver); db_write64(buf + 72, r->revision.edited_at);
    db_write32(buf + 80, (uint32_t)r->data_len); memcpy(buf + 84, r->data, r->data_len);
    memcpy(buf + 84 + r->data_len, r->signature, 64);
    return (int)(DB_WIRE_FIXED + r->data_len);
}
static int db_parse_record(struct DB_SYNC_INSTANCE* si, const uint8_t** ptr, const uint8_t* end, struct db_record* r) {
    if ((size_t)(end - *ptr) < DB_WIRE_FIXED) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history truncated record"); return -1; }
    const uint8_t* p = *ptr;
    memset(r, 0, sizeof(*r));
    r->timestamp = db_read64(p + 32); r->author = db_read64(p + 40); r->data_len = db_read32(p + 80);
    memcpy(r->revision.nonce, p + 48, 16); r->revision.ver = db_read64(p + 64); r->revision.edited_at = db_read64(p + 72);
    if (!r->data_len || r->data_len > DB_SYNC_MESSAGE_MAX || r->data_len > (size_t)(end - p) - DB_WIRE_FIXED) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid wire body=%zu", r->data_len); return -1;
    }
    r->data = (const char*)p + 84; r->sig_len = 64; memcpy(r->signature, p + 84 + r->data_len, 64);
    if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return -1;
    if (memcmp(r->message_id, p, 32)) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history wire identity mismatch"); return -1; }
    *ptr = p + DB_WIRE_FIXED + r->data_len;
    return 0;
}

static int db_index_clear(struct DB_SYNC_INSTANCE* si) {
    char ns[25]; snprintf(ns, sizeof(ns), "%u:%s", MT_DATA_MESSAGES, si->ns);
    sqlite3_stmt* st = NULL;
    int rc = sqlite3_prepare_v2(SI_DB(si), "DELETE FROM merkle_tree_hash WHERE namespace=?", -1, &st, NULL);
    if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 1, ns, -1, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    int end_rc = sqlite3_finalize(st);
    return rc == SQLITE_DONE && end_rc == SQLITE_OK ? 0 : db_error(si, "clear message index");
}

static int db_update_path(struct DB_SYNC_INSTANCE* si, const uint8_t key[32]) {
    if (si->index.pending || !merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) return 0;
    return merkle_sync_recompute_path(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, db_read64(key)) < 0 ? -1 : 0;
}

static void db_committed(struct DB_SYNC_INSTANCE* si, int changed) {
    if (si->index.pending || (changed & 2)) db_index_request(si);
    else merkle_sync_changed(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
}

static int db_same_record(const struct db_record* a, const struct db_record* b) {
    return a->timestamp == b->timestamp && a->author == b->author && a->data_len == b->data_len &&
        a->revision.ver == b->revision.ver && a->revision.edited_at == b->revision.edited_at &&
        !memcmp(a->revision.nonce, b->revision.nonce, 16) &&
        a->sig_len == b->sig_len && !memcmp(a->signature, b->signature, 64) && !memcmp(a->data, b->data, a->data_len);
}

/* Внешняя транзакция обязательна. Входящая подпись уже проверена до захвата write-lock. */
static int db_put_record(struct DB_SYNC_INSTANCE* si, struct db_record* r, enum db_record_check incoming,
        const char* attrs, int* changed) {
    sqlite3_stmt* st = NULL;
    if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE message_id=?") != SQLITE_OK)
        return db_error(si, "read duplicate");
    int rc = sqlite3_bind_blob(st, 1, r->message_id, 32, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    struct db_record stored;
    int exists = rc == SQLITE_ROW, remove = 0;
    if (exists) {
        enum db_record_check check = db_read_record(si, st, &stored);
        if (check == DB_CHECK_VALID && !(incoming == DB_CHECK_VALID && db_same_record(&stored, r)))
            check = db_verify_author_sig(si, &stored);
        if (check == DB_CHECK_ERROR || check == DB_CHECK_DEFERRED) { sqlite3_finalize(st); return -1; }
        remove = check == DB_CHECK_BAD_FORMAT || check == DB_CHECK_BAD_SIGNATURE;
        if (!remove && !stored.cache_valid) {
            sqlite3_finalize(st); db_index_request(si); return db_error(si, "derived identity needs rebuild");
        }
    } else if (rc != SQLITE_DONE) { sqlite3_finalize(st); return db_error(si, "duplicate step"); }
    int same = exists && !remove && db_same_record(&stored, r);
    if (sqlite3_finalize(st) != SQLITE_OK) return db_error(si, "duplicate finalize");
    if (remove) {
        if (db_delete_bad_row(si, stored.rowid) < 0) return -1;
        if (merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) && db_index_clear(si) < 0) return -1;
        *changed |= 3;
        DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history staged deletion of proven corrupt duplicate tbl=%s rowid=%lld",
            SI_TBL(si), (long long)stored.rowid);
    }
    if (incoming != DB_CHECK_VALID) return -2;
    if (exists && !remove) {
        if (r->revision.ver < stored.revision.ver || same) {
            DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history duplicate/stale tbl=%s id=%02x%02x%02x%02x incoming=%llu stored=%llu",
                SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
                (unsigned long long)r->revision.ver, (unsigned long long)stored.revision.ver); return 1;
        }
        if (r->revision.ver == stored.revision.ver) {
            DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history version conflict tbl=%s id=%02x%02x%02x%02x ver=%llu",
                SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
                (unsigned long long)r->revision.ver); return -2;
        }
        r->updated = 1; r->id = stored.id;
        if (si_prep(si, &st, "UPDATE \"%s\" SET data=?,author_signature=?,ver=?,edited_at=?,"
            "flags=flags & ~1,delivered_peers=0,delivery_chain='' WHERE message_id=?") != SQLITE_OK) return db_error(si, "update prepare");
        rc = sqlite3_bind_blob(st, 1, r->data, (int)r->data_len, SQLITE_STATIC);
        if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 2, r->signature, 64, SQLITE_STATIC);
        if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)r->revision.ver);
        if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 4, (sqlite3_int64)r->revision.edited_at);
        if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 5, r->message_id, 32, SQLITE_STATIC);
        if (rc == SQLITE_OK) rc = sqlite3_step(st);
        int end_rc = sqlite3_finalize(st);
        if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "update step");
        if (db_update_path(si, r->message_id) < 0) return -1;
        *changed |= 1;
        DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history updated tbl=%s id=%02x%02x%02x%02x ver=%llu->%llu edited=%llu",
            SI_TBL(si), r->message_id[0], r->message_id[1], r->message_id[2], r->message_id[3],
            (unsigned long long)stored.revision.ver, (unsigned long long)r->revision.ver,
            (unsigned long long)r->revision.edited_at); return 0;
    }
    if (si_prep(si, &st, "INSERT INTO \"%s\" (timestamp,node_id,id,flags,data,author_signature,message_id,bucket,local_attrs,nonce,ver,edited_at)"
        " VALUES(?,?,?,0,?,?,?,?,?,?,?,?)") != SQLITE_OK) return db_error(si, "insert prepare");
    rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)r->timestamp);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)r->author);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 3, (sqlite3_int64)r->id);
    if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 4, r->data, (int)r->data_len, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 5, r->signature, 64, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 6, r->message_id, 32, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 7, (sqlite3_int64)(db_read64(r->message_id) >> 39));
    if (rc == SQLITE_OK) rc = sqlite3_bind_text(st, 8, attrs ? attrs : "", -1, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_blob(st, 9, r->revision.nonce, 16, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 10, (sqlite3_int64)r->revision.ver);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 11, (sqlite3_int64)r->revision.edited_at);
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    int final_rc = sqlite3_finalize(st);
    if (rc != SQLITE_DONE || final_rc != SQLITE_OK) return db_error(si, "insert step");
    if (!(*changed & 2) && db_update_path(si, r->message_id) < 0) return -1;
    *changed |= 1; return 0;
}

/* Предварительная проверка идёт без write-lock; внутри короткой транзакции сверяем прочитанные байты. */
static int db_index_batch(struct DB_SYNC_INSTANCE* si) {
    sqlite3* db = SI_DB(si);
    sqlite3_stmt* st = NULL;
    if (!merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) return 1;
    if (si->index.restart) {
        merkle_sync_cancel_dataset_ns(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
        if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "index begin");
        if (db_index_clear(si) < 0 || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) {
            db_rollback(si, "index reset"); return -1;
        }
        si->index.restart = 0; si->index.phase = 0; si->index.after_valid = 0; si->index.after = 0;
    }
    if (si->index.phase == 0) {
        struct db_checked* rows = u_calloc(DB_BATCH_RECORDS, sizeof(*rows));
        if (!rows) return db_error(si, "preflight allocation");
        const char* query = si->index.after_valid
            ? "SELECT " DB_COLUMNS " FROM \"%s\" WHERE rowid>? ORDER BY rowid LIMIT 16"
            : "SELECT " DB_COLUMNS " FROM \"%s\" ORDER BY rowid LIMIT 16";
        int rc = si_prep(si, &st, query), count = 0, removed = 0;
        if (rc == SQLITE_OK && si->index.after_valid) rc = sqlite3_bind_int64(st, 1, si->index.after);
        if (rc != SQLITE_OK) goto verify_error;
        while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
            struct db_checked* item = &rows[count];
            item->check = db_read_record(si, st, &item->record);
            if (item->check == DB_CHECK_VALID) {
                memcpy(item->data, item->record.data, item->record.data_len); item->record.data = item->data;
                item->check = db_verify_author_sig(si, &item->record);
            }
            if (item->check == DB_CHECK_ERROR || item->check == DB_CHECK_DEFERRED) {
                si->index.waiting_key = item->check == DB_CHECK_DEFERRED;
                goto verify_error;
            }
            count++;
        }
        int final_rc = sqlite3_finalize(st); st = NULL;
        if (rc != SQLITE_DONE || final_rc != SQLITE_OK) goto verify_error;
        st = NULL;
        if (!count) {
            si->index.phase = 1; si->index.after_valid = 0; si->index.after = -1;
            u_free(rows); return 0;
        }
        if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) goto verify_error;
        for (int i = 0; i < count; i++) {
            struct db_checked* item = &rows[i]; struct db_record current;
            if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE rowid=?") != SQLITE_OK) goto verify_rollback;
            if (sqlite3_bind_int64(st, 1, item->record.rowid) != SQLITE_OK || sqlite3_step(st) != SQLITE_ROW) goto verify_rollback;
            enum db_record_check now = db_read_record(si, st, &current);
            int matches = now == DB_CHECK_BAD_FORMAT && item->check == DB_CHECK_BAD_FORMAT;
            if (now == DB_CHECK_VALID && item->check != DB_CHECK_BAD_FORMAT) matches = db_same_record(&item->record, &current);
            if (sqlite3_finalize(st) != SQLITE_OK) { st = NULL; goto verify_rollback; }
            st = NULL;
            if (!matches) {
                DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history preflight source changed tbl=%s; retry", SI_TBL(si)); goto verify_rollback;
            }
            if (item->check == DB_CHECK_BAD_FORMAT || item->check == DB_CHECK_BAD_SIGNATURE) {
                if (db_delete_bad_row(si, item->record.rowid) < 0) goto verify_rollback;
                removed++;
            } else {
                if (si_prep(si, &st, "UPDATE \"%s\" SET message_id=?,bucket=? WHERE rowid=?") != SQLITE_OK) goto verify_rollback;
                if (sqlite3_bind_blob(st, 1, item->record.message_id, 32, SQLITE_STATIC) != SQLITE_OK ||
                    sqlite3_bind_int64(st, 2, (sqlite3_int64)(db_read64(item->record.message_id) >> 39)) != SQLITE_OK ||
                    sqlite3_bind_int64(st, 3, item->record.rowid) != SQLITE_OK || sqlite3_step(st) != SQLITE_DONE) goto verify_rollback;
                if (sqlite3_finalize(st) != SQLITE_OK) { st = NULL; goto verify_rollback; }
                st = NULL;
            }
        }
        if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto verify_rollback;
        si->index.after = rows[count - 1].record.rowid; si->index.after_valid = 1; si->index.waiting_key = 0;
        DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history preflight committed tbl=%s checked=%d removed=%d after=%lld",
            SI_TBL(si), count, removed, (long long)si->index.after);
        if (removed) DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history removed %d proven invalid stored records tbl=%s", removed, SI_TBL(si));
        u_free(rows); return 0;
verify_rollback:
        sqlite3_finalize(st); st = NULL; db_rollback(si, "preflight commit");
verify_error:
        sqlite3_finalize(st); u_free(rows); return db_error(si, si->index.waiting_key ? "preflight waiting for author key" : "preflight failed");
    }

    /* Все сохранённые подписи проверены; в write-транзакции остаётся только хеширование ограниченной пачки листьев. */
    sqlite3_int64 buckets[DB_BATCH_RECORDS]; int count = 0, rc;
    if (si_prep(si, &st, "SELECT DISTINCT bucket FROM \"%s\" WHERE bucket>? ORDER BY bucket LIMIT 16") != SQLITE_OK)
        return db_error(si, "index buckets");
    if (sqlite3_bind_int64(st, 1, si->index.after) != SQLITE_OK) { sqlite3_finalize(st); return db_error(si, "bucket bind"); }
    while ((rc = sqlite3_step(st)) == SQLITE_ROW) buckets[count++] = sqlite3_column_int64(st, 0);
    int end_rc = sqlite3_finalize(st);
    if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "bucket step");
    if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "hash begin");
    for (int i = 0; i < count; i++)
        if (merkle_sync_recompute_path(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, (uint64_t)buckets[i] << 39) < 0) {
            db_rollback(si, "hash batch"); return -1;
        }
    if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "hash commit"); return -1; }
    if (count) si->index.after = buckets[count - 1];
    else {
        si->index.pending = 0;
        DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history verified index ready tbl=%s group=%s", SI_TBL(si), si->ns);
        merkle_sync_changed(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
        for (int i = 0; i < si->peer_count; i++) db_start_peer(si, si->peers[i].node_id);
    }
    return 0;
}

static void db_index_tick(void* arg) {
    struct DB_SYNC_INSTANCE* si = arg; si->index.timer = NULL;
    if (!si->enabled || !si->index.pending) return;
    int rc = db_index_batch(si);
    if (si->index.pending && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) {
        si->index.timer = uasync_set_timeout(si->db_sync->inst->ua, rc ? 50000 : 1, si, db_index_tick, "history_preflight");
        if (!si->index.timer) { db_error(si, "preflight timer"); db_sync_resume_peer_check(si->db_sync); }
    }
}
static void db_index_request(struct DB_SYNC_INSTANCE* si) {
    si->index.pending = 1; si->index.restart = 1; si->index.waiting_key = 0;
    if (!si->index.timer && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES))
        si->index.timer = uasync_set_timeout(si->db_sync->inst->ua, 0, si, db_index_tick, "history_preflight");
    db_sync_resume_peer_check(si->db_sync);
}

/* Чтение листа — одинаковый канонический отбор для корня и передачи. Проверка авторства уже выполнена preflight/put. */
static int db_message_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
    if (!si || level != MT_MAX_LEVEL) return -1;
    sqlite3_stmt* st = NULL;
    if (si_prep(si, &st, "SELECT " DB_COLUMNS " FROM \"%s\" WHERE bucket=? ORDER BY message_id") != SQLITE_OK)
        return db_error(si, "leaf prepare");
    int rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)(prefix >> 39)), count = 0, bad = 0;
    if (rc != SQLITE_OK) { sqlite3_finalize(st); return db_error(si, "leaf bind"); }
    while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
        struct db_record r; uint8_t wire[DB_WIRE_MAX];
        enum db_record_check check = db_read_record(si, st, &r);
        if (check != DB_CHECK_VALID || !r.cache_valid) { bad = check != DB_CHECK_ERROR; rc = SQLITE_ERROR; break; }
        int bytes = db_encode_record(&r, wire);
        if (EVP_DigestUpdate(hash, wire, (size_t)bytes) != 1) { rc = SQLITE_ERROR; break; }
        count++;
    }
    int end_rc = sqlite3_finalize(st);
    if (bad) db_index_request(si);
    if (rc != SQLITE_DONE || end_rc != SQLITE_OK) return db_error(si, "leaf read/hash");
    return count;
}

static int db_message_page(void* ctx, const char* ns, uint64_t prefix, int av, const merkle_key_t* after,
        uint8_t* buf, size_t* len, merkle_key_t* next, int* more) {
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
    if (!si || si->index.pending) return -1;
    sqlite3_stmt* st = NULL;
    const char* query = av
        ? "SELECT " DB_COLUMNS " FROM \"%s\" WHERE bucket=? AND message_id>? ORDER BY message_id LIMIT 17"
        : "SELECT " DB_COLUMNS " FROM \"%s\" WHERE bucket=? ORDER BY message_id LIMIT 17";
    int rc = si_prep(si, &st, query), count = 0, bad = 0;
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)(prefix >> 39));
    if (rc == SQLITE_OK && av) rc = sqlite3_bind_blob(st, 2, after->bytes, 32, SQLITE_STATIC);
    if (rc != SQLITE_OK) { sqlite3_finalize(st); return db_error(si, "page prepare"); }
    size_t used = 0; *more = 0; memset(next, 0, sizeof(*next));
    while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
        struct db_record r;
        enum db_record_check check = db_read_record(si, st, &r);
        if (check == DB_CHECK_VALID) check = db_verify_author_sig(si, &r);
        if (check != DB_CHECK_VALID || !r.cache_valid) {
            bad = check == DB_CHECK_BAD_SIGNATURE || check == DB_CHECK_BAD_FORMAT || (check == DB_CHECK_VALID && !r.cache_valid);
            rc = SQLITE_ERROR; break;
        }
        size_t bytes = DB_WIRE_FIXED + r.data_len;
        if (count == DB_BATCH_RECORDS || used + bytes > *len) {
            if (!count) { rc = SQLITE_TOOBIG; break; }
            *more = 1; break;
        }
        used += (size_t)db_encode_record(&r, buf + used); memcpy(next->bytes, r.message_id, 32); count++;
    }
    int end_rc = sqlite3_finalize(st);
    if (bad) db_index_request(si);
    if ((rc != SQLITE_DONE && !(rc == SQLITE_ROW && *more)) || end_rc != SQLITE_OK) return db_error(si, "page read");
    *len = used; return 0;
}

static void db_done_sweep(struct DB_SYNC* db) {
    if (db->dispatch_depth) return;
    struct db_sync_done_cbk_entry** link = &db->done_cbks;
    while (*link) {
        struct db_sync_done_cbk_entry* e = *link;
        if (e->removed) { *link = e->next; u_free(e); } else link = &e->next;
    }
}
int db_sync_add_done_cbk(struct UTUN_INSTANCE* inst, db_sync_done_fn fn, void* arg) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL;
    if (!db || !fn) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history done subscription requires context/callback"); return -1; }
    for (struct db_sync_done_cbk_entry* current = db->done_cbks; current; current = current->next)
        if (!current->removed && current->fn == fn && current->arg == arg) return 0;
    struct db_sync_done_cbk_entry* e = u_calloc(1, sizeof(*e));
    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history done subscription allocation"); return -1; }
    e->fn = fn; e->arg = arg; e->next = db->done_cbks; db->done_cbks = e;
    return 0;
}
void db_sync_remove_done_cbk(struct UTUN_INSTANCE* inst, db_sync_done_fn fn, void* arg) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL;
    if (!db) return;
    for (struct db_sync_done_cbk_entry* e = db->done_cbks; e; e = e->next)
        if (!e->removed && e->fn == fn && e->arg == arg) { e->removed = 1; break; }
    db_done_sweep(db);
}
static void db_message_result(void* ctx, const char* ns, uint64_t peer, int result) {
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
    if (!si) return;
    struct SI_PEER* p = si_peer_add(si, peer);
    if (!p) return;
    p->sync_state = result == MT_OK ? 2 : 0;
    if (result != MT_OK) { db_sync_resume_peer_check(si->db_sync); return; }
    struct DB_SYNC* db = si->db_sync;
    si->holds++; db->dispatch_depth++;
    for (struct db_sync_done_cbk_entry* e = db->done_cbks; e && si->enabled; e = e->next)
        if (!e->removed) e->fn(si, peer, e->arg);
    db->dispatch_depth--; db_done_sweep(db); db_release(si);
}

static int db_message_peer(void* ctx, const char* ns, uint64_t peer) {
    struct UTUN_INSTANCE* inst = ctx;
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(inst, strtoull(ns, NULL, 10));
    struct TOPO_GROUP* group = si ? topo_groups_find(inst->topo_groups, si->group_id) : NULL;
    struct ETCP_CONN* conn = instance_find_conn(inst, peer);
    return si && !si->sync_gated && !si->index.pending && topo_group_peer_ready(group, peer) && conn &&
        member_sync_validate_pubkey(inst, ns, conn->crypto_ctx.peer_public_key);
}
static int db_sync_peer_throttled(struct DB_SYNC* db, struct SI_PEER* p) {
    struct ETCP_CONN* conn = p && !p->sound_on ? instance_find_conn(db->inst, p->node_id) : NULL;
    return conn && conn->peer_sleep_phase;
}
static int db_message_can_run(void* ctx, const char* ns, uint64_t peer) {
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
    if (!si || !db_message_peer(ctx, ns, peer)) return 0;
    struct SI_PEER* p = si_peer_find(si, peer);
    return !p || (!p->paused && !db_sync_peer_throttled(si->db_sync, p));
}
static int db_start_peer(struct DB_SYNC_INSTANCE* si, uint64_t peer) {
    if (!merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) || !db_message_can_run(si->db_sync->inst, si->ns, peer)) return 0;
    struct SI_PEER* p = si_peer_add(si, peer);
    if (!p) return -1;
    if (merkle_sync_start(si->db_sync->inst, MT_DATA_MESSAGES, peer, si->ns, NULL, NULL) < 0) {
        p->sync_state = 0; db_sync_resume_peer_check(si->db_sync); return -1;
    }
    p->sync_state = 1; db_sync_resume_peer_check(si->db_sync); return 0;
}

static void db_notice_unlink(struct db_notice* n) {
    struct db_notice** link = &n->db->notices;
    while (*link && *link != n) link = &(*link)->next;
    if (*link) *link = n->next;
}
static void db_notice_run(void* arg) {
    struct db_notice* n = arg; n->work = NULL; db_notice_unlink(n);
    struct DB_SYNC_INSTANCE* si = db_instance_find(n->db, n->group_id);
    if (si) {
        si->holds++;
        struct db_record* records = (struct db_record*)(n + 1);
        for (unsigned i = 0; i < n->count && si->enabled; i++)
            if (si->on_insert) si->on_insert(si, records[i].message_id, records[i].timestamp, records[i].data,
                records[i].data_len, records[i].author, n->initial, records[i].updated, si->on_insert_arg);
        db_release(si);
    }
    u_free(n);
}
static void db_notice_abort(struct db_notice* n) {
    if (!n) return;
    uasync_call_soon_cancel(n->db->inst->ua, n->work); db_notice_unlink(n); u_free(n);
}
static int db_notice_stage(struct DB_SYNC_INSTANCE* si, struct db_record* records, int count, const uint8_t* added, struct db_notice** out) {
    size_t bodies = 0; unsigned total = 0; *out = NULL;
    for (int i = 0; i < count; i++) if (added[i]) { total++; bodies += records[i].data_len; }
    if (!total || !si->on_insert) return 0;
    struct db_notice* n = u_calloc(1, sizeof(*n) + total * sizeof(struct db_record) + bodies);
    if (!n) return db_error(si, "notification allocation");
    n->db = si->db_sync; n->group_id = si->group_id; n->initial = 1; n->count = total;
    struct db_record* saved = (struct db_record*)(n + 1); char* data = (char*)(saved + total); unsigned index = 0;
    for (int i = 0; i < count; i++) if (added[i]) {
        saved[index] = records[i]; saved[index].data = data; memcpy(data, records[i].data, records[i].data_len);
        data += records[i].data_len; index++;
    }
    n->work = uasync_call_soon(si->db_sync->inst->ua, n, db_notice_run);
    if (!n->work) { u_free(n); return db_error(si, "notification scheduling"); }
    n->next = si->db_sync->notices; si->db_sync->notices = n; *out = n; return 0;
}

/* Полная проверка входящей страницы — до BEGIN; write-lock охватывает только небольшой commit данных и индекса. */
static int db_message_apply(void* ctx, const char* ns, uint64_t peer, uint64_t prefix, int av,
        const merkle_key_t* after, const merkle_key_t* next_key, int more, const uint8_t* data, size_t len) {
    (void)peer;
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(ctx, strtoull(ns, NULL, 10));
    if (!si || si->index.pending) return -1;
    struct db_record records[DB_BATCH_RECORDS]; uint8_t added[DB_BATCH_RECORDS] = {0};
    const uint8_t* p = data; int count = 0;
    while (p < data + len) {
        if (count == DB_BATCH_RECORDS || db_parse_record(si, &p, data + len, &records[count]) < 0) return -1;
        struct db_record* r = &records[count];
        if (merkle_sync_level_prefix(db_read64(r->message_id), MT_MAX_LEVEL) != prefix ||
            (av && memcmp(r->message_id, after->bytes, 32) <= 0)) {
            DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page key outside requested range ns=%s", ns); return -1;
        }
        if (db_verify_author_sig(si, r) != DB_CHECK_VALID) return -1;
        if (count && memcmp(records[count - 1].message_id, r->message_id, 32) >= 0) {
            DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history incoming page not strictly ordered ns=%s", ns); return -1;
        }
        count++;
    }
    static const uint8_t empty[32];
    if ((!count && (more || memcmp(next_key->bytes, empty, 32))) ||
        (count && memcmp(next_key->bytes, records[count - 1].message_id, 32))) {
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history page cursor does not match records ns=%s", ns); return -1;
    }
    if (!count) return 0;
    sqlite3* db = SI_DB(si);
    if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "page begin");
    uint64_t next = si->next_id; int changed = 0; struct db_notice* notice = NULL;
    for (int i = 0; i < count; i++) {
        records[i].id = next;
        int result = db_put_record(si, &records[i], DB_CHECK_VALID, NULL, &changed);
        if (result < 0) goto rollback;
        if (!result) { added[i] = 1; if (!records[i].updated) next++; }
    }
    if (db_notice_stage(si, records, count, added, &notice) < 0) goto rollback;
    if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) goto rollback;
    si->next_id = next;
    if (changed) db_committed(si, changed);
    DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history page committed ns=%s records=%d changed=%d", ns, count, changed);
    return 0;
rollback:
    db_notice_abort(notice); db_rollback(si, "page apply"); return -1;
}

static const struct merkle_sync_data_ops db_message_ops = {
    .update_bucket_hash = db_message_hash, .get_page = db_message_page, .apply_page = db_message_apply,
    .validate_peer = db_message_peer, .can_run = db_message_can_run, .on_result = db_message_result
};

int db_sync_register_model(struct UTUN_INSTANCE* inst) {
    if (!inst || !inst->db_sync || !inst->db_sync->enabled) return -1;
    if (merkle_sync_registered(inst, MT_DATA_MESSAGES)) return 0;
    if (merkle_sync_register(inst, MT_DATA_MESSAGES, &db_message_ops, inst) < 0) return -1;
    for (int i = 0; i < inst->db_sync->instance_count; i++) db_index_request(inst->db_sync->instances[i]);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history registered shared Merkle model");
    return 0;
}

static int db_send(struct DB_SYNC_INSTANCE* si, uint64_t peer, uint8_t type, const uint8_t* data, size_t len) {
    if (!db_message_can_run(si->db_sync->inst, si->ns, peer)) return -1;
    struct ETCP_CONN* conn = instance_find_conn(si->db_sync->inst, peer);
    if (!conn || !conn->links_up) { DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history transport unavailable peer=%016llx", peer); return -1; }
    struct ll_entry* e = ll_alloc_lldgram((uint16_t)(10 + len));
    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history packet allocation"); return -1; }
    e->dgram[0] = ETCP_ID_DB_SYNC; db_write64(e->dgram + 1, si->group_id); e->dgram[9] = type;
    if (len) memcpy(e->dgram + 10, data, len);
    e->len = (uint16_t)(10 + len);
    int rc = etcp_send(conn, e);
    if (rc != 0) {
        queue_dgram_free(e); queue_entry_free(e);
        DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history send rejected peer=%016llx rc=%d", peer, rc);
    }
    return rc;
}

static void db_emit_push(struct DB_SYNC_INSTANCE* si, const struct db_record* r, uint64_t source) {
    si->holds++;
    if (si->on_insert) si->on_insert(si, r->message_id, r->timestamp, r->data, r->data_len, r->author, 0, r->updated, si->on_insert_arg);
    if (si->enabled) {
        uint8_t wire[DB_WIRE_MAX]; size_t bytes = (size_t)db_encode_record(r, wire);
        for (int i = 0; i < si->peer_count; i++)
            if (si->peers[i].node_id != source && si->peers[i].sync_state && !si->peers[i].paused &&
                !db_sync_peer_throttled(si->db_sync, &si->peers[i]))
                db_send(si, si->peers[i].node_id, DB_MSG_PUSH, wire, bytes);
    }
    db_release(si);
}
static int db_insert_one(struct DB_SYNC_INSTANCE* si, struct db_record* r, const char* attrs) {
    if (chat_message_id(si->group_id, r->timestamp, r->author, r->revision.nonce, r->message_id) < 0) return -1;
    enum db_record_check incoming = db_verify_author_sig(si, r);
    if (incoming == DB_CHECK_ERROR || incoming == DB_CHECK_DEFERRED) return -1;
    sqlite3* db = SI_DB(si);
    if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) return db_error(si, "insert begin");
    int changed = 0; r->id = si->next_id;
    int result = db_put_record(si, r, incoming, attrs, &changed);
    if (result == -1 || sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "insert"); return -1; }
    if (!result && !r->updated) si->next_id++;
    if (changed) db_committed(si, changed);
    return result;
}
/* ACK конкретного сообщения: повтор не увеличивает счётчик, очередь отправки не считается доставкой. */
static int si_delivery_update(struct DB_SYNC_INSTANCE* si, const uint8_t message_id[32], uint64_t ver, uint64_t peer_id)
{
    char hex[17]; snprintf(hex, sizeof(hex), "%016llx", (unsigned long long)peer_id);
    sqlite3_stmt* stmt = NULL;
    int rc = si_prep(si, &stmt,
        "UPDATE \"%s\" SET flags=flags | CASE WHEN node_id=?3 THEN 1 ELSE 0 END,"
        " delivered_peers=delivered_peers + (instr(',' || delivery_chain || ',', ',' || ?1 || ',')=0),"
        " delivery_chain=CASE WHEN instr(',' || delivery_chain || ',', ',' || ?1 || ',')>0 THEN delivery_chain"
        " WHEN delivery_chain='' THEN ?1 ELSE delivery_chain || ',' || ?1 END WHERE message_id=?2 AND ver=?4");
    if (rc == SQLITE_OK) rc = sqlite3_bind_text(stmt, 1, hex, -1, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_blob(stmt, 2, message_id, 32, SQLITE_STATIC);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 3, (sqlite3_int64)si->db_sync->inst->node_id);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(stmt, 4, (sqlite3_int64)ver);
    if (rc == SQLITE_OK) rc = sqlite3_step(stmt);
    int changed = sqlite3_changes(SI_DB(si)), final_rc = sqlite3_finalize(stmt);
    if (rc != SQLITE_DONE || final_rc != SQLITE_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history ACK update tbl=%s peer=%s rc=%d finalize=%d: %s",
            SI_TBL(si), hex, rc, final_rc, sqlite3_errmsg(SI_DB(si))); return -1;
    }
    if (!changed) DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK stale/unknown tbl=%s peer=%s ver=%llu",
        SI_TBL(si), hex, (unsigned long long)ver);
    else DEBUG_DEBUG(DEBUG_CATEGORY_CHAT_SYNC, "history ACK recorded tbl=%s peer=%s", SI_TBL(si), hex);
    return 0;
}


static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* e) {
    if (!e) return;
    struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL;
    struct DB_SYNC_INSTANCE* si = e->dgram && e->len >= 10 ? db_sync_instance_find(inst, db_read64(e->dgram + 1)) : NULL;
    if (!si || !db_message_peer(inst, si->ns, conn->peer_node_id)) goto done;
    const uint8_t* data = e->dgram + 10; size_t len = e->len - 10;
    si->holds++;
    if (e->dgram[9] == DB_MSG_ACK_PUSH) {
        if (len == 40) si_delivery_update(si, data, db_read64(data + 32), conn->peer_node_id);
        else DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history invalid ACK bytes=%zu", len);
    } else if (e->dgram[9] == DB_MSG_PUSH) {
        const uint8_t* p = data; struct db_record r;
        if (db_parse_record(si, &p, data + len, &r) == 0 && p == data + len) {
            int result = db_insert_one(si, &r, NULL);
            if (!result) db_emit_push(si, &r, conn->peer_node_id);
            if (result >= 0 && si->enabled) {
                uint8_t ack[40]; memcpy(ack, r.message_id, 32); db_write64(ack + 32, r.revision.ver);
                db_send(si, conn->peer_node_id, DB_MSG_ACK_PUSH, ack, sizeof(ack));
            }
        }
    } else DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "history unknown message type=%u", e->dgram[9]);
    db_release(si);
done:
    queue_dgram_free(e); queue_entry_free(e);
}

static uint64_t db_now_ms(struct DB_SYNC_INSTANCE* si) {
    int64_t us;
    if (si->db_sync->inst->ntp.synced) us = ntp_time_get_us(si->db_sync->inst);
    else { struct timeval tv; utun_gettimeofday(&tv, NULL); us = (int64_t)tv.tv_sec * 1000000LL + tv.tv_usec; }
    if (us < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history wall clock before epoch"); return 0; }
    return (uint64_t)us / 1000;
}
static void db_ttl_tick(void* arg) {
    struct DB_SYNC_INSTANCE* si = arg; si->ttl_timer = NULL;
    if (!si->enabled) return;
    sqlite3* db = SI_DB(si); sqlite3_stmt* st = NULL;
    uint64_t now = db_now_ms(si);
    uint64_t ttl = (uint64_t)si->db_sync->inst->config->global.db_sync_ttl * 1000;
    uint64_t cut = now > ttl ? now - ttl : 0;
    int removed = 0, committed = 0;
    if (sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL) != SQLITE_OK) { db_error(si, "TTL begin"); goto schedule; }
    char query[512];
    snprintf(query, sizeof(query), "DELETE FROM \"%s\" WHERE rowid IN (SELECT rowid FROM \"%s\""
        " WHERE node_id=? AND (flags & 1)=0 AND CASE WHEN edited_at>0 THEN edited_at ELSE timestamp END<? LIMIT 16)", SI_TBL(si), SI_TBL(si));
    int rc = sqlite3_prepare_v2(db, query, -1, &st, NULL);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)si->db_sync->inst->node_id);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, (sqlite3_int64)cut);
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    removed = sqlite3_changes(db);
    int end_rc = sqlite3_finalize(st); st = NULL;
    if (rc != SQLITE_DONE || end_rc != SQLITE_OK) { db_rollback(si, "TTL delete"); goto schedule; }
    if (removed && merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES) && db_index_clear(si) < 0) {
        db_rollback(si, "TTL index"); goto schedule;
    }
    if (sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) != SQLITE_OK) { db_rollback(si, "TTL commit"); goto schedule; }
    committed = 1;
    if (removed) {
        db_index_request(si);
        DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history TTL committed tbl=%s removed=%d cutoff_ms=%llu",
            SI_TBL(si), removed, (unsigned long long)cut);
    }
schedule:
    uint64_t delay = committed && removed == DB_BATCH_RECORDS ? 1 : DB_SYNC_TTL_INTERVAL * 10000ULL;
    si->ttl_timer = uasync_set_timeout(si->db_sync->inst->ua, delay, si, db_ttl_tick, "history_ttl");
    if (!si->ttl_timer) { db_error(si, "TTL timer"); db_sync_resume_peer_check(si->db_sync); }
}

static void db_sync_resume_peer_check(struct DB_SYNC* db) {
    if (!db || !db->enabled || db->peer_check_timer) return;
#ifdef UTUN_HAVE_STANDBY
    if (db->peer_check_wait) return;
#endif
    db->peer_check_timer = uasync_set_timeout(db->inst->ua, DB_SYNC_PEER_CHECK_INTERVAL * 10000ULL, db, db_sync_peer_check_cb, "history_retry");
    if (!db->peer_check_timer) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history retry timer allocation");
}
#ifdef UTUN_HAVE_STANDBY
/* standby_wait может вызвать callback синхронно при отказе памяти: освобождаем handle и ставим таймер без рекурсии. */
static void db_sync_peer_check_wake(void* arg) {
    struct DB_SYNC* db = arg; db->peer_check_wait = NULL;
    db_sync_resume_peer_check(db);
}
#endif
static void db_sync_peer_check_cb(void* arg) {
    struct DB_SYNC* db = arg; db->peer_check_timer = NULL; int pending = 0;
    if (!db->enabled) return;
#ifdef UTUN_HAVE_STANDBY
    db->peer_check_wait = NULL;
    if (standby_get_sleep_tb() > 0) {
        db->peer_check_wait = standby_wait(db, db_sync_peer_check_wake);
        if (!db->peer_check_wait) {
            DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history standby wait allocation; retry later");
            db_sync_resume_peer_check(db);
        }
        return;
    }
#endif
    for (int i = 0; i < db->instance_count; i++) {
        struct DB_SYNC_INSTANCE* si = db->instances[i];
        if (si->index.pending) {
            pending = 1;
            if (!si->index.timer) db_index_request(si);
        }
        for (int j = 0; j < si->peer_count; j++) {
            struct SI_PEER* p = &si->peers[j];
            if (p->sync_state != 2 && db_message_can_run(db->inst, si->ns, p->node_id)) {
                pending = 1;
                if (!p->sync_state) db_start_peer(si, p->node_id);
            }
        }
    }
    if (pending) db_sync_resume_peer_check(db);
}
static void db_sync_on_peer_ready(struct TOPO_GROUP* group, uint64_t peer, int ready, void* arg) {
    (void)group; struct DB_SYNC_INSTANCE* si = arg;
    if (!si->enabled) return;
    struct SI_PEER* p = ready ? si_peer_add(si, peer) : si_peer_find(si, peer);
    if (!p) return;
    if (!ready) {
        p->sync_state = 0;
        merkle_sync_cancel(si->db_sync->inst, MT_DATA_MESSAGES, peer, si->ns);
    } else db_start_peer(si, peer);
}

int db_sync_init(struct UTUN_INSTANCE* inst) {
    if (!inst || inst->db_sync) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid init"); return -1; }
    struct DB_SYNC* db = u_calloc(1, sizeof(*db));
    if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history allocation"); return -1; }
    db->inst = inst; db->db = inst->topo_sqlite_db; inst->db_sync = db;
    return inst->config->global.db_sync_enabled ? db_sync_enable(inst) : 0;
}
int db_sync_enable(struct UTUN_INSTANCE* inst) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL;
    if (!db || !inst->topo_sqlite_db) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history requires core SQLite"); return -1; }
    if (db->enabled) return 0;
    db->db = inst->topo_sqlite_db; db->enabled = 1;
    etcp_bind(inst, ETCP_ID_DB_SYNC, db_sync_recv_cb);
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history enabled on shared core SQLite");
    return 0;
}
struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst, const char* name, uint64_t group_id) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL;
    if (!db || !db->enabled || !si_name_valid(name) || !topo_groups_find(inst->topo_groups, group_id)) {
        DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid table registration"); return NULL;
    }
    struct DB_SYNC_INSTANCE* si = db_instance_find(db, group_id);
    if (si) return si;
    if (db->instance_count == db->instance_capacity) {
        int cap = db->instance_capacity ? db->instance_capacity * 2 : 4;
        void* entries = u_realloc(db->instances, (size_t)cap * sizeof(*db->instances));
        if (!entries) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history registry allocation"); return NULL; }
        db->instances = entries; db->instance_capacity = cap;
    }
    si = u_calloc(1, sizeof(*si));
    if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history table allocation"); return NULL; }
    si->db_sync = db; si->group_id = group_id; si->enabled = 1; si->sync_gated = 1;
    snprintf(si->table_name, sizeof(si->table_name), "%s", name);
    snprintf(si->ns, sizeof(si->ns), "%llu", (unsigned long long)group_id);
    db->instances[db->instance_count++] = si;
    if (si_table_init(si) < 0) goto error;
    char sql[512];
    snprintf(sql, sizeof(sql), "CREATE INDEX IF NOT EXISTS \"idx_%s_ttl\" ON \"%s\"(node_id,timestamp);"
        "CREATE INDEX IF NOT EXISTS \"idx_%s_bucket\" ON \"%s\"(bucket,message_id)", name, name, name, name);
    if (sqlite3_exec(db->db, sql, NULL, NULL, NULL) != SQLITE_OK) { db_error(si, "create indexes"); goto error; }
    sqlite3_stmt* st = NULL;
    int rc = si_prep(si, &st, "SELECT COALESCE(MAX(id),0),COALESCE(MAX(timestamp),0) FROM \"%s\"");
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    if (rc == SQLITE_ROW) {
        sqlite3_int64 max_id = sqlite3_column_int64(st, 0);
        if (max_id < 0 || max_id == INT64_MAX) rc = SQLITE_TOOBIG;
        else si->next_id = (uint64_t)max_id + 1;
        si->last_timestamp_ms = (uint64_t)sqlite3_column_int64(st, 1);
    }
    int end_rc = sqlite3_finalize(st);
    if (rc != SQLITE_ROW || end_rc != SQLITE_OK) { db_error(si, "load local sequence"); goto error; }
    if (topo_group_add_peer_ready_cbk(topo_groups_find(inst->topo_groups, group_id), db_sync_on_peer_ready, si) < 0) goto error;
    db_index_request(si);
    si->ttl_timer = uasync_set_timeout(inst->ua, DB_SYNC_TTL_INTERVAL * 10000ULL, si, db_ttl_tick, "history_ttl");
    if (!si->ttl_timer) { db_error(si, "initial TTL timer"); goto error; }
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history table registered tbl=%s group=%s; preflight pending", name, si->ns);
    return si;
error:
    db_sync_instance_remove(si); return NULL;
}
void db_sync_instance_remove(struct DB_SYNC_INSTANCE* si) {
    if (!si || !si->enabled) return;
    si->enabled = 0; struct DB_SYNC* db = si->db_sync;
    topo_group_remove_peer_ready_cbk(topo_groups_find(db->inst->topo_groups, si->group_id), db_sync_on_peer_ready, si);
    merkle_sync_cancel_dataset_ns(db->inst, MT_DATA_MESSAGES, si->ns);
    if (si->index.timer) uasync_cancel_timeout(db->inst->ua, si->index.timer);
    if (si->ttl_timer) uasync_cancel_timeout(db->inst->ua, si->ttl_timer);
    for (struct db_notice* n = db->notices; n;) {
        struct db_notice* next = n->next; if (n->group_id == si->group_id) db_notice_abort(n); n = next;
    }
    for (int i = 0; i < db->instance_count; i++) if (db->instances[i] == si) {
        memmove(db->instances + i, db->instances + i + 1, (size_t)(db->instance_count - i - 1) * sizeof(*db->instances));
        db->instance_count--; break;
    }
    u_free(si->peers); si->peers = NULL; si->peer_count = 0;
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history table detached tbl=%s; signed records preserved", SI_TBL(si));
    if (!si->holds) u_free(si);
}
void db_sync_destroy(struct UTUN_INSTANCE* inst) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL;
    if (!db) return;
    db->enabled = 0;
    if (db->peer_check_timer) uasync_cancel_timeout(inst->ua, db->peer_check_timer);
#ifdef UTUN_HAVE_STANDBY
    if (db->peer_check_wait) standby_wait_cancel(db->peer_check_wait);
#endif
    while (db->instance_count) db_sync_instance_remove(db->instances[db->instance_count - 1]);
    while (db->notices) db_notice_abort(db->notices);
    while (db->done_cbks) {
        struct db_sync_done_cbk_entry* e = db->done_cbks; db->done_cbks = e->next; u_free(e);
    }
    etcp_unbind(inst, ETCP_ID_DB_SYNC);
    inst->db_sync = NULL; u_free(db->instances); u_free(db);
}

int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* data, size_t len, const uint8_t* sig, size_t sig_len,
        uint64_t timestamp, const char* attrs, const struct db_message_revision* revision) {
    if (!si || !si->enabled || !sig || sig_len != 64 || !revision) {
        DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "history invalid insertion"); return -1;
    }
    struct db_record r = {.timestamp = timestamp, .author = si->db_sync->inst->node_id, .data = data, .data_len = len, .sig_len = sig_len};
    r.revision = *revision;
    memcpy(r.signature, sig, 64);
    int result = db_insert_one(si, &r, attrs);
    if (!result) db_emit_push(si, &r, r.author);
    return result;
}
uint32_t db_sync_count(struct DB_SYNC_INSTANCE* si) {
    if (!si || !si->enabled) return 0;
    sqlite3_stmt* st = NULL; int rc = si_prep(si, &st, "SELECT COUNT(*) FROM \"%s\"");
    if (rc == SQLITE_OK) rc = sqlite3_step(st);
    uint32_t count = rc == SQLITE_ROW ? (uint32_t)sqlite3_column_int64(st, 0) : 0;
    if (rc != SQLITE_ROW) db_error(si, "count"); sqlite3_finalize(st); return count;
}
uint64_t db_sync_get_last_timestamp(struct DB_SYNC_INSTANCE* si) { return si ? si->last_timestamp_ms : 0; }
uint64_t db_sync_next_timestamp(struct DB_SYNC_INSTANCE* si) {
    if (!si) return 0; uint64_t now = db_now_ms(si);
    if (now <= si->last_timestamp_ms) now = si->last_timestamp_ms + 1;
    return si->last_timestamp_ms = now;
}
void db_sync_set_insert_cb(struct DB_SYNC_INSTANCE* si, db_sync_insert_cb fn, void* arg) {
    if (si) { si->on_insert = fn; si->on_insert_arg = arg; }
}
int db_sync_select(struct DB_SYNC_INSTANCE* si, uint32_t offset, uint32_t limit, db_sync_select_cb cb, void* arg) {
    if (!si || !si->enabled || !cb) return -1;
    sqlite3_stmt* st = NULL; si->holds++;
    int rc = si_prep(si, &st, "SELECT id,timestamp,node_id,data,author_signature,delivered_peers,delivery_chain"
        " FROM \"%s\" ORDER BY timestamp,message_id LIMIT ? OFFSET ?");
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, limit ? limit : -1LL);
    if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 2, offset);
    int count = 0;
    if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW && si->enabled) {
        const char* data = sqlite3_column_blob(st, 3); int bytes = sqlite3_column_bytes(st, 3);
        const uint8_t* signature = sqlite3_column_blob(st, 4); int sig_len = sqlite3_column_bytes(st, 4);
        const char* chain = (const char*)sqlite3_column_text(st, 6);
        if (sqlite3_errcode(SI_DB(si)) == SQLITE_NOMEM) { rc = SQLITE_NOMEM; break; }
        cb(arg, (uint64_t)sqlite3_column_int64(st, 0), (uint64_t)sqlite3_column_int64(st, 1), data, (size_t)bytes,
            (uint64_t)sqlite3_column_int64(st, 2), signature, (size_t)sig_len, sqlite3_column_int(st, 5), chain ? chain : "");
        count++;
    }
    int end_rc = sqlite3_finalize(st);
    if ((rc != SQLITE_DONE && si->enabled) || end_rc != SQLITE_OK) { db_error(si, "select"); count = -1; }
    db_release(si); return count;
}

/* Диагностический синхронный обход: в рабочем обмене preflight идёт порциями по таймеру. */
int db_sync_verify(struct DB_SYNC_INSTANCE* si) {
    if (!si || !si->enabled || !merkle_sync_registered(si->db_sync->inst, MT_DATA_MESSAGES)) return -1;
    if (!si->index.pending) db_index_request(si);
    while (si->index.pending) if (db_index_batch(si) != 0) return si->index.waiting_key ? 1 : -1;
    if (si->index.timer) { uasync_cancel_timeout(si->db_sync->inst->ua, si->index.timer); si->index.timer = NULL; }
    uint8_t root[32]; return merkle_sync_read_hash(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, 0, 0, root);
}
int db_sync_read_root(struct DB_SYNC_INSTANCE* si, uint8_t root[32]) {
    if (!si || !si->enabled || !root) return -1;
    if (si->index.pending) return 1;
    return merkle_sync_read_hash(si->db_sync->inst, MT_DATA_MESSAGES, si->ns, 0, 0, root);
}
void db_sync_peer_set_paused(struct DB_SYNC_INSTANCE* si, uint64_t peer, int paused) {
    struct SI_PEER* p = si ? si_peer_add(si, peer) : NULL;
    if (!p) return;
    p->paused = !!paused; p->sync_state = 0;
    DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "history peer pause tbl=%s peer=%016llx paused=%u",
        SI_TBL(si), (unsigned long long)peer, p->paused);
    if (paused) merkle_sync_cancel(si->db_sync->inst, MT_DATA_MESSAGES, peer, si->ns);
    else db_start_peer(si, peer);
}
void db_sync_reinitiate(struct DB_SYNC_INSTANCE* si, uint64_t peer) {
    if (!si) return;
    struct SI_PEER* p = si_peer_add(si, peer); if (!p) return;
    p->paused = 0; p->sync_state = 0; db_start_peer(si, peer);
}
void db_sync_instance_set_gated(struct DB_SYNC_INSTANCE* si, int gated) {
    if (!si || !si->enabled) return;
    si->sync_gated = !!gated;
    if (gated) merkle_sync_cancel_dataset_ns(si->db_sync->inst, MT_DATA_MESSAGES, si->ns);
    struct TOPO_GROUP* group = topo_groups_find(si->db_sync->inst->topo_groups, si->group_id);
    for (struct ll_entry* e = group && group->senders_list ? group->senders_list->head : NULL; e; e = e->next) {
        struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
        struct SI_PEER* p = si_peer_add(si, item->node_id); if (p) p->sync_state = 0;
        if (!gated) db_start_peer(si, item->node_id);
    }
    db_sync_resume_peer_check(si->db_sync);
}
void db_sync_peer_policy_changed(struct UTUN_INSTANCE* inst, uint64_t peer) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL; if (!db) return;
    merkle_sync_recheck_peer(inst, peer);
    for (int i = 0; i < db->instance_count; i++) {
        struct DB_SYNC_INSTANCE* si = db->instances[i]; struct SI_PEER* p = si_peer_find(si, peer);
        if (p && !p->sync_state) db_start_peer(si, peer);
    }
}
void db_sync_set_peer_sound(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, int sound) {
    struct DB_SYNC_INSTANCE* si = db_sync_instance_find(inst, group);
    struct SI_PEER* p = si ? si_peer_add(si, peer) : NULL;
    if (p) { p->sound_on = !!sound; db_sync_peer_policy_changed(inst, peer); }
}
void db_sync_members_changed(struct UTUN_INSTANCE* inst, const char* ns) {
    struct DB_SYNC* db = inst ? inst->db_sync : NULL; if (!db) return;
    merkle_sync_revalidate_peers(inst, MT_DATA_MESSAGES, ns);
    struct DB_SYNC_INSTANCE* channel = db_sync_instance_find(inst, strtoull(ns, NULL, 10));
    if (channel) {
        for (int i = 0; i < channel->peer_count; i++)
            if (!channel->peers[i].sync_state) db_start_peer(channel, channel->peers[i].node_id);
    }
    for (int i = 0; i < db->instance_count; i++) {
        struct DB_SYNC_INSTANCE* si = db->instances[i];
        if (si->index.pending && si->index.waiting_key) {
            if (si->index.timer) uasync_cancel_timeout(inst->ua, si->index.timer);
            si->index.timer = uasync_set_timeout(inst->ua, 0, si, db_index_tick, "history_key_ready");
            if (!si->index.timer) db_sync_resume_peer_check(db);
        }
    }
}