#include <string.h>

#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../utun_instance.h"
#include "../transport_layer/node_conn_direct.h"
#include "stream_optimizer.h"
#include "stream_priv.h"

#define STREAM_PROBE_BURST 32
#define STREAM_PROBE_PERIOD_MS 1000
#define STREAM_PROBE_CYCLE_MS 30000
#define STREAM_OFFER_TTL_MS 60000
#define STREAM_MIGRATION_TIMEOUT_MS 5000
#define STREAM_BOOTSTRAP_RELAY_MS 100
#define STREAM_AVAILABILITY_RETRY_MS 100
#define STREAM_AVAILABILITY_WAIT_MS 5000
#define STREAM_PROBLEM_PENALTY_MS 1000
#define STREAM_HOP_PENALTY_MS 3

static uint64_t problem_penalty(const struct stream_metric_snapshot *metrics, int valid);

struct stream_offer {
    struct stream_receive_info info;
    uint64_t retry_ms;
    unsigned failures;
};
struct stream_supplier {
    struct stream_supplier *next;
    uint64_t node, cycle, retry_ms, availability_retry_ms, availability_until_ms;
    uint64_t local_epoch, peer_epoch;
    struct STREAM_RECEIVE *probe;
    struct stream_offer *offers;
    unsigned probe_sub, failures;
    int online, compatible, source, relay;
};
struct stream_selection {
    struct STREAM_RECEIVE *current, *replacement;
    uint64_t opened_ms, replacement_ms, better_ms, better_node;
    int opened;
};
struct stream_optimizer {
    struct UTUN_INSTANCE *inst;
    struct stream_descriptor descriptor;
    struct stream_supplier *suppliers;
    struct stream_selection *selected;
    struct stream_selection_settings settings;
    struct stream_selection_info info;
    struct stream_probe_schedule schedule;
    void *startup_timer;
    uint64_t started_ms, startup_peer;
    uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms, deficient_ms, relay_poll_ms;
    int progress_valid, draining, startup_complete;
};

/* Точный срок гонки не зависит от фазы общего 100-мс таймера; selector остаётся отложенным. */
static void startup_timeout(void *arg) {
    struct stream_optimizer *o = arg; o->startup_timer = NULL;
    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream bootstrap relay deadline group=%llx source=%llx epoch=%llx elapsed_ms=%llu delay_ms=%u",
        (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
        (unsigned long long)o->descriptor.source_epoch, (unsigned long long)(get_time_tb() / 10 - o->started_ms),
        STREAM_BOOTSTRAP_RELAY_MS);
    stream_network_wake(stream_network_get(o->inst));
}

unsigned stream_probe_schedule_tick(struct stream_probe_schedule *s, uint64_t now, unsigned pending, unsigned active) {
    if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream probe schedule missing"); return 0; }
    unsigned budget = 1;
    if (!s->cycle) { s->cycle = 1; s->started_ms = now; budget = STREAM_PROBE_BURST; }
    else if (now < s->next_ms) return 0;
    else if (!pending && !active && now - s->started_ms >= STREAM_PROBE_CYCLE_MS) { s->cycle++; s->started_ms = now; }
    s->next_ms = now + STREAM_PROBE_PERIOD_MS; return budget;
}

static struct stream_supplier *supplier_find(struct stream_optimizer *o, uint64_t node) {
    for (struct stream_supplier *s = o->suppliers; s; s = s->next) if (s->node == node) return s;
    return NULL;
}

static uint64_t backoff(unsigned failures) {
    return (uint64_t)STREAM_PROBE_CYCLE_MS << (failures > 4 ? 4 : failures - 1);
}

