/*
 * node_conn_direct.c — handle-based ETCP connection layer
 *
 * Тонкая прослойка над ETCP: управляет прямыми подключениями по node_id.
 * Несколько handle'ов могут разделять одно ETCP_CONN.
 * Conn закрывается при закрытии последнего handle (с протоколом CLOSE/KEEP_ALIVE).
 */

#include "node_conn_direct.h"
#include "etcp_api.h"
#include "etcp.h"
#include "etcp_connections.h"
#include "call_ring.h"
#include "secure_channel.h"
#include "utun_instance.h"
#include "topo_node.h"
#include "topo_node_sqlite.h"
#include "sock_match.h"
#include "nat_detection.h"
#include "topo_group.h"
#include "ntp_time.h"
#include "config_parser.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/u_async.h"
#include "../lib/ll_queue.h"
#include "../lib/socket_compat.h"
#include "../lib/platform_compat.h"
#include <string.h>
#include <limits.h>

#define DEBUG_CATEGORY_NCD DEBUG_CATEGORY_ETCP

/* ═══════════ локальные константы ═══════════ */

#define NCD_FIN_WAIT_TIMEOUT_TB 50000  /* 5 сек в 0.1ms */

/* ═══════════ внутренние структуры ═══════════ */

struct ncd_entry {
    uint64_t                    node_id;
    struct ETCP_CONN*           conn;
    struct UASYNC*              ua;
    struct UTUN_INSTANCE*      inst;
    unsigned                   dispatch_depth;
    uint8_t                    closed;
    uint8_t                    node_ref;
    uint64_t                   node_timestamp;
    int                         handle_count;

    struct NODE_CONN_DIRECT*    handles;           /* связный список всех handle'ов */

    void*                       connect_timer;     /* однократный таймер первого подъёма */
    void*                       fin_wait_timer;    /* таймер ожидания ответа на CLOSE */
    uint8_t                     up     : 1;        /* текущий статус: 1=есть живой линк  */
    uint8_t                     timed_out : 1;     /* connect_timer уже сработал          */

    struct ncd_entry*           next;              /* цепочка статического реестра        */
};

struct NODE_CONN_DIRECT {
    struct ncd_entry*           entry;
    ncd_callback                cb;
    void*                       cb_arg;
    struct NODE_CONN_DIRECT*    next;
    uint8_t                     closed;
    void*                       deliver_up_token;   /* handle uasync_call_soon отложенной доставки UP (для отмены при close) */
};

/* ─── forward declarations ─── */
static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event);
static void ncd_connect_timeout_cb(void* arg);
static void ncd_deliver_up_cb(void* arg);
static void ncd_init_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_up_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_down_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_delete_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_finish(struct ncd_entry* entry, int close_conn);
static void ncd_collect(struct ncd_entry* entry);
static void ncd_cancel_deliver_up(struct NODE_CONN_DIRECT* h, struct UASYNC* ua);
static void ncd_deferred_close_conn(void* arg);

/* ═══════════ реестр ═══════════ */
/*
 * Реестр ncd_entry по node_id — собственный связный список (не inst->connections).
 * ncd_entry хранит состояние, которого нет в ETCP_CONN: handles, таймеры, fin_wait.
 * Нужен чтобы при повторном open не создавать дублирующий conn для того же node_id.
 */

static struct ncd_entry* ncd_registry_find(struct UTUN_INSTANCE* inst, uint64_t node_id) {
    struct ncd_entry* e = (struct ncd_entry*)inst->ncd_registry;
    while (e) { if (e->node_id == node_id && !e->closed && !e->conn->close_requested && e->conn->state != 2) return e; e = e->next; }
    return NULL;
}

// Проверка владения по объекту conn исключает смешение старого и нового транспорта одного пира.
int node_conn_direct_handle_count(const struct ETCP_CONN* conn) {
    if (!conn || !conn->instance) return 0;
    for (const struct ncd_entry* entry = conn->instance->ncd_registry; entry; entry = entry->next)
        if (entry->conn == conn && !entry->closed) return entry->handle_count;
    return 0;
}

static int ncd_registry_add(struct UTUN_INSTANCE* inst, struct ncd_entry* entry) {
    entry->inst = inst;
    if (etcp_conn_ref_take(entry->conn) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] register closing conn"); return -1; }
    etcp_conn_add_cbk(entry->conn, ncd_delete_cb, entry, ETCP_CBK_EVENT_DELETE);
    entry->next = (struct ncd_entry*)inst->ncd_registry; inst->ncd_registry = entry;
    return 0;
}
static void ncd_registry_remove(struct UTUN_INSTANCE* inst, struct ncd_entry* entry) {
    struct ncd_entry** pp = (struct ncd_entry**)&inst->ncd_registry;
    while (*pp) { if (*pp == entry) { *pp = entry->next; return; } pp = &(*pp)->next; }
}

/* ═══════════ поиск узла ═══════════ */
/*
 * Загружает информацию об узле (адреса, pubkey) для создания линков.
 * Сначала ищет в памяти (node_registry — туда попадают узлы из BGP/topo),
 * если нет в памяти — подгружает из SQLite и помещает в реестр.
 * Возвращает владеющую ссылку (ref++), вызывающий обязан сделать topo_node_registry_unref.
 */

static struct TOPO_NODE* ncd_lookup_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
    if (inst->topo_groups && inst->topo_groups->node_registry) {
        struct ll_entry* e = queue_find_data_by_index(inst->topo_groups->node_registry, (const uint8_t*)&node_id);
        if (e) {
            struct TOPO_NODE* node;
            memcpy(&node, e->data + 8, sizeof(node));
            if (node) { topo_node_registry_ref(inst->topo_groups, node_id); return node; }
        }
    }
    if (inst->topo_sqlite_db && inst->topo_groups) {
        struct TOPO_NODE* ni = topo_node_sqlite_node_load(inst->topo_sqlite_db, inst->topo_groups, node_id);
        if (ni) return topo_node_registry_store(inst->topo_groups, ni);
    }
    return NULL;
}

/* ═══════════ создание линков (round‑robin, все сокеты кроме PRIVATE) ═══════════ */
/*
 * Создаёт по одному ETCP_LINK на каждый адрес пира (IPv4 и IPv6).
 *
 * Все адреса пира распределяются round-robin по локальным сокетам.
 * Если specific_sock задан — использует только его. PRIVATE-сокеты пропускаются.
 *
 * Особые случаи:
 *   - Нулевые адреса и нулевые порты пропускаются
 *   - Не-UDP протоколы пропускаются
 *   - IPv6 link-local: автоматически выставляется scope_id по netif_index сокета
 *   - Stale-линки: если на том же addr:port висит старый conn с другим pubkey
 *     и старый conn ещё не поднялся — stale-линк вытесняется (узел пересоздался
 *     с новым ключом, старый conn больше не нужен)
 */

/* Создаёт один UDP-линк на use_sock → sa, с вытеснением stale-линка и дедупом. */
#define NCD_MAX_SOCKET_VIEWS 32
static int ncd_add_udp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* use_sock, struct sockaddr_storage* sa) {
    struct ETCP_LINK* stale = etcp_link_find_by_addr(use_sock, sa, 0);
    if (stale && stale->etcp == conn) return 0;
    if (stale && stale->etcp != conn
        && memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE))
    {
        if (!stale->etcp->links_up || !stale->etcp->initialized) {
            struct ETCP_CONN* old = stale->etcp;
            DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "[ncd] evict stale link node=0x%016llx(old pk=%016llx) for new node=0x%016llx(pk=%016llx)",
                       (unsigned long long)old->peer_node_id, *(const uint64_t*)old->crypto_ctx.peer_public_key,
                       (unsigned long long)conn->peer_node_id, *(const uint64_t*)conn->crypto_ctx.peer_public_key);
            etcp_link_close(stale);
            etcp_cbk_fire(old, ETCP_CBK_EVENT_NODE_CHANGED);
            stale = NULL;
        } else {
            DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "[ncd] addr conflict: port already used by live conn 0x%016llx for node 0x%016llx",
                        (unsigned long long)stale->etcp->peer_node_id, (unsigned long long)conn->peer_node_id);
            return 0;
        }
    }
    if (stale && stale->etcp != conn
        && !memcmp(conn->crypto_ctx.peer_public_key, stale->etcp->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE))
    {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCP,
            "[ncd] add_udp_link: SAME-PUBKEY DUPLICATE conn at %s — new=%p [%s] state=%d / old=%p [%s] state=%d init=%d up=%d",
            sockaddr_storage_to_str(sa).str,
            (void*)conn, conn->log_name, conn->state,
            (void*)stale->etcp, stale->etcp->log_name, stale->etcp->state,
            stale->etcp->initialized, stale->etcp->links_up);
        return 0;
    }
    return etcp_link_new(conn, use_sock, sa, 0) ? 1 : 0;
}

