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

#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/platform_compat.h"
#include "../../lib/u_async.h"
#include "../utun_instance.h"
#include "../chat/chat_core.h"
#include "stream_member.h"
#include "stream_source.h"
#include "stream_priv.h"

struct stream_session {
    struct stream_session *next;
    struct stream_service *service;
    struct stream_descriptor descriptor;
    uint8_t public_key[SC_PUBKEY_SIZE];
    struct stream_source *source;
    struct stream_buffer *buffer;
    uint8_t *output;
    void *delivery;
    uint64_t next_seq;
    uint64_t start_seq, last_seq, passed_before;
    uint8_t end[STREAM_END_SIZE];
    int end_valid;
    unsigned owners, holds;
    int anchored, closing, invalidated, finished, end_delivered, reason;
};

struct STREAM_HANDLE {
    struct STREAM_HANDLE *next;
    struct stream_service *service;
    struct stream_session *session;
    int publisher;
    int reason;
};

struct stream_service {
    struct UTUN_INSTANCE *inst;
    struct stream_session *sessions;
    struct STREAM_HANDLE *handles;
    struct stream_network *network;
    stream_event_fn callback;
    void *arg;
    void *reconcile;
    int stopping, dispatching, registered, initializing;
};

static void session_deliver(void *arg);
static void service_reconcile(void *arg);

/* Активность CHAT проверяется до чтения профиля, выделения 32 МБ и создания handles. */
static struct TOPO_GROUP *stream_group(struct UTUN_INSTANCE *inst, uint64_t id) {
    struct TOPO_GROUP *group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, id) : NULL;
    return group && group->group_type == TOPO_GROUP_TYPE_CHAT && !group->stopping ? group : NULL;
}

/* Один локальный экземпляр на источник группы; закрывающийся старый запуск не мешает новому. */
static struct stream_session *session_find(struct stream_service *service, uint64_t group_id, uint64_t source_id) {
    for (struct stream_session *s = service->sessions; s; s = s->next)
        if (!s->closing && !s->invalidated && s->descriptor.group_id == group_id && s->descriptor.source_node_id == source_id) return s;
    return NULL;
}

/* Две опции заменяются одной подписанной записью; список отражает экземпляры, не отдельные handles. */
static int service_publish_options(struct stream_service *service, struct TOPO_GROUP *group) {
    if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream options group missing"); return STREAM_ERR_STATE; }
    struct stream_member_state state = {0};
    for (struct stream_session *s = service->sessions; s; s = s->next) {
        if (s->closing || s->invalidated || s->descriptor.group_id != group->group_id) continue;
        if (state.count == STREAM_MEMBER_MAX_SUBSCRIPTIONS) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local subscription list exceeds member limit group=%llx",
                        (unsigned long long)group->group_id); return STREAM_ERR_ARGUMENT;
        }
        struct stream_subscription_key entry = {s->descriptor.source_node_id, s->descriptor.source_epoch};
        size_t at = state.count++;
        while (at && (state.subscriptions[at - 1].source_node_id > entry.source_node_id
               || (state.subscriptions[at - 1].source_node_id == entry.source_node_id
                   && state.subscriptions[at - 1].source_epoch > entry.source_epoch))) {
            state.subscriptions[at] = state.subscriptions[at - 1]; at--;
        }
        state.subscriptions[at] = entry;
        if (s->source && !s->finished) { state.source = s->descriptor; state.published = 1; }
    }
    char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1];
    int result = stream_member_format(&state, source, subscriptions);
    if (result != STREAM_OK) return result;
    struct chat_profile_option options[] = {
        {STREAM_MEMBER_SOURCE, source[0] ? source : NULL},
        {STREAM_MEMBER_SUBSCRIPTIONS, subscriptions[0] ? subscriptions : NULL}
    };
    if (chat_core_set_my_profile_options(service->inst, group->channel_id, options, 2) < 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local member stream options update failed group=%llx owners=%zu",
                    (unsigned long long)group->group_id, state.count); return STREAM_ERR_STATE;
    }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream options group=%llx published=%d subscriptions=%zu",
               (unsigned long long)group->group_id, state.published, state.count);
    return STREAM_OK;
}

