#include "member_sync.h"
#include "topo_node_sqlite.h"

#include "../../../src/utun_instance.h"
#include "../../../src/etcp_router.h"
#include "../../../src/etcp_api.h"
#include "../../../src/etcp.h"
#include "../../../src/topo_group.h"
#include "../../../lib/debug_config.h"
#include "../../../lib/mem.h"
#include "../../../lib/u_async.h"

#include <string.h>
#include <stdlib.h>
#include <openssl/evp.h>
#include <sqlite3.h>

#define MS_ID "member_sync"
#define MS_SYNC_TIMEOUT_MS 10000
#define MS_BG_INTERVAL_MS  100

static struct member_sync* g_ms = NULL;

struct bucket_entry {
    uint8_t  level;
    uint8_t  prefix_bytes;
    uint64_t prefix;
};

struct addr_item {
    uint8_t family;
    uint8_t addr[16];
    uint16_t port;
};

struct ms_session {
    struct ms_session* next;
    char      ch_id[64];
    uint64_t  peer;
    uint8_t   active;
    void*     timer;
    uint32_t  started_tb;
    uint8_t   retries;
};

struct member_sync {
    struct UTUN_INSTANCE* inst;
    struct ms_session*    sessions;
    uint8_t               initialized;
    void*                 bg_timer;
};

static sqlite3* _db(struct UTUN_INSTANCE* inst) {
    return inst && inst->topo_groups ? inst->topo_groups->topo_sqlite_db : NULL;
}

static uint64_t _level_prefix(uint64_t member_id, uint8_t level) {
    int shift = 63 - (int)level * 5;
    if (shift < 0) shift = 0;
    return (member_id >> shift) << shift;
}

static uint8_t _prefix_bytes(uint8_t level) {
    int bits = level * 5;
    return (uint8_t)((bits + 7) / 8);
}

static void _prefix_write(uint8_t* out, uint64_t prefix, uint8_t pb) {
    for (int i = (int)pb - 1; i >= 0; i--)
        out[pb - 1 - i] = (uint8_t)(prefix >> (i * 8));
}

static uint64_t _prefix_read(const uint8_t* in, uint8_t pb) {
    uint64_t v = 0;
    for (uint8_t i = 0; i < pb && i < 8; i++) v = (v << 8) | in[i];
    return v;
}

static int _addr_cmp(const void* a, const void* b) {
    const struct addr_item* ia = (const struct addr_item*)a;
    const struct addr_item* ib = (const struct addr_item*)b;
    if (ia->family != ib->family) return (int)ia->family - (int)ib->family;
    int r = memcmp(ia->addr, ib->addr, (size_t)(ia->family == 4 ? 4 : 16));
    if (r) return r;
    return (int)ia->port - (int)ib->port;
}

static void _compute_member_hash(uint64_t node_id, const uint8_t* x25519,
                                  const uint8_t* ed25519, const uint8_t* join_sig,
                                  const uint8_t* addrs_data, int addr_count, int online,
                                  uint8_t hash_out[MS_HASH_SIZE]) {
    EVP_MD_CTX* ctx = EVP_MD_CTX_new();
    EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
    EVP_DigestUpdate(ctx, &node_id, 8);
    EVP_DigestUpdate(ctx, x25519, 32);
    EVP_DigestUpdate(ctx, ed25519, 32);
    EVP_DigestUpdate(ctx, join_sig, 64);
    uint8_t ac = (uint8_t)addr_count;
    EVP_DigestUpdate(ctx, &ac, 1);

    if (addr_count > 0 && addrs_data) {
        struct addr_item items[256]; int n = 0;
        const uint8_t* p = addrs_data;
        for (int i = 0; i < addr_count && n < 256; i++) {
            uint8_t fam = *p++;
            items[n].family = fam;
            int ip_len = fam == 4 ? 4 : 16;
            memcpy(items[n].addr, p, (size_t)ip_len); p += ip_len;
            items[n].port = ((uint16_t)p[0] << 8) | p[1]; p += 2;
            n++;
        }
        qsort(items, (size_t)n, sizeof(struct addr_item), _addr_cmp);
        for (int i = 0; i < n; i++) {
            EVP_DigestUpdate(ctx, &items[i].family, 1);
            EVP_DigestUpdate(ctx, items[i].addr, (size_t)(items[i].family == 4 ? 4 : 16));
            uint8_t port_be[2] = { (uint8_t)(items[i].port >> 8), (uint8_t)(items[i].port & 0xFF) };
            EVP_DigestUpdate(ctx, port_be, 2);
        }
    }

    uint8_t ol = (uint8_t)(online ? 1 : 0);
    EVP_DigestUpdate(ctx, &ol, 1);
    EVP_DigestFinal_ex(ctx, hash_out, NULL);
    EVP_MD_CTX_free(ctx);
}

static void _peers_table(const char* ch_id, char* buf, size_t sz) {
    size_t i = 0;
    while (*ch_id && i < sz - 1) {
        char c = *ch_id++;
        if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_')
            buf[i++] = c;
        else
            buf[i++] = '_';
    }
    buf[i] = '\0';
    char tbl[128]; snprintf(tbl, sizeof(tbl), "peers_%s", buf);
    snprintf(buf, sz, "%s", tbl);
}