static int ncd_same_address(const struct sockaddr_storage* a, const struct sockaddr_storage* b) {
    if (a->ss_family != b->ss_family) return 0;
    if (a->ss_family == AF_INET) {
        const struct sockaddr_in* x = (const struct sockaddr_in*)a;
        const struct sockaddr_in* y = (const struct sockaddr_in*)b;
        return x->sin_port == y->sin_port && x->sin_addr.s_addr == y->sin_addr.s_addr;
    }
    if (a->ss_family == AF_INET6) {
        const struct sockaddr_in6* x = (const struct sockaddr_in6*)a;
        const struct sockaddr_in6* y = (const struct sockaddr_in6*)b;
        return x->sin6_port == y->sin6_port && x->sin6_scope_id == y->sin6_scope_id &&
            !memcmp(&x->sin6_addr, &y->sin6_addr, sizeof(x->sin6_addr));
    }
    return 0;
}

static int ncd_add_tcp_link(struct ETCP_CONN* conn, struct ETCP_SOCKET* sock,
                            struct sockaddr_storage* addr, const struct TOPO_REALITY_SOCK* reality) {
    for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
        if (!link->is_tcp || link->is_server || link->conn != sock || !ncd_same_address(&link->remote_addr, addr)) continue;
        struct reality_client_config before; memcpy(&before, &link->reality, sizeof(before));
        int had_reality = link->reality_set;
        if (reality) topo_node_apply_reality(link, reality);
        else { memset(&link->reality, 0, sizeof(link->reality)); link->reality_set = 0; }
        if (had_reality != link->reality_set || memcmp(&before, &link->reality, sizeof(before))) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] TCP parameters updated node=%016llx link=%d up=%d",
                       (unsigned long long)conn->peer_node_id, link->local_link_id, link->link_status);
            if (!link->link_status) etcp_tcp_link_start_reconnect(link);
        }
        if (!link->link_status && !link->tcp_link && !link->tcp_reconnect_timer) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] resume idle TCP probe node=%016llx link=%u",
                       (unsigned long long)conn->peer_node_id, link->local_link_id);
            etcp_tcp_link_start_reconnect(link);
        }
        return 0;
    }
    struct ETCP_LINK* link = etcp_link_new(conn, sock, addr, 0);
    if (!link) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] TCP link allocation failed"); return -1; }
    topo_node_apply_reality(link, reality);
    etcp_tcp_link_start_connect(link, addr, 0);
    return 1;
}

/* Проверка совместимости и создание используют один обход пар, включая TCP fallback и IPv6. */
static int ncd_create_links(struct ncd_entry* entry, struct TOPO_NODE* ni,
                             struct ETCP_SOCKET* specific_sock, struct UTUN_INSTANCE* check_inst) {
    if (!ni) { DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] link matching without node snapshot"); return 0; }
    if (!check_inst && ni->timestamp < entry->node_timestamp) {
        DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] ignore stale addresses node=%016llx current=%llu incoming=%llu",
                    (unsigned long long)entry->node_id, (unsigned long long)entry->node_timestamp,
                    (unsigned long long)(ni ? ni->timestamp : 0)); return 0;
    }
    struct ETCP_CONN* conn = check_inst ? NULL : entry->conn;
    struct UTUN_INSTANCE* inst = check_inst ? check_inst : conn->instance;
    if (!check_inst && memcmp(ni->public_key, conn->crypto_ctx.peer_public_key, SC_PUBKEY_SIZE)) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] address update key mismatch node=%016llx", (unsigned long long)entry->node_id);
        return -1;
    }
    if (!check_inst && !entry->node_ref && topo_node_registry_find(inst->topo_groups, entry->node_id) == ni) {
        topo_node_registry_ref(inst->topo_groups, entry->node_id); entry->node_ref = 1;
    }
    if (!check_inst) entry->node_timestamp = ni->timestamp;
    int link_count = 0, v4_skip = 0, v6_skip = 0, v4_addrs = 0, v6_addrs = 0;
    uint64_t nid = check_inst ? ni->node_id : entry->node_id;

    /* ---- IPv4 ---- */
    struct sock_view views[NCD_MAX_SOCKET_VIEWS];
    int nviews = sock_collect_views(inst, views, NCD_MAX_SOCKET_VIEWS);
    for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
        v4_addrs++;
        if (a->port == 0 || !(a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP))) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v4 skip proto=%02x port=%d node=0x%016llx", a->protocol, (int)a->port, (unsigned long long)nid); v4_skip++; continue; }
        int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j] != 0) { zero = 0; break; }
        if (zero) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v4 skip zero-addr node=0x%016llx", (unsigned long long)nid); continue; }
        struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET;
        memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port);
        struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin));
        uint32_t tip; memcpy(&tip, a->addr, 4);
        const struct TOPO_SOCKMETA4* m = NULL;
        for (const struct TOPO_SOCKMETA4* mm = ni->v4_sock_meta; mm; mm = mm->next)
            if (mm->id == a->socket_id) { m = mm; break; }
        uint8_t tcfg = m ? m->config_type : CFG_SERVER_TYPE_UNKNOWN;
        const struct address_profile *profile = m ? &m->address_profile : NULL;
        uint8_t target_class = sock_target_class(a->type, tcfg, profile, ip_is_public(tip));
        if (target_class == SOCK_TC_UNREACH) { v4_skip++; continue; }

        if (a->protocol & TOPO_PROTO_TCP) {
            if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(ni, a->socket_id)) {
                DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] missing REALITY metadata node=%016llx socket=%u",
                           (unsigned long long)nid, a->socket_id); continue;
            }
            int tcp_sockets = 0;
            for (int vi = 0; vi < nviews; vi++) {
                const struct sock_view* v = &views[vi];
                if (!v->is_tcp) continue;
                tcp_sockets++;
                if (specific_sock ? v->sock != specific_sock : !sock_match(v, a->type, tcfg, profile, tip)) continue;
                if (check_inst) return 1;
                int added = ncd_add_tcp_link(conn, v->sock, &sa, topo_node_find_reality_sock(ni, a->socket_id));
                if (added > 0) link_count += added;
            }
            if (!specific_sock && !tcp_sockets) {
                if (check_inst) return 1;
                int added = ncd_add_tcp_link(conn, NULL, &sa, topo_node_find_reality_sock(ni, a->socket_id));
                if (added > 0) link_count += added;
            }
        }
        if (a->protocol & TOPO_PROTO_UDP) {
            if (specific_sock && !specific_sock->is_tcp && specific_sock->local_addr.ss_family == AF_INET) {
                if (check_inst) return 1;
                link_count += ncd_add_udp_link(conn, specific_sock, &sa);
            } else if (!specific_sock) {
                for (int vi = 0; vi < nviews; vi++) {
                    const struct sock_view* v = &views[vi];
                    if (v->is_tcp || !sock_match(v, a->type, tcfg, profile, tip)) continue;
                    if (check_inst) return 1;
                    link_count += ncd_add_udp_link(conn, v->sock, &sa);
                }
            }
        }
    }

    /* ---- IPv6 ---- */
    for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
        v6_addrs++;
        if (a->port == 0 || !(a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP))) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v6 skip proto=%02x port=%d node=0x%016llx", a->protocol, (int)a->port, (unsigned long long)nid); v6_skip++; continue; }
        int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j] != 0) { zero = 0; break; }
        if (zero) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[ncd] v6 skip zero-addr node=0x%016llx", (unsigned long long)nid); continue; }
        int is_ll = (a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80);
        if (a->addr[0] == 0xff) { v6_skip++; continue; }
        const struct TOPO_SOCKMETA6 *metadata = ni->v6_sock_meta;
        while (metadata && metadata->id != a->socket_id) metadata = metadata->next;
        int global = (a->addr[0] & 0xe0) == 0x20;
        uint8_t target_class = sock_target_class(a->type, metadata ? metadata->config_type : CFG_SERVER_TYPE_UNKNOWN,
            metadata ? &metadata->address_profile : NULL, global);
        if (target_class == SOCK_TC_UNREACH) { v6_skip++; continue; }
        struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6;
        memcpy(&sin6.sin6_addr, a->addr, 16); sin6.sin6_port = htons(a->port);

        if (a->protocol & TOPO_PROTO_TCP) {
            if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(ni, a->socket_id)) {
                DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] missing REALITY metadata node=%016llx socket=%u",
                           (unsigned long long)nid, a->socket_id); continue;
            }
            int tcp_sockets = 0;
            if (is_ll) { struct ETCP_SOCKET* sv = inst->etcp_sockets;
                while (sv) { if (!sv->probe_only && sv->local_addr.ss_family == AF_INET6 && sv->netif_index) { sin6.sin6_scope_id = sv->netif_index; break; } sv = sv->next; } }
            struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
            for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) {
                if (!s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6) continue;
                tcp_sockets++;
                if (specific_sock ? s != specific_sock : global && s->type == CFG_SERVER_TYPE_LOCAL) continue;
                if (check_inst) return 1;
                if (is_ll) { sin6.sin6_scope_id = s->netif_index; memcpy(&sa, &sin6, sizeof(sin6)); }
                int added = ncd_add_tcp_link(conn, s, &sa, topo_node_find_reality_sock(ni, a->socket_id));
                if (added > 0) link_count += added;
            }
            if (!specific_sock && !tcp_sockets && !is_ll) {
                if (check_inst) return 1;
                int added = ncd_add_tcp_link(conn, NULL, &sa, topo_node_find_reality_sock(ni, a->socket_id));
                if (added > 0) link_count += added;
            }
        }
        if (a->protocol & TOPO_PROTO_UDP) {
            if (specific_sock && !specific_sock->is_tcp && specific_sock->local_addr.ss_family == AF_INET6) {
                if (check_inst) return 1;
                if (is_ll) sin6.sin6_scope_id = specific_sock->netif_index;
                struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
                link_count += ncd_add_udp_link(conn, specific_sock, &sa);
            } else if (!specific_sock) {
                for (struct ETCP_SOCKET* s = inst->etcp_sockets; s; s = s->next) {
                    if (s->is_tcp || s->probe_only || s->local_addr.ss_family != AF_INET6 ||
                        (global && s->type == CFG_SERVER_TYPE_LOCAL)) continue;
                    if (check_inst) return 1;
                    if (is_ll) sin6.sin6_scope_id = s->netif_index;
                    struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
                    link_count += ncd_add_udp_link(conn, s, &sa);
                }
            }
        }
    }

    if (link_count > 0) {
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] create_links: node=0x%016llx v4=%d/skp=%d v6=%d/skp=%d → %d new links (UDP+TCP)",
                   (unsigned long long)nid, v4_addrs, v4_skip, v6_addrs, v6_skip, link_count);
    }
    return link_count;
}

