/* Агент стенда: настоящие JOIN/Merkle/ETCP и автоматический выбор поставщиков.
 * stdin/stdout — только JSONL RPC; диагностика ядра пишется в отдельный файл.
 * Заголовок нагрузки и SHA256 проверяют opaque DATA после восстановления FEC. */
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <openssl/sha.h>

#include "../../lib/debug_config.h"
#include "../../lib/json_flat.h"
#include "../../lib/mem.h"
#include "../../src/chat/chat_core.h"
#include "../../src/chat/chat_sync.h"
#include "../../src/chat/invite_build.h"
#include "../../src/chat/invite_link.h"
#include "../../src/stream/stream_priv.h"
#include "../../src/stream/stream_member.h"
#include "../../src/utun_instance.h"

#define LAB_STREAMS 4
#define LAB_HISTOGRAM 4096

struct lab_stream {
    struct STREAM_HANDLE *handle;
    uint64_t source, generated, received, bytes, gaps, gap_batches, errors, ends, closed, next, first, latency_max;
    uint64_t latency_sum, histogram[LAB_HISTOGRAM], last_data_ns, start_seq;
    uint64_t seed, interval_ns, due_ns;
    uint8_t *payload;
    size_t size;
    int publisher, started;
};
struct lab {
    struct UTUN_INSTANCE *inst;
    struct lab_stream streams[LAB_STREAMS];
    struct chat_invite_build *invite;
    uint64_t invite_id, group;
    int errors;
};
static volatile sig_atomic_t stopping;

static uint64_t now_ns(void) {
    struct timespec time;
    clock_gettime(CLOCK_MONOTONIC, &time);
    return (uint64_t)time.tv_sec * UINT64_C(1000000000) + time.tv_nsec;
}

static void stop_signal(int signal_number) { (void)signal_number; stopping = 1; }

static void json_string(const char *value) {
    putchar('"');
    for (const unsigned char *p = (const unsigned char *)value; *p; p++) {
        if (*p == '"' || *p == '\\') { putchar('\\'); putchar(*p); }
        else if (*p < 32) printf("\\u%04x", *p);
        else putchar(*p);
    }
    putchar('"');
}

static void reply(uint64_t id, int result) {
    printf("{\"id\":%" PRIu64 ",\"rc\":%d}\n", id, result);
}

static int number(const char *json, const char *key, uint64_t *value) {
    char buffer[32], *end;
    if (json_flat_get(json, key, buffer, sizeof(buffer)) || !buffer[0] || buffer[0] == '-') return -1;
    errno = 0;
    uint64_t parsed = strtoull(buffer, &end, 10);
    if (errno || *end) return -1;
    *value = parsed;
    return 0;
}

static struct lab_stream *slot(struct lab *lab, uint64_t source, int create) {
    for (unsigned i = 0; i < LAB_STREAMS; i++) if (lab->streams[i].source == source) return &lab->streams[i];
    for (unsigned i = 0; create && i < LAB_STREAMS; i++) {
        if (!lab->streams[i].source) { lab->streams[i].source = source; return &lab->streams[i]; }
    }
    return NULL;
}

