/**
 * @file topo_group.c
 * @brief BGP-like обмен топологией узлов (группы)
 */
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include "../lib/platform_compat.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/memory_pool.h"
#include "utun_instance.h"
#include "etcp_api.h"
#include "etcp.h"
#include "etcp_connections.h"
#include "etcp_debug.h"
#include "config_parser.h"
#include "topo_node.h"
#include "topo_node_sqlite.h"
#include "route_lib.h"
#include "topo_group.h"
#include "nat_detection.h"
#include "route_connectivity.h"
#include "control_server.h"
#include "conn_mgr.h"
#include "../transport_layer/node_conn_direct.h"
#include "topo_group_connect.h"
#include "topo_recovery.h"
#include "broadcast.h"
#include "../radio/radio.h"
#include "../chat/chat_setting.h"

#define TOPO_GROUP_UTUN  0x8000000000000000ULL


// ============================================================================
// Вспомогательные функции
// ============================================================================

/* Шлёт пиру запрос полной таблицы узлов группы (REQUEST_TABLE). */
static void topo_group_send_table_request(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) return;
    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "Sending table request to %s grp=%016llx", conn->log_name, (unsigned long long)group->group_id);
    struct TOPOMSG_TABLE_REQ* req = u_calloc(1, sizeof(struct TOPOMSG_TABLE_REQ));
    if (!req) return;
    req->cmd = ETCP_ID_TOPO_ENTRY;
    req->subcmd = TOPO_SUBCMD_REQUEST_TABLE;
    req->group_id = group->group_id;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) { u_free(req); return; }
    e->dgram = (uint8_t*)req; e->len = sizeof(struct TOPOMSG_TABLE_REQ);
    if (etcp_send(conn, e) != 0) { u_free(req); queue_entry_free(e); }
}

/* Запрашивает у пира членство в группе (JOIN_GROUP). */
static void topo_group_send_join_group(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) return;
    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "Sending join group request to %s grp=%016llx", conn->log_name, (unsigned long long)group->group_id);
    struct TOPOMSG_JOIN_GROUP* req = u_calloc(1, sizeof(struct TOPOMSG_JOIN_GROUP));
    if (!req) return;
    req->cmd = ETCP_ID_TOPO_ENTRY;
    req->subcmd = TOPO_SUBCMD_JOIN_GROUP;
    req->group_id = group->group_id;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) { u_free(req); return; }
    e->dgram = (uint8_t*)req; e->len = sizeof(struct TOPOMSG_JOIN_GROUP);
    if (etcp_send(conn, e) != 0) { u_free(req); queue_entry_free(e); }
}

/* Шлёт пиру RESYNC — просьбу заново анонсировать свои группы. */
static void topo_group_send_resync(struct ETCP_CONN* conn) {
    if (!conn) return;
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "Sending RESYNC to %s", conn->log_name);
    struct TOPOMSG_RESYNC* req = u_calloc(1, sizeof(struct TOPOMSG_RESYNC));
    if (!req) return;
    req->cmd = ETCP_ID_TOPO_ENTRY;
    req->subcmd = TOPO_SUBCMD_RESYNC;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) { u_free(req); return; }
    e->dgram = (uint8_t*)req; e->len = sizeof(struct TOPOMSG_RESYNC);
    if (etcp_send(conn, e) != 0) { u_free(req); queue_entry_free(e); }
}

/* Сигнализирует пиру об окончании начальной синхронизации таблицы. */
static void topo_group_send_table_complete(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) return;
    struct TOPOMSG_TABLE_REQ* req = u_calloc(1, sizeof(struct TOPOMSG_TABLE_REQ));
    if (!req) return;
    req->cmd = ETCP_ID_TOPO_ENTRY;
    req->subcmd = TOPO_SUBCMD_TABLE_COMPLETE;
    req->group_id = group->group_id;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) { u_free(req); return; }
    e->dgram = (uint8_t*)req; e->len = sizeof(struct TOPOMSG_TABLE_REQ);
    if (etcp_send(conn, e) != 0) { u_free(req); queue_entry_free(e); }
}

static int topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
static bool topo_group_should_send_to(const struct TOPO_GROUP_NODE* nq, uint64_t target_id);
static void topo_group_send_full_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
static void topo_group_handle_request_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
static void topo_group_handle_join_group(struct TOPO_GROUP* group, struct ETCP_CONN* conn);
static void topo_group_send_resync(struct ETCP_CONN* conn);
static void topo_group_handle_resync(struct UTUN_INSTANCE* instance, struct ETCP_CONN* conn);

/* Запоминает REQUEST_TABLE для группы, которой у нас ещё нет (CHAT-канал не загружен). */
static void topo_group_remember_table_req(struct TOPO_GROUPS* g, uint64_t node_id, uint64_t group_id) {
    if (!g) return;
    for (int i = 0; i < g->pending_table_req_count; i++)
        if (g->pending_table_reqs[i].node_id == node_id && g->pending_table_reqs[i].group_id == group_id) return;
    if (g->pending_table_req_count >= TOPO_MAX_PENDING_TABLE_REQS) {
        memmove(g->pending_table_reqs, g->pending_table_reqs + 1,
                (TOPO_MAX_PENDING_TABLE_REQS - 1) * sizeof(g->pending_table_reqs[0]));
        g->pending_table_req_count = TOPO_MAX_PENDING_TABLE_REQS - 1;
    }
    g->pending_table_reqs[g->pending_table_req_count].node_id = node_id;
    g->pending_table_reqs[g->pending_table_req_count].group_id = group_id;
    g->pending_table_req_count++;
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "remembered table request node=%016llx grp=%016llx (group not created yet)",
               (unsigned long long)node_id, (unsigned long long)group_id);
}

/* Когда группа создана — отвечаем тем, кто просил её таблицу раньше. */
static void topo_group_fulfill_table_reqs(struct TOPO_GROUPS* g, uint64_t group_id) {
    if (!g) return;
    struct TOPO_GROUP* group = topo_groups_find(g, group_id);
    if (!group) return;
    int w = 0;
    for (int i = 0; i < g->pending_table_req_count; i++) {
        struct topo_pending_table_req* req = &g->pending_table_reqs[i];
        if (req->group_id != group_id) { g->pending_table_reqs[w++] = *req; continue; }
        struct ETCP_CONN* conn = instance_find_conn(g->instance, req->node_id);
        if (conn && conn->links_up) {
            DEBUG_INFO(DEBUG_CATEGORY_BGP, "fulfilling remembered table request node=%016llx grp=%016llx",
                       (unsigned long long)req->node_id, (unsigned long long)group_id);
            topo_group_handle_request_table(group, conn);
        }
    }
    g->pending_table_req_count = w;
}

/* Краткий DEBUG-дамп принятого NODEINFO (id/ver/счётчики/hops). */
static void nodeinfo_dump_log(const uint8_t* data, size_t len) {
    if (!data || len < sizeof(struct TOPOMSG_NODEINFO_PKT)) return;
    struct TOPOMSG_NODEINFO_PKT* pkt = (struct TOPOMSG_NODEINFO_PKT*)data;
    struct TOPOMSG_NODE* ni = &pkt->node;
    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO nid=%016llx ver=%u v4s=%u v4a=%u hops=%u",
                (unsigned long long)ni->node_id, ni->ver,
                ni->local_v4_sockets, ni->local_v4_addrs, ni->hop_count);
}

/* Форматирует NODEINFO в читаемую строку для лога (имя, подсети, адреса). */
static char* nodeinfo_format(const uint8_t* data, size_t len) {
    if (!data || len < sizeof(struct TOPOMSG_NODEINFO_PKT)) return NULL;
    struct TOPOMSG_NODEINFO_PKT* pkt = (struct TOPOMSG_NODEINFO_PKT*)data;
    struct TOPOMSG_NODE* ni = &pkt->node;
    uint64_t node_id = ni->node_id;
    const uint8_t* dyn = data + sizeof(struct TOPOMSG_NODEINFO_PKT);
    size_t off = 0;
    char name_buf[64] = {0};
    size_t nl = ni->node_name_len;
    if (nl > 0 && nl < 63 && off + nl < len - sizeof(struct TOPOMSG_NODEINFO_PKT)) {
        memcpy(name_buf, dyn + off, nl); off += nl;
    }
    off += ni->local_v4_sockets * sizeof(struct TOPOMSG_SOCKMETA4);
    off += ni->local_v4_addrs * sizeof(struct TOPOMSG_ADDR4);
    off += ni->local_v6_sockets * sizeof(struct TOPOMSG_SOCKMETA6);
    off += ni->local_v6_addrs * sizeof(struct TOPOMSG_ADDR6);
    char subs_buf[512] = {0};
    if (ni->local_v4_subnets > 0) {
        const struct TOPOMSG_SUBNET4* subs = (const struct TOPOMSG_SUBNET4*)(dyn + off);
        size_t sl = 0;
        for (uint8_t i = 0; i < ni->local_v4_subnets && sl < 400; i++) {
            uint32_t addr = (subs[i].addr[0]<<24)|(subs[i].addr[1]<<16)|(subs[i].addr[2]<<8)|subs[i].addr[3];
            char tmp[64]; snprintf(tmp,sizeof(tmp),"%d.%d.%d.%d/%u ",(addr>>24)&255,(addr>>16)&255,(addr>>8)&255,addr&255,subs[i].prefix_length);
            strcat(subs_buf,tmp); sl += strlen(tmp);
        }
    }
    int need = snprintf(NULL,0,"NODEINFO nid=%016llx ver=%u name=\"%s\" v4subs=\"%s\" v4s=%u v4a=%u hop=%u", (unsigned long long)node_id,(unsigned)ni->ver,name_buf,subs_buf,(unsigned)ni->local_v4_sockets,(unsigned)ni->local_v4_addrs,(unsigned)ni->hop_count);
    char* buf = u_malloc(need+1); if (!buf) return NULL;
    snprintf(buf,need+1,"NODEINFO nid=%016llx ver=%u name=\"%s\" v4subs=\"%s\" v4s=%u v4a=%u hop=%u", (unsigned long long)node_id,(unsigned)ni->ver,name_buf,subs_buf,(unsigned)ni->local_v4_sockets,(unsigned)ni->local_v4_addrs,(unsigned)ni->hop_count);
    return buf;
}