int node_conn_direct_compatible(struct UTUN_INSTANCE* inst, uint64_t node_id) {
    if (!inst || !node_id) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] compatibility invalid arguments"); return 0; }
    struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
    if (conn && conn->state != 2 && !conn->close_requested && conn->links_up) return 1;
    struct TOPO_NODE* node = topo_node_registry_find(inst->topo_groups, node_id);
    if (!node) { DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] compatibility unknown snapshot node=%016llx", (unsigned long long)node_id); return 0; }
    if (ncd_create_links(NULL, node, NULL, inst) > 0) return 1;
    /* Входящее подключение тоже допустимо: REVERSE_REQ просит пира открыть наш опубликованный адрес.
     * TCP имеет собственный исходящий socket fallback; для UDP нужен известный UDP-сокет пира.
     * Это возможность попытки, не подтверждение маршрута/работоспособности скрытых сокетов пира. */
    const struct TOPO_NODE* self = topo_node_registry_find(inst->topo_groups, inst->node_id);
    if (!self) return 0;
    for (const struct TOPO_ADDR4* a = self->v4_addrs; a; a = a->next) {
        uint32_t ip; memcpy(&ip, a->addr, 4);
        if (!ip || !a->port) continue;
        const struct TOPO_SOCKMETA4* meta = self->v4_sock_meta;
        while (meta && meta->id != a->socket_id) meta = meta->next;
        uint8_t cfg = meta ? meta->config_type : CFG_SERVER_TYPE_UNKNOWN;
        const struct address_profile *profile = meta ? &meta->address_profile : NULL;
        uint8_t target_class = sock_target_class(a->type, cfg, profile, ip_is_public(ip));
        if (target_class == SOCK_TC_PRIV || target_class == SOCK_TC_UNREACH) continue;
        if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(self, a->socket_id)) continue;
        if (a->protocol & TOPO_PROTO_TCP) return 1;
        if (!(a->protocol & TOPO_PROTO_UDP)) continue;
        for (const struct TOPO_ADDR4* remote = node->v4_addrs; remote; remote = remote->next) {
            if (!(remote->protocol & TOPO_PROTO_UDP)) continue;
            const struct TOPO_SOCKMETA4* rm = node->v4_sock_meta;
            while (rm && rm->id != remote->socket_id) rm = rm->next;
            struct sock_view view = {.has_inet = 1};
            if (rm) view.address_profile = rm->address_profile;
            for (const struct TOPO_ADDR4* ra = node->v4_addrs; ra; ra = ra->next)
                if (ra->socket_id == remote->socket_id && ra->type == TOPO_ADDR_NAT) memcpy(&view.public_ip4, ra->addr, 4);
            if (sock_match(&view, a->type, cfg, profile, ip)) return 1;
        }
    }
    for (const struct TOPO_ADDR6* a = self->v6_addrs; a; a = a->next) {
        int nonzero = 0; for (unsigned i = 0; i < 16; i++) nonzero |= a->addr[i];
        if (!nonzero || !a->port || (a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) continue;
        const struct TOPO_SOCKMETA6* meta = self->v6_sock_meta;
        while (meta && meta->id != a->socket_id) meta = meta->next;
        if (meta && (meta->config_type == CFG_SERVER_TYPE_PRIVATE || meta->config_type == CFG_SERVER_TYPE_LOCAL)) continue;
        if ((a->protocol & TOPO_PROTO_REALITY) && !topo_node_find_reality_sock(self, a->socket_id)) continue;
        if (a->protocol & TOPO_PROTO_TCP) return 1;
        if (!(a->protocol & TOPO_PROTO_UDP)) continue;
        for (const struct TOPO_ADDR6* remote = node->v6_addrs; remote; remote = remote->next)
            if (remote->protocol & TOPO_PROTO_UDP) return 1;
    }
    return 0;
}

int node_conn_direct_update_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
    if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] update without instance"); return -1; }
    struct ncd_entry* entry = ncd_registry_find(inst, node_id);
    struct TOPO_NODE* ni = topo_node_registry_find(inst->topo_groups, node_id);
    if (!entry || !ni) return 0;
    int added = ncd_create_links(entry, ni, NULL, NULL);
    DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] apply node=%016llx timestamp=%llu up=%d added=%d",
                (unsigned long long)node_id, (unsigned long long)ni->timestamp, entry->up, added);
    return added;
}

/* ═══════════ единая диспетчеризация событий ═══════════ */
/*
 * Центральный диспетчер: принимает событие (UP/DOWN/TIMEOUT), проверяет что оно
 * действительно меняет состояние (повторный UP когда уже up — игнорируется),
 * обновляет entry->up/timed_out и рассылает callback всем handle'ам.
 *
 * Именно через эту функцию все потребители узнают об изменении состояния соединения.
 * Один вызов — один проход по всем handle'ам.
 */

static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
    if (!entry || (entry->closed && event != NCD_EVENT_CLOSED)) return;
    switch (event) {
    case NCD_EVENT_UP:
        if (entry->up) return;
        if (entry->timed_out) DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] connected after timeout node=%016llx",
                                       (unsigned long long)entry->node_id);
        entry->timed_out = 0;
        if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
        entry->up = 1; break;
    case NCD_EVENT_DOWN:
        if (!entry->up) return;
        entry->up = 0; break;
    case NCD_EVENT_TIMEOUT:
        if (entry->timed_out) return;
        entry->timed_out = 1; entry->connect_timer = NULL; break;
    case NCD_EVENT_CLOSED: entry->up = 0; break;
    }
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] event=%d node=%016llx handles=%d depth=%u", event,
               (unsigned long long)entry->node_id, entry->handle_count, entry->dispatch_depth);
    entry->dispatch_depth++;
    // Закрытые элементы не освобождаются до выхода из внешней рассылки. Новые добавляются в голову.
    for (struct NODE_CONN_DIRECT* h = entry->handles; h; h = h->next) {
        if (event != NCD_EVENT_UP) ncd_cancel_deliver_up(h, entry->ua);
        if (!h->closed && h->cb && (!entry->closed || event == NCD_EVENT_CLOSED)) h->cb(h, event, h->cb_arg);
    }
    entry->dispatch_depth--;
    ncd_collect(entry);
}

/* ═══════════ ETCP коллбэки (прокидывают в ncd_event_dispatch) ═══════════ */
/*
 * ETCP сообщил что соединение работает: INIT — handshake завершён,
 * UP — линки восстановились после DOWN. Транслируем в NCD_EVENT_UP.
 * TIMEOUT сообщает об истечении срока первой попытки, но удерживаемый handle
 * продолжает следить за транспортом (в том числе при REVERSE). FIN_WAIT подавляет UP.
 * Если INIT пришёл а линков нет — UP не доставляем: ncd_up_cb сам
 * доставит когда линки поднимутся. Так гарантируем что пользователь
 * получает UP только когда conn реально готов отправлять данные.
 */