/* Ресурсы живут до возврата пользовательского callback/write, даже если он закрыл последний handle. */
static void session_free(struct stream_session *s) {
    struct stream_session **link = &s->service->sessions;
    while (*link != s) link = &(*link)->next;
    *link = s->next;
    if (s->delivery) uasync_call_soon_cancel(s->service->inst->ua, s->delivery);
    if (s->source) stream_source_destroy(s->source);
    else stream_buffer_destroy(s->buffer);
    u_free(s->output);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream released group=%llx source=%llx epoch=%llx reason=%d",
               (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
               (unsigned long long)s->descriptor.source_epoch, s->reason);
    u_free(s);
}

/* Delivery выполняется отдельным callback, а не из source_ready при busy источнике. */
static void session_schedule(struct stream_session *s) {
    if (s->service->stopping || s->delivery) return;
    s->delivery = uasync_call_soon(s->service->inst->ua, s, session_deliver);
    if (!s->delivery) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream delivery scheduling failed source=%llx",
                                (unsigned long long)s->descriptor.source_node_id);
}

/* Снять локальные ссылки и публикацию; CLOSED также отложен и отменяется при stop всего сервиса. */
static void session_close(struct stream_session *s, int reason, int update_profile, int notify) {
    if (s->closing) return;
    s->closing = 1; s->reason = reason; s->owners = 0;
    stream_network_cancel(s->service->network, &s->descriptor);
    for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next)
        if (h->session == s) { h->session = NULL; h->reason = reason; }
    if (s->delivery) { uasync_call_soon_cancel(s->service->inst->ua, s->delivery); s->delivery = NULL; }
    struct TOPO_GROUP *group = s->service->inst->topo_groups
                               ? topo_groups_find(s->service->inst->topo_groups, s->descriptor.group_id) : NULL;
    if (update_profile && group && service_publish_options(s->service, group) != STREAM_OK)
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream closed with stale own member options group=%llx",
                   (unsigned long long)s->descriptor.group_id);
    if (notify && s->service->callback && !s->service->stopping) session_schedule(s);
    else if (!s->holds) session_free(s);
}

/* Уведомление использует собственный scratch экземпляра; close внутри callback не освобождает его преждевременно. */
static void session_deliver(void *arg) {
    struct stream_session *s = arg;
    struct stream_service *service = s->service;
    s->delivery = NULL; s->holds++;
    struct stream_event event = { .descriptor = s->descriptor };
    int emit = 0, again = 0;
    if (s->closing) { event.type = STREAM_EVENT_CLOSED; event.reason = s->reason; emit = 1; }
    else if (s->invalidated) {
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "invalidated stream awaits deferred close source=%llx",
                    (unsigned long long)s->descriptor.source_node_id);
    } else if (s->anchored) {
        struct stream_buffer_stats stats;
        stream_buffer_get_stats(s->buffer, &stats);
        size_t length = 0;
        int result = s->next_seq == UINT64_MAX ? STREAM_WAIT
                     : stream_buffer_decode(s->buffer, s->next_seq, s->output, stream_input_size(&s->descriptor), &length);
        if (result == STREAM_OK) {
            event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++;
            event.data = s->output; event.length = length; emit = 1; again = 1;
        } else if (result == STREAM_EXPIRED) {
            event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = stats.expired_before - 1;
            s->next_seq = stats.expired_before; emit = 1; again = 1;
        } else if ((result == STREAM_WAIT || result == STREAM_GAP) && s->next_seq < s->passed_before) {
            uint64_t next = s->passed_before, decoded;
            if (stream_buffer_next_decodable(s->buffer, s->next_seq + 1, &decoded) == STREAM_OK && decoded < next) next = decoded;
            event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = next - 1;
            s->next_seq = next; emit = 1; again = 1;
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream consumer GAP source=%llx first=%llu last=%llu next=%llu",
                       (unsigned long long)s->descriptor.source_node_id, (unsigned long long)event.first_seq,
                       (unsigned long long)event.last_seq, (unsigned long long)next);
        } else if (result < 0) {
            session_close(s, STREAM_CLOSE_FAILED, 1, 1);
        } else if (s->finished && s->end_valid && s->next_seq > s->last_seq && !s->end_delivered) {
            event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = s->last_seq;
            s->end_delivered = 1; emit = 1;
        }
    } else if (s->finished && s->end_valid && s->last_seq == UINT64_MAX && !s->end_delivered) {
        event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = UINT64_MAX;
        s->end_delivered = 1; emit = 1;
    }
    if (emit && service->callback && !service->stopping) {
        service->dispatching++;
        service->callback(service->inst, &event, service->arg);
        service->dispatching--;
    }
    s->holds--;
    if (s->closing) { if (!s->holds && !s->delivery) session_free(s); }
    else if (again) session_schedule(s);
}

