utun2/src/call/call.c

// call.c — P2P аудио-звонок: сигналинг + медиа (per-instance)
//
// Полный цикл: INVITE → RINGING → ACCEPT → conn_mgr (прямой линк) → ACTIVE →
// медиа (CALL_MEDIA) → HANGUP/ENDED. Звонок из CHAT-группы или DM — модуль
// оперирует (group_id, peer_node_id), маршрутизация через etcp_router.
//
// Сигналинг идёт с ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT, медиа — без подписи.
// Реестр сессий: ll_queue с hash-индексом по call_id (8 байт).
//
// Thread-модель: всё в uasync-потоке (single-threaded). Освобождение сессии
// отложено через uasync_call_soon, чтобы не было UAF при завершении из
// коллбэков conn_mgr/таймеров.

#include "call.h"
#include "call_proto.h"
#include "../utun_instance.h"
#include "../chat/chat_event.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/route_crypto.h"
#include "../routing_layer/conn_mgr.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_node.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
#include "../../lib/debug_config.h"
#include "../../lib/u_async.h"
#include "../../lib/platform_compat.h"

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

#define CALL_ID "call"

enum call_state {
    CALL_ST_IDLE = 0,
    CALL_ST_OUTGOING,    /* caller: INVITE отправлен */
    CALL_ST_RINGING,     /* caller: получил RINGING; callee: отправил RINGING */
    CALL_ST_CONNECTING,  /* обе стороны: conn_mgr_open в процессе */
    CALL_ST_ACTIVE,      /* обе стороны: соединение поднято */
    CALL_ST_ENDED,
};

enum call_role { CALL_ROLE_CALLER = 0, CALL_ROLE_CALLEE };

struct call_session {
    struct ll_entry ll;           /* dummy-заголовок; структура хранится в entry->data */
    uint64_t call_id;             /* hash-индекс (8 байт) */
    uint64_t group_id;
    uint64_t peer_node_id;
    uint8_t  role;                /* CALL_ROLE_* */
    uint8_t  state;               /* CALL_ST_* */
    uint8_t  reason;              /* CALL_REASON_* (при ENDED) */
    struct call_ctx* ctx;         /* обратная ссылка (для коллбэков conn_mgr/таймеров) */
    struct CONN_MGR_HANDLE* cm_handle;
    void*    ring_timer;          /* caller: 45с ожидания ответа */
    void*    traffic_timer;       /* ACTIVE: 20с без медиа (запускается по первому медиа) */
    void*    teardown_timer;      /* отложенный разрыв conn_mgr (500мс после завершения) */
    void*    stats_timer;         /* ACTIVE: раз в 1с — CALL_STATS (rtt/buffer/tempo) */
    uint64_t last_media_tb;       /* время последнего CALL_MEDIA (0.1ms) */
    uint16_t tx_media_seq;        /* seq следующего исходящего CALL_MEDIA */
    struct ll_queue* tx_q;        /* FIFO закодированных CALL_MEDIA (backpressure) */
    struct queue_waiter_handle send_waiter; /* waiter на send_q роутера */
    uint32_t c_tx_frames;         /* отправлено медиа-кадров */
    uint32_t c_rx_frames;         /* принято медиа-кадров */
    uint32_t c_tx_dropped;        /* дропнуто при переполнении tx_q */
    uint32_t c_rx_dropped;        /* дропнуто при переполнении jitter-буфера */
};

struct call_ctx {
    struct UTUN_INSTANCE* inst;
    int      initialized;
    struct ll_queue* sessions;    /* struct call_session, hash по call_id */
    uint32_t c_calls_started;     /* суммарно начато звонков */
    uint32_t c_calls_ended;       /* суммарно завершено звонков */
    call_media_recv_fn media_recv_fn;  /* приём Opus-кадров пира */
    void*              media_recv_arg;
    const struct call_audio_ops* audio_ops;  /* аудио-движок (GUI-сборки), NULL в демоне */
};

/* ───────────────────────────────────────────────────────────────────────── */

static struct call_ctx* call_of(struct UTUN_INSTANCE* inst) {
    return inst ? inst->call : NULL;
}

static const char* call_state_name(uint8_t st) {
    switch (st) {
    case CALL_ST_IDLE:       return "IDLE";
    case CALL_ST_OUTGOING:   return "OUTGOING";
    case CALL_ST_RINGING:    return "RINGING";
    case CALL_ST_CONNECTING: return "CONNECTING";
    case CALL_ST_ACTIVE:     return "ACTIVE";
    case CALL_ST_ENDED:      return "ENDED";
    default:                 return "?";
    }
}

static const char* call_subcmd_name(uint8_t sc) {
    switch (sc) {
    case CALL_SUBCMD_INVITE:  return "INVITE";
    case CALL_SUBCMD_RINGING: return "RINGING";
    case CALL_SUBCMD_ACCEPT:  return "ACCEPT";
    case CALL_SUBCMD_DECLINE: return "DECLINE";
    case CALL_SUBCMD_BUSY:    return "BUSY";
    case CALL_SUBCMD_CANCEL:  return "CANCEL";
    case CALL_SUBCMD_HANGUP:  return "HANGUP";
    case CALL_SUBCMD_MEDIA:   return "MEDIA";
    default:                  return "?";
    }
}

