// etcp_router.c — Сервисный слой маршрутизации поверх ETCP
// Упрощённый TCP: восстановление порядка (recv_q), дедупликация, без переповторов
// ACK: периодическая отправка rx_seq (не чаще 100ms), idle-таймер для последнего seq
// Inflight контроль через send_q: при переполнении — очередь + retry-таймер, без дропов
#include "etcp_router.h"
#include "etcp.h"
#include "utun_instance.h"
#include "route_bgp.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/ll_queue.h"
#include "../lib/u_async.h"
#include <string.h>

// ====================================================================
// Внутренние forward declarations
// ====================================================================
static void router_ack_timer_cb(void* arg);
static void router_idle_ack_timer_cb(void* arg);
static void router_send_ack(struct ETCP_ROUTER_CONN* rconn);
static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn);
static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn);
static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, const uint8_t* payload, size_t payload_len);
static int  router_send_one(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len);
static int  router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, uint32_t seq_flags);
static void router_send_to(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t svc_id, uint32_t seq_flags);
static void router_drain_send_q(struct ETCP_ROUTER_CONN* rconn);
static void router_send_resume_cb(void* arg);
static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn);
static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn);

// ====================================================================
// Управление ROUTER_CONN
// ====================================================================

static struct ETCP_ROUTER_CONN* router_conn_find(struct UTUN_INSTANCE* inst,
                                                  uint64_t remote_node_id, uint8_t svc_id) {
    if (!inst || !inst->router_conns) return NULL;
    uint8_t key[9];
    memcpy(key, &remote_node_id, 8);
    key[8] = svc_id;
    return (struct ETCP_ROUTER_CONN*)queue_find_data_by_index(inst->router_conns, key);
}

// ====================================================================
// Отправка: router_send_one + send_q
// ====================================================================

static int router_send_one(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) {
    return router_send_one_flags(rconn, payload, pl_len, 0);
}

static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len, uint32_t seq_flags) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    uint32_t seq = (seq_flags & (ROUTER_SEQ_CLOSE_FLAG | ROUTER_SEQ_RST_FLAG))
                   ? 0 : rconn->tx_seq++;
    seq |= seq_flags;
    size_t   total_len = SVC_ROUTE_HDR_SIZE + pl_len;
    uint8_t* dgram = u_malloc(total_len);
    if (!dgram) { if (!seq_flags) rconn->tx_seq--; return -1; }
    struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)dgram;
    hdr->cmd         = ETCP_ID_SVC_ROUTE;
    hdr->dst_node_id = rconn->remote_node_id;
    hdr->src_node_id = inst->node_id;
    hdr->seq         = seq;
    hdr->svc_id      = rconn->svc_id;
    if (pl_len > 0) memcpy(dgram + SVC_ROUTE_HDR_SIZE, payload, pl_len);

    struct ll_entry* entry = queue_entry_new(0);
    if (!entry) { u_free(dgram); if (!seq_flags) rconn->tx_seq--; return -1; }
    entry->dgram = dgram;
    entry->len   = total_len;

    struct ETCP_CONN* conn = route_bgp_find_conn_for_node(inst->bgp, rconn->remote_node_id);
    if (!conn) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "router_send_one: no route to %016llx svc_id=%u",
                   (unsigned long long)rconn->remote_node_id, rconn->svc_id);
        queue_entry_free(entry); queue_dgram_free(entry); if (!seq_flags) rconn->tx_seq--; return -1;
    }
    rconn->last_dgram_ts = get_current_timestamp();
    DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "router_send: seq=%u → %016llx svc_id=%u len=%zu inflight=%u",
                seq, (unsigned long long)rconn->remote_node_id, rconn->svc_id, pl_len,
                rconn->tx_seq - rconn->rx_acked);
    return etcp_send(conn, entry);
}

// Отправка без conn (RST для неизвестного src)
static void router_send_to(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t svc_id, uint32_t seq_flags) {
    struct ETCP_CONN* conn = route_bgp_find_conn_for_node(inst->bgp, dst);
    if (!conn) return;
    struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)u_malloc(SVC_ROUTE_HDR_SIZE);
    if (!hdr) return;
    hdr->cmd         = ETCP_ID_SVC_ROUTE;
    hdr->dst_node_id = dst;
    hdr->src_node_id = inst->node_id;
    hdr->seq         = seq_flags;
    hdr->svc_id      = svc_id;
    struct ll_entry* entry = queue_entry_new(0);
    if (!entry) { u_free(hdr); return; }
    entry->dgram = (uint8_t*)hdr;
    entry->len   = SVC_ROUTE_HDR_SIZE;
    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "router_send_to: seq_flags=0x%08x → %016llx svc_id=%u",
                seq_flags, (unsigned long long)dst, svc_id);
    etcp_send(conn, entry);
}