/* Читаемое имя sub-команды BGP для логов. */
static const char* group_subcmd_name(uint8_t subcmd) {
    switch (subcmd) {
        case TOPO_SUBCMD_NODEINFO:      return "NODEINFO";
        case TOPO_SUBCMD_REQUEST_TABLE: return "REQ_TABLE";
        case TOPO_SUBCMD_WITHDRAW:      return "WITHDRAW";
        case TOPO_SUBCMD_TABLE_COMPLETE: return "TABLE_COMPLETE";
        case TOPO_SUBCMD_ERR_GROUP_MISMATCH: return "ERR_GROUP_MISMATCH";
        case TOPO_SUBCMD_JOIN_GROUP:    return "JOIN_GROUP";
        case TOPO_SUBCMD_RESYNC:        return "RESYNC";
        default: return "?";
    }
}


// ============================================================================
// Broadcast / Withdraw
// ============================================================================

/* Рассылает WITHDRAW всем BGP-пирам (кроме exclude) об удалении узла. */
static void topo_group_broadcast_withdraw(struct TOPO_GROUP* group, uint64_t node_id, uint64_t wd_source, struct ETCP_CONN* exclude) {
    if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group is NULL"); return; }
    struct TOPOMSG_WITHDRAW_PKT* pkt = u_calloc(1, sizeof(struct TOPOMSG_WITHDRAW_PKT));
    if (!pkt) return;
    pkt->cmd = ETCP_ID_TOPO_ENTRY; pkt->subcmd = TOPO_SUBCMD_WITHDRAW;
    pkt->group_id = group->group_id; pkt->node_id = node_id; pkt->wd_source = wd_source;
    struct ll_entry* e = group->senders_list ? group->senders_list->head : NULL;
    while (e) {
        struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
        if (item && item->conn && item->conn != exclude) {
            struct ll_entry* copy = queue_entry_new(0);
            if (copy) { copy->dgram = u_malloc(sizeof(*pkt)); if (copy->dgram) { memcpy(copy->dgram, pkt, sizeof(*pkt)); copy->len = sizeof(*pkt); if (etcp_send(item->conn, copy) != 0) { u_free(copy->dgram); queue_entry_free(copy); } } else { queue_entry_free(copy); } }
        }
        e = e->next;
    }
    u_free(pkt);
}


// ============================================================================
// Приём пакетов
// ============================================================================

static int topo_group_has_sender(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    for (struct ll_entry* e = group->senders_list ? group->senders_list->head : NULL; e; e = e->next)
        if (((struct TOPO_GROUP_CONN_ITEM*)e->data)->conn == conn) return 1;
    return 0;
}

/* Транспорт не даёт членства: CHAT сверяет локальную таблицу участников. */
static int topo_group_peer_allowed(struct TOPO_GROUP* group, uint64_t node_id) {
    if (group->group_type != TOPO_GROUP_TYPE_CHAT) return 1;
    sqlite3* db = group->instance->topo_sqlite_db;
    return db && group->channel_id[0] && topo_node_sqlite_member_in_channel(db, group->channel_id, node_id);
}

/* Приёмник BGP-пакетов от ETCP: диспетчер по subcmd (NODEINFO/WITHDRAW/...). */
static void topo_group_receive_cbk(struct ETCP_CONN* from_conn, struct ll_entry* entry) {
    if (!from_conn || !entry || entry->len < 2) { if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return; }
    struct UTUN_INSTANCE* instance = from_conn->instance;
    if (!instance || !instance->topo_groups) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid instance/topo_groups"); queue_dgram_free(entry); queue_entry_free(entry); return; }
    uint8_t* data = entry->dgram; uint8_t cmd = data[0]; uint8_t subcmd = data[1];
    if (cmd != ETCP_ID_TOPO_ENTRY) { queue_dgram_free(entry); queue_entry_free(entry); return; }

    /* RESYNC — общий сигнал без group_id, обрабатывается до поиска группы */
    if (subcmd == TOPO_SUBCMD_RESYNC) {
        topo_group_handle_resync(instance, from_conn);
        queue_dgram_free(entry); queue_entry_free(entry); return;
    }

    uint64_t pkt_group_id = 0;
    if (subcmd == TOPO_SUBCMD_NODEINFO && entry->len >= 3) {
        pkt_group_id = ((struct TOPOMSG_NODEINFO_PKT*)data)->node.group_id;
    } else if (subcmd == TOPO_SUBCMD_WITHDRAW && entry->len >= sizeof(struct TOPOMSG_WITHDRAW_PKT)) {
        pkt_group_id = ((struct TOPOMSG_WITHDRAW_PKT*)data)->group_id;
    } else if (subcmd == TOPO_SUBCMD_REQUEST_TABLE && entry->len >= sizeof(struct TOPOMSG_TABLE_REQ)) {
        pkt_group_id = ((struct TOPOMSG_TABLE_REQ*)data)->group_id;
    } else if (subcmd == TOPO_SUBCMD_TABLE_COMPLETE && entry->len >= sizeof(struct TOPOMSG_TABLE_REQ)) {
        pkt_group_id = ((struct TOPOMSG_TABLE_REQ*)data)->group_id;
    } else if (subcmd == TOPO_SUBCMD_ERR_GROUP_MISMATCH && entry->len >= sizeof(struct TOPOMSG_ERR_GROUP_MISMATCH)) {
        pkt_group_id = ((struct TOPOMSG_ERR_GROUP_MISMATCH*)data)->group_id;
    } else if (subcmd == TOPO_SUBCMD_JOIN_GROUP && entry->len >= sizeof(struct TOPOMSG_JOIN_GROUP)) {
        pkt_group_id = ((struct TOPOMSG_JOIN_GROUP*)data)->group_id;
    }

    struct TOPO_GROUP* group = topo_groups_find(instance->topo_groups, pkt_group_id);
    if (!group) {
        if (subcmd == TOPO_SUBCMD_REQUEST_TABLE)
            topo_group_remember_table_req(instance->topo_groups, from_conn->peer_node_id, pkt_group_id);
        DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "BGP recv %s from %s: group %016llx not found, dropping", group_subcmd_name(subcmd), from_conn->log_name, (unsigned long long)pkt_group_id);
        queue_dgram_free(entry); queue_entry_free(entry); return;
    }

    DEBUG_DEBUG(DEBUG_CATEGORY_BGP,"BGP recv %s from=%s group=%016llx len=%u",group_subcmd_name(subcmd),
        from_conn->log_name,(unsigned long long)pkt_group_id,entry->len);

    if (!topo_group_peer_allowed(group, from_conn->peer_node_id)) {
        DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP deferred: member unknown peer=%016llx group=%016llx cmd=%u",
                   (unsigned long long)from_conn->peer_node_id, (unsigned long long)group->group_id, subcmd);
        queue_dgram_free(entry); queue_entry_free(entry); return;
    }

    if (subcmd == TOPO_SUBCMD_NODEINFO) { nodeinfo_dump_log(data, entry->len); topo_group_process_nodeinfo(group, from_conn, data, entry->len); }
    else if (subcmd == TOPO_SUBCMD_WITHDRAW) topo_group_process_withdraw(group, from_conn, data, entry->len);
    else if (subcmd == TOPO_SUBCMD_REQUEST_TABLE) topo_group_handle_request_table(group, from_conn);
    else if (subcmd == TOPO_SUBCMD_JOIN_GROUP) topo_group_handle_join_group(group, from_conn);
    else if (subcmd == TOPO_SUBCMD_TABLE_COMPLETE) { etcp_set_routing_exchange_state(from_conn, 3); DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP sync complete with %s: %d nodes grp=%016llx", from_conn->log_name, queue_entry_count(group->nodes), (unsigned long long)group->group_id); }
    else if (subcmd == TOPO_SUBCMD_ERR_GROUP_MISMATCH) {
        if (entry->len >= sizeof(struct TOPOMSG_ERR_GROUP_MISMATCH)) {
            struct TOPOMSG_ERR_GROUP_MISMATCH* err = (struct TOPOMSG_ERR_GROUP_MISMATCH*)data;
            DEBUG_ERROR(DEBUG_CATEGORY_BGP, "Group mismatch from %s: expected=%u received=0x%02x group=%016llx",
                        from_conn->log_name, err->expected_type, err->received_flags, (unsigned long long)err->group_id);
        }
    }

    queue_dgram_free(entry); queue_entry_free(entry);
}