/* ─────────────────── отправка ─────────────────── */

static void call_stats_timer_cb(void* arg);

/* data[0] = subcmd. etcp_route_send забирает владение entry. */
static int call_send(struct call_ctx* ctx, uint64_t group_id, uint64_t dst,
                     const uint8_t* data, size_t len, int mode) {
    struct ll_entry* e = queue_entry_new(0);
    if (!e) return -1;
    e->dgram = u_malloc(1 + len);
    if (!e->dgram) { queue_entry_free(e); return -1; }
    e->dgram[0] = ETCP_RT_ID_CALL;
    if (len) memcpy(e->dgram + 1, data, len);
    e->len = (uint16_t)(1 + len);
    int rc = etcp_route_send(ctx->inst, group_id, dst, e, 1, mode);
    if (rc != 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: send FAIL rc=%d subcmd=%s(0x%02x) dst=0x%016llx group=0x%016llx",
                    CALL_ID, rc, call_subcmd_name(len ? data[0] : 0), len ? data[0] : 0,
                    (unsigned long long)dst, (unsigned long long)group_id);
    }
    return rc;
}

/* ─────────────────── реестр сессий ─────────────────── */

static struct call_session* call_session_find(struct call_ctx* ctx, uint64_t call_id) {
    if (!ctx || !ctx->sessions) return NULL;
    struct ll_entry* e = queue_find_data_by_index(ctx->sessions, &call_id);
    return e ? (struct call_session*)e->data : NULL;
}

static struct call_session* call_session_new(struct call_ctx* ctx, uint64_t call_id,
                                             uint64_t group_id, uint64_t peer, uint8_t role) {
    if (!ctx || !ctx->sessions) return NULL;
    struct ll_entry* qe = queue_entry_new(sizeof(struct call_session));
    if (!qe) return NULL;
    struct call_session* s = (struct call_session*)qe->data;
    memset(s, 0, sizeof(*s));
    s->call_id = call_id;
    s->group_id = group_id;
    s->peer_node_id = peer;
    s->role = role;
    s->state = CALL_ST_IDLE;
    s->ctx = ctx;
    s->tx_q = queue_new(ctx->inst->ua, 0, 0, 0, "call_txq");
    if (!s->tx_q) { queue_entry_free(qe); return NULL; }
    if (queue_data_put_with_index(ctx->sessions, qe) != 0) {
        queue_free(s->tx_q);
        queue_entry_free(qe);
        return NULL;
    }
    return s;
}

static void call_session_free_cb(void* arg) {
    struct ll_entry* e = (struct ll_entry*)arg;
    queue_entry_free(e);
}

/* Освободить медиа-объекты сессии: отменить waiter на send_q роутера и
 * дренировать tx_q (без отправки). Безопасно вызывать из uasync-потока. */
static void call_session_cleanup(struct call_ctx* ctx, struct call_session* s) {
    etcp_router_cancel_send_ready(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL, &s->send_waiter);
    if (s->tx_q) {
        struct ll_entry* e;
        while ((e = queue_data_get(s->tx_q)) != NULL) {
            queue_dgram_free(e);
            queue_entry_free(e);
        }
        queue_free(s->tx_q);
        s->tx_q = NULL;
    }
}

/* Убрать сессию из реестра и освободить отложенно (не из коллбэков conn_mgr/таймеров). */
static void call_session_remove_and_free(struct call_ctx* ctx, struct call_session* s) {
    call_session_cleanup(ctx, s);
    struct ll_entry* e = queue_find_data_by_index(ctx->sessions, &s->call_id);
    if (e) {
        queue_remove_data(ctx->sessions, e);
        uasync_call_soon(ctx->inst->ua, e, call_session_free_cb);
    }
}

/* ─────────────────── события в GUI ─────────────────── */

static void call_event_id(struct call_ctx* ctx, int type, uint64_t call_id, uint8_t reason) {
    uint8_t buf[16];
    size_t off = 0;
    memcpy(buf + off, &call_id, 8); off += 8;
    buf[off++] = reason;
    chat_event_post(ctx->inst, type, buf, (int)off);
}

static void call_event_incoming(struct call_ctx* ctx, uint64_t group_id, uint64_t caller, uint64_t call_id) {
    char ch[32];
    snprintf(ch, sizeof(ch), "%llu", (unsigned long long)group_id);
    size_t ch_len = strlen(ch);
    uint8_t buf[1 + 32 + 8 + 8];
    size_t off = 0;
    buf[off++] = (uint8_t)ch_len;
    memcpy(buf + off, ch, ch_len); off += ch_len;
    memcpy(buf + off, &caller, 8); off += 8;
    memcpy(buf + off, &call_id, 8); off += 8;
    chat_event_post(ctx->inst, CHAT_EVT_CALL_INCOMING, buf, (int)off);
}

/* ─────────────────── завершение ─────────────────── */