static void roster_update(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
    struct stream_candidate *candidates = NULL; size_t count = 0;
    if (stream_get_candidates(h, &candidates, &count) != STREAM_OK) return;
    for (struct stream_supplier *s = o->suppliers; s; s = s->next) s->online = 0;
    for (size_t i = 0; i < count; i++) {
        struct stream_supplier *s = supplier_find(o, candidates[i].node_id);
        if (!s) {
            s = u_calloc(1, sizeof(*s));
            if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer candidate allocation failed"); continue; }
            s->node = candidates[i].node_id; s->next = o->suppliers; o->suppliers = s;
        }
        s->online = 1; s->source = candidates[i].source;
        s->relay = stream_relay_confirmed(o->inst, &o->descriptor, s->node);
        int compatible = node_conn_direct_compatible(o->inst, s->node);
        if (compatible && !s->compatible && s->cycle != o->schedule.cycle && now >= s->retry_ms) {
            o->schedule.next_ms = 0;
            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer new supplier wakes probe group=%llx source=%llx peer=%llx",
                        (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
                        (unsigned long long)s->node);
        }
        s->compatible = compatible;
    }
    u_free(candidates);
    struct stream_supplier **link = &o->suppliers;
    while (*link) {
        struct stream_supplier *s = *link;
        if (s->online) { link = &s->next; continue; }
        if (s->probe) stream_receive_close(s->probe);
        *link = s->next; u_free(s->offers); u_free(s);
    }
    o->roster_ms = now + STREAM_PROBE_PERIOD_MS;
}

/* Один NCD owner пира переиспользуется всеми пробами/выдачами, INFO не открывает DATA. */
static int probe_start(struct stream_optimizer *o, struct STREAM_HANDLE *h, struct stream_supplier *s, uint64_t now) {
    s->cycle = o->schedule.cycle; s->probe_sub = 0;
    if (!s->offers) s->offers = u_calloc(o->descriptor.n, sizeof(*s->offers));
    if (!s->offers) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer offers allocation failed"); return 0; }
    int result = stream_probe_open(h, s->node, 0, &s->probe);
    if (result != STREAM_OK) {
        s->failures++; s->retry_ms = now + backoff(s->failures);
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe refused peer=%llx result=%d retry_ms=%llu",
                   (unsigned long long)s->node, result, (unsigned long long)s->retry_ms); return 0;
    }
    o->info.probes_started++;
    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer probe peer=%llx cycle=%llu ordinal=%llu source=%d",
                (unsigned long long)s->node, (unsigned long long)o->schedule.cycle,
                (unsigned long long)o->info.probes_started, s->source); return 1;
}

static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
    for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
        if (!s->probe) continue;
        struct stream_receive_info info; stream_receive_get_info(s->probe, &info);
        if (!info.info_ready && info.state != STREAM_RECEIVE_CLOSED) continue;
        if (info.state == STREAM_RECEIVE_CLOSED) {
            s->offers[s->probe_sub].info.info_ms = 0;
            if (info.reason == STREAM_REASON_POSITION) {
                if (!s->availability_until_ms) s->availability_until_ms = now + STREAM_AVAILABILITY_WAIT_MS;
                if (now < s->availability_until_ms && !s->availability_retry_ms) {
                    s->availability_retry_ms = now + STREAM_AVAILABILITY_RETRY_MS;
                    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream supplier buffer pending source=%llx peer=%llx sub=%u retry_ms=%u",
                                (unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node,
                                s->probe_sub, STREAM_AVAILABILITY_RETRY_MS);
                }
            }
            if (info.reason == STREAM_REASON_TRANSPORT || info.reason == STREAM_REASON_MEMBER || info.reason == STREAM_REASON_PROTOCOL) {
                s->failures++; s->retry_ms = now + backoff(s->failures);
                DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe closed source=%llx peer=%llx sub=%u reason=%d"
                    " info_ready=%d local_epoch=%llx peer_epoch=%llx failures=%u retry_in_ms=%llu cycle=%llu",
                    (unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->probe_sub, info.reason,
                    info.info_ready, (unsigned long long)info.local_epoch, (unsigned long long)info.peer_epoch, s->failures,
                    (unsigned long long)(s->retry_ms - now), (unsigned long long)s->cycle);
                stream_receive_close(s->probe); s->probe = NULL; continue;
            }
        } else {
            s->offers[s->probe_sub].info = info; s->failures = 0; s->retry_ms = 0;
            s->local_epoch = info.local_epoch; s->peer_epoch = info.peer_epoch;
            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer offer peer=%llx sub=%u range=%llu..%llu hops=%u"
                " rtt=%u/%d chain=%u/%d slots=%u accepting=%d lag=%llu stopped=%d age=%llu/%d queue=%llu penalty_ms=%llu score_ms=%llu",
                (unsigned long long)s->node, s->probe_sub, (unsigned long long)info.first_available,
                (unsigned long long)info.last_available, info.path.count, info.rtt_ms, info.rtt_valid,
                info.path_rtt_ms, info.path_rtt_valid, info.free_slots, info.accepting,
                (unsigned long long)info.supplier_metrics.lag_ms, info.supplier_metrics.stopped,
                (unsigned long long)info.progress_age_ms, info.progress_valid, (unsigned long long)info.supplier_queue_bytes,
                (unsigned long long)problem_penalty(&info.supplier_metrics, info.supplier_metrics_valid),
                (unsigned long long)stream_supplier_score(&info, &info.supplier_metrics, info.supplier_metrics_valid));
        }
        unsigned next = s->probe_sub + 1;
        if (next == o->descriptor.n) { stream_receive_close(s->probe); s->probe = NULL; continue; }
        if (info.state == STREAM_RECEIVE_CLOSED) {
            stream_receive_close(s->probe); s->probe = NULL;
            if (stream_probe_open(h, s->node, (uint8_t)next, &s->probe) != STREAM_OK) {
                s->failures++; s->retry_ms = now + backoff(s->failures); continue;
            }
        } else if (stream_probe_next(s->probe, (uint8_t)next) != STREAM_OK) continue;
        s->probe_sub = next;
    }
}