// ============================================================================
// Init / Destroy / New / Remove conn
// ============================================================================

/* Callback статуса ETCP-соединения: UP — добавить пира в группы; DOWN — удалить. */
static void topo_group_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
    struct TOPO_GROUPS* groups = (struct TOPO_GROUPS*)arg;
    if (!conn || !groups) return;

    if (status == ETCP_CONN_STATUS_REINIT) {
        // Сброс normalizer мог уничтожить TABLE_REQUEST/NODEINFO. Повторяем обмен, сохраняя членство и пути.
        for (struct ll_entry* e = groups->group_list->head; e; e = e->next) {
            struct TOPO_GROUP* g = (struct TOPO_GROUP*)e;
            for (struct ll_entry* se = g->senders_list->head; se; se = se->next) {
                if (((struct TOPO_GROUP_CONN_ITEM*)se->data)->conn != conn) continue;
                DEBUG_INFO(DEBUG_CATEGORY_BGP,"BGP session resync: peer=%016llx group=%016llx",
                    (unsigned long long)conn->peer_node_id,(unsigned long long)g->group_id);
                topo_group_send_join_group(g,conn); topo_group_send_table_request(g,conn);
                break;
            }
        }
        return;
    }
    if (status == ETCP_CONN_STATUS_UP && conn->peer_node_id) {
        int is_client = conn->instance->config && config_peer_in_clients(conn->instance->config, conn->crypto_ctx.peer_public_key);
        int explicit_conn = conn->bgp_ready_cbk != NULL;

        /* пассивная сторона: просим пира заново анонсировать свои группы (после флэпа/переподключения) */
        if (!is_client && !explicit_conn)
            topo_group_send_resync(conn);

        /* non-CHAT (UTUN): узлы из конфига (clients) + явные подключения (etcp_connect). */
        struct ll_entry* fe = groups->group_list->head;
        while (fe) {
            struct TOPO_GROUP* g = (struct TOPO_GROUP*)fe; fe = fe->next;
            if (g->group_type != TOPO_GROUP_TYPE_CHAT) {
                if (is_client || explicit_conn) {
                    topo_group_new_conn(g, conn);          /* добавить + инициировать BGP */
                    topo_group_send_join_group(g, conn);   /* запросить членство у пира */
                }
            }
        }
        /* CHAT: только члены каналов (peers_* в БД). */
        sqlite3* db = conn->instance ? conn->instance->topo_sqlite_db : NULL;
        uint64_t* chs = NULL; int chn = 0;
        if (db && topo_node_sqlite_get_member_channels(db, conn->peer_node_id, &chs, &chn) == 0 && chn > 0) {
            for (int i = 0; i < chn; i++) {
                struct TOPO_GROUP* cg = topo_groups_find(groups, chs[i]);
                if (cg && cg->group_type == TOPO_GROUP_TYPE_CHAT)
                    topo_group_new_conn(cg, conn);
            }
        }
        u_free(chs);
    } else if (status == ETCP_CONN_STATUS_DOWN || status == ETCP_CONN_STATUS_DELETE) {
        struct ll_entry* fe = groups->group_list->head;
        while (fe) { struct TOPO_GROUP* g = (struct TOPO_GROUP*)fe; fe = fe->next;
            topo_group_remove_conn(g, conn);
        }
    }
}

/* Создаёт группу: очереди nodes/senders, local_node, conn_mgr, broadcast. */
static struct TOPO_GROUP* topo_group_create(struct UTUN_INSTANCE* instance, uint64_t group_id, uint8_t group_type) {
    if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "instance is NULL"); return NULL; }
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "group_id=%016llx node_id=%016llx", (unsigned long long)group_id, (unsigned long long)instance->node_id);

    struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP));
    if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "alloc failed"); return NULL; }
    struct TOPO_GROUP* group = (struct TOPO_GROUP*)qe;
    memset((uint8_t*)group + sizeof(struct ll_entry), 0, sizeof(*group) - sizeof(struct ll_entry));
    group->group_id = group_id;
    group->group_type = group_type;
    group->instance = instance;

    sc_derive_ed25519_pubkey(instance->my_keys.private_key, group->ed25519_public_key);
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "Ed25519 pubkey derived from X25519 private key");

    group->nodes = queue_new(instance->ua, BGP_NODES_HASH_SIZE, offsetof(struct TOPO_GROUP_NODE, node_id) - sizeof(struct ll_entry), 8, "group_nodes");
    if (!group->nodes) { queue_entry_free(qe); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "nodes queue creation failed"); return NULL; }

    group->senders_list = queue_new(instance->ua, 0, 0, 0, "group_senders");
    if (!group->senders_list) { queue_free(group->nodes); queue_entry_free(qe); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_list queue creation failed"); return NULL; }

    group->local_node = NULL;
    topo_group_update_my_nodeinfo(instance, group);

    group->conn_mgr = conn_mgr_init(group);

    broadcast_init(group);
    radio_init(group);

    DEBUG_INFO(DEBUG_CATEGORY_BGP, "TOPO group created: group_id=%016llx", (unsigned long long)group_id);
    return group;
}

/* Полностью разрушает группу: recovery, connect, broadcast, conn_mgr, узлы. */
static void topo_group_destroy(struct TOPO_GROUP* group) {
    if (!group) return;
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 0 enter grp=%p id=0x%016llx rec_list=%p connect=%p",
               group, (unsigned long long)group->group_id, group->recovery_list, group->connect);

    topo_recovery_cancel_all(group);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 1 recovery_cancel done grp=%p", group);
    topo_group_connect_destroy(group);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2 connect_destroy done");

    broadcast_destroy(group);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2b broadcast_destroy done");

    radio_destroy(group);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2c radio_destroy done");

    struct ll_entry* e;
    while ((e = queue_data_get(group->senders_list)) != NULL) {
        struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
        DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "group destroy: release peer=%016llx group=%016llx",
                    (unsigned long long)node_conn_direct_node_id(item->handle), (unsigned long long)group->group_id);
        if (item->handle) { node_conn_direct_close(item->handle); item->handle = NULL; }
        queue_entry_free(e);
    }
    queue_free(group->senders_list);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 3 senders_list done");

    if (group->conn_mgr) { conn_mgr_destroy(group->conn_mgr); group->conn_mgr = NULL; }
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 4 conn_mgr done");

    if (group->nodes) {
        int nc = 0;
        struct ll_entry* e = group->nodes->head;
        while (e) {
            struct ll_entry* next = e->next;
            topo_nodeq_remove_node(group, (struct TOPO_GROUP_NODE*)e);
            e = next; nc++;
        }
        queue_free(group->nodes);
        if (nc) DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 4b nodes cleaned (%d)", nc);
    }
    if (group->local_node) {
        if (group->local_node->paths) { queue_free(group->local_node->paths); group->local_node->paths = NULL; }
        topo_nodeq_free_group_fields(group->instance->topo_groups, group->local_node);
        u_free(group->local_node);
    }
    /* освободить подписчиков BGP-событий (member_sync и др.) */
    while (group->node_cbks) {
        struct topo_node_cbk_entry* c = group->node_cbks;
        group->node_cbks = c->next; u_free(c);
    }
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 5 done");
}

/* Удаляет группу из контейнера (кроме UTUN-группы по умолчанию). */
void topo_groups_remove_group(struct TOPO_GROUPS* g, uint64_t group_id) {
    if (!g || !g->group_list || group_id == TOPO_GROUP_UTUN) return;
    struct TOPO_GROUP* group = topo_groups_find(g, group_id);
    if (!group) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "remove_group: group %016llx not found", (unsigned long long)group_id); return; }

    DEBUG_INFO(DEBUG_CATEGORY_BGP, "remove_group: id=%016llx type=%d ch=%s",
               (unsigned long long)group_id, group->group_type, group->channel_id);

    queue_remove_data(g->group_list, &group->ll);
    topo_group_destroy(group);   /* node_cbks освобождаются внутри */
    queue_entry_free(&group->ll);
}