/* Отложенный разрыв: единственная точка, где закрывается conn_mgr (после того,
 * как финальный сигналинг — HANGUP/DECLINE — гарантированно долетел). */
static void call_teardown_timer_cb(void* arg) {
    struct call_session* s = (struct call_session*)arg;
    if (!s || !s->ctx) return;
    struct call_ctx* ctx = s->ctx;
    s->teardown_timer = NULL;
    if (s->cm_handle) { conn_mgr_close(s->cm_handle); s->cm_handle = NULL; }
    call_session_remove_and_free(ctx, s);
}

static void call_end(struct call_ctx* ctx, struct call_session* s, uint8_t reason) {
    if (!ctx || !s || s->state == CALL_ST_ENDED) return;

    s->state = CALL_ST_ENDED;
    s->reason = reason;
    ctx->c_calls_ended++;

    if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; }
    if (s->traffic_timer) { uasync_cancel_timeout(ctx->inst->ua, s->traffic_timer); s->traffic_timer = NULL; }
    if (s->stats_timer) { uasync_cancel_timeout(ctx->inst->ua, s->stats_timer); s->stats_timer = NULL; }

    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call ENDED id=%016llx reason=%d role=%s rx=%u tx=%u tx_dropped=%u",
               CALL_ID, (unsigned long long)s->call_id, reason,
               s->role == CALL_ROLE_CALLER ? "caller" : "callee",
               s->c_rx_frames, s->c_tx_frames, s->c_tx_dropped);

    call_event_id(ctx, CHAT_EVT_CALL_ENDED, s->call_id, reason);

    /* Разрыв conn_mgr — отложенно (500мс), чтобы HANGUP/DECLINE успел долететь.
     * Без cm_handle рвать нечего — освобождаем сразу (отложенно). */
    if (s->cm_handle) {
        s->teardown_timer = uasync_set_timeout(ctx->inst->ua, CALL_TEARDOWN_DELAY_TB, s, call_teardown_timer_cb, "call_teardown");
    } else {
        call_session_remove_and_free(ctx, s);
    }
}

/* ─────────────────── conn_mgr ─────────────────── */

static void call_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_id,
                         enum conn_mgr_event event, void* arg) {
    struct call_session* s = (struct call_session*)arg;
    if (!s || !s->ctx) return;
    struct call_ctx* ctx = s->ctx;
    if (s->state == CALL_ST_ENDED) {
        /* сессия завершена, ждёт teardown: conn_mgr сообщает об обрыве — handle больше
         * не трогаем (entry чистится самим conn_mgr / удалённой стороной) */
        if (event == CONN_EVENT_DOWN || event == CONN_EVENT_TIMEOUT) s->cm_handle = NULL;
        return;
    }

    if (event == CONN_EVENT_UP) {
        s->cm_handle = h;
        s->state = CALL_ST_ACTIVE;
        DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: conn UP id=%016llx peer=0x%016llx -> ACTIVE",
                   CALL_ID, (unsigned long long)s->call_id, (unsigned long long)node_id);
        call_event_id(ctx, CHAT_EVT_CALL_ACCEPTED, s->call_id, 0);
        if (!s->stats_timer)
            s->stats_timer = uasync_set_timeout(ctx->inst->ua, 10000, s, call_stats_timer_cb, "call_stats");
    } else if (event == CONN_EVENT_DOWN) {
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: conn DOWN id=%016llx peer=0x%016llx (temporary, keep alive)",
                   CALL_ID, (unsigned long long)s->call_id, (unsigned long long)node_id);
    } else { /* CONN_EVENT_TIMEOUT */
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: conn TIMEOUT id=%016llx peer=0x%016llx -> fail",
                   CALL_ID, (unsigned long long)s->call_id, (unsigned long long)node_id);
        /* entry уже чистится вызывающим (monitor) — handle не закрываем */
        s->cm_handle = NULL;
        call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
    }
}

/* ─────────────────── таймеры ─────────────────── */

static void call_ring_timeout_cb(void* arg) {
    struct call_session* s = (struct call_session*)arg;
    if (!s || !s->ctx) return;
    struct call_ctx* ctx = s->ctx;
    s->ring_timer = NULL;
    if (s->state == CALL_ST_ENDED) return;

    DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: ring timeout id=%016llx -> CANCEL",
               CALL_ID, (unsigned long long)s->call_id);
    uint8_t pkt[sizeof(struct call_pkt_cancel)];
    struct call_pkt_cancel* c = (struct call_pkt_cancel*)pkt;
    c->subcmd = CALL_SUBCMD_CANCEL;
    c->call_id = s->call_id;
    call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
    call_end(ctx, s, CALL_REASON_RING_TIMEOUT);
}

static void call_traffic_timeout_cb(void* arg) {
    struct call_session* s = (struct call_session*)arg;
    if (!s || !s->ctx) return;
    struct call_ctx* ctx = s->ctx;
    s->traffic_timer = NULL;
    if (s->state == CALL_ST_ENDED) return;

    DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: no traffic 20s id=%016llx -> HANGUP",
               CALL_ID, (unsigned long long)s->call_id);
    uint8_t pkt[sizeof(struct call_pkt_hangup)];
    struct call_pkt_hangup* hu = (struct call_pkt_hangup*)pkt;
    hu->subcmd = CALL_SUBCMD_HANGUP;
    hu->call_id = s->call_id;
    hu->reason = CALL_REASON_NO_TRAFFIC;
    call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
    call_end(ctx, s, CALL_REASON_NO_TRAFFIC);
}