static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
    unsigned pending = 0, active = 0;
    for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
        if (!s->probe && s->availability_retry_ms && now >= s->availability_retry_ms) {
            s->availability_retry_ms = 0;
            if (s->compatible && now < s->availability_until_ms && now >= s->retry_ms) {
                s->cycle = 0; o->schedule.next_ms = 0;
            }
        }
        if (s->probe) active++;
        if (now < s->retry_ms) s->cycle = o->schedule.cycle;
        if (s->compatible && s->cycle != o->schedule.cycle) pending++;
    }
    uint64_t previous = o->schedule.cycle;
    unsigned budget = stream_probe_schedule_tick(&o->schedule, now, pending, active);
    o->info.cycles = o->schedule.cycle;
    if (o->schedule.cycle != previous) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer cycle=%llu",
                                                 (unsigned long long)o->schedule.cycle);
    for (unsigned i = 0; i < budget; i++) {
        struct stream_supplier *chosen = NULL;
        for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
            if (!s->compatible || s->cycle == o->schedule.cycle || s->probe || now < s->retry_ms) continue;
            if (!chosen || s->relay > chosen->relay || (s->relay == chosen->relay && (s->source > chosen->source
                || (s->source == chosen->source && s->node < chosen->node)))) chosen = s;
        }
        if (!chosen) break;
        probe_start(o, h, chosen, now);
    }
}

static unsigned usage(struct stream_optimizer *o, uint64_t node, unsigned except) {
    unsigned count = 0;
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        struct stream_receive_info info;
        if (sub == except || !o->selected[sub].current) continue;
        stream_receive_get_info(o->selected[sub].current, &info);
        if (info.state != STREAM_RECEIVE_CLOSED && info.peer == node) count++;
    }
    return count;
}

/* Доля поздних блоков задаёт штраф в мс; текущая остановка/lag дают максимум, история остановок добавляет 10 мс. */
static uint64_t problem_penalty(const struct stream_metric_snapshot *m, int valid) {
    if (!valid) return 0;
    if (m->stopped || m->lag_ms > STREAM_LATE_MS) return STREAM_PROBLEM_PENALTY_MS;
    double fraction = m->blocks ? (double)m->late_blocks / (double)m->blocks : 0;
    double penalty = fraction * STREAM_PROBLEM_PENALTY_MS + (m->stops ? 10 : 0);
    return penalty >= STREAM_PROBLEM_PENALTY_MS ? STREAM_PROBLEM_PENALTY_MS : (uint64_t)penalty;
}

/* path.count включает автора и поставщика: с последним переходом к нам это число hops от автора до получателя. */
uint64_t stream_supplier_score(const struct stream_receive_info *info, const struct stream_metric_snapshot *metrics, int valid) {
    if (!info || !metrics) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream supplier score invalid arguments"); return UINT64_MAX; }
    if (!info->rtt_valid || !info->path_rtt_valid) return UINT64_MAX;
    return (uint64_t)info->path_rtt_ms + 2 * (uint64_t)info->rtt_ms + STREAM_HOP_PENALTY_MS * info->path.count
           + problem_penalty(metrics, valid);
}