/* Инициализация контейнера групп: реестр, пулы, SQLite, default-группа, подписки. */
struct TOPO_GROUPS* topo_groups_init(struct UTUN_INSTANCE* instance) {
    if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "instance is NULL"); return NULL; }
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "node_id=%016llx", (unsigned long long)instance->node_id);

    struct TOPO_GROUPS* g = u_calloc(1, sizeof(struct TOPO_GROUPS));
    if (!g) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "alloc failed"); return NULL; }
    g->instance = instance;

    g->group_list = queue_new(instance->ua, 16, 0, 8, "topo_groups");
    if (!g->group_list) { u_free(g); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group_list queue creation failed"); return NULL; }

    g->node_registry = queue_new(instance->ua, TOPO_NODE_REGISTRY_HASH_SIZE, 0, 8, "node_registry");
    if (!g->node_registry) { queue_free(g->group_list); u_free(g); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "node_registry queue creation failed"); return NULL; }

    g->v4_sock_meta_pool = memory_pool_init(sizeof(struct TOPO_SOCKMETA4), "to_v4sock");
    g->v4_addr_pool      = memory_pool_init(sizeof(struct TOPO_ADDR4), "to_v4addr");
    g->v6_sock_meta_pool = memory_pool_init(sizeof(struct TOPO_SOCKMETA6), "to_v6sock");
    g->v6_addr_pool      = memory_pool_init(sizeof(struct TOPO_ADDR6), "to_v6addr");
    g->v4_subnet_pool    = memory_pool_init(sizeof(struct TOPO_SUBNET4), "to_v4sub");
    g->v6_subnet_pool    = memory_pool_init(sizeof(struct TOPO_SUBNET6), "to_v6sub");

    if (instance->config && instance->config->global.db_path[0]) {
        char db_file[512];
        snprintf(db_file, sizeof(db_file), "%s/chats.db", instance->config->global.db_path);
        int rc = sqlite3_open_v2(db_file, &instance->topo_sqlite_db,
                                 SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX, NULL);
        if (rc == SQLITE_OK && instance->topo_sqlite_db) {
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA journal_mode=WAL", NULL, NULL, NULL);
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA foreign_keys=ON", NULL, NULL, NULL);
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA synchronous=NORMAL", NULL, NULL, NULL);
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA wal_autocheckpoint=10000", NULL, NULL, NULL);
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA cache_size=-32768", NULL, NULL, NULL);
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA mmap_size=134217728", NULL, NULL, NULL);
            sqlite3_exec(instance->topo_sqlite_db, "PRAGMA temp_store=MEMORY", NULL, NULL, NULL);
            topo_node_sqlite_init(instance->topo_sqlite_db);
            DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "SQLite opened: %s rc=%d db=%p", db_file, rc, (void*)instance->topo_sqlite_db);
        } else {
            DEBUG_ERROR(DEBUG_CATEGORY_BGP, "sqlite3_open(%s) failed: %s",
                        db_file, instance->topo_sqlite_db ? sqlite3_errmsg(instance->topo_sqlite_db) : "null db");
            if (instance->topo_sqlite_db) { sqlite3_close(instance->topo_sqlite_db); instance->topo_sqlite_db = NULL; }
        }
    }

    instance->topo_groups = g;

    struct TOPO_GROUP* default_group = topo_group_create(instance, TOPO_GROUP_UTUN, TOPO_GROUP_TYPE_UTUN);
    if (!default_group) {
        memory_pool_destroy(g->v4_sock_meta_pool); memory_pool_destroy(g->v4_addr_pool);
        memory_pool_destroy(g->v6_sock_meta_pool); memory_pool_destroy(g->v6_addr_pool);
        memory_pool_destroy(g->v4_subnet_pool); memory_pool_destroy(g->v6_subnet_pool);
        queue_free(g->node_registry); queue_free(g->group_list); u_free(g);
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "failed to create default group");
        return NULL;
    }
    queue_data_put_with_index(g->group_list, &default_group->ll);

    instance->conn_mgr = default_group->conn_mgr;  /* recv handler shortcut */

    etcp_bind(instance, ETCP_ID_TOPO_ENTRY, topo_group_receive_cbk);
    etcp_add_conn_status_cbk(instance, topo_group_conn_status, g);
    etcp_add_socket_cbk(instance, topo_node_on_socket_changed, NULL,
                        ETCP_SOCKET_EVENT_ADDR_CHANGED | ETCP_SOCKET_EVENT_STATUS_CHANGED);
    utun_add_activity_cbk(instance, topo_group_on_activity_change, NULL);

    DEBUG_INFO(DEBUG_CATEGORY_BGP, "TOPO groups initialized (with default group)");
    return g;
}

/* Callback смены active/standby: обновляет свой nodeinfo и рассылает его пирам. */
void topo_group_on_activity_change(struct UTUN_INSTANCE* instance, int active, void* arg) {
    (void)active; (void)arg;
    if (!instance || !instance->topo_groups || !instance->topo_groups->group_list) return;
    struct ll_entry* ge = instance->topo_groups->group_list->head;
    while (ge) {
        struct TOPO_GROUP* g = (struct TOPO_GROUP*)ge;
        topo_group_update_my_nodeinfo(instance, g);
        if (g->senders_list) {
            struct ll_entry* se = g->senders_list->head;
            while (se) {
                struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data;
                if (item && item->conn) topo_group_send_nodeinfo(g, g->local_node, item->conn, 0);
                se = se->next;
            }
        }
        ge = ge->next;
    }
}

/* Полный teardown контейнера: отписки, уничтожение групп, реестра, пулов. */
void topo_groups_destroy(struct UTUN_INSTANCE* instance) {
    if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "instance is NULL"); return; }
    if (!instance->topo_groups) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "topo_groups is NULL"); return; }
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "node_id=%016llx tgs=%p", (unsigned long long)instance->node_id, instance->topo_groups);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 0 enter");

    etcp_unbind(instance, ETCP_ID_TOPO_ENTRY);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 1 etcp_unbind done");
    etcp_remove_conn_status_cbk(instance, topo_group_conn_status, instance->topo_groups);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 2 conn_cbk_remove done");
    etcp_remove_socket_cbk(instance, topo_node_on_socket_changed, NULL);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 3 socket_cbk_remove done");
    utun_remove_activity_cbk(instance, topo_group_on_activity_change, NULL);
    route_connectivity_cancel_all(instance);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 4 connectivity_cancel done");

    struct TOPO_GROUPS* g = instance->topo_groups;
    struct ll_entry* ge;
    int gcount = 0;
    while ((ge = queue_data_get(g->group_list)) != NULL) { struct TOPO_GROUP* grp = (struct TOPO_GROUP*)ge; topo_group_destroy(grp); queue_entry_free(ge); gcount++; }
    queue_free(g->group_list);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 5 groups destroyed (%d)", gcount);

    if (g->node_registry) {
        int ncount = 0;
        struct ll_entry* re;
        while ((re = queue_data_get(g->node_registry)) != NULL) {
            struct TOPO_NODE* node;
            memcpy(&node, re->data + 8, sizeof(node));
            topo_node_destroy(g, node);
            queue_entry_free(re); ncount++;
        }
        queue_free(g->node_registry);
        DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 6 node_registry done (%d)", ncount);
    } else {
        DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 6 node_registry skip (NULL)");
    }

    memory_pool_destroy(g->v4_sock_meta_pool);
    memory_pool_destroy(g->v4_addr_pool);
    memory_pool_destroy(g->v6_sock_meta_pool);
    memory_pool_destroy(g->v6_addr_pool);
    memory_pool_destroy(g->v4_subnet_pool);
    memory_pool_destroy(g->v6_subnet_pool);
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 7 pools destroyed");

    /* Shared SQLite DB закрывается в utun_instance_destroy() в самом конце —
     * chat/media/db_sync ещё используют его во время teardown. */
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 8 sqlite deferred (закрывается в utun_instance_destroy)");

    u_free(g); instance->topo_groups = NULL; instance->conn_mgr = NULL;
    DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 9 done");
}

/* Возвращает UTUN-группу по умолчанию (group_id=TOPO_GROUP_UTUN). */
struct TOPO_GROUP* topo_groups_get_default(struct TOPO_GROUPS* g) {
    return topo_groups_find(g, TOPO_GROUP_UTUN);
}

/* Ищет группу по group_id (NULL, если нет). */
struct TOPO_GROUP* topo_groups_find(struct TOPO_GROUPS* g, uint64_t group_id) {
    if (!g || !g->group_list) return NULL;
    struct ll_entry* e = queue_find_data_by_index(g->group_list, &group_id);
    return e ? (struct TOPO_GROUP*)e : NULL;
}

/* Создаёт и регистрирует новую группу; добавляет существующие conn, для CHAT — автоподключение. */
struct TOPO_GROUP* topo_groups_create_group(struct TOPO_GROUPS* g, uint64_t group_id, uint8_t group_type, const char* channel_id) {
    if (!g || !g->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return NULL; }
    if (topo_groups_find(g, group_id)) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group_id=%016llx already exists", (unsigned long long)group_id); return NULL; }
    struct TOPO_GROUP* group = topo_group_create(g->instance, group_id, group_type);
    if (!group) return NULL;
    if (channel_id) { strncpy(group->channel_id, channel_id, sizeof(group->channel_id) - 1); group->channel_id[sizeof(group->channel_id) - 1] = '\0'; }
    queue_data_put_with_index(g->group_list, &group->ll);

    struct ll_queue* connections = g->instance->connections;
    if (connections) {
        sqlite3* db = g->instance->topo_sqlite_db;
        for (uint32_t slot = 0; slot < connections->hash_size; slot++) {
            struct ll_entry* entry = connections->hash_table[slot];
            while (entry) {
                struct conn_queue_entry* cqe = (struct conn_queue_entry*)entry;
                if (cqe->conn && cqe->peer_node_id != 0) {
                    if (group_type == TOPO_GROUP_TYPE_CHAT && db) {
                        /* CHAT: добавляем conn только если пир — член именно этого канала */
                        uint64_t* chs = NULL; int chn = 0;
                        if (topo_node_sqlite_get_member_channels(db, cqe->peer_node_id, &chs, &chn) == 0) {
                            for (int i = 0; i < chn; i++) {
                                if (chs[i] == group_id) { topo_group_new_conn(group, cqe->conn); break; }
                            }
                        }
                        u_free(chs);
                    } else {
                        topo_group_new_conn(group, cqe->conn);
                    }
                }
                entry = entry->hash_next;
            }
        }
    }

    DEBUG_INFO(DEBUG_CATEGORY_BGP, "Group created: group_id=%016llx type=%u ch_id=%s", (unsigned long long)group_id, group_type, group->channel_id);
    if (group_type == TOPO_GROUP_TYPE_CHAT && chat_setting_get_int(group->instance, "group_autoconnect", 1)) topo_group_connect_init(group);
    topo_group_fulfill_table_reqs(g, group_id);
    return group;
}