static int _recompute_bucket(struct UTUN_INSTANCE* inst, const char* ch_id,
                              uint8_t level, uint64_t prefix64) {
    sqlite3* db = _db(inst); if (!db) return -1;
    char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
    char sql[512];
    int mask_shift = 63 - (int)level * 5;
    uint64_t mask = (mask_shift >= 0) ? (~0ULL << mask_shift) : UINT64_MAX;
    snprintf(sql, sizeof(sql),
        "SELECT p.node_id, n.x25519_pubkey, n.ed25519_pubkey, p.join_sig, n.online"
        " FROM \"%s\" p JOIN nodes n ON p.node_id=n.node_id"
        " WHERE (p.node_id & %lld) == %lld ORDER BY p.node_id ASC",
        peers_tbl, (long long)mask, (long long)prefix64);

    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;

    EVP_MD_CTX* ctx = EVP_MD_CTX_new();
    EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
    int count = 0;

    while (sqlite3_step(stmt) == SQLITE_ROW) {
        uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
        const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
        const uint8_t* ed  = (const uint8_t*)sqlite3_column_blob(stmt, 2);
        const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
        int online = sqlite3_column_int(stmt, 4);
        if (!x25 || !ed || !sig) continue;

        uint8_t mh[MS_HASH_SIZE];
        _compute_member_hash(nid, x25, ed, sig, NULL, 0, online, mh);
        EVP_DigestUpdate(ctx, mh, MS_HASH_SIZE);
        count++;
    }
    sqlite3_finalize(stmt);

    if (count == 0) {
        EVP_MD_CTX_free(ctx);
        snprintf(sql, sizeof(sql),
            "DELETE FROM member_tree_hash WHERE channel_id=? AND level=? AND prefix64=?");
        sqlite3_stmt* ds = NULL;
        if (sqlite3_prepare_v2(db, sql, -1, &ds, NULL) == SQLITE_OK) {
            sqlite3_bind_text(ds, 1, ch_id, -1, SQLITE_STATIC);
            sqlite3_bind_int(ds, 2, level);
            sqlite3_bind_int64(ds, 3, (sqlite3_int64)prefix64);
            sqlite3_step(ds); sqlite3_finalize(ds);
        }
        return 0;
    }

    uint8_t hash[MS_HASH_SIZE];
    EVP_DigestFinal_ex(ctx, hash, NULL);
    EVP_MD_CTX_free(ctx);

    snprintf(sql, sizeof(sql),
        "INSERT OR REPLACE INTO member_tree_hash(channel_id, level, prefix64, hash, member_count)"
        " VALUES(?,?,?,?,?)");
    sqlite3_stmt* is = NULL;
    if (sqlite3_prepare_v2(db, sql, -1, &is, NULL) == SQLITE_OK) {
        sqlite3_bind_text(is, 1, ch_id, -1, SQLITE_STATIC);
        sqlite3_bind_int(is, 2, level);
        sqlite3_bind_int64(is, 3, (sqlite3_int64)prefix64);
        sqlite3_bind_blob(is, 4, hash, MS_HASH_SIZE, SQLITE_STATIC);
        sqlite3_bind_int(is, 5, count);
        sqlite3_step(is); sqlite3_finalize(is);
    }
    return 0;
}

static void _recompute_path(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id) {
    for (uint8_t level = 1; level <= MS_MAX_LEVEL; level++)
        _recompute_bucket(inst, ch_id, level, _level_prefix(member_id, level));
}

static void _ensure_table(void) {
    sqlite3* db = g_ms ? _db(g_ms->inst) : NULL; if (!db) return;
    sqlite3_exec(db,
        "CREATE TABLE IF NOT EXISTS member_tree_hash ("
        "  channel_id   TEXT NOT NULL,"
        "  level        INTEGER NOT NULL CHECK(level BETWEEN 1 AND 5),"
        "  prefix64     INTEGER NOT NULL,"
        "  hash         BLOB NOT NULL,"
        "  member_count INTEGER NOT NULL,"
        "  PRIMARY KEY (channel_id, level, prefix64))",
        NULL, NULL, NULL);
}

const uint8_t* member_sync_get_hash(struct UTUN_INSTANCE* inst, const char* ch_id,
                                     uint8_t level, uint64_t prefix64) {
    static uint8_t zero[MS_HASH_SIZE]; /* returns static — not thread-safe but single-threaded uasync */
    sqlite3* db = _db(inst); if (!db) { memset(zero, 0, MS_HASH_SIZE); return zero; }
    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db,
            "SELECT hash FROM member_tree_hash WHERE channel_id=? AND level=? AND prefix64=?",
            -1, &stmt, NULL) != SQLITE_OK) { memset(zero, 0, MS_HASH_SIZE); return zero; }
    sqlite3_bind_text(stmt, 1, ch_id, -1, SQLITE_STATIC);
    sqlite3_bind_int(stmt, 2, level);
    sqlite3_bind_int64(stmt, 3, (sqlite3_int64)prefix64);
    memset(zero, 0, MS_HASH_SIZE);
    if (sqlite3_step(stmt) == SQLITE_ROW) {
        const void* h = sqlite3_column_blob(stmt, 0);
        if (h) memcpy(zero, h, MS_HASH_SIZE);
    }
    sqlite3_finalize(stmt);
    return zero;
}