static unsigned shared_path(struct stream_optimizer *o, const struct stream_receive_info *offer, unsigned sub) {
    unsigned shared = 0;
    for (unsigned other = 0; other < o->descriptor.n; other++) {
        struct stream_receive_info info;
        if (other == sub || !o->selected[other].current) continue;
        stream_receive_get_info(o->selected[other].current, &info);
        for (unsigned i = 1; i < offer->path.count; i++)
            for (unsigned j = 1; j < info.path.count; j++) if (offer->path.nodes[i] == info.path.nodes[j]) shared++;
    }
    return shared;
}

/* Доступность позиции/путь уже проверены INFO. Ограничение r ослабляется только при отсутствии подходящей альтернативы. */
static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsigned sub, uint64_t seq, uint64_t exclude, uint64_t now) {
    struct stream_supplier *best = NULL;
    for (unsigned pass = 0; pass < 2 && !best; pass++) {
        for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
            if (!s->online || !s->compatible || s->node == exclude || !s->offers || now < s->retry_ms
                || (s->probe && s->probe_sub == sub)) continue;
            struct stream_offer *offer = &s->offers[sub]; struct stream_receive_info *info = &offer->info;
            if (now < offer->retry_ms || !info->info_ms || now - info->info_ms > STREAM_OFFER_TTL_MS || !info->accepting || !info->free_slots
                || (seq != UINT64_MAX && info->first_available != UINT64_MAX && seq < info->first_available)) continue;
            if (!pass && o->descriptor.r && usage(o, s->node, sub) >= o->descriptor.r) continue;
            if (!best) { best = s; continue; }
            struct stream_receive_info *previous = &best->offers[sub].info;
            if (!o->startup_complete && seq == UINT64_MAX) {
                if (info->info_order < previous->info_order) best = s;
                continue;
            }
            uint64_t score = stream_supplier_score(info, &info->supplier_metrics, info->supplier_metrics_valid);
            uint64_t old_score = stream_supplier_score(previous, &previous->supplier_metrics, previous->supplier_metrics_valid);
            int better = score < old_score;
            if (score == old_score) {
                unsigned shared = shared_path(o, info, sub), old_shared = shared_path(o, previous, sub);
                better = shared < old_shared || (shared == old_shared && s->node < best->node);
            }
            if (better) best = s;
        }
    }
    return best;
}

/* Первый пригодный INFO среди всех субпотоков выигрывает гонку; порядок приёма не зависит от порядка списка пиров. */
static struct stream_supplier *startup_choose(struct stream_optimizer *o, uint64_t now, unsigned *subblock) {
    struct stream_supplier *first = NULL; uint64_t order = UINT64_MAX;
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        struct stream_supplier *s = supplier_choose(o, sub, UINT64_MAX, 0, now);
        if (s && s->offers[sub].info.info_order < order) {
            first = s; order = s->offers[sub].info.info_order; *subblock = sub;
        }
    }
    return first;
}

static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **handle, unsigned sub, uint64_t now) {
    struct stream_receive_info info; stream_receive_get_info(*handle, &info);
    struct stream_supplier *s = supplier_find(o, info.peer);
    if (s && s->offers) {
        struct stream_offer *offer = &s->offers[sub];
        int resynchronized = info.reason == STREAM_REASON_TRANSPORT && info.local_epoch && info.peer_epoch
            && s->local_epoch && s->peer_epoch && (info.local_epoch != s->local_epoch || info.peer_epoch != s->peer_epoch);
        if (resynchronized) { offer->failures = 0; offer->retry_ms = 0; }
        else { offer->failures++; offer->retry_ms = now + backoff(offer->failures); }
        if (info.reason == STREAM_REASON_OVERLOAD) s->retry_ms = offer->retry_ms;
    }
    o->info.failures++;
    DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer supplier failed peer=%llx sub=%u reason=%d next=%llu",
               (unsigned long long)info.peer, sub, info.reason, (unsigned long long)info.next);
    stream_receive_close(*handle); *handle = NULL;
}