/* Разрешить/запретить NAT-проверку локальных подсетей (делегат в nat_detection). */
void topo_group_set_nat_check_local(struct TOPO_GROUP* group, int allow) {
    if (!group || !group->instance || !group->instance->nat_det) return;
    nat_detection_set_allow_local(group->instance->nat_det, allow);
}

/* ── Рация (PTT walkie-talkie, per-CHAT-группа) ── */

void topo_group_set_radio(struct TOPO_GROUP* group, int on) {
    if (!group || !group->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "set_radio: invalid args"); return; }
    if (group->group_type != TOPO_GROUP_TYPE_CHAT) {
        DEBUG_WARN(DEBUG_CATEGORY_BGP, "set_radio: only CHAT groups support radio grp=%016llx type=%u",
                   (unsigned long long)group->group_id, group->group_type);
        return;
    }
    uint8_t v = on ? 1 : 0;
    if (group->radio_active == v) {
        DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "set_radio: unchanged (%u) grp=%016llx ch=%s", v,
                    (unsigned long long)group->group_id, group->channel_id);
        return;
    }
    group->radio_active = v;
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "set_radio: %s grp=%016llx ch=%s",
               v ? "ON" : "OFF", (unsigned long long)group->group_id, group->channel_id);
    topo_group_update_my_nodeinfo(group->instance, group);
    if (group->senders_list) {
        struct ll_entry* se = group->senders_list->head;
        while (se) {
            struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data;
            if (item && item->conn && group->local_node)
                topo_group_send_nodeinfo(group, group->local_node, item->conn, 0);
            se = se->next;
        }
    }
}

int topo_group_radio_subscribers(struct TOPO_GROUP* group) {
    if (!group || !group->nodes) return 0;
    int n = 0;
    struct ll_entry* e = group->nodes->head;
    while (e) {
        struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
        if (nq->radio && nq->node_id != group->instance->node_id) n++;
        e = e->next;
    }
    return n;
}

/* Устанавливает колбэк сохранения узла в БД (используется chatgui/member_sync). */
void topo_groups_set_node_updated_cb(struct TOPO_GROUPS* groups, topo_node_updated_fn fn) {
    if (groups) groups->node_updated_cb = fn;
}

/* Добавляет пира в группу при UP: дедуп, добавление в senders, запрос таблицы. */
int topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return -1; }
    if (!conn->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance is NULL"); return -1; }
    if (!conn->instance->rt) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance->rt is NULL"); return -1; }
    if (!topo_group_peer_allowed(group, conn->peer_node_id)) {
        DEBUG_INFO(DEBUG_CATEGORY_BGP, "group connect deferred: member unknown peer=%016llx group=%016llx",
                   (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id);
        return -1;
    }

    /* дедуп: тот же conn стреляет ETCP_CONN_STATUS_UP дважды (UDP-линк, затем TCP-линк).
     * Если conn уже в senders_list — повторно не обрабатываем, иначе active_conn_count
     * задваивается и переподключение никогда не стартует. */
    {   struct ll_entry* se = group->senders_list ? group->senders_list->head : NULL;
        while (se) {
            if (((struct TOPO_GROUP_CONN_ITEM*)se->data)->conn == conn) {
                DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_new_conn: conn already in senders_list, skip double UP (%s)", conn->log_name);
                return 0;
            }
            se = se->next;
        }
    }


    struct TOPO_GROUP_NODE* peer_nq = topo_node_find_by_id(group, conn->peer_node_id);
    if (peer_nq) {
        peer_nq->connectivity.last_ping_time = get_time_tb();
        peer_nq->conn_presence |= NCONN_DIRECT;
        peer_nq->conn_up |= NCONN_DIRECT;
        if (conn->instance->topo_sqlite_db) topo_node_sqlite_nodeinfo_updated(conn->instance->topo_sqlite_db, conn->peer_node_id);
    }
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_new_conn: ADD conn=%p peer=%016llx state=%d grp=%016llx ch=%s",
               (void*)conn, (unsigned long long)conn->peer_node_id, conn->state,
               (unsigned long long)group->group_id, group->channel_id);
    if (topo_group_add_to_senders(group, conn) != 0) return -1;
    topo_recovery_cancel_for_node(group, conn->peer_node_id);

    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "topo_group_new_conn: peer=%016llx group=%016llx type=%d ch=%s", (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id);

    topo_group_send_join_group(group, conn);
    topo_group_send_table_request(group, conn);
    topo_group_connect_on_up(group, conn);
    return 0;
}

/* Обрабатывает DOWN пира: чистит пути, каскадно удаляет узлы, шлёт withdraw. */
void topo_group_remove_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return; }
    if (!conn->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid instance"); return; }

    bool found_in_list = false;
    struct ll_entry* e = group->senders_list->head;
    while (e) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; if (item->conn == conn) { found_in_list = true; break; } e = e->next; }
    if (!found_in_list) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: conn=%p peer=%016llx state=%d NOT in senders_list (already removed?), skip grp=%016llx ch=%s", (void*)conn, (unsigned long long)conn->peer_node_id, conn->state, (unsigned long long)group->group_id, group->channel_id); return; }

    DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: conn=%p peer=%016llx state=%d name='%s' grp=%016llx ch=%s",
               (void*)conn, (unsigned long long)conn->peer_node_id, conn->state, conn->log_name,
               (unsigned long long)group->group_id, group->channel_id);

    struct ROUTE_TABLE* rt = conn->instance->rt;
    int nodes_removed = 0;
    int cascaded = 0;
    struct ll_entry* node_entry = group->nodes ? group->nodes->head : NULL;
    while (node_entry) {
        struct ll_entry* next = node_entry->next;
        struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)node_entry;
        {
            uint64_t next_hop = nq->node_id;
            struct ll_entry* pe = nq->paths ? nq->paths->head : NULL;
            while (pe) { struct TOPO_NODEPATH* p = (struct TOPO_NODEPATH*)pe; if (p->conn == conn && p->hop_count >= 2) { uint64_t* hop = (uint64_t*)((uint8_t*)p + sizeof(struct TOPO_NODEPATH)); if (hop[p->hop_count - 1] == conn->peer_node_id) next_hop = hop[p->hop_count - 2]; break; } pe = pe->next; }
            if (topo_group_remove_path(nq, conn) == 1) {
                uint64_t key = nq->node_id;
                if (key != conn->peer_node_id) { topo_recovery_add_node(group, nq, next_hop); cascaded++; }
                nq->conn_presence = 0; nq->conn_up = 0;
                topo_fire_nodeinfo_cbk(conn->instance, group, nq);
                if (group->group_type != TOPO_GROUP_TYPE_CHAT && rt) route_delete(rt, nq);
                if (conn->instance && conn->instance->control_srv) control_server_notify_node_removed(conn->instance->control_srv, key);
                topo_nodeq_remove_node(group, nq);
                nodes_removed++;
                DEBUG_INFO(DEBUG_CATEGORY_BGP, "node removed from group: node=%016llx grp=%016llx type=%u ch=%s (link_down)",
                           (unsigned long long)key, (unsigned long long)group->group_id, group->group_type, group->channel_id);
                // REINIT временно убирает путь, но end-to-end данные роутера остаются неподтверждёнными.
                if (!conn->reinit_pending) etcp_router_conn_close_all_for_node(group->instance, group->group_id, key);
                uint64_t wd_src = (key == conn->peer_node_id) ? conn->instance->node_id : conn->peer_node_id;
                topo_group_broadcast_withdraw(group, key, wd_src, NULL);
                DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "Removed node %016llx after link down", (unsigned long long)key);
                {   struct topo_node_cbk_entry* c = group->node_cbks;
                    while (c) { c->fn(group, key, TOPO_NODE_EVENT_REMOVE, c->arg); c = c->next; }
                }
            } else {
                nq->conn_up &= ~NCONN_DIRECT;
                {   int has_direct = 0; struct ll_entry* pe2 = nq->paths ? nq->paths->head : NULL;
                    while (pe2) { struct TOPO_NODEPATH* p = (struct TOPO_NODEPATH*)pe2; if (p->conn->peer_node_id == nq->node_id) { has_direct = 1; break; } pe2 = pe2->next; }
                    if (!has_direct) nq->conn_presence &= ~NCONN_DIRECT; }
                topo_fire_nodeinfo_cbk(conn->instance, group, nq);
            }
        }
        node_entry = next;
    }

    /* освобождаем владение BGP: сначала убираем item из senders_list (чтобы
       ре-ентрантный remove_conn от conn DOWN не нашёл его), затем закрываем handle. */
    e = group->senders_list->head;
    while (e) {
        struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
        if (item->conn == conn) {
            struct NODE_CONN_DIRECT* h = item->handle;
            item->handle = NULL;
            queue_remove_data(group->senders_list, e);
            queue_entry_free(e);
            DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: REMOVED conn=%p peer=%016llx from senders_list grp=%016llx ch=%s",
                       (void*)conn, (unsigned long long)conn->peer_node_id,
                       (unsigned long long)group->group_id, group->channel_id);
            if (h) {
                node_conn_direct_close(h);
                DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: BGP handle released node=0x%016llx", (unsigned long long)conn->peer_node_id);
            }
            break;
        }
        e = e->next;
    }

    if (cascaded > 0) topo_recovery_start(group);

    topo_group_connect_on_down(group, conn);

    DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP peer removed: %s nodes=%d grp=%016llx", conn->log_name, nodes_removed, (unsigned long long)group->group_id);
}