int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
                    uint64_t member_id, const uint8_t* x25519,
                    const uint8_t* ed25519, const uint8_t* join_sig,
                    const uint8_t* addrs_data, int addr_count) {
    if (!inst || !ch_id) return -1;
    sqlite3* db = _db(inst); if (!db) return -1;
    _ensure_table();

    sqlite3_stmt* ns = NULL;
    sqlite3_prepare_v2(db,
        "INSERT OR REPLACE INTO nodes(node_id, name, x25519_pubkey, ed25519_pubkey, online)"
        " VALUES(?,COALESCE((SELECT name FROM nodes WHERE node_id=?),''),?,?,"
        " COALESCE((SELECT online FROM nodes WHERE node_id=?),0))",
        -1, &ns, NULL);
    if (ns) {
        sqlite3_bind_int64(ns, 1, (sqlite3_int64)member_id);
        sqlite3_bind_int64(ns, 2, (sqlite3_int64)member_id);
        sqlite3_bind_blob(ns, 3, x25519, 32, SQLITE_STATIC);
        sqlite3_bind_blob(ns, 4, ed25519, 32, SQLITE_STATIC);
        sqlite3_bind_int64(ns, 5, (sqlite3_int64)member_id);
        sqlite3_step(ns); sqlite3_finalize(ns);
    }

    topo_node_sqlite_member_put(db, ch_id, member_id, join_sig, NULL);

    if (addrs_data && addr_count > 0) {
        sqlite3_exec(db, "BEGIN", NULL, NULL, NULL);
        sqlite3_stmt* ds = NULL;
        sqlite3_prepare_v2(db, "DELETE FROM node_addresses WHERE node_id=? AND is_nat=0", -1, &ds, NULL);
        if (ds) { sqlite3_bind_int64(ds, 1, (sqlite3_int64)member_id); sqlite3_step(ds); sqlite3_finalize(ds); }

        sqlite3_stmt* as = NULL;
        sqlite3_prepare_v2(db,
            "INSERT INTO node_addresses(node_id,family,protocol,address,port,is_nat)"
            " VALUES(?,?,1,?,?,0)", -1, &as, NULL);
        if (as) {
            const uint8_t* p = addrs_data;
            for (int i = 0; i < addr_count; i++) {
                uint8_t fam = *p++; int ip_sz = fam == 4 ? 4 : 16;
                sqlite3_bind_int64(as, 1, (sqlite3_int64)member_id);
                sqlite3_bind_int(as, 2, fam);
                sqlite3_bind_blob(as, 3, p, ip_sz, SQLITE_STATIC);
                p += ip_sz;
                uint16_t port = ((uint16_t)p[0] << 8) | p[1]; p += 2;
                sqlite3_bind_int(as, 4, port);
                sqlite3_step(as); sqlite3_reset(as);
            }
            sqlite3_finalize(as);
        }
        sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
    }

    _recompute_path(inst, ch_id, member_id);
    return 0;
}

int member_sync_del(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id) {
    if (!inst || !ch_id) return -1;
    sqlite3* db = _db(inst); if (!db) return -1;
    topo_node_sqlite_member_del(db, ch_id, member_id);
    _recompute_path(inst, ch_id, member_id);
    return 0;
}

void member_sync_set_online(struct UTUN_INSTANCE* inst, uint64_t node_id, int online) {
    if (!inst) return;
    sqlite3* db = _db(inst); if (!db) return;
    topo_node_sqlite_node_set_online(db, node_id, online);
    if (!g_ms || !g_ms->initialized) return;
    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db, "SELECT channel_id FROM channels", -1, &stmt, NULL) != SQLITE_OK) return;
    while (sqlite3_step(stmt) == SQLITE_ROW) {
        const char* ch_id = (const char*)sqlite3_column_text(stmt, 0);
        if (!ch_id) continue;
        char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
        char buf[256]; snprintf(buf, sizeof(buf),
            "SELECT 1 FROM \"%s\" WHERE node_id=?", peers_tbl);
        sqlite3_stmt* cs = NULL;
        if (sqlite3_prepare_v2(db, buf, -1, &cs, NULL) == SQLITE_OK) {
            sqlite3_bind_int64(cs, 1, (sqlite3_int64)node_id);
            if (sqlite3_step(cs) == SQLITE_ROW) _recompute_path(inst, ch_id, node_id);
            sqlite3_finalize(cs);
        }
    }
    sqlite3_finalize(stmt);
}

int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id) {
    sqlite3* db = _db(inst);
    if (!db || !ch_id) return 0;
    char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
    char sql[256]; snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM \"%s\"", peers_tbl);
    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return 0;
    int c = 0;
    if (sqlite3_step(stmt) == SQLITE_ROW) c = sqlite3_column_int(stmt, 0);
    sqlite3_finalize(stmt);
    return c;
}

int member_sync_get_level_hashes(struct UTUN_INSTANCE* inst, const char* ch_id,
                                  uint8_t level, uint64_t prefix, uint8_t prefix_bytes,
                                  uint32_t* bitmap, uint8_t hashes[MS_BUCKETS][MS_HASH_SIZE]) {
    (void)prefix_bytes;
    memset(hashes, 0, sizeof(uint8_t) * MS_BUCKETS * MS_HASH_SIZE);
    *bitmap = 0;
    if (level >= MS_MAX_LEVEL) return 0;
    int next_shift = 63 - ((int)level + 1) * 5;

    for (int i = 0; i < MS_BUCKETS; i++) {
        uint64_t child_prefix = prefix | ((uint64_t)i << (next_shift > 0 ? next_shift : 0));
        const uint8_t* h = member_sync_get_hash(inst, ch_id, (uint8_t)(level + 1), child_prefix);
        int empty = 1;
        for (int j = 0; j < MS_HASH_SIZE; j++) { if (h[j] != 0) { empty = 0; break; } }
        if (!empty) { *bitmap |= (1u << i); memcpy(hashes[i], h, MS_HASH_SIZE); }
    }
    return 0;
}