/* Готовая source-пачка уже полностью в кольце. Пользовательский callback здесь не вызывается. */
static void source_ready(struct stream_source *source, uint64_t seq, void *arg) {
    (void)source;
    struct stream_session *s = arg;
    if (s->closing) return;
    if (!s->anchored) { s->anchored = 1; s->start_seq = s->next_seq = seq; }
    session_schedule(s);
    stream_network_changed(s->service->network);
}

/* Отложенная сверка не разрушает экземпляры из BGP/member callbacks. */
static void service_reconcile(void *arg) {
    struct stream_service *service = arg;
    service->reconcile = NULL;
    struct stream_session *snapshot = service->sessions;
    for (struct stream_session *s = snapshot; s; s = s->next) s->holds++;
    struct stream_session *s = snapshot;
    while (s) {
        struct stream_session *next = s->next;
        if (!s->closing && s->invalidated) session_close(s, s->reason, 1, 1);
        else if (!s->closing && !s->source) {
            struct stream_member_state state; uint8_t key[32];
            int result = stream_member_load(service->inst, s->descriptor.group_id, s->descriptor.source_node_id, &state, key);
            if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT)
                session_close(s, STREAM_CLOSE_MEMBER, 1, 1);
            else if (result == STREAM_OK && (memcmp(key, s->public_key, 32)
                     || (state.published && (state.source.source_epoch != s->descriptor.source_epoch
                         || state.source.n != s->descriptor.n || state.source.r != s->descriptor.r
                         || state.source.block_size != s->descriptor.block_size))))
                session_close(s, STREAM_CLOSE_SOURCE_CHANGED, 1, 1);
        }
        s = next;
    }
    s = snapshot;
    while (s) {
        struct stream_session *next = s->next;
        s->holds--;
        if (s->closing && !s->holds && !s->delivery) session_free(s);
        s = next;
    }
}

/* Общий props сигнал относится и к Merkle commit, и к BGP. Payload adm_tags здесь не используется. */
static void member_changed(uint64_t node_id, const char *tags, const char *channel, void *arg) {
    (void)tags;
    struct stream_service *service = arg;
    if (service->stopping || !service->sessions) return;
    stream_network_membership_changed(service->network, node_id, channel);
    for (struct stream_session *s = service->sessions; s; s = s->next) {
        int own = node_id == service->inst->node_id;
        if (s->closing || (!own && (s->source || s->descriptor.source_node_id != node_id))) continue;
        struct TOPO_GROUP *group = stream_group(service->inst, s->descriptor.group_id);
        if (!group || !channel || strcmp(group->channel_id, channel)) continue;
        struct stream_member_state state; uint8_t key[32];
        int result = stream_member_load(service->inst, group->group_id, node_id, &state, key);
        if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT) {
            s->invalidated = 1; s->reason = STREAM_CLOSE_MEMBER;
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream membership invalidated group=%llx member=%llx own=%d result=%d",
                       (unsigned long long)group->group_id, (unsigned long long)node_id, own, result);
        } else if (own && result == STREAM_OK && memcmp(key, service->inst->my_ed25519_pubkey, 32)) {
            s->invalidated = 1; s->reason = STREAM_CLOSE_MEMBER;
            DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream own member key changed group=%llx", (unsigned long long)group->group_id);
        } else if (!own && result == STREAM_OK && (memcmp(key, s->public_key, 32)
                   || (state.published && (state.source.source_epoch != s->descriptor.source_epoch
                       || state.source.n != s->descriptor.n || state.source.r != s->descriptor.r
                       || state.source.block_size != s->descriptor.block_size)))) {
            s->invalidated = 1; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
        }
    }
    if (service->reconcile) return;
    service->reconcile = uasync_call_soon(service->inst->ua, service, service_reconcile);
    if (!service->reconcile) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream member reconciliation scheduling failed");
}