/* Периодическое событие CALL_STATS (раз в 1с): rtt_ms, buffer_ms, tempo_x100. */
static void call_stats_timer_cb(void* arg) {
    struct call_session* s = (struct call_session*)arg;
    if (!s || !s->ctx) return;
    struct call_ctx* ctx = s->ctx;
    s->stats_timer = NULL;
    if (s->state != CALL_ST_ACTIVE) return;

    int buffer_ms = 0, tempo_x100 = 100;
    uint32_t dropped = 0, underruns = 0;
    if (ctx->audio_ops && ctx->audio_ops->get_stats)
        ctx->audio_ops->get_stats(s->call_id, &buffer_ms, &tempo_x100, &dropped, &underruns);

    uint16_t rtt = 0;
    struct ETCP_ROUTER_CONN* rconn =
        etcp_router_conn_get(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL);
    if (rconn && rconn->recv_conn) rtt = rconn->recv_conn->rtt_last;

    uint8_t buf[18];
    size_t off = 0;
    memcpy(buf + off, &s->call_id, 8); off += 8;
    buf[off++] = (uint8_t)(rtt >> 8); buf[off++] = (uint8_t)(rtt & 0xFF);
    uint16_t bms = (uint16_t)(buffer_ms < 0 ? 0 : (buffer_ms > 0xFFFF ? 0xFFFF : buffer_ms));
    buf[off++] = (uint8_t)(bms >> 8); buf[off++] = (uint8_t)(bms & 0xFF);
    uint16_t t = (uint16_t)(tempo_x100 < 0 ? 0 : (tempo_x100 > 0xFFFF ? 0xFFFF : tempo_x100));
    buf[off++] = (uint8_t)(t >> 8); buf[off++] = (uint8_t)(t & 0xFF);
    uint16_t dr = (uint16_t)(dropped > 0xFFFF ? 0xFFFF : dropped);
    buf[off++] = (uint8_t)(dr >> 8); buf[off++] = (uint8_t)(dr & 0xFF);
    uint16_t un = (uint16_t)(underruns > 0xFFFF ? 0xFFFF : underruns);
    buf[off++] = (uint8_t)(un >> 8); buf[off++] = (uint8_t)(un & 0xFF);
    chat_event_post(ctx->inst, CHAT_EVT_CALL_STATS, buf, (int)off);

    s->stats_timer = uasync_set_timeout(ctx->inst->ua, 10000, s, call_stats_timer_cb, "call_stats");
}

/* ─────────────────── потеря соединения ─────────────────── */

/* На обрыв ETCP-соединения с пиром завершаем ACTIVE-звонки (robust fallback:
 * HANGUP может не успеть долететь, если закрывающая сторона рвёт conn_mgr). */
static void call_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
    struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized || !conn) return;
    if (status != ETCP_CONN_STATUS_DOWN && status != ETCP_CONN_STATUS_DELETE) return;
    uint64_t peer = conn->peer_node_id;
    if (!peer) return;

    struct ll_entry* e = ctx->sessions ? ctx->sessions->head : NULL;
    while (e) {
        struct ll_entry* next = e->next;
        struct call_session* s = (struct call_session*)e->data;
        if (s->peer_node_id == peer) {
            if (s->state == CALL_ST_ACTIVE) {
                DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: peer conn lost id=%016llx peer=0x%016llx -> end",
                           CALL_ID, (unsigned long long)s->call_id, (unsigned long long)peer);
                call_end(ctx, s, CALL_REASON_REMOTE_HANGUP);
            } else if (s->state == CALL_ST_ENDED) {
                /* удалённая сторона уже порвала (DISCONNECT чистит entry без коллбэка) */
                s->cm_handle = NULL;
            }
        }
        e = next;
    }
}

/* ─────────────────── обработчики приёма ─────────────────── */