int member_sync_get_bucket_members(struct UTUN_INSTANCE* inst, const char* ch_id,
                                    uint8_t level, uint64_t prefix, uint8_t prefix_bytes,
                                    uint8_t* buf, size_t* len) {
    (void)prefix_bytes;
    sqlite3* db = _db(inst); if (!db || !buf || !len) return -1;
    char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
    char sql[512];
    int mask_shift = 63 - (int)level * 5;
    uint64_t mask = (mask_shift >= 0) ? (~0ULL << mask_shift) : UINT64_MAX;
    snprintf(sql, sizeof(sql),
        "SELECT p.node_id, n.x25519_pubkey, n.ed25519_pubkey, p.join_sig, n.online"
        " FROM \"%s\" p JOIN nodes n ON p.node_id=n.node_id"
        " WHERE (p.node_id & %lld) == %lld ORDER BY p.node_id ASC",
        peers_tbl, (long long)mask, (long long)prefix);

    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1;

    size_t off = 0;
    if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; }
    uint16_t* cnt = (uint16_t*)(buf + off); off += 2; *cnt = 0;

    while (sqlite3_step(stmt) == SQLITE_ROW) {
        uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
        const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
        const uint8_t* ed  = (const uint8_t*)sqlite3_column_blob(stmt, 2);
        const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
        int online = sqlite3_column_int(stmt, 4);
        if (!x25 || !ed || !sig) continue;

        sqlite3_stmt* as = NULL;
        sqlite3_prepare_v2(db,
            "SELECT family, address, port FROM node_addresses WHERE node_id=? AND is_nat=0"
            " ORDER BY family, address, port", -1, &as, NULL);
        uint8_t addrs[2048]; int addr_off = 0; int addr_count = 0;
        if (as) {
            sqlite3_bind_int64(as, 1, (sqlite3_int64)nid);
            while (sqlite3_step(as) == SQLITE_ROW && addr_off < (int)sizeof(addrs) - 7) {
                int fam = sqlite3_column_int(as, 0);
                addrs[addr_off++] = (uint8_t)fam;
                int ip_sz = fam == 4 ? 4 : 16;
                memcpy(addrs + addr_off, sqlite3_column_blob(as, 1), (size_t)ip_sz);
                addr_off += ip_sz;
                uint16_t p = (uint16_t)sqlite3_column_int(as, 2);
                addrs[addr_off++] = (uint8_t)(p >> 8);
                addrs[addr_off++] = (uint8_t)(p & 0xFF);
                addr_count++;
            }
            sqlite3_finalize(as);
        }

        size_t need = 8 + 32 + 32 + 64 + 1 + 1 + (size_t)addr_off;
        if (off + need > *len) { sqlite3_finalize(stmt); return -2; }
        memcpy(buf + off, &nid, 8); off += 8;
        memcpy(buf + off, x25, 32); off += 32;
        memcpy(buf + off, ed, 32); off += 32;
        memcpy(buf + off, sig, 64); off += 64;
        buf[off++] = (uint8_t)(online ? 1 : 0);
        buf[off++] = (uint8_t)addr_count;
        memcpy(buf + off, addrs, (size_t)addr_off); off += (size_t)addr_off;
        (*cnt)++;
    }
    sqlite3_finalize(stmt);
    *len = off;
    return 0;
}

static int _send_msg(struct UTUN_INSTANCE* inst, uint64_t peer,
                     const uint8_t* payload, size_t len) {
    if (!inst || len < 1) return -1;
    uint8_t* buf = u_malloc(1 + len);
    if (!buf) return -1;
    buf[0] = ETCP_RT_ID_MEMBER_SYNC;
    memcpy(buf + 1, payload, len);
    struct ll_entry* entry = queue_entry_new(0);
    if (!entry) { u_free(buf); return -1; }
    entry->dgram = buf; entry->len = 1 + len;
    int r = etcp_route_send(inst, peer, entry, 0);
    if (r != 0) { u_free(buf); queue_entry_free(entry); }
    return r;
}

static int _popcount_u32(uint32_t v) {
    return __builtin_popcount(v);
}
#ifndef __has_builtin
#define __has_builtin(x) 0
#endif
#if !defined(__GNUC__) && !defined(__clang__)
static int _popcount_u32(uint32_t v) {
    v = v - ((v >> 1) & 0x55555555);
    v = (v & 0x33333333) + ((v >> 2) & 0x33333333);
    return ((v + (v >> 4) & 0x0F0F0F0F) * 0x01010101) >> 24;
}
#endif

static int _send_hashes(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id,
                         uint8_t level, uint64_t prefix, uint8_t prefix_bytes, int is_data) {
    uint8_t ch_len = (uint8_t)strlen(ch_id);
    size_t max_sz = 1 + 1 + ch_len + 1 + 1 + prefix_bytes + 1 + 4 + MS_BUCKETS * MS_HASH_SIZE + 2 + 65536;
    uint8_t* buf = u_malloc(max_sz);
    if (!buf) return -1;
    uint8_t* p = buf;
    *p++ = MS_MSG_HASHES;
    *p++ = ch_len; memcpy(p, ch_id, ch_len); p += ch_len;
    *p++ = level;
    *p++ = prefix_bytes; _prefix_write(p, prefix, prefix_bytes); p += prefix_bytes;
    *p++ = (uint8_t)(is_data ? 1 : 0);

    if (is_data) {
        size_t mlen = 65536;
        uint8_t* mbuf = u_malloc(mlen);
        if (mbuf) {
            if (member_sync_get_bucket_members(inst, ch_id, level, prefix, prefix_bytes, mbuf, &mlen) == 0)
                { memcpy(p, mbuf, mlen); p += mlen; }
            else { uint16_t zero = 0; memcpy(p, &zero, 2); p += 2; }
            u_free(mbuf);
        }
    } else {
        uint32_t bitmap; uint8_t hashes[MS_BUCKETS][MS_HASH_SIZE];
        member_sync_get_level_hashes(inst, ch_id, level, prefix, prefix_bytes, &bitmap, hashes);
        memcpy(p, &bitmap, 4); p += 4;
        for (int i = 0; i < MS_BUCKETS; i++)
            if (bitmap & (1u << i)) { memcpy(p, hashes[i], MS_HASH_SIZE); p += MS_HASH_SIZE; }
    }

    int r = _send_msg(inst, peer, buf, (size_t)(p - buf));
    u_free(buf);
    return r;
}