static void ncd_init_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
    struct ncd_entry* entry = (struct ncd_entry*)arg;
    if (!entry || conn->fin_wait) return;
    if (conn->links_up > 0) ncd_event_dispatch(entry, NCD_EVENT_UP);
}
static void ncd_up_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
    struct ncd_entry* entry = (struct ncd_entry*)arg;
    if (!entry || conn->fin_wait) return;
    ncd_event_dispatch(entry, NCD_EVENT_UP);
}
// Каждый отложенный запрос удерживает ETCP до завершения callback, в том числе при повторном close.
static void ncd_deferred_close_conn(void* arg) {
    struct ETCP_CONN* conn = arg;
    etcp_connection_close(conn);
    etcp_conn_ref_free(conn);
}
static void ncd_schedule_close(struct ETCP_CONN* conn) {
    if (!conn || conn->close_requested || conn->state == 2) return;
    if (etcp_conn_ref_take(conn) != 0) return;
    conn->close_requested = 1;
    uasync_call_soon(conn->instance->ua, conn, ncd_deferred_close_conn);
}

static void ncd_collect(struct ncd_entry* entry) {
    if (entry->dispatch_depth) return;
    struct NODE_CONN_DIRECT** pp = &entry->handles;
    while (*pp) {
        struct NODE_CONN_DIRECT* h = *pp;
        if (h->closed) { *pp = h->next; u_free(h); } else pp = &h->next;
    }
    if (entry->closed && !entry->handle_count) {
        DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] free node=%016llx", (unsigned long long)entry->node_id);
        u_free(entry);
    }
}

// Сначала исключение из реестра и отсоединение транспорта, затем уведомление владельцев.
static void ncd_finish(struct ncd_entry* entry, int close_conn) {
    if (entry->closed) return;
    entry->closed = 1;
    ncd_registry_remove(entry->inst, entry);
    if (entry->node_ref) {
        topo_node_registry_unref(entry->inst->topo_groups, entry->node_id); entry->node_ref = 0;
    }
    if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
    if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
    struct ETCP_CONN* conn = entry->conn; entry->conn = NULL;
    if (conn) {
        conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
        etcp_conn_remove_cbk(conn, ncd_init_cb, entry);
        etcp_conn_remove_cbk(conn, ncd_up_cb, entry);
        etcp_conn_remove_cbk(conn, ncd_down_cb, entry);
        etcp_conn_remove_cbk(conn, ncd_delete_cb, entry);
        if (close_conn) ncd_schedule_close(conn);
        etcp_conn_ref_free(conn);
    }
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] detached node=%016llx handles=%d", (unsigned long long)entry->node_id, entry->handle_count);
    ncd_event_dispatch(entry, NCD_EVENT_CLOSED); // collect может освободить entry
}
static void ncd_delete_cb(struct ETCP_CONN* conn, int event, void* arg) {
    (void)conn; (void)event; ncd_finish(arg, 0);
}

/*
 * ETCP сообщил что все линки упали.
 * Если conn в fin_wait и handle'ов нет — peer подтвердил закрытие,
 * делаем немедленную очистку. Иначе транслируем NCD_EVENT_DOWN потребителям.
 */
static void ncd_down_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
    struct ncd_entry* entry = (struct ncd_entry*)arg;
    if (!entry) return;
    if (conn->fin_wait && entry->handle_count <= 0) {
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] DOWN during fin_wait, cleaning up node=0x%016llx", (unsigned long long)entry->node_id);
        conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
        if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
        ncd_finish(entry, 1);
        return;
    }
    ncd_event_dispatch(entry, NCD_EVENT_DOWN);
}

/* ═══════════ таймер подключения ═══════════ */
/*
 * Таймер первого подключения истёк — соединение не установилось за etcp_connect_timeout_tb.
 * Транслирует NCD_EVENT_TIMEOUT один раз. Владелец может закрыть handle или
 * оставить его для позднего UP; окончательное завершение обозначается CLOSED.
 */

static void ncd_connect_timeout_cb(void* arg) {
    struct ncd_entry* entry = (struct ncd_entry*)arg;
    if (!entry || entry->timed_out) return;
    entry->connect_timer = NULL;
    DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] connect TIMEOUT: node=0x%016llx now_tb=%llu handles=%d",
               (unsigned long long)entry->node_id, (unsigned long long)get_time_tb(), entry->handle_count);
    nat_detection_connection_failed(entry->inst->nat_det, entry->conn, 0);
    ncd_event_dispatch(entry, NCD_EVENT_TIMEOUT);
}

/* ═══════════ асинхронная доставка UP ═══════════ */
/*
 * Доставляет NCD_EVENT_UP одному конкретному handle'у (не всем).
 * Используется когда handle добавляется к уже работающему conn: доставка отложенная
 * (uasync_call_soon), чтобы вызывающий успел сохранить указатель handle
 * до вызова callback'а. Синхронная доставка привела бы к гонке.
 */

static void ncd_deliver_up_cb(void* arg) {
    struct NODE_CONN_DIRECT* h = (struct NODE_CONN_DIRECT*)arg;
    h->deliver_up_token = NULL;   /* токен consumed — close больше не отменит освобождённый узел */
    if (!h->closed && !h->entry->closed && h->entry->up && !h->entry->timed_out && h->cb)
        h->cb(h, NCD_EVENT_UP, h->cb_arg);
}

/* Планирует отложенную доставку UP и сохраняет токен для возможной отмены при close. */
static void ncd_schedule_deliver_up(struct UASYNC* ua, struct NODE_CONN_DIRECT* h) {
    ncd_cancel_deliver_up(h, ua);
    h->deliver_up_token = uasync_call_soon(ua, h, ncd_deliver_up_cb);
}

/* Отменяет pending-доставку UP (если ещё не сработала). Без этого h освобождается,
 * а отложенный ncd_deliver_up_cb читает freed-память — UAF. */
static void ncd_cancel_deliver_up(struct NODE_CONN_DIRECT* h, struct UASYNC* ua) {
    if (h->deliver_up_token) { uasync_call_soon_cancel(ua, h->deliver_up_token); h->deliver_up_token = NULL; }
}

/* ═══════════ FIN_WAIT ═══════════ */
/*
 * fin_wait — состояние ожидания подтверждения закрытия от пира.
 * Вызывающая сторона: мы отправили CLOSE и ждём ответа (KEEP_ALIVE или DOWN).
 * Обычное ожидание — NCD_FIN_WAIT_TIMEOUT_TB (5 сек); объявленный сон продлевает
 * его до RX deadline живого линка. По истечении закрываем транспорт.
 */

/*
 * ETCP отменяет наш fin_wait — peer переподключился и прислал новый INIT
 * (или другая причина отмены на уровне ETCP). Отменяем локальный таймер,
 * соединение продолжает работать.
 */
static void ncd_fin_wait_cancelled(struct ETCP_CONN* conn, void* arg) {
    struct ncd_entry* entry = (struct ncd_entry*)arg;
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait cancelled node=0x%016llx", (unsigned long long)entry->node_id);
    if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
}

/*
 * Таймаут fin_wait: peer не ответил на CLOSE в отведённый срок с учётом сна.
 * Форсированно закрываем conn и освобождаем entry — соединение разорвано
 * без подтверждения от пира.
 */
static void ncd_fin_wait_timeout_cb(void* arg) {
    struct ncd_entry* entry = (struct ncd_entry*)arg;
    if (!entry || !entry->conn) return;
    entry->fin_wait_timer = NULL;
    if (!entry->conn->fin_wait) return;
    DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] fin_wait timeout, force close node=0x%016llx", (unsigned long long)entry->node_id);
    ncd_finish(entry, 1);
}

/* ═══════════ приём CLOSE / KEEP_ALIVE ═══════════ */
/*
 * Обрабатывает управляющие сообщения от пира по протоколу graceful shutdown.
 *
 * CLOSE (пир хочет закрыть conn — у него закончились handle'ы):
 *   - Если у нас ещё есть handle'ы — отказываем: шлём KEEP_ALIVE, conn живёт.
 *   - Если handle'ов нет — соглашаемся, закрываем conn.
 *
 * KEEP_ALIVE (пир просит не закрывать conn — у него ещё есть handle'ы):
 *   - Если мы в fin_wait — отменяем его, conn продолжает работать.
 *   - Отсутствие наших handle'ов не отменяет потребность пира: FIN_WAIT снимается полностью.
 */