static int improved(struct stream_optimizer *o, unsigned sub, struct stream_supplier *s, const struct stream_receive_info *current) {
    struct stream_receive_info *offer = &s->offers[sub].info;
    uint64_t before = stream_supplier_score(current, &current->metrics, current->metrics_valid);
    uint64_t after = stream_supplier_score(offer, &offer->supplier_metrics, offer->supplier_metrics_valid);
    if (after < before && (before == UINT64_MAX || before - after >= o->settings.score_hysteresis_ms)) return 1;
    return o->descriptor.r && usage(o, current->peer, sub) >= o->descriptor.r
           && usage(o, s->node, sub) < o->descriptor.r
           && (after <= before || after - before <= o->settings.score_hysteresis_ms);
}

int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, struct stream_optimizer **out) {
    if (!inst || !d || !out || *out) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer create invalid arguments"); return STREAM_ERR_ARGUMENT; }
    struct stream_optimizer *o = u_calloc(1, sizeof(*o));
    if (o) o->selected = u_calloc(d->n, sizeof(*o->selected));
    if (!o || !o->selected) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer allocation failed"); if (o) u_free(o); return STREAM_ERR_MEMORY; }
    o->inst = inst; o->descriptor = *d; o->settings = (struct stream_selection_settings){2000, 5000, 20};
    o->started_ms = get_time_tb() / 10;
    o->startup_timer = uasync_set_timeout(inst->ua, STREAM_BOOTSTRAP_RELAY_MS * 10, o, startup_timeout, "stream_bootstrap");
    if (!o->startup_timer) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bootstrap timer allocation failed");
        u_free(o->selected); u_free(o); return STREAM_ERR_MEMORY;
    }
    o->info.automatic = 1; *out = o; return STREAM_OK;
}

/* Освобождает основные и заменяющие выдачи; повторный вызов безопасен, статистика истории сохраняется. */
static unsigned selections_close(struct stream_optimizer *o) {
    unsigned closed = 0;
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        struct stream_selection *pick = &o->selected[sub];
        closed += (pick->current != NULL) + (pick->replacement != NULL);
        stream_receive_close(pick->current); stream_receive_close(pick->replacement);
        memset(pick, 0, sizeof(*pick));
    }
    o->info.subflows = o->info.overlaps = o->info.providers = o->info.protected_by_fec = 0;
    return closed;
}

void stream_optimizer_destroy(struct stream_optimizer *o) {
    if (!o) return;
    if (o->startup_timer) uasync_cancel_timeout(o->inst->ua, o->startup_timer);
    selections_close(o);
    while (o->suppliers) {
        struct stream_supplier *s = o->suppliers; o->suppliers = s->next;
        stream_receive_close(s->probe); u_free(s->offers); u_free(s);
    }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer stopped source=%llx probes=%llu migrations=%llu failures=%llu restarts=%llu",
               (unsigned long long)o->descriptor.source_node_id, (unsigned long long)o->info.probes_started,
               (unsigned long long)o->info.migrations, (unsigned long long)o->info.failures, (unsigned long long)o->info.restarts);
    u_free(o->selected); u_free(o);
}

int stream_optimizer_configure(struct stream_optimizer *o, const struct stream_selection_settings *settings) {
    if (!o || !settings || settings->migration_confirm_ms > 60000 || settings->minimum_hold_ms > 60000
        || settings->score_hysteresis_ms > 60000) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer settings invalid"); return STREAM_ERR_ARGUMENT;
    }
    o->settings = *settings; return STREAM_OK;
}

void stream_optimizer_info(struct stream_optimizer *o, struct stream_selection_info *info) { *info = o->info; }

/* Несколько изменений до одного прохода объединяются; только новый доступный поставщик ускоряет расписание пробы. */
void stream_optimizer_roster_changed(struct stream_optimizer *o, uint64_t peer) {
    if (o->roster_ms) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer roster invalidated group=%llx source=%llx peer=%llx",
        (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer);
    o->roster_ms = 0;
}

/* READY отменяет ожидание старого цикла/ошибки только для подтверждённого поставщика. */
void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) {
    o->roster_ms = o->schedule.next_ms = 0;
    struct stream_supplier *s = supplier_find(o, peer);
    if (s) {
        s->cycle = s->retry_ms = 0; s->failures = 0;
        if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) s->offers[sub].retry_ms = 0;
    }
    DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer relay wake source=%llx peer=%llx",
                (unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer);
}