static int _send_batch(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id,
                       struct bucket_entry* buckets, int count) {
    uint8_t ch_len = (uint8_t)strlen(ch_id);
    size_t max_sz = 1 + 1 + ch_len + 1 + (size_t)count * (1 + 1 + 8 + 1 + 4 + MS_BUCKETS * MS_HASH_SIZE + 65536);
    uint8_t* buf = u_malloc(max_sz);
    if (!buf) return -1;
    uint8_t* p = buf;
    *p++ = MS_MSG_BATCH;
    *p++ = ch_len; memcpy(p, ch_id, ch_len); p += ch_len;
    *p++ = (uint8_t)count;

    for (int i = 0; i < count; i++) {
        *p++ = buckets[i].level;
        *p++ = buckets[i].prefix_bytes;
        _prefix_write(p, buckets[i].prefix, buckets[i].prefix_bytes); p += buckets[i].prefix_bytes;

        int bcount = 0;
        uint64_t mask = 0;
        int mask_shift = 63 - (int)buckets[i].level * 5;
        if (mask_shift >= 0) mask = ~0ULL << mask_shift; else mask = UINT64_MAX;
        char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
        sqlite3* db = _db(inst);
        if (db) {
            char sql[256]; snprintf(sql, sizeof(sql),
                "SELECT COUNT(*) FROM \"%s\" p WHERE (p.node_id & %lld)==%lld",
                peers_tbl, (long long)mask, (long long)buckets[i].prefix);
            sqlite3_stmt* cs = NULL;
            if (sqlite3_prepare_v2(db, sql, -1, &cs, NULL) == SQLITE_OK) {
                if (sqlite3_step(cs) == SQLITE_ROW) bcount = sqlite3_column_int(cs, 0);
                sqlite3_finalize(cs);
            }
        }

        int is_terminal = (buckets[i].level >= MS_MAX_LEVEL || bcount < 8);
        *p++ = (uint8_t)(is_terminal ? 1 : 0);

        if (is_terminal) {
            size_t mlen = 65536;
            uint8_t* mbuf = u_malloc(mlen);
            if (mbuf) {
                if (member_sync_get_bucket_members(inst, ch_id, buckets[i].level,
                        buckets[i].prefix, buckets[i].prefix_bytes, mbuf, &mlen) == 0)
                    { memcpy(p, mbuf, mlen); p += mlen; }
                else { uint16_t z = 0; memcpy(p, &z, 2); p += 2; }
                u_free(mbuf);
            }
        } else {
            uint32_t bm; uint8_t hs[MS_BUCKETS][MS_HASH_SIZE];
            member_sync_get_level_hashes(inst, ch_id, buckets[i].level,
                                          buckets[i].prefix, buckets[i].prefix_bytes, &bm, hs);
            memcpy(p, &bm, 4); p += 4;
            for (int j = 0; j < MS_BUCKETS; j++)
                if (bm & (1u << j)) { memcpy(p, hs[j], MS_HASH_SIZE); p += MS_HASH_SIZE; }
        }
    }

    int r = _send_msg(inst, peer, buf, (size_t)(p - buf));
    u_free(buf);
    return r;
}

/* ── sessions ── */

static void _session_start_timer(struct ms_session* s);

static void _session_timeout_cb(void* arg) {
    struct ms_session* s = (struct ms_session*)arg;
    if (!s || !s->active || !g_ms || !g_ms->initialized) return;
    s->timer = NULL;
    s->retries++;
    if (s->retries > 3) {
        DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: sync timeout peer=%016llx ch=%s", MS_ID,
                   (unsigned long long)s->peer, s->ch_id);
        s->active = 0;
        return;
    }
    DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: retry %d peer=%016llx ch=%s", MS_ID,
               s->retries, (unsigned long long)s->peer, s->ch_id);
    _send_hashes(g_ms->inst, s->peer, s->ch_id, 1, 0, 1, 0);
    _session_start_timer(s);
}

static void _session_start_timer(struct ms_session* s) {
    if (!g_ms || !g_ms->inst) return;
    s->timer = uasync_set_timeout(g_ms->inst->ua, (uint32_t)(MS_SYNC_TIMEOUT_MS * 10),
                                   s, _session_timeout_cb, "ms_sync");
}

static struct ms_session* _session_find(uint64_t peer, const char* ch_id) {
    for (struct ms_session* s = g_ms ? g_ms->sessions : NULL; s; s = s->next)
        if (s->peer == peer && strcmp(s->ch_id, ch_id) == 0) return s;
    return NULL;
}

void member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) {
    if (!inst || !ch_id || !g_ms || !g_ms->initialized) return;
    _ensure_table();

    struct ms_session* s = _session_find(peer, ch_id);
    if (!s) {
        s = u_calloc(1, sizeof(*s));
        if (!s) return;
        snprintf(s->ch_id, sizeof(s->ch_id), "%s", ch_id);
        s->peer = peer;
        s->next = g_ms->sessions;
        g_ms->sessions = s;
    }
    s->active = 1; s->retries = 0;

    _send_hashes(inst, peer, ch_id, 1, 0, 1, 0);
    _session_start_timer(s);
}