static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) {
    if (!conn || !e || e->len < NCD_CONTROL_MSG_SIZE) {
        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] malformed control conn=%p size=%u", (void*)conn, e ? e->len : 0);
        if (e) { queue_dgram_free(e); queue_entry_free(e); }
        return;
    }
    struct ncd_control_msg* msg = (struct ncd_control_msg*)e->dgram;
    uint64_t sender_id = msg->node_id;
    struct ncd_entry* entry = ncd_registry_find(conn->instance, sender_id);
    if (sender_id != conn->peer_node_id || (entry && entry->conn != conn)) {
        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] control from wrong transport node=%016llx", (unsigned long long)sender_id);
        queue_dgram_free(e); queue_entry_free(e); return;
    }

    switch (msg->subcmd) {
    case NCD_SUBCMD_CLOSE:
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] recv CLOSE from 0x%016llx", (unsigned long long)sender_id);
        if (!entry) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] no entry for CLOSE sender 0x%016llx, deferred close conn", (unsigned long long)sender_id);
            ncd_schedule_close(conn);
            break;
        }
        if (entry->handle_count > 0) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] refusing CLOSE, still have %d handles for 0x%016llx",
                       entry->handle_count, (unsigned long long)sender_id);
            {   struct ncd_control_msg resp = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, conn->instance->node_id };
                struct ll_entry* qe = ll_alloc_lldgram(sizeof(resp));
                if (!qe) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE allocation failed");
                else {
                    memcpy(qe->dgram, &resp, sizeof(resp)); qe->len = sizeof(resp);
                    if (etcp_send(conn, qe) != 0) {
                        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE send failed node=%016llx", (unsigned long long)sender_id);
                        queue_dgram_free(qe); queue_entry_free(qe);
                    }
                }
            }
        } else {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] deferred close entry for CLOSE from 0x%016llx", (unsigned long long)sender_id);
            ncd_finish(entry, 1);
        }
        break;
    case NCD_SUBCMD_KEEP_ALIVE:
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] recv KEEP_ALIVE from 0x%016llx", (unsigned long long)sender_id);
        if (entry && entry->conn->fin_wait) {
            entry->conn->fin_wait = 0; entry->conn->fin_wait_clear_cb = NULL; entry->conn->fin_wait_clear_arg = NULL;
            if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE: fin_wait cleared, conn stays alive handles=%d", entry->handle_count);
        }
        break;
    default:
        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] unknown control subcmd=%u node=%016llx", msg->subcmd, (unsigned long long)sender_id);
        break;
    }
    queue_dgram_free(e); queue_entry_free(e);
}

/* ═══════════ инициализация глобального обработчика ═══════════ */
/*
 * Однократная привязка обработчика NCD control-сообщений к ETCP (etcp_bind).
 * Вызывается при запуске сети и node_conn_direct_open, повторные вызовы безвредны
 * (защита через inst->ncd_control_bound).
 */

int node_conn_direct_init(struct UTUN_INSTANCE* inst) {
    if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] init without instance"); return -1; }
    if (inst->ncd_control_bound) return 0;
    if (etcp_bind(inst, ETCP_RT_ID_NCD_CONTROL, ncd_recv_control_handler) < 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot bind CLOSE/KEEP_ALIVE handler"); return -1;
    }
    inst->ncd_control_bound = 1;
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] bound ETCP_RT_ID_NCD_CONTROL handler");
    return 0;
}

/* ═══════════ оживление «зависшего» (timed_out) entry ═══════════ */
/*
 * Entry не поднялся за connect-таймер (timed_out) и «завис»: conn живёт в
 * реестре, но линки мертвы. При новом open переиспользуем conn и обновляем
 * линки: убираем старые, добавляем новые по свежим адресам, заново заводим
 * таймер подключения. Если линк всё же успел подняться — просто отдаём UP.
 */
static void ncd_revive_entry(struct ncd_entry* entry, struct UTUN_INSTANCE* inst,
                             struct TOPO_NODE* ni, struct ETCP_SOCKET* specific_sock) {
    if (!entry) return;
    if (entry->conn && entry->conn->links_up > 0) {
        entry->timed_out = 0;
        entry->up = 1;
        for (struct NODE_CONN_DIRECT* h = entry->handles; h; h = h->next)
            if (!h->closed) ncd_schedule_deliver_up(entry->ua, h);
        return;
    }
    struct ETCP_LINK* l = entry->conn ? entry->conn->links : NULL;
    while (l) { struct ETCP_LINK* nx = l->next; etcp_link_close(l); l = nx; }
    if (!ni) {
        /* оставляем timed_out=1 — следующий open снова попробует revive со свежим lookup */
        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] revive: node=0x%016llx no node info — cannot recreate links",
                   (unsigned long long)entry->node_id);
        return;
    }
    entry->timed_out = 0;
    ncd_create_links(entry, ni, specific_sock, NULL);
    entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb,
                                               entry, ncd_connect_timeout_cb, "ncd_connect");
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] revive: node=0x%016llx links recreated, connect timer restarted",
               (unsigned long long)entry->node_id);
}

/* ═══════════ API ═══════════ */

/*
 * Открыть handle для связи с удалённым узлом.
 *
 * Три сценария (прозрачно для вызывающего):
 *   1. Узел уже в NCD-реестре — другой модуль уже открыл соединение.
 *      Добавляем ещё один handle. Если conn работает — сразу шлём UP.
 *   2. ETCP-соединение существует (входящее / etcp_connect), но NCD о нём не знает.
 *      Оборачиваем в ncd_entry, подписываемся на события.
 *   3. Ничего нет — загружаем адреса/pupkey узла, создаём новый ETCP_CONN,
 *      инициализируем шифрование, создаём линки, запускаем таймер подключения.
 *
 * Возвращает NCD_NEW (новое), NCD_REUSED (переиспользовано) или NCD_ERR.
 * specific_sock=NULL — авто-подбор всех локальных сокетов.
 */
