utun2/tools/chatgui/transport/member_sync.c

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

#include "../../../src/utun_instance.h"
#include "../../../src/topo_group.h"
#include "../../../lib/debug_config.h"
#include "../../../lib/mem.h"

#include <string.h>
#include <stdlib.h>
#include <sqlite3.h>

#define MS_ID "member_sync"

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

/* ── DB access ── */

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

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);
}

/* ── Member hash (identical to old _compute_member_hash) ── */

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[MT_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);
}

/* ──  merkle_sync_data_ops implementation ── */

static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
                                       uint64_t prefix64, EVP_MD_CTX* sha_ctx) {
    struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
    sqlite3* db = _db(inst); if (!db) return -1;
    char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));

    int mask_shift = 63 - (int)level * 5;
    uint64_t mask = (mask_shift >= 0) ? (~0ULL << mask_shift) : UINT64_MAX;

    char sql[512]; 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;

    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[MT_HASH_SIZE];
        _compute_member_hash(nid, x25, ed, sig, NULL, 0, online, mh);
        EVP_DigestUpdate(sha_ctx, mh, MT_HASH_SIZE);
        count++;
    }
    sqlite3_finalize(stmt);
    return count;
}

static int _member_get_items(void* ctx, const char* ns, uint8_t level,
                              uint64_t prefix, uint8_t prefix_bytes,
                              uint8_t* buf, size_t* len) {
    struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
    sqlite3* db = _db(inst); if (!db || !buf || !len) return -1;
    char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));

    int mask_shift = 63 - (int)level * 5;
    uint64_t mask = (mask_shift >= 0) ? (~0ULL << mask_shift) : UINT64_MAX;

    char sql[512]; 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 _member_apply_items(void* ctx, const char* ns,
                                const uint8_t* data, size_t len) {
    struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
    if (len < 2) return -1;

    uint16_t count; memcpy(&count, data, 2);
    const uint8_t* mp = data + 2; size_t mrem = len - 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, ns, nid, x25, ed, sig, addrs, (int)ac);
        if (online) member_sync_set_online(inst, nid, 1);
        mp += consumed; mrem -= (size_t)consumed;
    }
    return 0;
}

static const struct merkle_sync_data_ops g_member_ops = {
    .update_bucket_hash = _member_update_bucket_hash,
    .get_items          = _member_get_items,
    .apply_items        = _member_apply_items,
};

/* ──  node_updated callback ── */

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) return;
    sqlite3* db = _db(inst); if (!db) return;
    int rc = 0;
    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) {
                merkle_sync_recompute_path(inst, ch, node_id);
                rc++;
            }
            sqlite3_finalize(ps);
        }
    }
    sqlite3_finalize(cs);
    DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: _on_node_updated node=%016llx channels_recomputed=%d", MS_ID, (unsigned long long)node_id, rc);
}

/* ──  Public API ── */

int member_sync_init(struct UTUN_INSTANCE* inst) {
    if (!inst) return -1;
    int rc = merkle_sync_init(inst, 0x31, &g_member_ops, inst);
    if (rc != 0) return rc;
    topo_groups_set_node_updated_cb(inst->topo_groups, _on_node_updated);
    DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized, merkle_rc=%d cb_registered=%d", MS_ID, rc, inst->topo_groups && inst->topo_groups->node_updated_cb ? 1 : 0);
    return 0;
}

void member_sync_destroy(struct UTUN_INSTANCE* inst) {
    if (!inst) return;
    topo_groups_set_node_updated_cb(inst->topo_groups, NULL);
    merkle_sync_destroy(inst);
}

int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
                       const char* ch_id, merkle_sync_done_cb done_cb, void* arg) {
    return merkle_sync_start(inst, peer, ch_id, done_cb, arg);
}

void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) {
    merkle_sync_cancel(inst, peer, ch_id);
}

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;

    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);
    }

    merkle_sync_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);
    merkle_sync_recompute_path(inst, ch_id, member_id);
    return 0;
}

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;
}

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);
    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)
                merkle_sync_recompute_path(inst, ch_id, node_id);
            sqlite3_finalize(cs);
        }
    }
    sqlite3_finalize(stmt);
}

const uint8_t* member_sync_get_hash(struct UTUN_INSTANCE* inst, const char* ch_id,
                                     uint8_t level, uint64_t prefix64) {
    return merkle_sync_get_hash(inst, ch_id, level, prefix64);
}