void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
    struct stream_local local;
    if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return;
    if (local.decoded && !o->startup_complete) {
        o->startup_complete = 1;
        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap complete group=%llx source=%llx epoch=%llx peer=%llx elapsed_ms=%llu start=%llu",
            (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
            (unsigned long long)o->descriptor.source_epoch, (unsigned long long)o->startup_peer,
            (unsigned long long)(now - o->started_ms), (unsigned long long)local.start);
    }
    if (local.finished) {
        /* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */
        if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) {
            stream_relay_demand(o->inst, &o->descriptor, 0, now);
            unsigned closed = selections_close(o);
            for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; }
            if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; }
            if (closed || o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM,
                "stream optimizer END drained source=%llx next=%llu last=%llu released=%u probes=%u",
                (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next,
                (unsigned long long)local.last, closed, o->info.probes_active);
            o->draining = 0; o->info.probes_active = 0; return;
        }
        if (!o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer draining END source=%llx next=%llu last=%llu",
            (unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last);
        o->draining = 1;
    }
    if (now >= o->roster_ms) roster_update(o, h, now);
    probes_poll(o, h, now); probes_schedule(o, h, now);
    unsigned busy = 0, received = 0;
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        struct stream_selection *pick = &o->selected[sub]; struct stream_receive_info info;
        if (pick->current) {
            stream_receive_get_info(pick->current, &info);
            if (info.state == STREAM_RECEIVE_CLOSED || !supplier_find(o, info.peer)) receive_failed(o, &pick->current, sub, now);
            else {
                if (info.blocks && info.metrics_valid && !info.metrics.stopped) received++;
                busy++;
                struct stream_supplier *supplier = supplier_find(o, info.peer);
                if (supplier->offers && info.metrics_valid) {
                    supplier->offers[sub].info.supplier_metrics = info.metrics;
                    supplier->offers[sub].info.supplier_metrics_valid = 1;
                    supplier->offers[sub].info.last_available = info.next - 1;
                }
            }
        }
        if (!pick->replacement) continue;
        struct stream_receive_info replacement; stream_receive_get_info(pick->replacement, &replacement);
        if (replacement.state == STREAM_RECEIVE_CLOSED || !supplier_find(o, replacement.peer)
            || now - pick->replacement_ms >= STREAM_MIGRATION_TIMEOUT_MS) {
            receive_failed(o, &pick->replacement, sub, now); continue;
        }
        if (replacement.blocks && (!pick->current || replacement.next >= info.next)) {
            uint64_t previous = pick->current ? info.peer : 0;
            stream_receive_close(pick->current); pick->current = pick->replacement; pick->replacement = NULL;
            pick->opened_ms = now; pick->better_ms = pick->better_node = 0; o->info.migrations++;
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer moved sub=%u from=%llx to=%llx verified_blocks=%llu next=%llu",
                       sub, (unsigned long long)previous, (unsigned long long)replacement.peer,
                       (unsigned long long)replacement.blocks, (unsigned long long)replacement.next);
        }
    }
    if (!o->progress_valid || o->next_seq != local.next) {
        o->progress_valid = 1; o->next_seq = local.next; o->progress_ms = now;
    }
    if (local.anchored && now - o->progress_ms >= STREAM_MIGRATION_TIMEOUT_MS && received < (unsigned)(o->descriptor.n - o->descriptor.r)) {
        int newer = 0;
        for (struct stream_supplier *s = o->suppliers; s; s = s->next)
            if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
                struct stream_receive_info *offer = &s->offers[sub].info;
                if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS && offer->last_available > local.next) newer = 1;
            }
        if (newer) {
            selections_close(o);
            if (stream_local_restart(o->inst, &o->descriptor) == STREAM_OK) {
                o->info.restarts++; o->progress_ms = now; busy = 0; stream_local_get(o->inst, &o->descriptor, &local);
            }
        }
    }
    int migrating = 0;
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) if (o->selected[sub].replacement) migrating = 1;
    if (!local.anchored && !busy && !o->startup_complete) {
        o->startup_peer = 0;
        unsigned sub = 0; struct stream_supplier *s = startup_choose(o, now, &sub);
        if (s && stream_receive_open(h, s->node, (uint8_t)sub, UINT64_MAX, &o->selected[sub].current) == STREAM_OK) {
            o->startup_peer = s->node; o->selected[sub].opened = 1; o->selected[sub].opened_ms = now; busy++;
            if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; }
            /* Остальные INFO нужны для недостающих частей: победитель определяет только общий DATA старт. */
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap winner group=%llx source=%llx epoch=%llx peer=%llx sub=%u"
                " order=%llu elapsed_ms=%llu relay=%d",
                (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
                (unsigned long long)o->descriptor.source_epoch, (unsigned long long)s->node, sub,
                (unsigned long long)s->offers[sub].info.info_order, (unsigned long long)(now - o->started_ms), s->relay);
        }
    }
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        struct stream_selection *pick = &o->selected[sub]; struct stream_receive_info info = {0};
        if (pick->replacement || (!local.anchored && busy)) continue;
        if (pick->current) stream_receive_get_info(pick->current, &info);
        uint64_t seq = local.anchored ? (pick->opened ? local.next : local.start) : UINT64_MAX;
        struct stream_supplier *s = supplier_choose(o, sub, seq, info.peer, now);
        if (!s) { pick->better_ms = pick->better_node = 0; continue; }
        if (!pick->current) {
            if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->current) == STREAM_OK) {
                pick->opened_ms = now; pick->opened = 1; busy++;
                struct stream_receive_info *offer = &s->offers[sub].info;
                DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer selected peer=%llx sub=%u index=%llu score_ms=%llu"
                    " rtt_ms=%u chain_ms=%u hops=%u penalty_ms=%llu",
                    (unsigned long long)s->node, sub, (unsigned long long)seq,
                    (unsigned long long)stream_supplier_score(offer, &offer->supplier_metrics, offer->supplier_metrics_valid),
                    offer->rtt_ms, offer->path_rtt_ms, offer->path.count,
                    (unsigned long long)problem_penalty(&offer->supplier_metrics, offer->supplier_metrics_valid));
            }
            continue;
        }
        if (info.state != STREAM_RECEIVE_ACTIVE || info.close_requested || migrating
            || now - pick->opened_ms < o->settings.minimum_hold_ms || !improved(o, sub, s, &info)) {
            pick->better_ms = pick->better_node = 0; continue;
        }
        if (pick->better_node != s->node) { pick->better_node = s->node; pick->better_ms = now; }
        if (now - pick->better_ms < o->settings.migration_confirm_ms) continue;
        struct stream_receive_info *offer = &s->offers[sub].info;
        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer improvement sub=%u old=%llx new=%llx old_score_ms=%llu new_score_ms=%llu"
            " old_penalty_ms=%llu new_penalty_ms=%llu old_rtt_ms=%u/%d new_rtt_ms=%u/%d old_chain_ms=%u/%d new_chain_ms=%u/%d"
            " old_hops=%u new_hops=%u old_usage=%u new_usage=%u confirmed_ms=%llu",
            sub, (unsigned long long)info.peer, (unsigned long long)s->node,
            (unsigned long long)stream_supplier_score(&info, &info.metrics, info.metrics_valid),
            (unsigned long long)stream_supplier_score(offer, &offer->supplier_metrics, offer->supplier_metrics_valid),
            (unsigned long long)problem_penalty(&info.metrics, info.metrics_valid),
            (unsigned long long)problem_penalty(&offer->supplier_metrics, offer->supplier_metrics_valid),
            info.rtt_ms, info.rtt_valid, offer->rtt_ms, offer->rtt_valid,
            info.path_rtt_ms, info.path_rtt_valid, offer->path_rtt_ms, offer->path_rtt_valid, info.path.count, offer->path.count,
            usage(o, info.peer, sub), usage(o, s->node, sub), (unsigned long long)(now - pick->better_ms));
        if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->replacement) == STREAM_OK) {
            pick->replacement_ms = now; migrating = 1;
            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer overlap sub=%u old=%llx new=%llx index=%llu",
                       sub, (unsigned long long)info.peer, (unsigned long long)s->node, (unsigned long long)seq);
        }
    }
    o->info.candidates = o->info.probes_active = o->info.subflows = o->info.overlaps = o->info.providers = 0;
    o->info.protected_by_fec = o->descriptor.r != 0;
    for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
        o->info.candidates++; if (s->probe) o->info.probes_active++;
        unsigned used = usage(o, s->node, o->descriptor.n);
        if (used) o->info.providers++;
        if (used > o->descriptor.r) o->info.protected_by_fec = 0;
        if (used || s->probe || !s->offers) continue;
        uint64_t keep_until = 0;
        for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
            struct stream_offer *offer = &s->offers[sub];
            uint64_t until = offer->info.info_ms + STREAM_OFFER_TTL_MS;
            if (until < offer->retry_ms) until = offer->retry_ms;
            if (until > keep_until) keep_until = until;
        }
        if (now >= keep_until) { u_free(s->offers); s->offers = NULL; }
    }
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        if (o->selected[sub].current) {
            struct stream_receive_info info; stream_receive_get_info(o->selected[sub].current, &info);
            o->info.subflows++;
            if (info.state != STREAM_RECEIVE_ACTIVE || !info.blocks || !info.metrics_valid || info.metrics.stopped || info.close_requested)
                o->info.protected_by_fec = 0;
        }
        if (o->selected[sub].replacement) o->info.overlaps++;
    }
    if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0;
    /* Помощь нужна для получения k субпотоков, а не для идеальной защиты n/r; пауза источника не является отказом пути. */
    unsigned active = 0, viable = 0; int newer = 0;
    for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
        struct STREAM_RECEIVE *current = o->selected[sub].current;
        if (!current) continue;
        struct stream_receive_info info; stream_receive_get_info(current, &info);
        if (info.state != STREAM_RECEIVE_ACTIVE || info.close_requested) continue;
        active++;
        if ((info.blocks && info.metrics_valid && !info.metrics.stopped) || (!info.blocks && info.last_available == UINT64_MAX)) viable++;
    }
    for (struct stream_supplier *s = o->suppliers; s; s = s->next)
        if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
            struct stream_receive_info *offer = &s->offers[sub].info;
            if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS
                && offer->last_available != UINT64_MAX && offer->last_available > local.next) newer = 1;
        }
    unsigned required = (unsigned)(o->descriptor.n - o->descriptor.r);
    int deficient = active < required || (viable < required && newer);
    if (!deficient) o->deficient_ms = 0;
    else if (!o->deficient_ms) o->deficient_ms = now;
    if (!o->startup_complete || now >= o->relay_poll_ms) {
        o->relay_poll_ms = now + STREAM_PROBE_PERIOD_MS;
        int needed = deficient && now - o->deficient_ms >= STREAM_MIGRATION_TIMEOUT_MS;
        if (!o->startup_complete && !o->startup_peer && now - o->started_ms >= STREAM_BOOTSTRAP_RELAY_MS) needed = 1;
        if (!o->startup_complete && o->startup_peer && stream_relay_confirmed(o->inst, &o->descriptor, o->startup_peer)) needed = 1;
        stream_relay_demand(o->inst, &o->descriptor, needed, now);
    }
    /* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */
    if (!local.anchored && now >= o->diagnostic_ms) {
        o->diagnostic_ms = now + STREAM_PROBE_PERIOD_MS;
        DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stream bootstrap waits group=%llx source=%llx epoch=%llx candidates=%u probes=%u"
            " subflows=%u cycle=%llu cycle_age_ms=%llu",
            (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
            (unsigned long long)o->descriptor.source_epoch, o->info.candidates, o->info.probes_active, o->info.subflows,
            (unsigned long long)o->schedule.cycle, (unsigned long long)(now - o->schedule.started_ms));
        for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
            struct stream_receive_info probe = {0};
            if (s->probe) stream_receive_get_info(s->probe, &probe);
            DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stream bootstrap supplier source=%llx peer=%llx online=%d compatible=%d source_peer=%d"
                " cycle=%llu failures=%u retry_in_ms=%llu probe=%d sub=%u state=%d info_ready=%d reason=%d",
                (unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->online, s->compatible, s->source,
                (unsigned long long)s->cycle, s->failures, (unsigned long long)(s->retry_ms > now ? s->retry_ms - now : 0),
                s->probe != NULL, s->probe_sub, probe.state, probe.info_ready, probe.reason);
        }
    }
}