static void call_handle_invite(struct call_ctx* ctx, uint64_t from, uint64_t group,
                               const uint8_t* b, size_t n) {
    if (n < sizeof(struct call_pkt_invite) - sizeof(struct call_pkt_track)) {
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: INVITE too short n=%zu", CALL_ID, n);
        return;
    }
    uint64_t call_id, inv_group;
    memcpy(&call_id, b + 1, 8);
    memcpy(&inv_group, b + 9, 8);

    /* повторная доставка INVITE (тот же call_id) — идемпотентно RINGING */
    struct call_session* s = call_session_find(ctx, call_id);
    if (s && s->state != CALL_ST_ENDED) {
        DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: INVITE retransmit id=%016llx -> re-RINGING", CALL_ID, (unsigned long long)call_id);
        uint8_t pkt[sizeof(struct call_pkt_ringing)];
        struct call_pkt_ringing* r = (struct call_pkt_ringing*)pkt;
        r->subcmd = CALL_SUBCMD_RINGING;
        r->call_id = call_id;
        call_send(ctx, s->group_id, from, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
        return;
    }

    /* занятость: есть другая активная сессия */
    struct ll_entry* e = ctx->sessions ? ctx->sessions->head : NULL;
    while (e) {
        struct call_session* other = (struct call_session*)e->data;
        if (other->state != CALL_ST_ENDED) {
            DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: busy — active call id=%016llx, reply BUSY to id=%016llx",
                       CALL_ID, (unsigned long long)other->call_id, (unsigned long long)call_id);
            uint8_t pkt[sizeof(struct call_pkt_busy)];
            struct call_pkt_busy* bz = (struct call_pkt_busy*)pkt;
            bz->subcmd = CALL_SUBCMD_BUSY;
            bz->call_id = call_id;
            call_send(ctx, group, from, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
            return;
        }
        e = e->next;
    }

    s = call_session_new(ctx, call_id, group, from, CALL_ROLE_CALLEE);
    if (!s) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: session alloc failed id=%016llx", CALL_ID, (unsigned long long)call_id);
        return;
    }
    s->state = CALL_ST_RINGING;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: INVITE id=%016llx from=0x%016llx group=0x%016llx (declared group=0x%016llx)",
               CALL_ID, (unsigned long long)call_id, (unsigned long long)from,
               (unsigned long long)group, (unsigned long long)inv_group);

    call_event_incoming(ctx, group, from, call_id);

    uint8_t pkt[sizeof(struct call_pkt_ringing)];
    struct call_pkt_ringing* r = (struct call_pkt_ringing*)pkt;
    r->subcmd = CALL_SUBCMD_RINGING;
    r->call_id = call_id;
    call_send(ctx, group, from, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
}

static void call_handle_ringing(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < 9) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->role != CALL_ROLE_CALLER || s->state == CALL_ST_ENDED) return;
    if (s->state == CALL_ST_OUTGOING) {
        s->state = CALL_ST_RINGING;
        DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: RINGING id=%016llx", CALL_ID, (unsigned long long)call_id);
        call_event_id(ctx, CHAT_EVT_CALL_RINGING, call_id, 0);
    }
    (void)from;
}

static void call_handle_accept(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < 9) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->role != CALL_ROLE_CALLER || s->state == CALL_ST_ENDED || s->state == CALL_ST_CONNECTING) return;

    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: ACCEPT id=%016llx -> conn_mgr_open", CALL_ID, (unsigned long long)call_id);
    s->state = CALL_ST_CONNECTING;
    if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; }

    int rc = conn_mgr_open(ctx->inst, s->group_id, s->peer_node_id, call_conn_cb, s, &s->cm_handle);
    if (rc != 0) {
        /* либо уже получили TIMEOUT (сессия ENDED), либо ошибка без коллбэка */
        if (s->state != CALL_ST_ENDED) call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
    }
}

static void call_handle_decline(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < 10) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    uint8_t reason = b[9];
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state == CALL_ST_ENDED) return;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: DECLINE id=%016llx reason=%d", CALL_ID, (unsigned long long)call_id, reason);
    call_event_id(ctx, CHAT_EVT_CALL_DECLINED, call_id, reason);
    call_end(ctx, s, CALL_REASON_DECLINE);
    (void)from;
}

static void call_handle_busy(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < 9) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state == CALL_ST_ENDED) return;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: BUSY id=%016llx", CALL_ID, (unsigned long long)call_id);
    call_end(ctx, s, CALL_REASON_BUSY);
    (void)from;
}

static void call_handle_cancel(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < 9) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state == CALL_ST_ENDED) return;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: CANCEL id=%016llx", CALL_ID, (unsigned long long)call_id);
    call_end(ctx, s, CALL_REASON_CANCEL);
    (void)from;
}

static void call_handle_hangup(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < 10) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    uint8_t reason = b[9];
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state == CALL_ST_ENDED) return;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: HANGUP id=%016llx reason=%d", CALL_ID, (unsigned long long)call_id, reason);
    call_end(ctx, s, CALL_REASON_REMOTE_HANGUP);
    (void)from;
}

static void call_handle_media(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
    if (n < CALL_MEDIA_HDR_SIZE) return;
    uint64_t call_id; memcpy(&call_id, b + 1, 8);
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state != CALL_ST_ACTIVE) {
        /* медиа вне ACTIVE — игнорируем (гонка с завершением) */
        return;
    }
    s->c_rx_frames++;
    s->last_media_tb = get_time_tb();

    uint16_t seq = 0;
    uint32_t ts_ms = 0;
    memcpy(&seq, b + 10, 2);
    memcpy(&ts_ms, b + 12, 4);

    /* аудио-движок (декод + джиттер-буфер) — если зарегистрирован */
    if (ctx->audio_ops && ctx->audio_ops->on_media)
        ctx->audio_ops->on_media(ctx->inst, call_id, seq, ts_ms,
                                 b + CALL_MEDIA_HDR_SIZE, (int)(n - CALL_MEDIA_HDR_SIZE));

    /* релей Opus-кадра потребителю (headless-аудио-сокет) */
    if (ctx->media_recv_fn)
        ctx->media_recv_fn(ctx->inst, call_id, b + CALL_MEDIA_HDR_SIZE,
                           (int)(n - CALL_MEDIA_HDR_SIZE), ctx->media_recv_arg);

    /* watchdog: пере-взводим таймер no-traffic на каждый принятый кадр */
    if (s->traffic_timer) { uasync_cancel_timeout(ctx->inst->ua, s->traffic_timer); s->traffic_timer = NULL; }
    s->traffic_timer = uasync_set_timeout(ctx->inst->ua, CALL_NO_TRAFFIC_TB, s, call_traffic_timeout_cb, "call_traffic");
    (void)from;
}