static void event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) {
    (void)inst;
    struct lab *lab = arg;
    struct lab_stream *stream = slot(lab, event->descriptor.source_node_id, 0);
    if (!stream) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: event without owner source=%016" PRIx64, event->descriptor.source_node_id);
        lab->errors++; return;
    }
    if (event->type == STREAM_EVENT_CLOSED) { stream->closed++; return; }
    if (event->type == STREAM_EVENT_END) { stream->ends++; return; }
    if (event->type == STREAM_EVENT_GAP) {
        if (!stream->started) stream->start_seq = event->first_seq;
        stream->gaps++; stream->gap_batches += event->last_seq - event->first_seq + 1;
        if (stream->started && event->first_seq != stream->next) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: GAP cursor expected=%" PRIu64 " first=%" PRIu64, stream->next, event->first_seq);
            stream->errors++;
        }
        stream->started = 1; stream->next = event->last_seq + 1;
        DEBUG_WARN(DEBUG_CATEGORY_STREAM, "lab: GAP source=%016" PRIx64 " first=%" PRIu64 " last=%" PRIu64,
                   stream->source, event->first_seq, event->last_seq);
        return;
    }
    const uint8_t *data = event->data;
    uint8_t digest[32];
    int valid = event->length >= 80 && !memcmp(data, "SLABv001", 8);
    if (valid) {
        SHA256(data, event->length - 32, digest);
        valid = !memcmp(digest, data + event->length - 32, 32) && stream_read_be(data + 8, 8) == event->first_seq &&
                stream_read_be(data + 32, 8) == stream->source && stream_read_be(data + 40, 4) == event->length;
    }
    if (!valid || (stream->started && event->first_seq != stream->next)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: DATA validation source=%016" PRIx64 " seq=%" PRIu64
                    " expected=%" PRIu64 " bytes=%zu checksum=%d", stream->source, event->first_seq, stream->next, event->length, valid);
        stream->errors++;
    }
    if (!stream->started) stream->start_seq = event->first_seq;
    if (!stream->received) stream->first = event->first_seq;
    stream->started = 1; stream->next = event->last_seq + 1;
    stream->received++; stream->bytes += event->length; stream->last_data_ns = now_ns();
    if (valid) {
        uint64_t generated_ns = stream_read_be(data + 16, 8);
        if (generated_ns > stream->last_data_ns) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: CLOCK_MONOTONIC moved backwards"); stream->errors++; return;
        }
        uint64_t delay = (stream->last_data_ns - generated_ns) / UINT64_C(1000000);
        stream->latency_sum += delay;
        if (delay > stream->latency_max) stream->latency_max = delay;
        stream->histogram[delay < LAB_HISTOGRAM ? delay : LAB_HISTOGRAM - 1]++;
    }
}

static uint64_t percentile(const struct lab_stream *stream, unsigned percent) {
    uint64_t target = (stream->received * percent + 99) / 100, count = 0;
    if (!target) return 0;
    for (unsigned i = 0; i < LAB_HISTOGRAM; i++) { count += stream->histogram[i]; if (count >= target) return i; }
    return LAB_HISTOGRAM - 1;
}