int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
                          ncd_callback cb, void* cb_arg,
                          struct NODE_CONN_DIRECT** out_handle,
                          struct ETCP_SOCKET* specific_sock) {
    if (!inst || !out_handle) return NCD_ERR;
    *out_handle = NULL;
    if (node_conn_direct_init(inst) < 0) return NCD_ERR;

    /* 1. Ищем в реестре — conn уже есть */
    struct ncd_entry* entry = ncd_registry_find(inst, node_id);
    if (entry) {
        /* снять fin_wait если был */
        if (entry->conn && entry->conn->fin_wait) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED clearing fin_wait node=0x%016llx", (unsigned long long)node_id);
            entry->conn->fin_wait = 0; entry->conn->fin_wait_clear_cb = NULL; entry->conn->fin_wait_clear_arg = NULL;
            if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
        }
        struct NODE_CONN_DIRECT* h = u_calloc(1, sizeof(*h));
        if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] alloc handle failed node=0x%016llx", (unsigned long long)node_id); return NCD_ERR; }
        h->entry = entry; h->cb = cb; h->cb_arg = cb_arg;
        h->next = entry->handles; entry->handles = h;
        entry->handle_count++;
        *out_handle = h;
        if (entry->conn && entry->conn->state == 1 && entry->up) {
            ncd_schedule_deliver_up(inst->ua, h);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED (ready) node=0x%016llx handles=%d", (unsigned long long)node_id, entry->handle_count);
        } else if (entry->timed_out) {
            struct TOPO_NODE* ni = ncd_lookup_node(inst, node_id);
            ncd_revive_entry(entry, inst, ni, specific_sock);
            if (ni) topo_node_registry_unref(inst->topo_groups, ni->node_id);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED (revived) node=0x%016llx handles=%d", (unsigned long long)node_id, entry->handle_count);
        } else {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED (pending) node=0x%016llx handles=%d", (unsigned long long)node_id, entry->handle_count);
        }
        return NCD_REUSED;
    }

    /* 2. Ищем conn через instance_find_conn (входящее / созданное etcp_connect) */
    struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
    if (conn && conn->state != 2 && !conn->close_requested) {
        /* снять fin_wait если был */
        if (conn->fin_wait) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED clearing fin_wait (existing conn) node=0x%016llx", (unsigned long long)node_id);
            conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
        }
        entry = u_calloc(1, sizeof(*entry));
        if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] alloc entry failed node=0x%016llx", (unsigned long long)node_id); return NCD_ERR; }
        entry->node_id = node_id; entry->conn = conn; entry->ua = inst->ua;
        entry->up = (uint8_t)(conn->links_up ? 1 : 0);
        if (ncd_registry_add(inst, entry) < 0) { u_free(entry); return NCD_ERR; }

        struct NODE_CONN_DIRECT* h = u_calloc(1, sizeof(*h));
        if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] alloc handle failed node=0x%016llx", (unsigned long long)node_id);
                  ncd_finish(entry, 0); return NCD_ERR; }
        h->entry = entry; h->cb = cb; h->cb_arg = cb_arg;
        h->next = entry->handles; entry->handles = h;
        entry->handle_count = 1;
        *out_handle = h;

        etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
        etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
        etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);

        if (conn->state == 1 && entry->up) {
            ncd_schedule_deliver_up(inst->ua, h);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED new-entry (ready) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
        } else {
            {   struct TOPO_NODE* ni = ncd_lookup_node(inst, node_id);
                if (ni) { ncd_create_links(entry, ni, specific_sock, NULL); topo_node_registry_unref(inst->topo_groups, ni->node_id); }
            }
            DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "[ncd] connect timer SET: node=0x%016llx value_tb=%u now_tb=%llu path=REUSED_entry_pending",
                       (unsigned long long)node_id, inst->etcp_connect_timeout_tb, (unsigned long long)get_time_tb());
            entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb, entry, ncd_connect_timeout_cb, "ncd_connect");
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED new-entry (pending) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
        }
        return NCD_REUSED;
    }

    /* 3. Новое подключение */
    struct TOPO_NODE* ni = ncd_lookup_node(inst, node_id);
    if (!ni) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] node not found id=0x%016llx", (unsigned long long)node_id);
        return NCD_ERR;
    }

    {   char conn_name[MAX_CONN_NAME_LEN];
        if (ni->node_name && ni->node_name[0]) strncpy(conn_name, ni->node_name, MAX_CONN_NAME_LEN - 1);
        else snprintf(conn_name, sizeof(conn_name), "n_%016llx", (unsigned long long)node_id);
        conn = etcp_connection_create(inst, conn_name); }

    if (!conn) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] etcp_connection_create failed node=0x%016llx", (unsigned long long)node_id);
        topo_node_registry_unref(inst->topo_groups, ni->node_id);
        return NCD_ERR;
    }
    conn->peer_node_id = node_id;
    etcp_update_log_name(conn);

    if (sc_init_ctx(&conn->crypto_ctx, &inst->my_keys) != SC_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] sc_init_ctx failed node=0x%016llx", (unsigned long long)node_id);
        etcp_connection_close(conn); topo_node_registry_unref(inst->topo_groups, ni->node_id); return NCD_ERR;
    }
    if (sc_set_peer_public_key(&conn->crypto_ctx, ni->public_key, 0) != SC_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] sc_set_peer_public_key failed node=0x%016llx", (unsigned long long)node_id);
        etcp_connection_close(conn); topo_node_registry_unref(inst->topo_groups, ni->node_id); return NCD_ERR;
    }

    /* Подключаем conn в inst->connections с индексом по node_id */
    if (conn->conn_queue && conn->conn_queue_entry) {
        queue_remove_data(conn->conn_queue, conn->conn_queue_entry);
        queue_entry_free(conn->conn_queue_entry);
    }
    {   struct ll_entry* qe = queue_entry_new(sizeof(struct conn_queue_entry));
        if (qe) { struct conn_queue_entry* ce = (struct conn_queue_entry*)qe->data;
            ce->peer_node_id = node_id; ce->conn = conn;
            conn->conn_queue_entry = qe; conn->conn_queue = inst->connections;
            queue_data_put_with_index(inst->connections, qe); }
    }

    entry = u_calloc(1, sizeof(*entry));
    if (!entry) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] alloc entry failed node=0x%016llx", (unsigned long long)node_id);
        etcp_connection_close(conn); topo_node_registry_unref(inst->topo_groups, ni->node_id); return NCD_ERR;
    }
    entry->node_id = node_id; entry->conn = conn; entry->ua = inst->ua; entry->up = 0;
    if (ncd_registry_add(inst, entry) < 0) {
        u_free(entry); etcp_connection_close(conn); topo_node_registry_unref(inst->topo_groups, ni->node_id); return NCD_ERR;
    }

    struct NODE_CONN_DIRECT* h = u_calloc(1, sizeof(*h));
    if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] alloc handle failed node=0x%016llx", (unsigned long long)node_id);
              ncd_finish(entry, 0); etcp_connection_close(conn); topo_node_registry_unref(inst->topo_groups, ni->node_id); return NCD_ERR; }
    h->entry = entry; h->cb = cb; h->cb_arg = cb_arg;
    h->next = entry->handles; entry->handles = h;
    entry->handle_count = 1;
    *out_handle = h;

    etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
    etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
    etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);

    int link_count = ncd_create_links(entry, ni, specific_sock, NULL);

    if (link_count == 0)
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] no links created (UDP only, TCP added by conn_mgr) node=0x%016llx", (unsigned long long)node_id);

    DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "[ncd] connect timer SET: node=0x%016llx value_tb=%u now_tb=%llu path=NEW_node_conn_direct_open",
               (unsigned long long)node_id, inst->etcp_connect_timeout_tb, (unsigned long long)get_time_tb());
    entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb, entry, ncd_connect_timeout_cb, "ncd_connect");
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open NEW node=0x%016llx conn=%p links=%d handles=%d",
               (unsigned long long)node_id, (void*)conn, link_count, entry->handle_count);
    topo_node_registry_unref(inst->topo_groups, ni->node_id);
    return NCD_NEW;
}

/*
 * То же что open, но TOPO_NODE (адреса + pubkey) уже загружен вызывающим.
 * Экономит поиск узла (ncd_lookup_node) — полезно когда адреса/pupkey
 * известны заранее (например из BGP-анонса). ni не владеем — можно
 * передать временную структуру, копия не делается.
 */
static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
                               ncd_callback cb, void* cb_arg,
                               struct NODE_CONN_DIRECT** out_handle,
                               struct TOPO_NODE* ni,
                               struct ETCP_SOCKET* specific_sock) {
    if (!inst || !out_handle || !ni) return NCD_ERR;
    *out_handle = NULL;
    if (node_conn_direct_init(inst) < 0) return NCD_ERR;

    /* 1. Ищем в реестре — conn уже есть */
    { struct ncd_entry* entry = ncd_registry_find(inst, node_id);
    if (entry) {
        if (entry->conn && entry->conn->fin_wait) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED clearing fin_wait node=0x%016llx", (unsigned long long)node_id);
            entry->conn->fin_wait = 0; entry->conn->fin_wait_clear_cb = NULL; entry->conn->fin_wait_clear_arg = NULL;
            if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
        }
        struct NODE_CONN_DIRECT* h = u_calloc(1, sizeof(*h));
        if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node alloc handle failed node=0x%016llx", (unsigned long long)node_id); return NCD_ERR; }
        h->entry = entry; h->cb = cb; h->cb_arg = cb_arg;
        h->next = entry->handles; entry->handles = h;
        entry->handle_count++;
        *out_handle = h;
        if (entry->conn && entry->conn->state == 1 && entry->up) {
            ncd_create_links(entry, ni, specific_sock, NULL);
            ncd_schedule_deliver_up(inst->ua, h);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED (ready) node=0x%016llx handles=%d", (unsigned long long)node_id, entry->handle_count);
        } else if (entry->timed_out) {
            ncd_revive_entry(entry, inst, ni, specific_sock);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED (revived) node=0x%016llx handles=%d", (unsigned long long)node_id, entry->handle_count);
        } else {
            ncd_create_links(entry, ni, specific_sock, NULL);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED (pending) node=0x%016llx handles=%d", (unsigned long long)node_id, entry->handle_count);
        }
        return NCD_REUSED;
    }}

    /* 2. Ищем conn через instance_find_conn */
    { struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
    if (conn && conn->state != 2 && !conn->close_requested) {
        if (conn->fin_wait) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED clearing fin_wait (existing conn) node=0x%016llx", (unsigned long long)node_id);
            conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
        }
        struct ncd_entry* entry = u_calloc(1, sizeof(*entry));
        if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node alloc entry failed node=0x%016llx", (unsigned long long)node_id); return NCD_ERR; }
        entry->node_id = node_id; entry->conn = conn; entry->ua = inst->ua;
        entry->up = (uint8_t)(conn->links_up ? 1 : 0);
        if (ncd_registry_add(inst, entry) < 0) { u_free(entry); return NCD_ERR; }

        struct NODE_CONN_DIRECT* h = u_calloc(1, sizeof(*h));
        if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node alloc handle failed node=0x%016llx", (unsigned long long)node_id);
                  ncd_finish(entry, 0); return NCD_ERR; }
        h->entry = entry; h->cb = cb; h->cb_arg = cb_arg;
        h->next = entry->handles; entry->handles = h;
        entry->handle_count = 1;
        *out_handle = h;

        etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
        etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
        etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);

        if (conn->state == 1 && entry->up) {
            ncd_create_links(entry, ni, specific_sock, NULL);
            ncd_schedule_deliver_up(inst->ua, h);
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED new-entry (ready) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
        } else {
            ncd_create_links(entry, ni, specific_sock, NULL);
            DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "[ncd] connect timer SET: node=0x%016llx value_tb=%u now_tb=%llu path=REUSED_conn_pending",
                       (unsigned long long)node_id, inst->etcp_connect_timeout_tb, (unsigned long long)get_time_tb());
            entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb, entry, ncd_connect_timeout_cb, "ncd_connect_node");
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node REUSED new-entry (pending) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
        }
        return NCD_REUSED;
    }}

    /* 3. Новое подключение — используем переданный ni (временный, не владеем) */
    { char conn_name[MAX_CONN_NAME_LEN];
    if (ni->node_name && ni->node_name[0]) strncpy(conn_name, ni->node_name, MAX_CONN_NAME_LEN - 1);
    else snprintf(conn_name, sizeof(conn_name), "n_%016llx", (unsigned long long)node_id);
    struct ETCP_CONN* conn = etcp_connection_create(inst, conn_name);
    if (!conn) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node etcp_connection_create failed node=0x%016llx", (unsigned long long)node_id);
        return NCD_ERR;
    }
    conn->peer_node_id = node_id;
    etcp_update_log_name(conn);

    if (sc_init_ctx(&conn->crypto_ctx, &inst->my_keys) != SC_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node sc_init_ctx failed node=0x%016llx", (unsigned long long)node_id);
        etcp_connection_close(conn); return NCD_ERR;
    }
    if (sc_set_peer_public_key(&conn->crypto_ctx, ni->public_key, 0) != SC_OK) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node sc_set_peer_public_key failed node=0x%016llx", (unsigned long long)node_id);
        etcp_connection_close(conn); return NCD_ERR;
    }

    if (conn->conn_queue && conn->conn_queue_entry) {
        queue_remove_data(conn->conn_queue, conn->conn_queue_entry);
        queue_entry_free(conn->conn_queue_entry);
    }
    {   struct ll_entry* qe = queue_entry_new(sizeof(struct conn_queue_entry));
        if (qe) { struct conn_queue_entry* ce = (struct conn_queue_entry*)qe->data;
            ce->peer_node_id = node_id; ce->conn = conn;
            conn->conn_queue_entry = qe; conn->conn_queue = inst->connections;
            queue_data_put_with_index(inst->connections, qe); }
    }

    struct ncd_entry* entry = u_calloc(1, sizeof(*entry));
    if (!entry) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node alloc entry failed node=0x%016llx", (unsigned long long)node_id);
        etcp_connection_close(conn); return NCD_ERR;
    }
    entry->node_id = node_id; entry->conn = conn; entry->ua = inst->ua; entry->up = 0;
    if (ncd_registry_add(inst, entry) < 0) { u_free(entry); etcp_connection_close(conn); return NCD_ERR; }

    struct NODE_CONN_DIRECT* h = u_calloc(1, sizeof(*h));
    if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] open_node alloc handle failed node=0x%016llx", (unsigned long long)node_id);
              ncd_finish(entry, 0); etcp_connection_close(conn); return NCD_ERR; }
    h->entry = entry; h->cb = cb; h->cb_arg = cb_arg;
    h->next = entry->handles; entry->handles = h;
    entry->handle_count = 1;
    *out_handle = h;

    etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
    etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
    etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);

    int link_count = ncd_create_links(entry, ni, specific_sock, NULL);

    if (link_count == 0)
        DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] open_node: no new compatible links node=0x%016llx", (unsigned long long)node_id);

    DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "[ncd] connect timer SET: node=0x%016llx value_tb=%u now_tb=%llu path=NEW_node_conn_direct_open_node",
               (unsigned long long)node_id, inst->etcp_connect_timeout_tb, (unsigned long long)get_time_tb());
    entry->connect_timer = uasync_set_timeout(inst->ua, (int)inst->etcp_connect_timeout_tb, entry, ncd_connect_timeout_cb, "ncd_connect_node");
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open_node NEW node=0x%016llx conn=%p links=%d handles=%d",
               (unsigned long long)node_id, (void*)conn, link_count, entry->handle_count);
    return NCD_NEW;
    }
}