void member_sync_cancel_peer(struct UTUN_INSTANCE* inst, uint64_t peer) {
    (void)inst;
    struct ms_session** p = g_ms ? &g_ms->sessions : NULL;
    while (p && *p) {
        struct ms_session* s = *p;
        if (s->peer == peer) {
            if (s->timer) { uasync_cancel_timeout(g_ms->inst->ua, s->timer); s->timer = NULL; }
            *p = s->next; u_free(s);
        } else {
            p = &(*p)->next;
        }
    }
}

/* ── recv handlers ── */

static void _handle_hashes(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id,
                            const uint8_t* pl, size_t plen) {
    if (plen < 3) return;
    uint8_t level = pl[0]; uint8_t pb = pl[1];
    uint64_t prefix = _prefix_read(pl + 2, pb);
    uint8_t is_data = pl[2 + pb];
    const uint8_t* payload = pl + 3 + pb;
    size_t paylen = plen - 3 - pb;

    struct ms_session* s = _session_find(peer, ch_id);
    if (s && s->timer) { uasync_cancel_timeout(g_ms->inst->ua, s->timer); s->timer = NULL; }

    if (is_data) {
        if (paylen < 2) return;
        uint16_t count; memcpy(&count, payload, 2);
        const uint8_t* mp = payload + 2; size_t mrem = paylen - 2;
        for (uint16_t i = 0; i < count && mrem >= 137; i++) {
            uint64_t nid; memcpy(&nid, mp, 8); mp += 8; mrem -= 8;
            const uint8_t* x25 = mp; mp += 32; mrem -= 32;
            const uint8_t* ed  = mp; mp += 32; mrem -= 32;
            const uint8_t* sig = mp; mp += 64; mrem -= 64;
            uint8_t online = *mp++; mrem--;
            uint8_t ac = *mp++; mrem--;
            const uint8_t* addrs = mp;
            int consumed = 0;
            for (int a = 0; a < (int)ac && mrem >= (size_t)(1 + consumed); a++) {
                uint8_t fam = mp[consumed]; consumed++;
                int sz = fam == 4 ? 4 : 16;
                consumed += sz + 2;
            }
            member_sync_put(inst, ch_id, nid, x25, ed, sig, addrs, (int)ac);
            mp += consumed; mrem -= (size_t)consumed;
        }
    } else {
        if (!s) { s = u_calloc(1, sizeof(*s)); if (!s) return;
            snprintf(s->ch_id, sizeof(s->ch_id), "%s", ch_id); s->peer = peer;
            s->next = g_ms->sessions; g_ms->sessions = s; }
        s->active = 1;

        if (paylen < 4) return;
        uint32_t remote_bm; memcpy(&remote_bm, payload, 4);
        const uint8_t* rp = payload + 4;

        uint32_t local_bm; uint8_t lh[MS_BUCKETS][MS_HASH_SIZE];
        member_sync_get_level_hashes(inst, ch_id, level, prefix, pb, &local_bm, lh);

        uint32_t differs = remote_bm ^ local_bm;
        for (int i = 0; i < MS_BUCKETS; i++) {
            if (!(local_bm & (1u << i)) && !(remote_bm & (1u << i))) continue;
            if ((local_bm & (1u << i)) && (remote_bm & (1u << i))) {
                const uint8_t* rh_ptr = rp;
                int rh_idx = 0;
                for (int j = 0; j < i; j++) if (remote_bm & (1u << j)) rh_idx++;
                if (memcmp(rh_ptr + (size_t)rh_idx * MS_HASH_SIZE, lh[i], MS_HASH_SIZE) == 0)
                    differs &= ~(1u << i);
            }
        }

        if (differs == 0) { s->active = 0; return; }

        int next_shift = 63 - ((int)level + 1) * 5;
        struct bucket_entry requests[MS_BUCKETS]; int rcount = 0;
        for (int i = 0; i < MS_BUCKETS && rcount < MS_MAX_BATCH; i++) {
            if (!(differs & (1u << i))) continue;
            uint8_t nl = (uint8_t)(level + 1);
            if (nl > MS_MAX_LEVEL) nl = MS_MAX_LEVEL;
            requests[rcount].level = nl;
            requests[rcount].prefix_bytes = _prefix_bytes(nl);
            requests[rcount].prefix = prefix | ((uint64_t)i << (next_shift > 0 ? next_shift : 0));
            rcount++;
        }

        if (rcount > 0) {
            uint8_t ch_len = (uint8_t)strlen(ch_id);
            size_t rs = 1 + 1 + ch_len + 1;
            for (int i = 0; i < rcount; i++) rs += 1 + 1 + requests[i].prefix_bytes + 1;
            uint8_t* rbuf = u_malloc(rs);
            if (rbuf) {
                uint8_t* wr = rbuf;
                *wr++ = MS_MSG_REQUEST; *wr++ = ch_len;
                memcpy(wr, ch_id, ch_len); wr += ch_len;
                *wr++ = (uint8_t)rcount;
                for (int i = 0; i < rcount; i++) {
                    *wr++ = requests[i].level;
                    *wr++ = requests[i].prefix_bytes;
                    _prefix_write(wr, requests[i].prefix, requests[i].prefix_bytes);
                    wr += requests[i].prefix_bytes;
                    int cnt = 0; sqlite3* db = _db(inst);
                    if (db) {
                        char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
                        char sql[256]; snprintf(sql, sizeof(sql),
                            "SELECT COUNT(*) FROM \"%s\" WHERE (node_id & "
                            "(CASE WHEN %d>=0 THEN ~0<<%d ELSE ~0 END)) == %lld",
                            peers_tbl, 63-(int)requests[i].level*5, 63-(int)requests[i].level*5,
                            (long long)requests[i].prefix);
                        sqlite3_stmt* cst = NULL;
                        if (sqlite3_prepare_v2(db, sql, -1, &cst, NULL) == SQLITE_OK) {
                            if (sqlite3_step(cst) == SQLITE_ROW) cnt = sqlite3_column_int(cst, 0);
                            sqlite3_finalize(cst);
                        }
                    }
                    *wr++ = (uint8_t)(cnt < 8 || requests[i].level >= MS_MAX_LEVEL ? 1 : 0);
                }
                _send_msg(inst, peer, rbuf, (size_t)(wr - rbuf));
                u_free(rbuf);
            }
        }
    }
}