static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id];
    if (!cb) return;
    struct ll_entry* e = queue_entry_new(0);
    if (!e) return;
    e->dgram = u_malloc(1);
    if (!e->dgram) { queue_entry_free(e); return; }
    e->dgram[0] = rconn->svc_id;
    e->len = 1;
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "router_close_svc: svc_id=%u remote=%016llx",
               rconn->svc_id, (unsigned long long)rconn->remote_node_id);
    cb(NULL, e);
}

// Закрыть conn: отправить CLOSE удалённой стороне, уведомить локальный сервис, очистить
static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) {
    if (!rconn) return;
    struct UTUN_INSTANCE* inst = rconn->inst;
    router_send_close_to_service(rconn);
    // Отправляем CLOSE удалённой стороне
    struct ETCP_CONN* conn = route_bgp_find_conn_for_node(inst->bgp, rconn->remote_node_id);
    if (conn) router_send_one_flags(rconn, NULL, 0, ROUTER_SEQ_CLOSE_FLAG);
    // Таймеры
    if (rconn->ack_timer)          { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; }
    if (rconn->idle_ack_timer)     { uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; }
    if (rconn->send_resume_timer)  { uasync_cancel_timeout(inst->ua, rconn->send_resume_timer); rconn->send_resume_timer = NULL; }
    if (rconn->recv_q) {
        struct ll_entry* f;
        while ((f = queue_data_get(rconn->recv_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
        queue_free(rconn->recv_q); rconn->recv_q = NULL;
    }
    if (rconn->send_q) {
        struct ll_entry* f;
        while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
        queue_free(rconn->send_q); rconn->send_q = NULL;
    }
    if (inst->router_conns)
        queue_remove_data(inst->router_conns, &rconn->ll);
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "router_conn: closed remote=%016llx svc_id=%u",
               (unsigned long long)rconn->remote_node_id, rconn->svc_id);
    queue_entry_free(&rconn->ll);
}

static void router_drain_send_q(struct ETCP_ROUTER_CONN* rconn) {
    while (1) {
        if (rconn->tx_seq - rconn->rx_acked >= ROUTER_MAX_INFLIGHT) break;
        struct ll_entry* e = queue_data_get(rconn->send_q);
        if (!e) { rconn->send_blocked = 0; break; }
        router_send_one(rconn, e->dgram, e->len);
        queue_dgram_free(e); queue_entry_free(e);
    }
    if (rconn->send_blocked && !rconn->send_resume_timer)
        rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua,
            ROUTER_SEND_RESUME_TB, rconn, router_send_resume_cb, "router_send_resume");
    if (!rconn->send_blocked && rconn->send_resume_timer) {
        uasync_cancel_timeout(rconn->inst->ua, rconn->send_resume_timer);
        rconn->send_resume_timer = NULL;
    }
}

static void router_send_resume_cb(void* arg) {
    struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg;
    rconn->send_resume_timer = NULL;
    router_drain_send_q(rconn);
}

static void router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) {
    struct ll_entry* qe = queue_entry_new(0);
    if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_enqueue_send: queue_entry_new failed"); return; }
    qe->dgram = u_malloc(pl_len);
    if (!qe->dgram) { queue_entry_free(qe); return; }
    qe->len = pl_len;
    if (pl_len > 0) memcpy(qe->dgram, payload, pl_len);
    queue_data_put(rconn->send_q, qe);
    rconn->send_blocked = 1;
    if (!rconn->send_resume_timer)
        rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua,
            ROUTER_SEND_RESUME_TB, rconn, router_send_resume_cb, "router_send_resume");
    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "router_send_q: queued svc_id=%u inflight=%u send_q=%d",
                rconn->svc_id, rconn->tx_seq - rconn->rx_acked, queue_entry_count(rconn->send_q));
}