/* ─────────────────── приём (etcp_router) ─────────────────── */

static void call_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
    if (!entry || !conn || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF + 1) {
        if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
        return;
    }
    struct call_ctx* ctx = call_of(conn->instance);
    if (!ctx || !ctx->initialized) { queue_dgram_free(entry); queue_entry_free(entry); return; }

    const uint8_t* d = entry->dgram;
    uint64_t from;  memcpy(&from, d + ROUTER_SVC_SRC_OFF, 8);
    uint64_t group; memcpy(&group, d + ROUTER_SVC_GROUP_OFF, 8);
    uint8_t subcmd = d[ROUTER_SVC_PAYLOAD_OFF];
    const uint8_t* body = d + ROUTER_SVC_PAYLOAD_OFF;
    size_t blen = entry->len - ROUTER_SVC_PAYLOAD_OFF;

    switch (subcmd) {
    case CALL_SUBCMD_INVITE:  call_handle_invite(ctx, from, group, body, blen); break;
    case CALL_SUBCMD_RINGING: call_handle_ringing(ctx, from, body, blen); break;
    case CALL_SUBCMD_ACCEPT:  call_handle_accept(ctx, from, body, blen); break;
    case CALL_SUBCMD_DECLINE: call_handle_decline(ctx, from, body, blen); break;
    case CALL_SUBCMD_BUSY:    call_handle_busy(ctx, from, body, blen); break;
    case CALL_SUBCMD_CANCEL:  call_handle_cancel(ctx, from, body, blen); break;
    case CALL_SUBCMD_HANGUP:  call_handle_hangup(ctx, from, body, blen); break;
    case CALL_SUBCMD_MEDIA:   call_handle_media(ctx, from, body, blen); break;
    default:
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: unknown subcmd=0x%02x from=0x%016llx len=%zu",
                   CALL_ID, subcmd, (unsigned long long)from, blen);
        break;
    }
    queue_dgram_free(entry);
    queue_entry_free(entry);
}

/* ─────────────────── публичный API ─────────────────── */

int call_init(struct UTUN_INSTANCE* inst) {
    if (!inst) return -1;
    if (inst->call && inst->call->initialized) return 0;

    struct call_ctx* ctx = u_calloc(1, sizeof(*ctx));
    if (!ctx) return -1;
    ctx->inst = inst;
    inst->call = ctx;

    ctx->sessions = queue_new(inst->ua, 256, offsetof(struct call_session, call_id), 8, "call_sessions");
    if (!ctx->sessions) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: queue_new(sessions) failed", CALL_ID);
        u_free(ctx); inst->call = NULL;
        return -1;
    }
    if (etcp_router_bind(inst, ETCP_RT_ID_CALL, call_etcp_recv) != 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: etcp_router_bind(0x%02x) failed", CALL_ID, ETCP_RT_ID_CALL);
        queue_free(ctx->sessions);
        u_free(ctx); inst->call = NULL;
        return -1;
    }
    etcp_add_conn_status_cbk(inst, call_on_conn_status, inst);

    ctx->initialized = 1;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: initialized (svc=0x%02x)", CALL_ID, ETCP_RT_ID_CALL);
    return 0;
}

void call_destroy(struct UTUN_INSTANCE* inst) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized) return;

    /* завершить все активные сессии (без отправки — мы выключаемся) */
    while (ctx->sessions && ctx->sessions->head) {
        struct ll_entry* e = ctx->sessions->head;
        struct call_session* s = (struct call_session*)e->data;
        s->state = CALL_ST_ENDED;
        if (s->ring_timer) { uasync_cancel_timeout(inst->ua, s->ring_timer); s->ring_timer = NULL; }
        if (s->traffic_timer) { uasync_cancel_timeout(inst->ua, s->traffic_timer); s->traffic_timer = NULL; }
        if (s->stats_timer) { uasync_cancel_timeout(inst->ua, s->stats_timer); s->stats_timer = NULL; }
        if (s->teardown_timer) { uasync_cancel_timeout(inst->ua, s->teardown_timer); s->teardown_timer = NULL; }
        if (s->cm_handle) { conn_mgr_close(s->cm_handle); s->cm_handle = NULL; }
        call_session_cleanup(ctx, s);
        queue_remove_data(ctx->sessions, e);
        queue_entry_free(e);
    }
    queue_free(ctx->sessions);
    etcp_router_unbind(inst, ETCP_RT_ID_CALL);
    etcp_remove_conn_status_cbk(inst, call_on_conn_status, inst);
    inst->call = NULL;
    uint32_t started = ctx->c_calls_started, ended = ctx->c_calls_ended;
    u_free(ctx);
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: destroyed (started=%u ended=%u)", CALL_ID, started, ended);
}