/* Снять старые собственные опции после аварии; неполный/отсутствующий профиль не является активной подпиской. */
int stream_service_init(struct UTUN_INSTANCE *inst) {
    if (!inst || !inst->core_started || !chat_core_get_db(inst) || !inst->topo_groups) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream init requires core and chat DB"); return STREAM_ERR_STATE;
    }
    if (inst->stream) return STREAM_OK;
    struct stream_service *service = u_calloc(1, sizeof(*service));
    if (!service) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream service allocation failed"); return STREAM_ERR_MEMORY; }
    service->inst = inst; service->initializing = 1; inst->stream = service;
    for (struct ll_entry *e = inst->topo_groups->group_list->head; e; e = e->next) {
        struct TOPO_GROUP *group = (struct TOPO_GROUP *)e;
        if (group->group_type != TOPO_GROUP_TYPE_CHAT) continue;
        struct stream_member_state state;
        int result = stream_member_load(inst, group->group_id, inst->node_id, &state, NULL);
        if (result == STREAM_OK && !state.published && !state.count) continue;
        if (result == STREAM_WAIT || result == STREAM_ERR_NOT_MEMBER) continue;
        if (service_publish_options(service, group) != STREAM_OK) {
            service->initializing = 0; stream_service_destroy(inst); return STREAM_ERR_STATE;
        }
    }
    if (member_sync_add_props_cbk(inst, member_changed, service) < 0) {
        service->initializing = 0; stream_service_destroy(inst); return STREAM_ERR_MEMORY;
    }
    service->registered = 1; service->initializing = 0;
    if (stream_network_init(inst, &service->network) != STREAM_OK) { stream_service_destroy(inst); return STREAM_ERR_STATE; }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream service initialized node=%llx", (unsigned long long)inst->node_id);
    return STREAM_OK;
}

/* Detached handles принадлежат callers и остаются безопасными для close после остановки чата/ядра. */
void stream_service_destroy(struct UTUN_INSTANCE *inst) {
    struct stream_service *service = inst ? inst->stream : NULL;
    if (!service) return;
    if (service->dispatching || service->initializing) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream service destroy inside callback/init rejected"); return;
    }
    for (struct stream_session *s = service->sessions; s; s = s->next)
        if (s->holds) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream stop during active operation rejected"); return; }
    service->stopping = 1;
    stream_network_destroy(service->network); service->network = NULL;
    if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service);
    if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile);
    while (service->sessions) {
        struct stream_session *s = service->sessions;
        if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0);
        else session_free(s);
    }
    for (struct STREAM_HANDLE *h = service->handles; h; h = h->next) { h->service = NULL; h->session = NULL; h->reason = STREAM_CLOSE_SERVICE; }
    inst->stream = NULL;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream service stopped node=%llx", (unsigned long long)inst->node_id);
    u_free(service);
}

void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) {
    struct stream_service *service = inst ? inst->stream : NULL;
    if (!service) return;
    struct stream_session *s = service->sessions;
    while (s) {
        struct stream_session *next = s->next;
        if (!s->closing && s->descriptor.group_id == group_id) session_close(s, STREAM_CLOSE_GROUP, 1, 0);
        s = next;
    }
}

void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callback, void *arg) {
    if (!inst || !inst->stream || inst->stream->stopping) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream callback before init"); return; }
    inst->stream->callback = callback; inst->stream->arg = arg;
    if (callback) for (struct stream_session *s = inst->stream->sessions; s; s = s->next)
        if (!s->closing) session_schedule(s);
}

/* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */
static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher,
                         struct STREAM_HANDLE **out) {
    struct stream_service *service = inst->stream;
    struct stream_session *s = session_find(service, d->group_id, d->source_node_id);
    int created = s == NULL;
    if (s && (publisher || s->descriptor.source_epoch != d->source_epoch)) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source already owned with another publication/epoch"); return STREAM_ERR_STATE;
    }
    struct STREAM_HANDLE *h = u_calloc(1, sizeof(*h));
    if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream handle allocation failed"); return STREAM_ERR_MEMORY; }
    if (created) {
        s = u_calloc(1, sizeof(*s));
        if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session allocation failed"); u_free(h); return STREAM_ERR_MEMORY; }
        s->service = service; s->descriptor = *d; memcpy(s->public_key, key, 32);
        s->output = u_malloc((uint32_t)stream_input_size(d));
        int result = s->output ? STREAM_OK : STREAM_ERR_MEMORY;
        if (result == STREAM_OK && publisher) {
            result = stream_source_create(&s->source, inst->ua, d, inst->my_ed25519_privkey, source_ready, s);
            if (result == STREAM_OK) s->buffer = stream_source_buffer(s->source);
        } else if (result == STREAM_OK) result = stream_buffer_create(&s->buffer, d, key, STREAM_BUFFER_BYTES);
        s->next = service->sessions; service->sessions = s;
        if (result != STREAM_OK) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session initialization failed result=%d", result);
            session_free(s); u_free(h); return result;
        }
    }
    h->service = service; h->session = s; h->publisher = publisher;
    h->next = service->handles; service->handles = h; s->owners++;
    s->holds++;
    int result = created ? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK;
    if (s->closing && result == STREAM_OK) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream open cancelled during member callbacks"); result = STREAM_ERR_STATE;
    }
    if (result != STREAM_OK) {
        struct STREAM_HANDLE **link = &service->handles;
        while (*link != h) link = &(*link)->next;
        *link = h->next;
        if (!s->closing) { s->owners--; session_close(s, STREAM_CLOSE_FAILED, 0, 0); }
        u_free(h); s->holds--;
        if (!s->holds && !s->delivery) session_free(s);
        return result;
    }
    s->holds--;
    *out = h;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream handle opened group=%llx source=%llx epoch=%llx publisher=%d owners=%u",
               (unsigned long long)d->group_id, (unsigned long long)d->source_node_id,
               (unsigned long long)d->source_epoch, publisher, s->owners);
    return STREAM_OK;
}