// ====================================================================
// ACK timers
// ====================================================================

static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn) {
    if (rconn->idle_ack_timer) {
        uasync_cancel_timeout(rconn->inst->ua, rconn->idle_ack_timer);
        rconn->idle_ack_timer = NULL;
    }
    if (!rconn->ack_timer)
        rconn->ack_timer = uasync_set_timeout(rconn->inst->ua,
            ROUTER_ACK_INTERVAL_TB, rconn, router_ack_timer_cb, "router_ack");
}

static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    struct ETCP_CONN* conn = route_bgp_find_conn_for_node(inst->bgp, rconn->remote_node_id);
    if (!conn) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "router_ack: no route to %016llx svc_id=%u",
                   (unsigned long long)rconn->remote_node_id, rconn->svc_id);
        return;
    }
    struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)u_malloc(SVC_ROUTE_HDR_SIZE);
    if (!hdr) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_ack: u_malloc failed"); return; }
    hdr->cmd         = ETCP_ID_SVC_ROUTE;
    hdr->dst_node_id = rconn->remote_node_id;
    hdr->src_node_id = inst->node_id;
    hdr->seq         = rconn->rx_seq;
    hdr->svc_id      = rconn->svc_id;

    struct ll_entry* entry = queue_entry_new(0);
    if (!entry) { u_free(hdr); return; }
    entry->dgram = (uint8_t*)hdr;
    entry->len   = SVC_ROUTE_HDR_SIZE;

    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "router_ack: → %016llx svc_id=%u rx_seq=%u",
                (unsigned long long)rconn->remote_node_id, rconn->svc_id, rconn->rx_seq);
    etcp_send(conn, entry);
}

static void router_ack_timer_cb(void* arg) {
    struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg;
    rconn->ack_timer = NULL;
    if (rconn->rx_seq != rconn->rx_acked) {
        router_send_ack(rconn);
        rconn->rx_acked = rconn->rx_seq;
    }
    if (rconn->rx_seq != rconn->rx_acked)
        rconn->ack_timer = uasync_set_timeout(rconn->inst->ua,
            ROUTER_ACK_INTERVAL_TB, rconn, router_ack_timer_cb, "router_ack");
    else
        rconn->idle_ack_timer = uasync_set_timeout(rconn->inst->ua,
            ROUTER_ACK_IDLE_TB, rconn, router_idle_ack_timer_cb, "router_idle_ack");
}

static void router_idle_ack_timer_cb(void* arg) {
    struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg;
    rconn->idle_ack_timer = NULL;
    if (rconn->rx_seq != rconn->rx_acked) {
        router_send_ack(rconn);
        rconn->rx_acked = rconn->rx_seq;
    }
}

// ====================================================================
// Reorder / assembly (аналог etcp_output_try_assembly)
// ====================================================================

static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn,
                           const uint8_t* payload, size_t payload_len) {
    struct UTUN_INSTANCE* inst = rconn->inst;
    etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id];
    if (!cb) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "router_deliver: no handler for svc_id=%u", rconn->svc_id);
        return;
    }
    size_t entry_len = 1 + payload_len;
    struct ll_entry* svc_entry = queue_entry_new(0);
    if (!svc_entry) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_deliver: queue_entry_new failed"); return; }
    svc_entry->len   = entry_len;
    svc_entry->dgram = u_malloc(entry_len);
    if (!svc_entry->dgram) { queue_entry_free(svc_entry); return; }
    svc_entry->dgram[0] = rconn->svc_id;
    if (payload_len > 0) memcpy(svc_entry->dgram + 1, payload, payload_len);

    DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "router_deliver: svc_id=%u len=%zu from remote=%016llx",
                rconn->svc_id, payload_len, (unsigned long long)rconn->remote_node_id);
    cb(conn, svc_entry);
}

static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn) {
    uint32_t next = rconn->rx_seq;
    while (1) {
        struct ll_entry* entry = queue_find_data_by_index(rconn->recv_q, &next);
        if (!entry) break;
        queue_remove_data(rconn->recv_q, entry);
        router_deliver(rconn, conn, entry->dgram, entry->len);
        rconn->rx_seq = next + 1;
        next++;
        queue_dgram_free(entry);
        queue_entry_free(entry);
    }
}

// ====================================================================
// etcp_router_recv_cb — обработчик ETCP_ID_SVC_ROUTE
// ====================================================================