// ============================================================================
// NODEINFO BASED IMPLEMENTATION
// ============================================================================

/* Ищет лучшее ETCP-соединение до узла (мин. hop_count среди живых путей). */
struct ETCP_CONN* topo_group_find_conn_for_node(struct TOPO_GROUP* group, uint64_t node_id) {
    if (!group) return NULL;
    struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, node_id);
    if (!nq || !nq->paths || !nq->paths->head) return NULL;
    struct ll_entry* e = nq->paths->head;
    struct TOPO_NODEPATH* best_live = NULL; uint8_t best_live_hops = 255;
    struct TOPO_NODEPATH* best_any = NULL; uint8_t best_any_hops = 255;
    while (e) {
        struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e;
        if (path->conn) { if (path->conn->links_up && path->hop_count < best_live_hops) { best_live = path; best_live_hops = path->hop_count; } if (path->hop_count < best_any_hops) { best_any = path; best_any_hops = path->hop_count; } }
        e = e->next;
    }
    if (!best_live && best_any) DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "find_conn_for_node %016llx: no live paths, fallback to best (hops=%d links_up=%d)", (unsigned long long)node_id, best_any_hops, best_any->conn->links_up);
    return best_live ? best_live->conn : (best_any ? best_any->conn : NULL);
}

/* Добавляет путь (conn + hop_list + rtt) в paths узла. */
int topo_group_add_path(struct TOPO_GROUP_NODE* nq, struct ETCP_CONN* conn, uint64_t* hop_list, uint8_t hop_count, uint16_t cumulative_rtt) {
    if (!nq || !conn || hop_count > MAX_HOPS || !hop_list) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "add_path: invalid args"); return -1; }
    if (!nq->paths) { nq->paths = queue_new(conn->instance->ua, 0, 0, 0, "node_paths"); if (!nq->paths) return -1; }
    size_t path_size = sizeof(struct TOPO_NODEPATH) - sizeof(struct ll_entry) + hop_count * 8;
    struct ll_entry* pe = queue_entry_new(path_size); if (!pe) return -1;
    struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)pe;
    path->conn = conn; path->hop_count = hop_count; path->cumulative_rtt = cumulative_rtt;
    memcpy((uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH)), hop_list, hop_count * 8);
    queue_data_put(nq->paths, pe);
    return 0;
}

/* Удаляет пути, содержащие wd_source в hop_list; 1 — если путей не осталось. */
static int topo_group_remove_path_by_hop(struct TOPO_GROUP_NODE* nq, uint64_t wd_source) {
    if (!nq || !nq->paths) return 0;
    int removed = 0;
    struct ll_entry* e = nq->paths->head;
    while (e) {
        struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e;
        uint64_t* hop = (uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH));
        bool has_wd = false; for (uint8_t i = 0; i < path->hop_count; i++) if (hop[i] == wd_source) { has_wd = true; break; }
        if (has_wd) { struct ll_entry* next = e->next; queue_remove_data(nq->paths, e); queue_entry_free(e); removed++; e = next; continue; }
        e = e->next;
    }
    if (removed > 0 && nq->paths && queue_entry_count(nq->paths) == 0) { queue_free(nq->paths); nq->paths = NULL; return 1; }
    return removed;
}

/* Удаляет путь через конкретный conn; 1 — если путей не осталось. */
int topo_group_remove_path(struct TOPO_GROUP_NODE* nq, struct ETCP_CONN* conn) {
    if (!nq || !conn || !nq->paths) return -1;
    struct ll_entry* e = nq->paths->head;
    while (e) { struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e; if (path->conn == conn) { queue_remove_data(nq->paths, e); queue_entry_free(e); break; } e = e->next; }
    int remaining = nq->paths ? queue_entry_count(nq->paths) : 0;
    if (remaining == 0) { if (nq->paths) { queue_free(nq->paths); nq->paths = NULL; } return 1; }
    return 0;
}

// ===== NODEINFO process =====

/* Шлёт пиру ошибку несоответствия типа группы (ERR_GROUP_MISMATCH). */
static void topo_group_send_err_group_mismatch(struct TOPO_GROUP* group, struct ETCP_CONN* conn, uint8_t expected_type, uint8_t received_flags) {
    if (!group || !conn) return;
    struct TOPOMSG_ERR_GROUP_MISMATCH* pkt = u_calloc(1, sizeof(struct TOPOMSG_ERR_GROUP_MISMATCH));
    if (!pkt) return;
    pkt->cmd = ETCP_ID_TOPO_ENTRY; pkt->subcmd = TOPO_SUBCMD_ERR_GROUP_MISMATCH;
    pkt->group_id = group->group_id;
    pkt->expected_type = expected_type; pkt->received_flags = received_flags;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) { u_free(pkt); return; }
    e->dgram = (uint8_t*)pkt; e->len = sizeof(struct TOPOMSG_ERR_GROUP_MISMATCH);
    if (etcp_send(conn, e) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "etcp_send ERR_GROUP_MISMATCH failed"); u_free(pkt); queue_entry_free(e); }
}