/* Контекст не создаётся для неизвестной группы, неавторизованного self или неподтверждённого ключа. */
static int stream_open_check(struct UTUN_INSTANCE *inst, uint64_t group_id, struct STREAM_HANDLE **out) {
    if (!out || *out || !inst || !inst->chat_started || !inst->stream || inst->stream->stopping || !stream_group(inst, group_id)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream open invalid state/group=%llx", (unsigned long long)group_id); return STREAM_ERR_STATE;
    }
    struct stream_member_state own; uint8_t key[32];
    int result = stream_member_load(inst, group_id, inst->node_id, &own, key);
    if (result != STREAM_OK) return result;
    if (memcmp(key, inst->my_ed25519_pubkey, 32)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream own member key differs from core identity"); return STREAM_ERR_SIGNATURE;
    }
    return STREAM_OK;
}

int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const struct stream_parameters *parameters,
                        struct STREAM_HANDLE **out) {
    int result = stream_open_check(inst, group_id, out);
    if (result != STREAM_OK) return result;
    struct stream_descriptor d = {.group_id = group_id, .source_node_id = inst->node_id,
        .n = parameters ? parameters->n : 8, .r = parameters ? parameters->r : 2,
        .block_size = parameters ? parameters->block_size : STREAM_RS_DEFAULT_SIZE};
    do {
        if (random_bytes((uint8_t *)&d.source_epoch, sizeof(d.source_epoch)) != 0) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source epoch generation failed"); return STREAM_ERR_CRYPTO;
        }
    } while (!d.source_epoch);
    if (stream_descriptor_validate(&d) != STREAM_OK) return STREAM_ERR_ARGUMENT;
    return session_open(inst, &d, inst->my_ed25519_pubkey, 1, out);
}

int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) {
    int result = stream_open_check(inst, group_id, out);
    if (result != STREAM_OK) return result;
    struct stream_session *s = session_find(inst->stream, group_id, source_id);
    if (s) return session_open(inst, &s->descriptor, s->public_key, 0, out);
    struct stream_member_state state; uint8_t key[32];
    result = stream_member_load(inst, group_id, source_id, &state, key);
    if (result != STREAM_OK) return result;
    if (!state.published) {
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id);
        return STREAM_WAIT;
    }
    return session_open(inst, &state.source, key, 0, out);
}

/* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */
void stream_close(struct STREAM_HANDLE *h) {
    if (!h) return;
    struct stream_service *service = h->service;
    if (service) {
        struct STREAM_HANDLE **link = &service->handles;
        while (*link != h) link = &(*link)->next;
        *link = h->next;
    }
    struct stream_session *s = h->session;
    if (s && (h->publisher || --s->owners == 0)) session_close(s, STREAM_CLOSE_STOPPED, 1, 1);
    u_free(h);
}

int stream_handle_active(const struct STREAM_HANDLE *h) { return h && h->session && !h->session->closing && !h->session->invalidated; }
int stream_handle_reason(const struct STREAM_HANDLE *h) {
    return h && h->session && h->session->invalidated ? h->session->reason : h ? h->reason : STREAM_CLOSE_STOPPED;
}