static void statistics(struct lab *lab, uint64_t id, int loads) {
    struct TOPO_GROUP *group = topo_groups_find(lab->inst->topo_groups, lab->group);
    printf("{\"id\":%" PRIu64 ",\"rc\":0,\"node\":\"%" PRIu64 "\",\"group\":\"%" PRIu64
           "\",\"errors\":%d,\"online\":[", id, lab->inst->node_id, lab->group, lab->errors);
    int separator = 0;
    if (stream_member_online(lab->inst, lab->group, lab->inst->node_id)) {
        printf("\"%" PRIu64 "\"", lab->inst->node_id); separator = 1;
    }
    for (struct ll_entry *e = group ? group->nodes->head : NULL; e; e = e->next) {
        struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e;
        if (stream_member_online(lab->inst, lab->group, node->node_id)) {
            printf("%s\"%" PRIu64 "\"", separator++ ? "," : "", node->node_id);
        }
    }
    printf("],\"profile_status\":{"); separator = 0;
    if (group) {
        struct stream_member_state state;
        printf("\"%" PRIu64 "\":%d", lab->inst->node_id,
               stream_member_load(lab->inst, lab->group, lab->inst->node_id, &state, NULL)); separator = 1;
        for (struct ll_entry *e = group->nodes->head; e; e = e->next) {
            struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e;
            printf("%s\"%" PRIu64 "\":%d", separator++ ? "," : "", node->node_id,
                   stream_member_load(lab->inst, lab->group, node->node_id, &state, NULL));
        }
    }
    printf("},\"streams\":["); separator = 0;
    for (unsigned i = 0; i < LAB_STREAMS; i++) {
        struct lab_stream *stream = &lab->streams[i];
        if (!stream->source) continue;
        printf("%s{\"source\":\"%" PRIu64 "\",\"publisher\":%d,\"generated\":%" PRIu64 ",\"received\":%" PRIu64
               ",\"bytes\":%" PRIu64 ",\"gaps\":%" PRIu64 ",\"gap_batches\":%" PRIu64 ",\"errors\":%" PRIu64
               ",\"ends\":%" PRIu64 ",\"closed\":%" PRIu64 ",\"first_seq\":%" PRIu64 ",\"next_seq\":%" PRIu64
               ",\"start_seq\":%" PRIu64 ",\"age_ms\":%" PRIu64 ",\"latency_p50_ms\":%" PRIu64 ",\"latency_p95_ms\":%" PRIu64
               ",\"latency_p99_ms\":%" PRIu64 ",\"latency_max_ms\":%" PRIu64,
               separator++ ? "," : "", stream->source, stream->publisher, stream->generated, stream->received,
               stream->bytes, stream->gaps, stream->gap_batches, stream->errors, stream->ends, stream->closed,
               stream->first, stream->next, stream->start_seq,
               stream->last_data_ns ? (now_ns() - stream->last_data_ns) / UINT64_C(1000000) : 0,
               percentile(stream, 50), percentile(stream, 95), percentile(stream, 99), stream->latency_max);
        struct stream_info info;
        if (stream->handle && stream_get_info(stream->handle, &info) == STREAM_OK) {
            struct stream_selection_info selection;
            memset(&selection, 0, sizeof(selection));
            stream_selection_get_info(stream->handle, &selection);
            printf(",\"active\":%d,\"epoch\":\"%" PRIu64 "\",\"stored_bytes\":%zu,\"capacity_bytes\":%zu"
                   ",\"evicted_batches\":%" PRIu64 ",\"subflows\":%u,\"providers\":%u,\"probes\":%" PRIu64
                   ",\"cycles\":%" PRIu64 ",\"migrations\":%" PRIu64 ",\"failures\":%" PRIu64 ",\"restarts\":%" PRIu64
                   ",\"protected\":%d,\"paths\":[", stream_handle_active(stream->handle), info.descriptor.source_epoch,
                   info.buffer.stored_bytes, info.buffer.capacity_bytes, info.buffer.evicted_batches, selection.subflows,
                   selection.providers, selection.probes_started, selection.cycles, selection.migrations, selection.failures,
                   selection.restarts, selection.protected_by_fec);
            struct stream_local local;
            int path_separator = 0;
            if (stream_local_get(lab->inst, &info.descriptor, &local) == STREAM_OK) {
                for (uint8_t sub = 0; sub < info.descriptor.n; sub++) {
                    uint64_t first, last;
                    struct stream_path path;
                    if (stream_buffer_range(local.buffer, sub, &first, &last) != STREAM_OK ||
                        stream_buffer_path(local.buffer, last, sub, &path) != STREAM_OK) continue;
                    printf("%s{\"sub\":%u,\"seq\":%" PRIu64 ",\"nodes\":[", path_separator++ ? "," : "", sub, last);
                    for (unsigned p = 0; p < path.count; p++) printf("%s\"%" PRIu64 "\"", p ? "," : "", path.nodes[p]);
                    printf("]");
                    const uint8_t *wire; size_t length; uint64_t actual;
                    if (stream_buffer_read(local.buffer, last, sub, &wire, &length, &actual) == STREAM_OK)
                        printf(",\"source_timestamp_ms\":%u", (unsigned)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2));
                    struct stream_metric_snapshot metric;
                    if (stream_local_metrics(lab->inst, &info.descriptor, sub, get_time_tb() / 10, &metric) == STREAM_OK)
                        printf(",\"lag_ms\":%" PRIu64 ",\"source_rate\":%" PRIu64 ",\"receive_rate\":%" PRIu64
                               ",\"late_blocks\":%" PRIu64 ",\"stops\":%" PRIu64 ",\"stopped\":%d",
                               metric.lag_ms, metric.source_rate_bytes_sec, metric.receive_rate_bytes_sec,
                               metric.late_blocks, metric.stops, metric.stopped);
                    printf("}");
                }
            }
            printf("]");
        }
        printf("}");
    }
    printf("],\"peer_load\":["); separator = 0;
    for (struct ll_entry *e = loads && group ? group->nodes->head : NULL; e; e = e->next) {
        struct TOPO_GROUP_NODE *node = (struct TOPO_GROUP_NODE *)e;
        struct stream_peer_load_info load;
        if (stream_peer_get_load(lab->inst, node->node_id, &load) != STREAM_OK) continue;
        printf("%s{\"peer\":\"%" PRIu64 "\",\"queued_bytes\":%" PRIu64 ",\"growth_rate\":%" PRIu64
               ",\"transport_bytes\":%" PRIu64 ",\"inflight_bytes\":%" PRIu64 ",\"pressure\":%d,\"confirmed\":%d}",
               separator++ ? "," : "", node->node_id, load.load.queued.bytes, load.load.growth_rate,
               load.transport_queued_bytes, load.inflight_bytes, load.load.pressure, load.load.confirmed);
    }
    printf("]}\n");
}

static void invite_ready(void *arg, int result, const char *link) {
    struct lab *lab = arg;
    lab->invite = NULL;
    printf("{\"id\":%" PRIu64 ",\"rc\":%d,\"link\":", lab->invite_id, result);
    json_string(link ? link : ""); printf("}\n");
}