static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
    if (!entry) return;
    struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL;
    if (!inst || entry->len < SVC_ROUTE_HDR_SIZE) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "etcp_router: invalid packet inst=%p len=%zu min=%u",
                   (void*)inst, entry->len, (unsigned)SVC_ROUTE_HDR_SIZE);
        if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
        return;
    }
    struct SVC_ROUTE_HDR* hdr    = (struct SVC_ROUTE_HDR*)entry->dgram;
    size_t                pl_len = entry->len - SVC_ROUTE_HDR_SIZE;
    uint8_t*              pl     = entry->dgram + SVC_ROUTE_HDR_SIZE;

    if (hdr->dst_node_id == inst->node_id) {
        // ========== Мы — целевая нода ==========
        struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, hdr->src_node_id, hdr->svc_id);

        // CLOSE или RST — pl_len==0 и флаг в seq (закрываем conn)
        if (pl_len == 0 && (hdr->seq & (ROUTER_SEQ_CLOSE_FLAG | ROUTER_SEQ_RST_FLAG))) {
            if (rconn) router_close_and_notify(rconn);
            queue_dgram_free(entry); queue_entry_free(entry);
            return;
        }

        if (pl_len == 0) {
            // ACK-пакет: обновляем rx_acked, drain send_q, выходим
            if (rconn) {
                rconn->rx_acked = hdr->seq;
                rconn->last_dgram_ts = get_current_timestamp();
                if (rconn->send_blocked) router_drain_send_q(rconn);
            }
            queue_dgram_free(entry); queue_entry_free(entry);
            return;
        }

        // Data-пакет — авто-создаём conn если ещё нет
        if (!rconn) {
            rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id);
            if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return; }
        }

        // Seq-состояние есть — reorder
        rconn->last_dgram_ts = get_current_timestamp();
        uint32_t seq = hdr->seq;

        // Проверка границ (32-bit circular, аналог ETCP MAX_INFLIGHT_SIZE)
        {
            int32_t d = (int32_t)(seq - rconn->rx_seq);
            if (d > ROUTER_MAX_INFLIGHT || d < -ROUTER_MAX_INFLIGHT) {
                DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router: seq=%u out of bounds, rx_seq=%u (d=%d), dropping",
                            seq, rconn->rx_seq, d);
                queue_dgram_free(entry); queue_entry_free(entry);
                return;
            }
        }

        // Дубликат (seq уже доставлен или в recv_q)
        if (((int32_t)(rconn->rx_seq - seq) > 0) ||
            queue_find_data_by_index(rconn->recv_q, &seq)) {
            DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "router: dup seq=%u rx_seq=%u, dropping", seq, rconn->rx_seq);
            queue_dgram_free(entry); queue_entry_free(entry);
            return;
        }

        // Кладём в recv_q
        struct ll_entry* qe = queue_entry_new(4);
        if (!qe) { queue_dgram_free(entry); queue_entry_free(entry); return; }
        *(uint32_t*)qe->data = seq;
        qe->dgram = u_malloc(pl_len);
        if (!qe->dgram) { queue_entry_free(qe); queue_dgram_free(entry); queue_entry_free(entry); return; }
        qe->len = pl_len;
        memcpy(qe->dgram, pl, pl_len);

        queue_data_put_with_index(rconn->recv_q, qe);
        DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "router: queued seq=%u rx_seq=%u recv_q=%d from %016llx svc_id=%u",
                    seq, rconn->rx_seq, queue_entry_count(rconn->recv_q),
                    (unsigned long long)hdr->src_node_id, rconn->svc_id);

        if (seq == rconn->rx_seq)
            router_try_assembly(rconn, conn);

        router_schedule_ack(rconn);
        queue_dgram_free(entry); queue_entry_free(entry);
    } else {
        // ========== Транзит ==========
        struct ETCP_CONN* next = route_bgp_find_conn_for_node(inst->bgp, hdr->dst_node_id);
        if (!next) {
            DEBUG_WARN(DEBUG_CATEGORY_ETCP, "etcp_router: no route to %016llx svc_id=%u, dropping",
                       (unsigned long long)hdr->dst_node_id, hdr->svc_id);
            queue_dgram_free(entry); queue_entry_free(entry);
            return;
        }
        DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "etcp_router: forwarding svc_id=%u → %016llx via %s",
                    hdr->svc_id, (unsigned long long)hdr->dst_node_id, next->log_name);
        etcp_send(next, entry);
    }
}