int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) {
    if (!info || !stream_handle_active(h)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream info detached/invalid handle"); return STREAM_ERR_STATE; }
    struct stream_session *s = h->session;
    struct stream_info result = {.descriptor = s->descriptor, .owners = s->owners, .publisher = s->source != NULL, .finished = s->finished};
    stream_buffer_get_stats(s->buffer, &result.buffer);
    *info = result;
    return STREAM_OK;
}

size_t stream_publish_capacity(const struct STREAM_HANDLE *h) {
    if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream capacity not a publisher"); return 0; }
    return stream_source_input_capacity(h->session->source);
}

int stream_publish_write(struct STREAM_HANDLE *h, const uint8_t *data, size_t length, size_t *accepted) {
    if (accepted) *accepted = 0;
    if (!stream_handle_active(h) || !h->publisher || !accepted) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream write not an active publisher"); return STREAM_ERR_STATE;
    }
    struct stream_session *s = h->session; s->holds++;
    int result = stream_source_write(s->source, data, length, accepted);
    if (result != STREAM_OK && result != STREAM_ERR_ARGUMENT) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
    s->holds--;
    if (s->closing && !s->holds && !s->delivery) session_free(s);
    return result;
}

int stream_publish_finish(struct STREAM_HANDLE *h) {
    if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream finish not a publisher"); return STREAM_ERR_STATE; }
    struct stream_session *s = h->session;
    if (s->finished) return STREAM_OK;
    s->holds++;
    s->finished = 1;
    int result = service_publish_options(s->service, stream_group(s->service->inst, s->descriptor.group_id));
    if (result != STREAM_OK || s->closing) {
        if (!s->closing) s->finished = 0;
        s->holds--;
        if (s->closing && !s->holds && !s->delivery) session_free(s);
        return result != STREAM_OK ? result : STREAM_ERR_STATE;
    }
    result = stream_source_finish(s->source);
    if (result == STREAM_OK) {
        struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
        s->last_seq = stats.blocks ? stats.highest_seq : UINT64_MAX;
        result = stream_end_sign(&s->descriptor, s->last_seq, s->service->inst->my_ed25519_privkey, s->end);
        s->end_valid = result == STREAM_OK;
    }
    if (result != STREAM_OK) session_close(s, STREAM_CLOSE_FAILED, 1, 1);
    else { session_schedule(s); stream_network_changed(s->service->network); }
    s->holds--;
    if (s->closing && !s->holds && !s->delivery) session_free(s);
    return result;
}

/* Сначала только ключ/наличие подписки, затем crypto/FEC. Парсер пути и peer authorization — выше этого входа. */
int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length) {
    struct stream_path path = {.count = 1};
    if (wire && length >= STREAM_BLOCK_RS_OFF) path.nodes[0] = stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8);
    return stream_ingest_path(inst, wire, length, &path);
}

/* Кольцо хранит уже дополненный путь: источник ... отправитель ... этот узел. */
int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length, const struct stream_path *path) {
    if (!inst || !wire || length < STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream ingest invalid length=%zu", length); return STREAM_ERR_ARGUMENT;
    }
    if (!inst->stream || inst->stream->stopping) return STREAM_WAIT;
    uint64_t group = stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8), source = stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8);
    struct stream_session *s = session_find(inst->stream, group, source);
    if (!s) { DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "unsubscribed stream ignored group=%llx source=%llx",
                         (unsigned long long)group, (unsigned long long)source); return STREAM_WAIT; }
    if (s->source) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "incoming DATA to own publisher rejected"); return STREAM_ERR_STATE; }
    if (!path || !path->count || path->count >= STREAM_PATH_MAX
        || stream_path_validate(path, source, path->nodes[path->count - 1], inst->node_id) != STREAM_OK)
        return STREAM_ERR_FORMAT;
    if (s->end_valid && (s->last_seq == UINT64_MAX || stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8) > s->last_seq)) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA beyond signed END source=%llx", (unsigned long long)source);
        return STREAM_ERR_CONFLICT;
    }
    struct stream_path retained = *path; retained.nodes[retained.count++] = inst->node_id;
    int result = stream_buffer_put_path(s->buffer, wire, length, &retained);
    if (result == STREAM_OK || (result == STREAM_DUPLICATE && !s->anchored)) {
        if (!s->anchored) { s->anchored = 1; s->start_seq = s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); }
        session_schedule(s);
        stream_network_changed(s->service->network);
    }
    return result;
}

struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst) { return inst && inst->stream ? inst->stream->network : NULL; }