static void command(struct lab *lab, const char *json) {
    uint64_t id = 0, source = 0, value = 0;
    char operation[32];
    if (number(json, "id", &id) || json_flat_get(json, "op", operation, sizeof(operation))) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: invalid RPC"); reply(id, -1); return;
    }
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "lab: RPC id=%" PRIu64 " op=%s", id, operation);
    if (!strcmp(operation, "stats")) { number(json, "loads", &value); statistics(lab, id, value != 0); return; }
    if (!strcmp(operation, "quit")) { stopping = 1; reply(id, 0); return; }
    if (!strcmp(operation, "create")) {
        if (lab->group) { reply(id, -1); return; }
        chat_core_create_channel_auto(lab->inst, "stream-netns");
        for (struct ll_entry *e = lab->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) { lab->group = group->group_id; break; }
        }
        printf("{\"id\":%" PRIu64 ",\"rc\":%d,\"group\":\"%" PRIu64 "\"}\n", id, lab->group ? 0 : -1, lab->group); return;
    }
    if (!strcmp(operation, "invite")) {
        if (lab->invite || !lab->group) { reply(id, -1); return; }
        lab->invite_id = id;
        lab->invite = chat_invite_build_link(lab->inst, lab->group, lab->inst->node_id, NULL, invite_ready, lab);
        if (!lab->invite) reply(id, -1);
        return;
    }
    if (!strcmp(operation, "join")) {
        char link[2048], error[256];
        struct InviteData invite;
        uint8_t addresses[2048];
        if (json_flat_get(json, "link", link, sizeof(link)) || invite_link_decode(link, strlen(link), &invite, error, sizeof(error))) {
            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: invalid invite"); reply(id, -1); return;
        }
        int size = invite_serialize_addrs(&invite, addresses, sizeof(addresses));
        if (size <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: invite has no endpoint"); reply(id, -1); return; }
        lab->group = invite.channelId;
        chat_sync_connect_from_invite(lab->inst, invite.channelId, invite.nodeId, invite.pubkey, addresses,
                                     invite.addrCount, size, invite.password_len ? invite.password : NULL, invite.join_key);
        reply(id, 0); return;
    }
    int publish = !strcmp(operation, "publish");
    if (publish) source = lab->inst->node_id;
    else if (number(json, "source", &source)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: missing source"); reply(id, -1); return; }
    struct lab_stream *stream = slot(lab, source, publish || !strcmp(operation, "subscribe"));
    if (!stream) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: no stream slot"); reply(id, -1); return; }
    if (publish || !strcmp(operation, "subscribe")) {
        if (stream->handle) { reply(id, 0); return; }
        int result = publish ? stream_publish_open(lab->inst, lab->group, NULL, &stream->handle) :
                               stream_subscribe_open(lab->inst, lab->group, source, &stream->handle);
        if (result == STREAM_OK && publish) {
            struct stream_info info;
            stream_get_info(stream->handle, &info);
            stream->size = stream_input_size(&info.descriptor);
            stream->payload = u_malloc(stream->size);
            stream->publisher = 1;
            if (!stream->payload) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: payload allocation failed"); stopping = 1; lab->errors++; }
            if (!number(json, "seed", &value)) stream->seed = value;
        }
        reply(id, result); return;
    }
    if (!strcmp(operation, "rate") && stream->publisher && !number(json, "bps", &value)) {
        stream->interval_ns = value ? stream->size * UINT64_C(8000000000) / value : 0;
        if (value && !stream->interval_ns) stream->interval_ns = 1;
        stream->due_ns = now_ns(); reply(id, 0); return;
    }
    if (!strcmp(operation, "finish") && stream->publisher) {
        stream->interval_ns = 0; reply(id, stream_publish_finish(stream->handle)); return;
    }
    if (!strcmp(operation, "close")) {
        stream->interval_ns = 0;
        if (stream->handle) stream_close(stream->handle);
        stream->handle = NULL; reply(id, 0); return;
    }
    DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: unsupported RPC op=%s", operation); reply(id, -1);
}

static void generate(struct lab *lab) {
    for (unsigned i = 0; i < LAB_STREAMS; i++) {
        struct lab_stream *stream = &lab->streams[i];
        for (unsigned burst = 0; stream->interval_ns && now_ns() >= stream->due_ns && burst < 8; burst++) {
            uint8_t *data = stream->payload;
            memset(data, 0, 48); memcpy(data, "SLABv001", 8);
            stream_write_be(data + 8, stream->generated, 8); stream_write_be(data + 16, now_ns(), 8);
            stream_write_be(data + 24, stream->seed, 8); stream_write_be(data + 32, stream->source, 8);
            stream_write_be(data + 40, stream->size, 4);
            uint64_t random = stream->seed ^ stream->generated ^ stream->source;
            for (size_t p = 48; p < stream->size - 32; p++) {
                random ^= random << 13; random ^= random >> 7; random ^= random << 17; data[p] = (uint8_t)random;
            }
            SHA256(data, stream->size - 32, data + stream->size - 32);
            size_t accepted = 0;
            int result = stream_publish_write(stream->handle, data, stream->size, &accepted);
            if (result != STREAM_OK || accepted != stream->size) {
                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: write failed rc=%d accepted=%zu expected=%zu", result, accepted, stream->size);
                lab->errors++; stream->interval_ns = 0; break;
            }
            stream->generated++; stream->due_ns += stream->interval_ns;
        }
    }
}

int main(int argc, char **argv) {
    if (argc == 3 && !strcmp(argv[1], "--identity")) {
        uint8_t private_key[32], public_key[32];
        SHA256((const unsigned char *)argv[2], strlen(argv[2]), private_key);
        if (sc_compute_public_key_from_private(private_key, public_key) != SC_OK) return 1;
        printf("{\"private\":\"");
        for (unsigned i = 0; i < 32; i++) printf("%02x", private_key[i]);
        printf("\",\"public\":\"");
        for (unsigned i = 0; i < 32; i++) printf("%02x", public_key[i]);
        printf("\",\"node\":\"%" PRIu64 "\"}\n", sc_derive_node_id_from_pubkey(public_key)); return 0;
    }
    if (argc != 3) { fprintf(stderr, "Usage: %s CONFIG LOG | --identity SEED\n", argv[0]); return 1; }
    setvbuf(stdout, NULL, _IOLBF, 0);
    signal(SIGTERM, stop_signal); signal(SIGINT, stop_signal); signal(SIGPIPE, SIG_IGN);
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_enable_console(0); debug_enable_file_output(argv[2], 1);
    utun_instance_set_tun_init_enabled(0);
    struct UASYNC *async = uasync_create();
    struct lab *lab = u_calloc(1, sizeof(*lab));
    if (!async || !lab) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: initialization allocation failed"); return 1; }
    lab->inst = utun_instance_create(async, argv[1]);
    if (!lab->inst || utun_core_start(lab->inst) || chat_service_start(lab->inst)) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: service startup failed"); return 1;
    }
    debug_enable_console(0);
    stream_set_event_callback(lab->inst, event, lab);
    if (fcntl(STDIN_FILENO, F_SETFL, fcntl(STDIN_FILENO, F_GETFL) | O_NONBLOCK) < 0) {
        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: stdin nonblocking failed: %s", strerror(errno)); return 1;
    }
    printf("{\"ready\":true,\"node\":\"%" PRIu64 "\"}\n", lab->inst->node_id);
    char input[8192]; size_t used = 0;
    while (!stopping) {
        uasync_poll(async, 100); generate(lab);
        ssize_t size = read(STDIN_FILENO, input + used, sizeof(input) - used - 1);
        if (size == 0) break;
        if (size < 0) {
            if (errno != EAGAIN && errno != EINTR) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: stdin failed: %s", strerror(errno)); lab->errors++; break; }
            continue;
        }
        used += (size_t)size; input[used] = 0;
        char *line;
        while ((line = memchr(input, '\n', used))) {
            *line = 0; command(lab, input);
            size_t consumed = (size_t)(line - input) + 1;
            memmove(input, input + consumed, used - consumed); used -= consumed; input[used] = 0;
        }
        if (used == sizeof(input) - 1) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "lab: RPC exceeds input limit"); lab->errors++; break; }
    }
    if (lab->invite) chat_invite_build_cancel(lab->invite);
    for (unsigned i = 0; i < LAB_STREAMS; i++) {
        if (lab->streams[i].handle) stream_close(lab->streams[i].handle);
        u_free(lab->streams[i].payload);
        if (lab->streams[i].errors) lab->errors++;
    }
    int errors = lab->errors;
    utun_instance_destroy(lab->inst); uasync_destroy(async, 1); u_free(lab);
    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "lab: stopped errors=%d", errors);
    debug_disable_file_output(); return errors ? 1 : 0;
}