#include "fake.h"
#include "session_broker.c"
#define printf capture_printf
#include "session_console.c"
#undef printf

static uint8_t payload[SESSION_BROKER_OUTPUT_SIZE + 256];
static void feed(size_t size) {
    assert(size <= sizeof(payload));
    serial_input = payload; serial_remaining = size;
    if (setjmp(task_exit) == 0) task_entry(NULL);
    assert(!mutex && serial_remaining == 0);
}
static session_broker_client_snapshot_t snapshot(session_broker_client_id_t id) {
    session_broker_client_snapshot_t s;
    assert(session_broker_get_client_snapshot(id, &s) == ESP_OK); return s;
}
static session_broker_global_snapshot_t global(void) {
    session_broker_global_snapshot_t s;
    assert(session_broker_get_global_snapshot(&s) == ESP_OK); return s;
}
static size_t drain(session_broker_client_id_t id, size_t size) {
    uint8_t data[sizeof(payload)]; size_t received;
    assert(size <= sizeof(data));
    assert(session_broker_read(id, data, size, &received) == ESP_OK); return received;
}
static session_broker_client_id_t connect_type(session_broker_client_type_t type) {
    session_broker_client_id_t id;
    assert(session_broker_connect(type, "SECRET-NAME-\033[2J", &id) == ESP_OK); return id;
}
static int command(const char *operation, session_broker_client_id_t id, const char *size) {
    char id_text[16]; snprintf(id_text, sizeof(id_text), "%u", id);
    char *argv[] = {"broker", (char *)operation, id_text, (char *)size};
    console_used = 0; console_output[0] = 0;
    return registered_command(id ? (size ? 4 : 3) : 2, argv);
}
static void disconnect_all(void) {
    session_broker_client_snapshot_t clients[SESSION_BROKER_MAX_CLIENTS];
    size_t n = session_broker_list_clients(clients, SESSION_BROKER_MAX_CLIENTS);
    for (size_t i = 0; i < n; ++i) assert(session_broker_disconnect(clients[i].id) == ESP_OK);
}
static void comparison(unsigned browsers) {
    session_broker_client_id_t ids[4];
    ids[0] = connect_type(SESSION_BROKER_CLIENT_USB);
    ids[1] = connect_type(SESSION_BROKER_CLIENT_SSH);
    for (unsigned i = 0; i < browsers; ++i) ids[2+i] = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(session_broker_clear_counters() == ESP_OK);
    size_t total = 0;
    while (total < 71292) {
        size_t n = 256;
        if (n > 71292 - total) n = 71292 - total;
        /* Deterministic slow-browser window, not a claim about real scheduling. */
        if (browsers == 2 && total < 23412 && n > 23412 - total) n = 23412 - total;
        feed(n); total += n;
        for (unsigned i = 0; i < 2+browsers; ++i) {
            if (browsers == 2 && i == 3 && total < 23412) continue;
            drain(ids[i], sizeof(payload));
        }
    }
    session_broker_global_snapshot_t g = global();
    uint64_t expected = 71292U * (2+browsers);
    uint64_t drops = browsers == 2 ? 19316 : 0;
    assert(g.counters.uart_rx_bytes == 71292 && g.counters.disconnections == 0);
    assert(g.counters.output_queued_bytes == expected-drops);
    assert(g.counters.output_read_bytes == expected-drops && g.counters.output_dropped_bytes == drops);
    for (unsigned i = 0; i < 2+browsers; ++i) {
        session_broker_client_snapshot_t s = snapshot(ids[i]);
        assert(s.counters.output_dropped_bytes == (i == 3 ? drops : 0));
        assert(s.counters.output_high_water_bytes == (i == 3 ? 4096 : 256));
    }
    printf("PASS synthetic %u-browser comparison: UART=71292 copies=%" PRIu64 " queued=%" PRIu64 " read=%" PRIu64 " dropped=%" PRIu64 " disconnect=0\n",
           browsers, expected, g.counters.output_queued_bytes, g.counters.output_read_bytes, drops);
    disconnect_all();
}
int main(void) {
    for (size_t i = 0; i < sizeof(payload); ++i) payload[i] = (uint8_t)i;
    assert(session_broker_get_global_snapshot(&(session_broker_global_snapshot_t){0}) == ESP_ERR_INVALID_STATE);
    assert(session_broker_init() == ESP_OK);
    assert(session_console_register_commands() == ESP_OK);
    size_t initial_allocations = allocations;
    session_broker_client_id_t fast = connect_type(SESSION_BROKER_CLIENT_USB);
    session_broker_client_id_t slow = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(session_broker_request_writer(fast) == ESP_OK);
    for (unsigned i = 0; i < 17; ++i) { feed(256); assert(drain(fast, 256) == 256); }
    session_broker_client_snapshot_t s = snapshot(slow);
    assert(s.output_bytes_pending == 4096 && s.counters.output_high_water_bytes == 4096);
    assert(s.counters.output_queued_bytes == 4096 && s.counters.output_dropped_bytes == 256);
    s = snapshot(fast);
    assert(s.counters.output_read_bytes == 4352 && s.counters.output_high_water_bytes == 256);
    assert(s.counters.output_dropped_bytes == 0 && s.is_writer);
    puts("PASS isolated fanout overflow and exact 4096-byte high-water");

    assert(drain(slow, 4000) == 4000);
    assert(snapshot(slow).counters.output_high_water_bytes == 4096);
    session_broker_global_snapshot_t before = global();
    assert(session_broker_clear_client_counters(slow) == ESP_OK);
    s = snapshot(slow);
    assert(s.output_bytes_pending == 96 && s.counters.output_high_water_bytes == 96);
    assert(s.counters.output_queued_bytes == 0 && s.counters.output_read_bytes == 0 && s.counters.output_dropped_bytes == 0);
    assert(global().counters.output_queued_bytes == before.counters.output_queued_bytes);
    feed(256); assert(snapshot(slow).counters.output_high_water_bytes == 352);
    assert(session_broker_clear_counters() == ESP_OK);
    s = snapshot(slow);
    assert(s.output_bytes_pending == 352 && s.counters.output_high_water_bytes == 352);
    assert(global().writer_id == fast && global().latest_event_sequence == before.latest_event_sequence);
    assert(global().counters.uart_rx_bytes == 0 && s.counters.uart_rx_bytes == 0);
    assert(drain(slow, 352) == 352);
    assert(snapshot(slow).counters.output_read_bytes == 352);
    feed(sizeof(payload));
    assert(snapshot(slow).counters.output_dropped_bytes == 256);
    puts("PASS per-client/global clears seed pending; subsequent read/drop/HWM accounting");

    before = global();
    assert(session_broker_disconnect(slow) == ESP_OK);
    assert(global().counters.output_dropped_bytes == before.counters.output_dropped_bytes + 4096);
    assert(session_broker_get_client_snapshot(slow, &s) == ESP_ERR_NOT_FOUND);
    session_broker_client_id_t replacement = connect_type(SESSION_BROKER_CLIENT_WEB);
    assert(replacement != slow && (replacement & 7) == (slow & 7));
    s = snapshot(replacement);
    assert(s.output_bytes_pending == 0 && s.counters.output_high_water_bytes == 0);
    assert(s.counters.output_queued_bytes == 0 && s.counters.output_read_bytes == 0 && s.counters.output_dropped_bytes == 0);
    assert(session_broker_clear_client_counters(slow) == ESP_ERR_NOT_FOUND);
    puts("PASS disconnect discard retained globally and generation-safe reuse resets diagnostics");

    while (session_broker_list_clients(NULL, 0) < SESSION_BROKER_MAX_CLIENTS)
        connect_type(SESSION_BROKER_CLIENT_INTERNAL);
    struct { session_broker_client_snapshot_t row; uint64_t guard; } bounded = {.guard = UINT64_MAX};
    assert(session_broker_list_clients(&bounded.row, 1) == 1 && bounded.guard == UINT64_MAX);
    before = global();
    assert(command("counters", 0, NULL) == 0);
    assert(strstr(console_output, "pending HWM") && !strstr(console_output, "SECRET-NAME") && !strchr(console_output, '\033'));
    unsigned lines = 0;
    for (const char *p = strstr(console_output, "ID         type"); *p; ++p) lines += *p == '\n';
    assert(lines == 1 + SESSION_BROKER_MAX_CLIENTS && console_used < 4096);
    assert(global().counters.output_read_bytes == before.counters.output_read_bytes);
    puts("PASS bounded eight-row metadata-only console counters; snapshots do not consume output");

    feed(1024);
    assert(command("read", replacement, "513") == 1);
    assert(snapshot(replacement).output_bytes_pending == 1024);
    assert(command("read", replacement, "0") == 1);
    assert(command("read", replacement, NULL) == 0);
    assert(strstr(console_output, "read 512 bytes: 00010203") && !strchr(console_output, '\033'));
    assert(snapshot(replacement).output_bytes_pending == 512);
    assert(command("read", replacement, "1") == 0);
    assert(snapshot(replacement).output_bytes_pending == 511);
    puts("PASS console read bounds and binary-safe hexadecimal rendering");
    disconnect_all();
    assert(command("counters", 0, NULL) == 0);
    comparison(1); comparison(2);
    assert(allocations == initial_allocations);
    puts("PASS no post-init broker allocations; 7 diagnostic groups passed");
    cleanup_allocations();
    return 0;
}