static void _handle_request(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id,
                             const uint8_t* pl, size_t plen) {
    if (plen < 1) return;
    uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t off = 0;

    struct bucket_entry buckets[MS_MAX_BATCH]; int bc = 0;
    for (uint8_t i = 0; i < count && bc < MS_MAX_BATCH; i++) {
        if (off + 2 > plen - 1) break;
        uint8_t lvl = bp[off++];
        uint8_t pb = bp[off++];
        if (off + pb > plen - 1) break;
        uint64_t pr = _prefix_read(bp + off, pb); off += pb;
        buckets[bc].level = lvl;
        buckets[bc].prefix_bytes = pb;
        buckets[bc].prefix = pr;
        bc++;
        off++; /* skip full_data */
    }

    if (bc > 0) _send_batch(inst, peer, ch_id, buckets, bc);
}

static void _handle_batch(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id,
                           const uint8_t* pl, size_t plen) {
    if (plen < 1) return;
    uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t rem = plen - 1;

    for (uint8_t i = 0; i < count && rem >= 3; i++) {
        uint8_t lvl = bp[0]; uint8_t pb = bp[1]; rem -= 2; bp += 2;
        if (rem < pb + 1) break;
        uint64_t pr = _prefix_read(bp, pb); bp += pb; rem -= pb;
        uint8_t is_data = *bp++; rem--;

        if (is_data && rem >= 2) {
            uint16_t mc; memcpy(&mc, bp, 2); bp += 2; rem -= 2;
            for (uint16_t j = 0; j < mc && rem >= 137; j++) {
                uint64_t nid; memcpy(&nid, bp, 8); bp += 8; rem -= 8;
                const uint8_t* x25 = bp; bp += 32; rem -= 32;
                const uint8_t* ed  = bp; bp += 32; rem -= 32;
                const uint8_t* sig = bp; bp += 64; rem -= 64;
                uint8_t online = *bp++; rem--;
                uint8_t ac = *bp++; rem--;
                const uint8_t* addrs = bp;
                int consumed = 0;
                for (int a = 0; a < (int)ac && rem >= (size_t)(1 + consumed); a++) {
                    uint8_t fam = bp[consumed]; consumed++;
                    int sz = fam == 4 ? 4 : 16;
                    consumed += sz + 2;
                }
                member_sync_put(inst, ch_id, nid, x25, ed, sig, addrs, (int)ac);
                bp += consumed; rem -= (size_t)consumed;
            }
        } else if (!is_data && rem >= 4) {
            uint8_t sub_pl[4096]; size_t sub_len = 0;
            uint8_t next_lvl = (uint8_t)(lvl < MS_MAX_LEVEL ? lvl + 1 : lvl);
            sub_pl[sub_len++] = next_lvl;
            sub_pl[sub_len++] = pb;
            _prefix_write(sub_pl + sub_len, pr, pb); sub_len += pb;
            sub_pl[sub_len++] = 0;
            uint32_t bm; memcpy(&bm, bp, 4); bp += 4; rem -= 4;
            memcpy(sub_pl + sub_len, &bm, 4); sub_len += 4;
            int nh = _popcount_u32(bm);
            if (rem >= (size_t)nh * MS_HASH_SIZE) {
                memcpy(sub_pl + sub_len, bp, (size_t)nh * MS_HASH_SIZE);
                sub_len += (size_t)nh * MS_HASH_SIZE;
                bp += (size_t)nh * MS_HASH_SIZE; rem -= (size_t)nh * MS_HASH_SIZE;
            }
            _handle_hashes(inst, peer, ch_id, sub_pl, sub_len);
        }
    }
}

static void _recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
    if (!entry || entry->len < 4) {
        if (entry) { if (entry->dgram) u_free(entry->dgram); queue_entry_free(entry); }
        return;
    }
    if (!g_ms || !g_ms->initialized) { u_free(entry->dgram); queue_entry_free(entry); return; }
    uint64_t peer = conn ? conn->peer_node_id : 0;
    const uint8_t* d = entry->dgram;
    size_t dlen = entry->len;

    uint8_t ch_len = d[1];
    if (dlen < (size_t)(2 + ch_len + 1)) { u_free(entry->dgram); queue_entry_free(entry); return; }
    char ch_id[64]; memcpy(ch_id, d + 2, ch_len); ch_id[ch_len] = '\0';
    uint8_t type = d[2 + ch_len];
    const uint8_t* pl = d + 3 + ch_len;
    size_t plen = dlen - 3 - ch_len;

    switch (type) {
    case MS_MSG_HASHES:  _handle_hashes(g_ms->inst, peer, ch_id, pl, plen); break;
    case MS_MSG_REQUEST: _handle_request(g_ms->inst, peer, ch_id, pl, plen); break;
    case MS_MSG_BATCH:   _handle_batch(g_ms->inst, peer, ch_id, pl, plen); break;
    }

    u_free(entry->dgram); queue_entry_free(entry);
}