int call_is_online(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_node_id) {
    if (!inst || !ch_id || !ch_id[0] || !inst->topo_groups) return 0;
    if (peer_node_id == inst->node_id) return 1;
    uint64_t group_id = strtoull(ch_id, NULL, 10);
    struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, group_id);
    if (!g) return 0;
    return topo_node_find_by_id(g, peer_node_id) != NULL;
}

void call_set_media_recv_cb(struct UTUN_INSTANCE* inst, call_media_recv_fn fn, void* arg) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx) return;
    ctx->media_recv_fn = fn;
    ctx->media_recv_arg = arg;
}

void call_set_audio_ops(struct UTUN_INSTANCE* inst, const struct call_audio_ops* ops) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx) return;
    ctx->audio_ops = ops;
}

/* ─────────────────── TX медиа: tx_q + backpressure ─────────────────── */

static void call_tx_flush(struct call_session* s);

static void call_tx_drain_cb(struct ll_queue* q, void* arg) {
    (void)q;
    struct call_session* s = (struct call_session*)arg;
    if (!s || !s->ctx) return;
    if (s->state == CALL_ST_ENDED) return;
    call_tx_flush(s);
}

/* Дренируем tx_q, пока роутер готов; иначе — одноразовый waiter (re-arm в drain_cb). */
static void call_tx_flush(struct call_session* s) {
    struct call_ctx* ctx = s->ctx;
    if (s->state == CALL_ST_ENDED) return;
    /* гарантируем router conn — иначе has_room/on_send_ready не видят его */
    etcp_router_conn_get(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL);
    while (s->tx_q && s->tx_q->head) {
        if (!etcp_router_send_q_has_room(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL)) {
            etcp_router_on_send_ready(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL,
                                      &s->send_waiter, call_tx_drain_cb, s);
            return;
        }
        struct ll_entry* e = queue_data_get(s->tx_q);
        if (etcp_route_send(ctx->inst, s->group_id, s->peer_node_id, e, 1, 0) == 0)
            s->c_tx_frames++;
    }
}

int call_send_media(struct UTUN_INSTANCE* inst, uint64_t call_id, const uint8_t* opus, int len) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized || !opus || len <= 0) return -1;
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state != CALL_ST_ACTIVE) return -1;
    if (CALL_MEDIA_HDR_SIZE + (size_t)len > 1024) return -1;

    uint8_t buf[1024];
    struct call_pkt_media* m = (struct call_pkt_media*)buf;
    m->subcmd = CALL_SUBCMD_MEDIA;
    m->call_id = call_id;
    m->track_id = CALL_TRACK_AUDIO;
    m->seq = s->tx_media_seq++;
    m->ts_ms = (uint32_t)(get_time_tb() / 10);
    memcpy(buf + CALL_MEDIA_HDR_SIZE, opus, (size_t)len);
    size_t mlen = CALL_MEDIA_HDR_SIZE + (size_t)len;

    /* entry: [svc_id][media-пакет] — кладём в tx_q (переполнение → дроп старейшего) */
    struct ll_entry* e = queue_entry_new(0);
    if (!e) return -1;
    e->dgram = u_malloc(1 + mlen);
    if (!e->dgram) { queue_entry_free(e); return -1; }
    e->dgram[0] = ETCP_RT_ID_CALL;
    memcpy(e->dgram + 1, buf, mlen);
    e->len = (uint16_t)(1 + mlen);

    if (queue_entry_count(s->tx_q) >= CALL_MAX_TX_FRAMES) {
        struct ll_entry* old = queue_data_get(s->tx_q);
        if (old) { queue_dgram_free(old); queue_entry_free(old); s->c_tx_dropped++; }
    }
    queue_data_put(s->tx_q, e);

    call_tx_flush(s);
    return 0;
}