// ====================================================================
// Public API
// ====================================================================

int etcp_router_init(struct UTUN_INSTANCE* inst) {
    if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_router_init: NULL instance"); return -1; }
    memset(&inst->router_bindings, 0, sizeof(inst->router_bindings));
    inst->router_conns = queue_new(inst->ua,
        ROUTER_CONN_HASH_SIZE, 0, 9, "router_conns");
    if (!inst->router_conns) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_router_init: queue_new(router_conns) failed");
        return -1;
    }
    int ret = etcp_bind(inst, ETCP_ID_SVC_ROUTE, etcp_router_recv_cb);
    if (ret != 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_router_init: etcp_bind failed, ret=%d", ret);
        queue_free(inst->router_conns); inst->router_conns = NULL;
        return -1;
    }
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "etcp_router initialized for node %016llx", (unsigned long long)inst->node_id);
    return 0;
}

void etcp_router_destroy(struct UTUN_INSTANCE* inst) {
    if (!inst) return;
    etcp_unbind(inst, ETCP_ID_SVC_ROUTE);
    if (inst->router_conns) {
        struct ll_entry* entry;
        while ((entry = queue_data_get(inst->router_conns)) != NULL) {
            struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry;
            router_send_close_to_service(rconn);
            if (rconn->ack_timer)          uasync_cancel_timeout(inst->ua, rconn->ack_timer);
            if (rconn->idle_ack_timer)     uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer);
            if (rconn->send_resume_timer)  uasync_cancel_timeout(inst->ua, rconn->send_resume_timer);
            if (rconn->recv_q) {
                struct ll_entry* f;
                while ((f = queue_data_get(rconn->recv_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
                queue_free(rconn->recv_q);
            }
            if (rconn->send_q) {
                struct ll_entry* f;
                while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
                queue_free(rconn->send_q);
            }
            queue_entry_free(entry);
        }
        queue_free(inst->router_conns);
        inst->router_conns = NULL;
    }
    memset(&inst->router_bindings, 0, sizeof(inst->router_bindings));
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "etcp_router destroyed");
}

int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn callback) {
    if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_router_bind: NULL instance"); return -1; }
    if (!callback) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_router_bind: NULL callback for svc_id=%u", svc_id); return -1; }
    if (inst->router_bindings.callbacks[svc_id])
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "etcp_router_bind: overwriting svc_id=%u", svc_id);
    inst->router_bindings.callbacks[svc_id] = callback;
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "etcp_router_bind: svc_id=%u → cb=%p", svc_id, (void*)callback);
    return 0;
}

int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id) {
    if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_router_unbind: NULL instance"); return -1; }
    if (!inst->router_bindings.callbacks[svc_id]) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "etcp_router_unbind: svc_id=%u not bound", svc_id);
        return -1;
    }
    inst->router_bindings.callbacks[svc_id] = NULL;
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "etcp_router_unbind: svc_id=%u", svc_id);
    return 0;
}

int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_entry* entry) {
    if (!inst || !entry) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_route_send: NULL inst=%p entry=%p", (void*)inst, (void*)entry);
        if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return -1;
    }
    if (!entry->dgram || entry->len < 1) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_route_send: empty entry"); queue_dgram_free(entry); queue_entry_free(entry); return -1;
    }
    uint8_t  svc_id      = entry->dgram[0];
    size_t   payload_len = entry->len - 1;

    if (dst_node_id == inst->node_id) {
        DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "etcp_route_send: loopback svc_id=%u len=%zu", svc_id, payload_len);
        if (inst->router_bindings.callbacks[svc_id])
            inst->router_bindings.callbacks[svc_id](NULL, entry);
        else { queue_dgram_free(entry); queue_entry_free(entry); }
        return 0;
    }

    struct ETCP_CONN* etcp_conn = route_bgp_find_conn_for_node(inst->bgp, dst_node_id);
    if (!etcp_conn) {
        DEBUG_WARN(DEBUG_CATEGORY_ETCP, "etcp_route_send: no BGP route to %016llx svc_id=%u, dropping",
                   (unsigned long long)dst_node_id, svc_id);
        queue_dgram_free(entry); queue_entry_free(entry);
        return -1;
    }

    struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, dst_node_id, svc_id);
    if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; }

    if (rconn->tx_seq - rconn->rx_acked >= ROUTER_MAX_INFLIGHT) {
        router_enqueue_send(rconn, entry->dgram + 1, payload_len);
        queue_dgram_free(entry); queue_entry_free(entry);
        return 0;
    }

    int ret = router_send_one(rconn, entry->dgram + 1, payload_len);
    queue_dgram_free(entry); queue_entry_free(entry);
    return ret;
}