/* Обрабатывает NODEINFO: проверка подписи/версии, обновление узла, пути, роутинг, форвард. */
int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from, const uint8_t* data, size_t len) {
    if (!group || !from || len < sizeof(struct TOPOMSG_NODEINFO_PKT)) return -1;

    struct TOPOMSG_NODEINFO_PKT* pkt = (struct TOPOMSG_NODEINFO_PKT*)data;
    struct TOPOMSG_NODE* ni = &pkt->node;

    if (ni->group_id != group->group_id) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO group_id mismatch from %s: expected=%016llx got=%016llx nid=%016llx",
                    from->log_name, (unsigned long long)group->group_id, (unsigned long long)ni->group_id, (unsigned long long)ni->node_id);
        return -1;
    }

    // Проверка соответствия типа группы
    uint8_t expected_flags = (group->group_type == TOPO_GROUP_TYPE_CHAT) ? 0 : TOPO_FLAG_SEND_SUBNETS;
    if ((ni->flags & TOPO_FLAG_SEND_SUBNETS) != expected_flags) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "Group type mismatch with %s: expected_flags=0x%02x received_flags=0x%02x (local_type=%u)",
                    from->log_name, expected_flags, ni->flags, group->group_type);
        topo_group_send_err_group_mismatch(group, from, group->group_type, ni->flags);
        return -1;
    }

    int dyn = topo_node_dyn_size(ni);
    if ((size_t)dyn + sizeof(struct TOPOMSG_NODEINFO_PKT) != len) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "Incorrect packet size (%s)", from->log_name); return -1; }

    uint64_t node_id = ni->node_id;
    if (ni->hop_count >= MAX_HOPS) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "NODEINFO from %s dropped: too many hops (%d)", from->log_name, ni->hop_count); return -1; }
    if (node_id == 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO node_id=0 from %s grp=%016llx — rejected (malformed)",
                    from->log_name, (unsigned long long)group->group_id);
        return -1;
    }
    if (node_id == group->instance->node_id) return 0;

    struct TOPO_GROUP_NODE* nodeinfo1 = topo_node_find_by_id(group, node_id);
    uint8_t new_ver = ni->ver;

    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO: node=%016llx ver=%d v4s=%d v4a=%d v6s=%d v6a=%d hops=%d from=%s",
                (unsigned long long)node_id, new_ver, ni->local_v4_sockets, ni->local_v4_addrs,
                ni->local_v6_sockets, ni->local_v6_addrs, ni->hop_count, from->log_name);

    struct ll_queue* paths = NULL;
    int is_new_node = 0;
    if (nodeinfo1) {
        paths = nodeinfo1->paths;
    } else { is_new_node = 1; }

    const uint8_t* ser_data = data + 2;
    size_t ser_len = len - 2;
    struct TOPO_NODE* new_ni = NULL;
    struct TOPO_NODESUBNETS* new_subnets = NULL;
    uint64_t* new_hop_list = NULL; uint8_t new_hop_count = 0;
    uint16_t incoming_cumulative_rtt = 0;
    if (topo_node_deserialize(group, ser_data, ser_len, &new_ni, &new_subnets, &new_hop_list, &new_hop_count, &incoming_cumulative_rtt) != 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO deserialize failed from %s nid=%016llx", from->log_name, (unsigned long long)node_id);
        return -1;
    }
    {   int v4c = topo_list_count((struct _topo_head*)new_ni->v4_addrs);
        int v6c = topo_list_count((struct _topo_head*)new_ni->v6_addrs);
        DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO deser: nid=%016llx v4a=%d v6a=%d from=%s",
                   (unsigned long long)node_id, v4c, v6c, from->log_name); }

    /* Проверяем целую запись до изменения маршрутов и общего реестра. */
    struct TOPO_NODE* current = topo_node_registry_find(group->instance->topo_groups, node_id);
    int conflict = current && current->timestamp == new_ni->timestamp &&
        memcmp(current->x25519_self_sig, new_ni->x25519_self_sig, 64);
    if (topo_node_verify(new_ni) < 0 || conflict) {
        DEBUG_WARN(DEBUG_CATEGORY_BGP, "NODEINFO rejected node=%016llx timestamp=%llu conflict=%d",
                   (unsigned long long)node_id, (unsigned long long)new_ni->timestamp, conflict);
        topo_node_destroy(group->instance->topo_groups, new_ni);
        struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets };
        topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected);
        u_free(new_hop_list);
        return -1;
    }
    if (nodeinfo1 && new_ni->timestamp <= nodeinfo1->last_timestamp) {
        DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO unchanged/stale node=%016llx current=%llu incoming=%llu",
                    (unsigned long long)node_id, (unsigned long long)nodeinfo1->last_timestamp,
                    (unsigned long long)new_ni->timestamp);
        uint64_t hops[MAX_HOPS];
        if (new_hop_count) memcpy(hops, new_hop_list, new_hop_count * sizeof(*hops));
        hops[new_hop_count] = from->peer_node_id;
        topo_group_remove_path(nodeinfo1, from);
        topo_group_add_path(nodeinfo1, from, hops, new_hop_count + 1, incoming_cumulative_rtt);
        topo_node_destroy(group->instance->topo_groups, new_ni);
        struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets };
        topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected);
        u_free(new_hop_list);
        return 0;
    }
    uint64_t incoming_timestamp = new_ni->timestamp;
    if (!nodeinfo1) {
        struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP_NODE));
        if (!qe) {
            DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group node allocation failed");
            topo_node_destroy(group->instance->topo_groups, new_ni);
            struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets };
            topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected);
            u_free(new_hop_list); return -1;
        }
        nodeinfo1 = (struct TOPO_GROUP_NODE*)qe;
        memset((uint8_t*)nodeinfo1 + sizeof(struct ll_entry), 0, sizeof(*nodeinfo1) - sizeof(struct ll_entry));
    }
    struct TOPO_NODE* stored = topo_node_registry_store(group->instance->topo_groups, new_ni);
    if (!stored) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "node registry update failed");
        topo_node_destroy(group->instance->topo_groups, new_ni);
        struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets };
        topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected);
        if (is_new_node) queue_entry_free(&nodeinfo1->ll);
        u_free(new_hop_list); return -1;
    }
    new_ni = stored;
    node_conn_direct_update_node(group->instance, node_id);
    if (!is_new_node) topo_nodeq_free_group_fields(group->instance->topo_groups, nodeinfo1);
    nodeinfo1->node_id = node_id;
    nodeinfo1->subnets = new_subnets;
    if (is_new_node) queue_data_put_with_index(group->nodes, &nodeinfo1->ll);
    if (is_new_node)
        DEBUG_INFO(DEBUG_CATEGORY_BGP, "node added to group: node=%016llx grp=%016llx type=%u ch=%s (bgp)",
                   (unsigned long long)node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id);

    nodeinfo1->paths = paths; nodeinfo1->last_ver = new_ver;
    nodeinfo1->last_timestamp = incoming_timestamp;
    nodeinfo1->radio = (ni->flags & TOPO_FLAG_RADIO) ? 1 : 0;

    uint64_t hop_list[MAX_HOPS];
    if (new_hop_count > 0) memcpy(hop_list, new_hop_list, new_hop_count * 8);
    hop_list[new_hop_count] = from->peer_node_id;
    uint8_t extended_count = new_hop_count + 1;

    u_free(new_hop_list);

    topo_group_remove_path_by_hop(nodeinfo1, from->peer_node_id);
    topo_group_add_path(nodeinfo1, from, hop_list, extended_count, incoming_cumulative_rtt);

    nodeinfo1->conn_presence |= NCONN_BGP; nodeinfo1->conn_up |= NCONN_BGP;
    if (from->peer_node_id == node_id) { nodeinfo1->conn_presence |= NCONN_DIRECT; nodeinfo1->conn_up |= NCONN_DIRECT; }

    if (group->group_type != TOPO_GROUP_TYPE_CHAT && group->instance->rt) route_insert(group->instance->rt, nodeinfo1);
    if (node_id != group->instance->node_id) {
        sqlite3* sdb = group->instance->topo_sqlite_db;
        if (sdb) {
            if (topo_node_sqlite_snapshot_put(sdb, new_ni) < 0)
                DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot persist received NODEINFO node=%016llx", (unsigned long long)node_id);
            /* узлы в nodes пишет только merkle-путь; BGP лишь дополняет IP для мемберов */
            if (topo_node_sqlite_member_exists(sdb, node_id))
                topo_node_sqlite_addrs_put(sdb, node_id, new_ni);
            if (group->group_type == TOPO_GROUP_TYPE_CHAT && group->channel_id[0]) {
                /* упрощённый плейсхолдер (source=1): pubkeys из BGP, без подписей.
                   не участвует в merkle-хеше, дополняется полной merkle-записью */
                topo_node_sqlite_member_placeholder_put(sdb, group->channel_id, node_id,
                                                        new_ni->public_key, new_ni->ed25519_public_key);
            }
            topo_node_sqlite_nodeinfo_updated(sdb, node_id);
        }
    }
    if (group->instance->control_srv) control_server_notify_node_change(group->instance->control_srv, nodeinfo1);

    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "node_updated_cb: node=%016llx has_cb=%d ch=%s type=%d", (unsigned long long)node_id, group->instance->topo_groups->node_updated_cb ? 1 : 0, group->channel_id, group->group_type);
    if (group->instance->topo_groups->node_updated_cb)
        group->instance->topo_groups->node_updated_cb(group->instance, node_id, ni->public_key, ni->ed25519_public_key);

    {   /* fire BGP node event callbacks */
        int ev = is_new_node ? TOPO_NODE_EVENT_NEW : TOPO_NODE_EVENT_UPDATE;
        struct topo_node_cbk_entry* c = group->node_cbks;
        while (c) { c->fn(group, node_id, ev, c->arg); c = c->next; }
    }
    topo_fire_nodeinfo_cbk(group->instance, group, nodeinfo1);

    int hop_count = extended_count;
    struct ll_entry* se = group->senders_list ? group->senders_list->head : NULL;
    while (se) {
        struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data;
        if (item->conn) {
            uint64_t id = item->conn->peer_node_id;
            if (id != node_id) {
                int found = 0; for (int i = 0; i < hop_count; i++) if (hop_list[i] == id) found = 1;
                if (found == 0) {
                    uint16_t out_rtt = topo_get_chain_rtt(nodeinfo1);
                    topo_group_send_nodeinfo(group, nodeinfo1, item->conn, out_rtt);
                }
            }
        }
        se = se->next;
    }

    int socks_changed = 1;
    if (!is_new_node) socks_changed = 1;

    if (node_id != group->instance->node_id && (is_new_node || socks_changed)) {
        struct TOPO_NODE* tni = topo_node_registry_find(group->instance->topo_groups, nodeinfo1->node_id);
        int v4a = tni ? topo_list_count((struct _topo_head*)tni->v4_addrs) : 0;
        int v6a = tni ? topo_list_count((struct _topo_head*)tni->v6_addrs) : 0;
        if (v4a > 0 || v6a > 0) route_connectivity_probe_node(group->instance, group, nodeinfo1);
    }

    return 0;
}

/* Обрабатывает WITHDRAW: удаляет узел/пути, чистит роутинг, форвардит withdraw. */
int topo_group_process_withdraw(struct TOPO_GROUP* group, struct ETCP_CONN* sender, const uint8_t* data, size_t len) {
    if (!group || len < sizeof(struct TOPOMSG_WITHDRAW_PKT)) return -1;
    struct TOPOMSG_WITHDRAW_PKT* wp = (struct TOPOMSG_WITHDRAW_PKT*)data;
    uint64_t node_id = wp->node_id, wd_source = wp->wd_source;

    struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, node_id);
    if (!nq) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "node not found"); return 0; }
    int ret = topo_group_remove_path_by_hop(nq, wd_source);
    if (ret == 1) {
        nq->conn_presence = 0; nq->conn_up = 0;
        topo_fire_nodeinfo_cbk(group->instance, group, nq);
        if (group->group_type != TOPO_GROUP_TYPE_CHAT && group->instance && group->instance->rt) route_delete(group->instance->rt, nq);
        if (group->instance->control_srv) control_server_notify_node_removed(group->instance->control_srv, node_id);
        topo_nodeq_remove_node(group, nq);
        DEBUG_INFO(DEBUG_CATEGORY_BGP, "node removed from group: node=%016llx grp=%016llx type=%u ch=%s (withdraw)",
                   (unsigned long long)node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id);
        topo_group_broadcast_withdraw(group, node_id, wd_source, sender);
        /* BGP не удаляет member-запись: он работает только со своей RAM-таблицей.
           Удаление из peers_* — только битая подпись при верификации (member_sync). */
        {   /* fire BGP REMOVE callback */
            struct topo_node_cbk_entry* c = group->node_cbks;
            while (c) { c->fn(group, node_id, TOPO_NODE_EVENT_REMOVE, c->arg); c = c->next; }
        }
    }
    return 0;
}