/* Доступ network только к существующей локальной подписке точной эпохи. */
int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local) {
    if (!key || !local || !inst || !inst->stream || inst->stream->stopping) return STREAM_WAIT;
    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
    if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT;
    *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq,
        .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, .end = s->end_valid ? s->end : NULL};
    return STREAM_OK;
}

/* Выданный локальный handle проверяется до создания сетевых ресурсов. */
int stream_handle_context(struct STREAM_HANDLE *h, struct UTUN_INSTANCE **inst, struct stream_descriptor *d) {
    if (!inst || !d || !stream_handle_active(h) || h->session->source) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active reader handle"); return STREAM_ERR_STATE;
    }
    *inst = h->service->inst; *d = h->session->descriptor; return STREAM_OK;
}

/* END ретранслируется без изменений и не может завершить поток по заявлению поставщика без подписи источника. */
int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, const uint8_t *wire, size_t length) {
    struct stream_local local;
    if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) return STREAM_WAIT;
    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
    uint64_t last; int result = stream_end_verify(&s->descriptor, s->public_key, wire, length, &last);
    if (result != STREAM_OK) return result;
    struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats);
    if ((s->end_valid && s->last_seq != last) || (stats.blocks && (last == UINT64_MAX || stats.highest_seq > last))) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "signed END conflicts with retained stream source=%llx", (unsigned long long)key->source_node_id);
        return STREAM_ERR_CONFLICT;
    }
    memcpy(s->end, wire, STREAM_END_SIZE); s->end_valid = s->finished = 1; s->last_seq = last;
    session_schedule(s); stream_network_changed(inst->stream->network); return STREAM_OK;
}

/* Все выбранные upstream прошли эту границу; дальше старую неполную пачку они уже не выдадут. */
void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t before) {
    struct stream_local local;
    if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) return;
    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
    if (before > s->passed_before) { s->passed_before = before; session_schedule(s); }
}

/* Явный новый старт не смешивается с DATA старых upstream handles; кольцо остаётся пригодным для relay. */
int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key) {
    struct stream_local local;
    if (stream_local_get(inst, key, &local) != STREAM_OK || local.publisher) {
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream restart requires local receiver"); return STREAM_ERR_STATE;
    }
    struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
    if (s->delivery) { uasync_call_soon_cancel(inst->ua, s->delivery); s->delivery = NULL; }
    s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u",
               (unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
               (unsigned long long)key->source_epoch, local.buffer != NULL); return STREAM_OK;
}

/* Выборка идёт из live BGP-путей группы, не из nodes.online или глобальных соединений. */
int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate **out, size_t *count) {
    if (!out || *out || !count || !stream_handle_active(h)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates invalid handle/output"); return STREAM_ERR_ARGUMENT;
    }
    struct stream_session *s = h->session;
    struct UTUN_INSTANCE *inst = s->service->inst;
    struct TOPO_GROUP *group = stream_group(inst, s->descriptor.group_id);
    if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates group missing"); return STREAM_ERR_STATE; }
    size_t capacity = (size_t)group->nodes->count;
    struct stream_candidate *candidates = capacity ? u_calloc((uint32_t)capacity, sizeof(*candidates)) : NULL;
    if (capacity && !candidates) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream candidates allocation failed count=%zu", capacity);
        return STREAM_ERR_MEMORY;
    }
    size_t found = 0;
    for (struct ll_entry *e = group->nodes->head; e; e = e->next) {
        struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e;
        if (node->node_id == inst->node_id || !stream_member_online(inst, group->group_id, node->node_id)) continue;
        struct TOPO_NODE *identity = topo_node_registry_find(inst->topo_groups, node->node_id);
        int source = node->node_id == s->descriptor.source_node_id;
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1",
                    (unsigned long long)node->node_id, identity ? identity->client_type : UINT8_MAX, source);
        if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) {
            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay",
                        (unsigned long long)node->node_id); continue;
        }
        struct stream_member_state state;
        if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK
            || !stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch)) continue;
        candidates[found++] = (struct stream_candidate){ .node_id = node->node_id, .source = (uint8_t)source,
            .client_type = identity ? identity->client_type : UINT8_MAX };
    }
    if (!found) { u_free(candidates); candidates = NULL; }
    *out = candidates; *count = found;
    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream candidates source=%llx eligible=%zu BGP=%zu",
                (unsigned long long)s->descriptor.source_node_id, found, capacity);
    return STREAM_OK;
}