/* ── bg_check ── */

static void _bg_timer_cb(void* arg) {
    struct member_sync* ms = (struct member_sync*)arg;
    if (!ms || !ms->initialized || !ms->inst) return;
    sqlite3* db = _db(ms->inst); if (!db) return;

    sqlite3_stmt* cs = NULL;
    sqlite3_prepare_v2(db, "SELECT channel_id FROM channels", -1, &cs, NULL);
    if (!cs) return;
    while (sqlite3_step(cs) == SQLITE_ROW) {
        const char* ch_id = (const char*)sqlite3_column_text(cs, 0);
        if (!ch_id) continue;
        int rc = member_sync_bg_check(ms->inst, ch_id);
        if (rc < 0) continue;
        break;
    }
    sqlite3_finalize(cs);

    ms->bg_timer = uasync_set_timeout(ms->inst->ua, (uint32_t)(MS_BG_INTERVAL_MS * 10),
                                       ms, _bg_timer_cb, "ms_bg");
}

int member_sync_bg_check(struct UTUN_INSTANCE* inst, const char* ch_id) {
    sqlite3* db = _db(inst); if (!db || !ch_id) return -1;

    sqlite3_stmt* stmt = NULL;
    if (sqlite3_prepare_v2(db,
            "SELECT prefix64 FROM member_tree_hash WHERE channel_id=? AND level=? LIMIT 1",
            -1, &stmt, NULL) != SQLITE_OK) return -1;
    sqlite3_bind_text(stmt, 1, ch_id, -1, SQLITE_STATIC);
    sqlite3_bind_int(stmt, 2, MS_MAX_LEVEL);

    if (sqlite3_step(stmt) != SQLITE_ROW) { sqlite3_finalize(stmt); return -1; }
    uint64_t pref = (uint64_t)sqlite3_column_int64(stmt, 0);
    sqlite3_finalize(stmt);

    uint8_t old_hash[MS_HASH_SIZE];
    memcpy(old_hash, member_sync_get_hash(inst, ch_id, MS_MAX_LEVEL, pref), MS_HASH_SIZE);

    _recompute_bucket(inst, ch_id, MS_MAX_LEVEL, pref);

    const uint8_t* new_hash = member_sync_get_hash(inst, ch_id, MS_MAX_LEVEL, pref);
    if (memcmp(old_hash, new_hash, MS_HASH_SIZE) != 0) {
        for (uint8_t lv = (uint8_t)(MS_MAX_LEVEL - 1); lv >= 1; lv--)
            _recompute_bucket(inst, ch_id, lv, _level_prefix(pref, lv));
        return 1;
    }
    return 0;
}

/* called when topo_group stores updated node info (pubkeys/addresses) to DB */
static void _on_node_updated(struct UTUN_INSTANCE* inst, uint64_t node_id,
                              const uint8_t* x25519, const uint8_t* ed25519) {
    (void)x25519; (void)ed25519;
    if (!inst || !g_ms || !g_ms->initialized) return;
    sqlite3* db = _db(inst); if (!db) return;
    sqlite3_stmt* cs = NULL;
    if (sqlite3_prepare_v2(db, "SELECT channel_id FROM channels", -1, &cs, NULL) != SQLITE_OK) return;
    while (sqlite3_step(cs) == SQLITE_ROW) {
        const char* ch = (const char*)sqlite3_column_text(cs, 0);
        if (!ch) continue;
        char peers_tbl[128]; _peers_table(ch, peers_tbl, sizeof(peers_tbl));
        char buf[256]; snprintf(buf, sizeof(buf), "SELECT 1 FROM \"%s\" WHERE node_id=?", peers_tbl);
        sqlite3_stmt* ps = NULL;
        if (sqlite3_prepare_v2(db, buf, -1, &ps, NULL) == SQLITE_OK) {
            sqlite3_bind_int64(ps, 1, (sqlite3_int64)node_id);
            if (sqlite3_step(ps) == SQLITE_ROW) _recompute_path(inst, ch, node_id);
            sqlite3_finalize(ps);
        }
    }
    sqlite3_finalize(cs);
}

int member_sync_init(struct UTUN_INSTANCE* inst) {
    if (!inst) return -1;
    struct member_sync* ms = u_calloc(1, sizeof(*ms));
    if (!ms) return -1;
    ms->inst = inst;
    ms->initialized = 1;
    g_ms = ms;

    _ensure_table();

    etcp_router_bind(inst, ETCP_RT_ID_MEMBER_SYNC, _recv_cb);

    topo_groups_set_node_updated_cb(inst->topo_groups, _on_node_updated);

    ms->bg_timer = uasync_set_timeout(inst->ua, (uint32_t)(MS_BG_INTERVAL_MS * 10),
                                       ms, _bg_timer_cb, "ms_bg");

    DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized", MS_ID);
    return 0;
}

void member_sync_destroy(struct UTUN_INSTANCE* inst) {
    if (!g_ms || !inst) return;
    g_ms->initialized = 0;

    etcp_router_bind(inst, ETCP_RT_ID_MEMBER_SYNC, NULL);

    if (g_ms->bg_timer) { uasync_cancel_timeout(inst->ua, g_ms->bg_timer); g_ms->bg_timer = NULL; }

    struct ms_session* s = g_ms->sessions;
    while (s) { struct ms_session* next = s->next;
        if (s->timer) uasync_cancel_timeout(inst->ua, s->timer);
        u_free(s); s = next; }

    u_free(g_ms); g_ms = NULL;
}