/* Сериализует и шлёт NODEINFO узла конкретному пиру. */
void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* node, struct ETCP_CONN* conn, uint16_t cumulative_rtt) {
    if (!group || !node || !conn) return;

    size_t max_sz = sizeof(struct TOPOMSG_NODEINFO_PKT) + 4096;
    uint8_t* p = u_malloc(max_sz);
    if (!p) return;

    p[0] = ETCP_ID_TOPO_ENTRY; p[1] = TOPO_SUBCMD_NODEINFO;

    struct TOPO_NODE* sni = topo_node_registry_find(group->instance->topo_groups, node->node_id);
    if (!sni) {
        DEBUG_WARN(DEBUG_CATEGORY_BGP,
            "send_nodeinfo: node %016llx NOT in registry — skip forward to %s grp=%016llx type=%d ch=%s is_local=%d paths=%d nodes_total=%d",
            (unsigned long long)node->node_id, conn->log_name,
            (unsigned long long)group->group_id, group->group_type,
            group->channel_id[0] ? group->channel_id : "-",
            (node == group->local_node) ? 1 : 0,
            node->paths ? queue_entry_count(node->paths) : -1,
            group->nodes ? queue_entry_count(group->nodes) : -1);
        u_free(p);
        return;
    }
    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "send_nodeinfo: node %016llx ver=%d grp=%016llx to conn=%p name='%s'", (unsigned long long)node->node_id, sni->ver, (unsigned long long)group->group_id, (void*)conn, conn->log_name);
    uint8_t sflags = (group->group_type == TOPO_GROUP_TYPE_CHAT) ? 0 : TOPO_FLAG_SEND_SUBNETS;
    int ser_len = topo_node_serialize(sni, node, group->group_id, sflags, p + 2, max_sz - 2, cumulative_rtt);
    if (ser_len < 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: serialize failed for node %016llx", (unsigned long long)node->node_id); u_free(p); return; }

    struct ll_entry* e = queue_entry_new(0);
    if (!e) { u_free(p); return; }
    e->dgram = p; e->len = (size_t)ser_len + 2;

    if (etcp_send(conn, e) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: etcp_send FAILED for node %016llx to %s", (unsigned long long)node->node_id, conn->log_name); u_free(p); queue_entry_free(e); }
}

/* Добавляет conn в senders_list (дедуп), если его там ещё нет. */
static int topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn || !group->senders_list) return -1;
    for (struct ll_entry* e = group->senders_list->head; e; e = e->next)
        if (((struct TOPO_GROUP_CONN_ITEM*)e->data)->conn == conn) return 0;
    struct ll_entry* e = queue_entry_new(sizeof(struct TOPO_GROUP_CONN_ITEM));
    if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_add: allocation failed"); return -1; }
    struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
    item->conn = conn; item->handle = NULL;
    if (node_conn_direct_open(conn->instance, conn->peer_node_id, NULL, NULL, &item->handle, NULL) == NCD_ERR) {
        DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_add: cannot acquire NCD handle peer=%016llx", (unsigned long long)conn->peer_node_id);
        queue_entry_free(e); return -1;
    }
    queue_data_put(group->senders_list, e);
    DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "senders_add: acquired peer=%016llx group=%016llx",
                (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id);
    return 0;
}

/* Решает, надо ли слать узел пиру (не шлём, если пир уже есть в hop_list пути). */
static bool topo_group_should_send_to(const struct TOPO_GROUP_NODE* nq, uint64_t target_id) {
    if (!nq || !nq->paths) return false;
    if (target_id == nq->node_id) return false;
    struct ll_entry* e = nq->paths->head;
    while (e) {
        struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e;
        uint64_t* hop = (uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH));
        bool has_id = false; for (uint8_t i = 0; i < path->hop_count; i++) if (hop[i] == target_id) { has_id = true; break; }
        if (!has_id) return true;
        e = e->next;
    }
    return false;
}

/* Шлёт пиру полную таблицу узлов группы (все узлы, кроме достижимых через него). */
static void topo_group_send_full_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) return;
    uint64_t target = conn->peer_node_id;
    struct ll_entry* e = group->nodes ? group->nodes->head : NULL;
    while (e) {
        struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
        if (nq->node_id == group->instance->node_id) { e = e->next; continue; }
        if (!nq->paths) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "node has no paths"); e = e->next; continue; }
        if (topo_group_should_send_to(nq, target)) topo_group_send_nodeinfo(group, nq, conn, topo_get_chain_rtt(nq));
        e = e->next;
    }
}

/* Обрабатывает REQUEST_TABLE: шлёт свой nodeinfo + полную таблицу + TABLE_COMPLETE. */
static void topo_group_handle_request_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) return;
    if (!topo_group_peer_allowed(group, conn->peer_node_id)) {
        DEBUG_INFO(DEBUG_CATEGORY_BGP, "table request deferred: member unknown peer=%016llx group=%016llx",
                   (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id);
        return;
    }
    topo_group_send_nodeinfo(group, group->local_node, conn, 0);
    topo_group_send_full_table(group, conn);
    /* senders_list заполняется только через topo_group_new_conn (add + BGP);
       здесь не добавляем, иначе new_conn (на JOIN_GROUP) упирается в дедуп и не шлёт TABLE_REQ обратно. */
    topo_group_send_table_complete(group, conn);
}

/* Обрабатывает JOIN_GROUP: для UTUN — добавить запросившего и инициировать BGP обратно. */
static void topo_group_handle_join_group(struct TOPO_GROUP* group, struct ETCP_CONN* conn) {
    if (!group || !conn) return;
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "handle_join_group: from %s grp=%016llx type=%d", conn->log_name, (unsigned long long)group->group_id, group->group_type);
    /* Проверка таблицы участников общая для входящих и локальных запросов. */
    int existing = topo_group_has_sender(group, conn);
    if (topo_group_new_conn(group, conn) == 0 && existing)
        topo_group_send_table_request(group, conn);
}

/* Обрабатывает RESYNC: если мы VPN-клиент пира — заново шлём JOIN_GROUP по группам. */
static void topo_group_handle_resync(struct UTUN_INSTANCE* instance, struct ETCP_CONN* conn) {
    if (!instance || !conn || !conn->instance || !instance->topo_groups) return;
    /* общий сигнал от пассивной стороны: рестартуем обмен, если мы для этого пира VPN-клиент */
    int is_client = conn->instance->config && config_peer_in_clients(conn->instance->config, conn->crypto_ctx.peer_public_key);
    if (!is_client && conn->bgp_ready_cbk == NULL) return;
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "handle_resync: from %s — re-announce groups", conn->log_name);
    struct ll_entry* fe = instance->topo_groups->group_list->head;
    while (fe) {
        struct TOPO_GROUP* g = (struct TOPO_GROUP*)fe; fe = fe->next;
        if (g->group_type != TOPO_GROUP_TYPE_CHAT)
            topo_group_send_join_group(g, conn);
    }
}

/* ── BGP node event callbacks ── */

/* Регистрирует подписчика на события узлов группы (NEW/UPDATE/REMOVE). */
void topo_group_add_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg) {
    if (!group || !fn) return;
    struct topo_node_cbk_entry* e = u_malloc(sizeof(*e));
    if (!e) return;
    e->fn = fn; e->arg = arg;
    e->next = group->node_cbks;
    group->node_cbks = e;
}

/* Снимает подписчика событий узлов группы. */
void topo_group_remove_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg) {
    if (!group || !fn) return;
    struct topo_node_cbk_entry** p = &group->node_cbks;
    while (*p) {
        if ((*p)->fn == fn && (*p)->arg == arg) {
            struct topo_node_cbk_entry* r = *p;
            *p = r->next; u_free(r);
            return;
        }
        p = &(*p)->next;
    }
}

/* ── Global nodeinfo callbacks (instance-level, any node change) ── */

/* Регистрирует глобальный колбэк изменения nodeinfo (уровень инстанса). */
void utun_add_nodeinfo_cbk(struct UTUN_INSTANCE* instance, nodeinfo_cbk_fn fn, void* arg) {
    if (!instance || !fn) return;
    struct nodeinfo_cbk_entry* e = u_malloc(sizeof(*e));
    if (!e) return;
    e->fn = fn; e->arg = arg;
    e->next = instance->nodeinfo_cbks;
    instance->nodeinfo_cbks = e;
}

/* Снимает глобальный колбэк nodeinfo. */
void utun_remove_nodeinfo_cbk(struct UTUN_INSTANCE* instance, nodeinfo_cbk_fn fn, void* arg) {
    if (!instance || !fn) return;
    struct nodeinfo_cbk_entry** p = &instance->nodeinfo_cbks;
    while (*p) {
        if ((*p)->fn == fn && (*p)->arg == arg) {
            struct nodeinfo_cbk_entry* r = *p;
            *p = r->next; u_free(r);
            return;
        }
        p = &(*p)->next;
    }
}

/* Вызывает все глобальные nodeinfo-колбэки для узла. */
void topo_fire_nodeinfo_cbk(struct UTUN_INSTANCE* instance, struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* node) {
    if (!instance || !group || !node) return;
    struct nodeinfo_cbk_entry* c = instance->nodeinfo_cbks;
    while (c) { c->fn(group, node, c->arg); c = c->next; }
}

/* Публичная рассылка WITHDRAW об удалении узла. */
void topo_group_send_withdraw(struct TOPO_GROUP* group, uint64_t node_id) {
    if (!group) return;
    DEBUG_INFO(DEBUG_CATEGORY_BGP, "node=%016llx", (unsigned long long)node_id);
    topo_group_broadcast_withdraw(group, node_id, group->instance->node_id, NULL);
}