int node_conn_direct_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
                               ncd_callback cb, void* cb_arg, struct NODE_CONN_DIRECT** out_handle,
                               struct TOPO_NODE* ni, struct ETCP_SOCKET* specific_sock) {
    if (out_handle) *out_handle = NULL;
    if (!inst || !out_handle || !ni || node_id != ni->node_id ||
        node_id != sc_derive_node_id_from_pubkey(ni->public_key) || (ni->timestamp && topo_node_verify(ni) < 0)) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] invalid node passed to open"); return NCD_ERR;
    }
    struct TOPO_NODE* known = ncd_lookup_node(inst, node_id);
    if (ni->timestamp && (!known || ni->timestamp > known->timestamp)) {
        uint8_t wire[TOPO_NODE_WIRE_MAX_SIZE];
        struct TOPO_GROUP_NODE nq = {0};
        struct TOPO_GROUP group = { .instance = inst };
        struct TOPO_NODE* copy = NULL;
        int len = topo_node_serialize(ni, &nq, 0, 0, wire, sizeof(wire), 0);
        if (len < 0 || !inst->topo_groups ||
            topo_node_deserialize(&group, wire, (size_t)len, &copy, NULL, NULL, NULL, NULL) < 0) {
            DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot copy node snapshot");
            if (known) topo_node_registry_unref(inst->topo_groups, node_id);
            return NCD_ERR;
        }
        struct TOPO_NODE* stored = topo_node_registry_store(inst->topo_groups, copy);
        if (known) topo_node_registry_unref(inst->topo_groups, node_id);
        known = stored;
        if (!stored) { topo_node_destroy(inst->topo_groups, copy); return NCD_ERR; }
        if (inst->topo_sqlite_db && topo_node_sqlite_snapshot_put(inst->topo_sqlite_db, stored) < 0)
            DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot persist received node snapshot");
    }
    if (known && ni->timestamp && known->timestamp == ni->timestamp &&
        memcmp(known->x25519_self_sig, ni->x25519_self_sig, 64)) {
        DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] conflicting node snapshot node=%016llx", (unsigned long long)node_id);
        topo_node_registry_unref(inst->topo_groups, node_id); return NCD_ERR;
    }
    struct TOPO_NODE* selected = known && known->timestamp ? known : ni;
    int result = ncd_open_node(inst, node_id, cb, cb_arg, out_handle, selected, specific_sock);
    if (known) topo_node_registry_unref(inst->topo_groups, node_id);
    return result;
}

/*
 * Закрыть handle (graceful shutdown).
 *
 * Если остались другие handle'ы — просто удаляется из списка, conn продолжает работу.
 *
 * Если это был последний handle (handle_count стал 0):
 *   - Запускается протокол graceful shutdown: пиру отправляется CLOSE, conn
 *     переводится в fin_wait, ставится таймер на 5 сек либо до пробуждения пира.
 *   - Если пир ответит KEEP_ALIVE (у него ещё есть handle'ы) — conn останется жив,
 *     просто без наших handle'ов.
 *   - Если пир не ответит — conn закрывается форсированно по таймауту.
 */
static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) {
    if (!h || h->closed) return;
    struct ncd_entry* entry = h->entry;
    ncd_cancel_deliver_up(h, entry->ua);
    h->closed = 1; h->cb = NULL; h->cb_arg = NULL;
    entry->handle_count--;
    DEBUG_DEBUG(DEBUG_CATEGORY_NCD, "[ncd] handle close node=%016llx remaining=%d force=%d",
                (unsigned long long)entry->node_id, entry->handle_count, force);
    if (entry->handle_count || entry->closed) { ncd_collect(entry); return; }
    struct ETCP_CONN* conn = entry->conn;
    if (force || !conn || conn->close_requested || conn->state == 2 || entry->timed_out || (!conn->state && !conn->links_up)) {
        ncd_finish(entry, 1); return;
    }
    if (!conn->links_up) {
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] last handle released while DOWN, retaining until ETCP timeout node=%016llx",
                   (unsigned long long)entry->node_id);
        ncd_finish(entry, 0);
        if (etcp_conn_schedule_down_close(conn) < 0) etcp_connection_close(conn);
        return;
    }
    if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
    conn->fin_wait = 1; conn->fin_wait_clear_cb = ncd_fin_wait_cancelled; conn->fin_wait_clear_arg = entry;
    struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_CLOSE, entry->inst->node_id };
    struct ll_entry* e = ll_alloc_lldgram(sizeof(msg));
    if (!e) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] CLOSE allocation failed");
    else {
        memcpy(e->dgram, &msg, sizeof(msg)); e->len = sizeof(msg);
        if (etcp_send(conn,e) != 0) {
            DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] CLOSE send failed"); queue_dgram_free(e); queue_entry_free(e);
        }
    }
    uint64_t now = get_time_tb(), wait_tb = NCD_FIN_WAIT_TIMEOUT_TB;
    for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
        if (!link->link_status || (!link->ka_my_sleeping && !link->ka_peer_sleeping)) continue;
        if (link->rx_deadline_tb > now && link->rx_deadline_tb - now > wait_tb) wait_tb = link->rx_deadline_tb - now;
    }
    int delay = wait_tb > INT_MAX ? INT_MAX : (int)wait_tb;
    entry->fin_wait_timer = uasync_set_timeout(entry->ua, delay, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait");
    if (!entry->fin_wait_timer) {
        DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot schedule FIN_WAIT node=%016llx", (unsigned long long)entry->node_id);
        ncd_finish(entry, 1); return;
    }
    DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait node=%016llx timeout_ms=%d", (unsigned long long)entry->node_id, delay / 10);
    ncd_collect(entry);
}
void node_conn_direct_close(struct NODE_CONN_DIRECT* h) { ncd_handle_close(h, 0); }
void node_conn_direct_force_close(struct NODE_CONN_DIRECT* h) { ncd_handle_close(h, 1); }