void call_start(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_node_id, uint64_t out_call_id) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized || !ch_id || !ch_id[0]) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: call_start bad args (ctx=%p ch_id=%s)", CALL_ID, (void*)ctx, ch_id ? ch_id : "(null)");
        return;
    }
    if (peer_node_id == inst->node_id) {
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: call_start self-call rejected", CALL_ID);
        return;
    }

    uint64_t group_id = strtoull(ch_id, NULL, 10);
    if (!call_is_online(inst, ch_id, peer_node_id)) {
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: peer 0x%016llx offline in group 0x%016llx — call not started",
                   CALL_ID, (unsigned long long)peer_node_id, (unsigned long long)group_id);
        uint8_t buf[9];
        memcpy(buf, &out_call_id, 8);
        buf[8] = 1; /* err: peer offline */
        chat_event_post(inst, CHAT_EVT_CALL_ERROR, buf, sizeof(buf));
        return;
    }

    struct call_session* s = call_session_new(ctx, out_call_id, group_id, peer_node_id, CALL_ROLE_CALLER);
    if (!s) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: session alloc failed id=%016llx", CALL_ID, (unsigned long long)out_call_id);
        return;
    }
    s->state = CALL_ST_OUTGOING;
    ctx->c_calls_started++;

    uint8_t buf[sizeof(struct call_pkt_invite)];
    struct call_pkt_invite* inv = (struct call_pkt_invite*)buf;
    inv->subcmd = CALL_SUBCMD_INVITE;
    inv->call_id = out_call_id;
    inv->group_id = group_id;
    inv->num_tracks = 1;
    inv->tracks[0].type = CALL_TRACK_AUDIO;
    inv->tracks[0].codec = CALL_CODEC_OPUS;
    inv->tracks[0].sample_rate = CALL_OPUS_SAMPLE_RATE;
    inv->tracks[0].channels = 1;
    inv->tracks[0].bitrate = 32000;

    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_start id=%016llx peer=0x%016llx group=0x%016llx",
               CALL_ID, (unsigned long long)out_call_id, (unsigned long long)peer_node_id, (unsigned long long)group_id);

    if (call_send(ctx, group_id, peer_node_id, buf, sizeof(buf), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) != 0) {
        call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
        return;
    }
    s->ring_timer = uasync_set_timeout(inst->ua, CALL_RING_TIMEOUT_TB, s, call_ring_timeout_cb, "call_ring");
}

void call_accept(struct UTUN_INSTANCE* inst, uint64_t call_id) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized) return;
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->role != CALL_ROLE_CALLEE || s->state == CALL_ST_ENDED || s->state == CALL_ST_CONNECTING) return;

    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_accept id=%016llx -> ACCEPT + conn_mgr_open",
               CALL_ID, (unsigned long long)call_id);
    s->state = CALL_ST_CONNECTING;

    uint8_t pkt[sizeof(struct call_pkt_accept)];
    struct call_pkt_accept* a = (struct call_pkt_accept*)pkt;
    a->subcmd = CALL_SUBCMD_ACCEPT;
    a->call_id = call_id;
    call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);

    int rc = conn_mgr_open(inst, s->group_id, s->peer_node_id, call_conn_cb, s, &s->cm_handle);
    if (rc != 0 && s->state != CALL_ST_ENDED) {
        call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
    }
}

void call_decline(struct UTUN_INSTANCE* inst, uint64_t call_id) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized) return;
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->role != CALL_ROLE_CALLEE || s->state == CALL_ST_ENDED) return;

    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_decline id=%016llx", CALL_ID, (unsigned long long)call_id);
    uint8_t pkt[sizeof(struct call_pkt_decline)];
    struct call_pkt_decline* d = (struct call_pkt_decline*)pkt;
    d->subcmd = CALL_SUBCMD_DECLINE;
    d->call_id = call_id;
    d->reason = CALL_REASON_DECLINE;
    call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);

    call_event_id(ctx, CHAT_EVT_CALL_DECLINED, call_id, CALL_REASON_DECLINE);
    call_end(ctx, s, CALL_REASON_DECLINE);
}

void call_hangup(struct UTUN_INSTANCE* inst, uint64_t call_id) {
    struct call_ctx* ctx = call_of(inst);
    if (!ctx || !ctx->initialized) return;
    struct call_session* s = call_session_find(ctx, call_id);
    if (!s || s->state == CALL_ST_ENDED) return;

    DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_hangup id=%016llx", CALL_ID, (unsigned long long)call_id);
    uint8_t pkt[sizeof(struct call_pkt_hangup)];
    struct call_pkt_hangup* hu = (struct call_pkt_hangup*)pkt;
    hu->subcmd = CALL_SUBCMD_HANGUP;
    hu->call_id = call_id;
    hu->reason = CALL_REASON_LOCAL_HANGUP;
    call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);

    call_end(ctx, s, CALL_REASON_LOCAL_HANGUP);
}

/* ─────────────────── trampoline-функции (GUI → uasync) ─────────────────── */

void call_start_trampoline(void* arg) {
    struct call_start_arg* a = (struct call_start_arg*)arg;
    call_start(a->inst, a->ch_id, a->peer_node_id, a->call_id);
    u_free(a);
}

void call_accept_trampoline(void* arg) {
    struct call_id_arg* a = (struct call_id_arg*)arg;
    call_accept(a->inst, a->call_id);
    u_free(a);
}

void call_decline_trampoline(void* arg) {
    struct call_id_arg* a = (struct call_id_arg*)arg;
    call_decline(a->inst, a->call_id);
    u_free(a);
}

void call_hangup_trampoline(void* arg) {
    struct call_id_arg* a = (struct call_id_arg*)arg;
    call_hangup(a->inst, a->call_id);
    u_free(a);
}

void call_send_media_trampoline(void* arg) {
    struct call_media_arg* a = (struct call_media_arg*)arg;
    call_send_media(a->inst, a->call_id, a->opus, a->len);
    u_free(a);
}