int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t node_id) {
    if (!inst || !inst->bgp) return -1;
    struct ETCP_CONN* conn = route_bgp_find_conn_for_node(inst->bgp, node_id);
    if (!conn || !conn->normalizer || !conn->normalizer->input) return -1;
    return queue_entry_count(conn->normalizer->input);
}

// ====================================================================
// Seq-connection API
// ====================================================================

struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst,
                                               uint64_t remote_node_id, uint8_t svc_id) {
    if (!inst || !inst->router_conns) return NULL;
    struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, remote_node_id, svc_id);
    if (rconn) return rconn;

    size_t data_size = sizeof(struct ETCP_ROUTER_CONN) - sizeof(struct ll_entry);
    rconn = (struct ETCP_ROUTER_CONN*)queue_entry_new(data_size);
    if (!rconn) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_conn_get: queue_entry_new failed"); return NULL; }
    rconn->remote_node_id = remote_node_id;
    rconn->svc_id         = svc_id;
    rconn->last_dgram_ts  = get_current_timestamp();
    rconn->tx_seq         = 0;
    rconn->rx_seq         = 0;
    rconn->rx_acked       = 0;
    rconn->inst           = inst;
    rconn->ack_timer      = NULL;
    rconn->idle_ack_timer = NULL;
    rconn->send_blocked   = 0;
    rconn->send_resume_timer = NULL;
    rconn->recv_q = queue_new(inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q");
    if (!rconn->recv_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_conn_get: queue_new(recv_q) failed"); queue_entry_free(&rconn->ll); return NULL; }
    rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q");
    if (!rconn->send_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_conn_get: queue_new(send_q) failed"); queue_free(rconn->recv_q); queue_entry_free(&rconn->ll); return NULL; }
    queue_data_put_with_index(inst->router_conns, &rconn->ll);
    DEBUG_INFO(DEBUG_CATEGORY_ETCP, "router_conn: new conn remote=%016llx svc_id=%u",
               (unsigned long long)remote_node_id, svc_id);
    return rconn;
}

int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn,
                          const uint8_t* data, size_t len) {
    if (!rconn || !data || len == 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "router_conn_send: invalid args rconn=%p data=%p len=%zu",
                    (void*)rconn, (const void*)data, len);
        return -1;
    }
    if (rconn->tx_seq - rconn->rx_acked >= ROUTER_MAX_INFLIGHT) {
        router_enqueue_send(rconn, data, len);
        return 0;
    }
    return router_send_one(rconn, data, len);
}

void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn) {
    router_close_and_notify(rconn);
}

void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id) {
    if (!inst || !inst->router_conns) return;
    struct ll_entry* entry;
    while ((entry = queue_data_get(inst->router_conns)) != NULL) {
        struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry;
        if (rconn->remote_node_id == remote_node_id) {
            router_send_close_to_service(rconn);
            if (rconn->ack_timer)          { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; }
            if (rconn->idle_ack_timer)     { uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; }
            if (rconn->send_resume_timer)  { uasync_cancel_timeout(inst->ua, rconn->send_resume_timer); rconn->send_resume_timer = NULL; }
            if (rconn->recv_q) {
                struct ll_entry* f;
                while ((f = queue_data_get(rconn->recv_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
                queue_free(rconn->recv_q);
            }
            if (rconn->send_q) {
                struct ll_entry* f;
                while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
                queue_free(rconn->send_q);
            }
            DEBUG_INFO(DEBUG_CATEGORY_ETCP, "router_conn: closed by node remote=%016llx svc_id=%u",
                       (unsigned long long)rconn->remote_node_id, rconn->svc_id);
            queue_entry_free(&rconn->ll);
        } else {
            queue_data_put(inst->router_conns, entry);
        }
    }
}