/*
 * Сменить или сбросить (cb=NULL) callback на уже открытом handle.
 * Не влияет на refcounting и состояние conn.
 */
void node_conn_direct_set_callback(struct NODE_CONN_DIRECT* h, ncd_callback cb, void* cb_arg) {
    if (!h || !h->entry) return;
    h->cb = cb;
    h->cb_arg = cb_arg;
}

void node_conn_direct_transfer(struct NODE_CONN_DIRECT* h, ncd_callback new_cb, void* new_cb_arg) {
    if (!h || !h->entry) return;
    h->cb = new_cb;
    h->cb_arg = new_cb_arg;
}

/*
 * Прямой доступ к ETCP_CONN из handle.
 * Нужен для отправки данных (etcp_send), проверки статуса, и т.д.
 */
uint64_t node_conn_direct_node_id(const struct NODE_CONN_DIRECT* h) {
    return h && h->entry ? h->entry->node_id : 0;
}

struct ETCP_CONN* node_conn_direct_get_conn(struct NODE_CONN_DIRECT* h) {
    if (!h || h->closed || !h->entry || h->entry->closed) return NULL;
    return h->entry->conn;
}

/* ═══════════ Управление линками на конкретном сокете (для auto_socket) ═══════════ */

/* ncd_add_udp_link_random / ncd_add_tcp_link_random — собирают совместимые адреса пира,
 * случайно выбирают один, проверяют что линк на (sock, addr, proto) ещё не существует,
 * и создают новый. */
static int ncd_add_udp_link_random(struct ETCP_CONN* conn, struct TOPO_NODE* ni, struct ETCP_SOCKET* sock) {
    int family = sock->local_addr.ss_family;
    if (family == AF_INET) {
        struct TOPO_ADDR4* compat[64]; int count = 0;
        for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
            if (!a->port || !(a->protocol & TOPO_PROTO_UDP)) continue;
            int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j]) { zero = 0; break; }
            if (zero) continue;
            if (count < 64) compat[count++] = (struct TOPO_ADDR4*)a;
        }
        if (count == 0) return 0;
        uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
        struct TOPO_ADDR4* a = compat[r % count];
        struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET;
        memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port);
        struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin));
        if (etcp_link_find_by_addr(sock, &sa, 0)) return 0;
        if (etcp_link_new(conn, sock, &sa, 0)) return 1;
        return 0;
    }
    if (family == AF_INET6) {
        struct TOPO_ADDR6* compat[64]; int count = 0;
        for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
            if (!a->port || !(a->protocol & TOPO_PROTO_UDP)) continue;
            int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j]) { zero = 0; break; }
            if (zero) continue;
            if (count < 64) compat[count++] = (struct TOPO_ADDR6*)a;
        }
        if (count == 0) return 0;
        uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
        struct TOPO_ADDR6* a = compat[r % count];
        struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6;
        memcpy(&sin6.sin6_addr, a->addr, 16); sin6.sin6_port = htons(a->port);
        if ((a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) sin6.sin6_scope_id = sock->netif_index;
        struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
        if (etcp_link_find_by_addr(sock, &sa, 0)) return 0;
        if (etcp_link_new(conn, sock, &sa, 0)) return 1;
        return 0;
    }
    return 0;
}

static int ncd_add_tcp_link_random(struct ETCP_CONN* conn, struct TOPO_NODE* ni, struct ETCP_SOCKET* sock) {
    int family = sock->local_addr.ss_family;
    if (family == AF_INET) {
        struct TOPO_ADDR4* compat[64]; int count = 0;
        for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
            if (!a->port || !(a->protocol & TOPO_PROTO_TCP)) continue;
            int zero = 1; for (int j = 0; j < 4; j++) if (a->addr[j]) { zero = 0; break; }
            if (zero) continue;
            if (count < 64) compat[count++] = (struct TOPO_ADDR4*)a;
        }
        if (count == 0) return 0;
        uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
        struct TOPO_ADDR4* a = compat[r % count];
        struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET;
        memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port);
        struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin));
        if (etcp_link_find_by_addr(sock, &sa, 1)) return 0;
        struct ETCP_LINK* tlink = etcp_link_new(conn, sock, &sa, 0);
        if (!tlink) return 0;
        tlink->is_tcp = 1;
        topo_node_apply_reality(tlink, topo_node_find_reality_sock(ni, a->socket_id));
        etcp_tcp_link_start_connect(tlink, &sa, a->port);
        return 1;
    }
    if (family == AF_INET6) {
        struct TOPO_ADDR6* compat[64]; int count = 0;
        for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
            if (!a->port || !(a->protocol & TOPO_PROTO_TCP)) continue;
            int zero = 1; for (int j = 0; j < 16; j++) if (a->addr[j]) { zero = 0; break; }
            if (zero) continue;
            if (count < 64) compat[count++] = (struct TOPO_ADDR6*)a;
        }
        if (count == 0) return 0;
        uint32_t r; random_bytes((uint8_t*)&r, sizeof(r));
        struct TOPO_ADDR6* a = compat[r % count];
        struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6;
        memcpy(&sin6.sin6_addr, a->addr, 16); sin6.sin6_port = htons(a->port);
        if ((a->addr[0] == 0xfe && (a->addr[1] & 0xc0) == 0x80)) sin6.sin6_scope_id = sock->netif_index;
        struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6));
        if (etcp_link_find_by_addr(sock, &sa, 1)) return 0;
        struct ETCP_LINK* tlink = etcp_link_new(conn, sock, &sa, 0);
        if (!tlink) return 0;
        tlink->is_tcp = 1;
        topo_node_apply_reality(tlink, topo_node_find_reality_sock(ni, a->socket_id));
        etcp_tcp_link_start_connect(tlink, &sa, a->port);
        return 1;
    }
    return 0;
}

int ncd_add_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) {
    if (!inst || !sock) return 0;
    int added = 0, total_conns = 0, skipped_closed = 0, skipped_no_addr = 0;

    struct ll_entry* qe = inst->connections ? inst->connections->head : NULL;
    while (qe) {
        struct conn_queue_entry* ce = (struct conn_queue_entry*)qe->data;
        struct ETCP_CONN* conn = ce->conn;
        qe = qe->next;
        if (!conn) continue;
        total_conns++;

        if (conn->state == 2) { skipped_closed++; continue; }

        struct TOPO_NODE* ni = ncd_lookup_node(inst, ce->peer_node_id);
        if (!ni) { skipped_no_addr++; continue; }

        int n;
        if (sock->is_tcp) n = ncd_add_tcp_link_random(conn, ni, sock);
        else              n = ncd_add_udp_link_random(conn, ni, sock);
        if (n > 0) {
            DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] add_socket_links sock=%s fd=%d conn=[%s] node=0x%016llx addr=%s is_tcp=%d",
                       sock->name, (int)sock->fd, conn->log_name,
                       (unsigned long long)ce->peer_node_id,
                       sockaddr_storage_to_str(&sock->local_addr).str, sock->is_tcp);
            added += n;
        }
        topo_node_registry_unref(inst->topo_groups, ni->node_id);
    }

    DEBUG_INFO(DEBUG_CATEGORY_NCD,
        "[ncd] add_socket_links done: sock=%s is_tcp=%d total_conns=%d skipped(cl=%d,no_addr=%d) added=%d",
        sock->name, sock->is_tcp, total_conns, skipped_closed, skipped_no_addr, added);
    return added;
}

int ncd_remove_socket_links(struct UTUN_INSTANCE* inst, struct ETCP_SOCKET* sock) {
    if (!inst || !sock) return 0;
    int removed = 0;
    size_t count = 0;
    for (struct ncd_entry* e = inst->ncd_registry; e; e = e->next) count++;
    if (!count) return 0;
    struct ETCP_CONN** conns = u_calloc(count,sizeof(*conns));
    if (!conns) { DEBUG_ERROR(DEBUG_CATEGORY_NCD,"[ncd] remove_socket_links: snapshot allocation failed"); return -1; }
    size_t held = 0;
    for (struct ncd_entry* e = inst->ncd_registry; e; e = e->next)
        if (etcp_conn_ref_take(e->conn) == 0) conns[held++] = e->conn;
    // DOWN callback может удалить текущий и следующий entry. Снимок удерживает только транспорт.
    for (size_t i = 0; i < held; i++) {
        struct ETCP_LINK* link = conns[i]->links;
        while (link) {
            struct ETCP_LINK* next = link->next;
            if (link->conn == sock) { etcp_link_close(link); removed++; }
            link = next;
        }
        etcp_conn_ref_free(conns[i]);
    }
    u_free(conns);
    if (removed > 0) {
        DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] remove_socket_links socket=%s closed=%d", sock->name, removed);
    }
    return removed;
}