/* Реальные callbacks, БД, lifecycle и uasync; наблюдаем только границу запуска member_sync.
 * Private-реализация включена как в test_merkle_protocol: публичный API не расширяем ради теста. */
#include "chat/member_sync.h"
#include "transport_layer/etcp_api.h"
#include "../lib/u_async.h"
#include "../lib/ll_queue.h"
#include "test_clock.h"
#include <assert.h>
#include <stdio.h>
#include <string.h>

static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg);
static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry);
static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name);
#define member_sync_start observe_start
#define etcp_send discard_send
#define uasync_set_timeout schedule_timeout
#include "../src/chat/chat_sync.c"
#undef member_sync_start
#undef etcp_send
#undef uasync_set_timeout

struct fixture {
    struct UASYNC* ua;
    struct UTUN_INSTANCE* inst;
    struct ETCP_CONN* peers[3];
    char channel[2][21];
    uint64_t gid[2];
};

static struct { uint64_t peer, channel; } starts[128];
static unsigned start_count;
static int fail_timer;
static struct fixture* injecting;

static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg) {
    (void)inst; (void)cb; (void)arg;
    assert(start_count < sizeof(starts) / sizeof(starts[0]));
    starts[start_count].peer = peer; starts[start_count++].channel = strtoull(channel, NULL, 10);
    if (injecting) {
        struct fixture* f = injecting; injecting = NULL;
        cs_on_conn_down(f->peers[1], ETCP_CBK_EVENT_DOWN, f->inst->chat_sync);
        cs_schedule_peer_sync(f->inst->chat_sync, f->peers[2]->peer_node_id, "during_flush");
    }
    return 0;
}

static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry) {
    (void)conn;
    queue_dgram_free(entry); queue_entry_free(entry);
    return 0;
}

static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name) {
    if (fail_timer && !strcmp(name, "cs_sync")) { fail_timer = 0; return NULL; }
    return uasync_set_timeout(ua, tb, arg, cb, name);
}

static void run_ms(struct fixture* f, unsigned ms) { test_clock_run_for(f->ua, ms * 10); }

/* Два канала: A состоит только в первом, B в обоих, C только во втором. */
static void setup(struct fixture* f) {
    memset(f, 0, sizeof(*f)); start_count = 0; fail_timer = 0; injecting = NULL;
    f->ua = uasync_create(); assert(f->ua);
    f->inst = utun_instance_create_from_str(f->ua,
        "[global]\nmy_node_name=schedule test\n[server: udp]\naddr=127.0.0.1:0\ntype=public\n"
        "[ntp]\nenabled=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n");
    assert(f->inst && utun_core_start(f->inst) == 0 && chat_service_start(f->inst) == 0);
    chat_core_create_channel_auto(f->inst, "schedule 1"); chat_core_create_channel_auto(f->inst, "schedule 2");
    unsigned channels = 0;
    for (struct ll_entry* e = f->inst->topo_groups->group_list->head; e; e = e->next) {
        struct TOPO_GROUP* group = (struct TOPO_GROUP*)e;
        assert(group->group_type == TOPO_GROUP_TYPE_CHAT && channels < 2);
        snprintf(f->channel[channels], sizeof(f->channel[channels]), "%llu", (unsigned long long)group->group_id);
        f->gid[channels++] = group->group_id;
    }
    assert(channels == 2);
    for (unsigned i = 0; i < 3; i++) {
        struct SC_MYKEYS keys; assert(sc_generate_keypair(&keys) == SC_OK);
        struct ETCP_CONN* conn = etcp_connection_create(f->inst, "schedule peer"); assert(conn);
        conn->peer_node_id = sc_derive_node_id_from_pubkey(keys.public_key);
        conn->initialized = 1; conn->links_up = 1; conn->state = 1;
        assert(sc_init_ctx(&conn->crypto_ctx, &f->inst->my_keys) == SC_OK);
        assert(sc_set_peer_public_key(&conn->crypto_ctx, keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK);
        assert(etcp_conn_index_peer(conn) == 0); f->peers[i] = conn;
        uint8_t signature[64] = {0};
        for (unsigned ch = 0; ch < 2; ch++) {
            if ((i == 0 && ch == 1) || (i == 2 && ch == 0)) continue;
            assert(topo_node_sqlite_member_block_put(f->inst->topo_sqlite_db, f->channel[ch], conn->peer_node_id,
                signature, 1, signature, 1, keys.public_key, keys.public_key, "{}", NULL, 0, signature) == 0);
        }
    }
    /* Кеш намеренно не обновляем: задания должны видеть актуальное членство в БД. */
}

static void cleanup(struct fixture* f) {
    utun_instance_destroy(f->inst); uasync_poll(f->ua, 0);
    assert(f->ua->timer_alloc_count == f->ua->timer_free_count);
    assert(f->ua->socket_alloc_count == f->ua->socket_free_count);
    uasync_destroy(f->ua, 0);
}

static unsigned peer_starts(struct fixture* f, unsigned peer) {
    unsigned count = 0;
    for (unsigned i = 0; i < start_count; i++) if (starts[i].peer == f->peers[peer]->peer_node_id) count++;
    return count;
}

static void test_reinit_isolation(void) {
    struct fixture f; setup(&f);
    for (unsigned i = 0; i < 3; i++) etcp_fire_conn_status(f.peers[i], ETCP_CONN_STATUS_UP);
    assert(!start_count); run_ms(&f, 1100);
    assert(start_count == 4 && peer_starts(&f, 0) == 1 && peer_starts(&f, 1) == 2 && peer_starts(&f, 2) == 1);
    start_count = 0;
    for (unsigned i = 0; i < 12; i++) {
        etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_REINIT); run_ms(&f, 200);
        assert(!peer_starts(&f, 1) && !peer_starts(&f, 2));
    }
    run_ms(&f, 1100); assert(start_count == 1 && starts[0].channel == f.gid[0]);
    run_ms(&f, 4000); assert(start_count == 1);
    cleanup(&f); puts("PASS: repeated REINIT is isolated; unchanged peers and idle service start no new rounds");
}

static void test_coalesce_and_deadline(void) {
    struct fixture f; setup(&f);
    etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    for (unsigned i = 0; i < 7; i++) {
        run_ms(&f, 100); etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_REINIT);
        etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    }
    run_ms(&f, 400);
    assert(start_count == 3 && peer_starts(&f, 0) == 1 && peer_starts(&f, 1) == 2 && !peer_starts(&f, 2));
    cleanup(&f); puts("PASS: duplicate pairs coalesce and other peers cannot move the first deadline");
}

static void test_sleep_and_cancel(void) {
    struct fixture f; setup(&f);
    etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    f.peers[0]->peer_sleep_phase = 1; utun_fire_peer_sleep_cbk(f.inst, f.peers[0]->peer_node_id, 1);
    run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 1) == 2);
    start_count = 0; f.peers[0]->peer_sleep_phase = 0;
    utun_fire_peer_sleep_cbk(f.inst, f.peers[0]->peer_node_id, 0);
    etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); run_ms(&f, 1100);
    assert(start_count == 1 && peer_starts(&f, 0) == 1);
    start_count = 0;
    etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    f.peers[0]->links_up = 0; etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_DOWN);
    run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 1) == 2);
    start_count = 0;
    etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    char sql[160]; snprintf(sql, sizeof(sql), "UPDATE peers_%s SET deleted=1 WHERE node_id=%lld", f.channel[0],
        (long long)f.peers[1]->peer_node_id);
    assert(sqlite3_exec(f.inst->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK);
    run_ms(&f, 1100); assert(start_count == 1 && starts[0].channel == f.gid[1]);
    cleanup(&f); puts("PASS: sleep, DOWN and changed membership cancel only the affected requests");
}

static void test_join_existing_transport(void) {
    struct fixture f; setup(&f);
    cs_send_join_ready(f.inst->chat_sync, f.channel[0], f.peers[0]->peer_node_id);
    assert(!start_count); run_ms(&f, 1100);
    assert(start_count == 1 && starts[0].peer == f.peers[0]->peer_node_id && starts[0].channel == f.gid[0]);
    start_count = 0;
    cs_handle_join_ready(f.inst->chat_sync, f.peers[1]->peer_node_id, f.channel[0], (uint8_t*)&f.gid[0], 8);
    run_ms(&f, 1100);
    assert(start_count == 1 && starts[0].peer == f.peers[1]->peer_node_id && starts[0].channel == f.gid[0]);
    cleanup(&f); puts("PASS: both JOIN_READY sides start only their pair without a new transport UP");
}

static void test_reentrant_cancel(void) {
    struct fixture f; setup(&f);
    etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    injecting = &f; run_ms(&f, 1100);
    assert(start_count == 1 && peer_starts(&f, 0) == 1);
    run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 2) == 1 && !peer_starts(&f, 1));
    cleanup(&f); puts("PASS: callbacks can cancel remaining pairs and defer newly queued work");
}

static void test_lifecycle_and_timer_failure(void) {
    struct fixture f; setup(&f);
    fail_timer = 1; etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP);
    assert(!f.inst->chat_sync->sync_pending && !f.inst->chat_sync->sync_timer);
    etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    chat_sync_on_channel_deleted(f.inst, f.channel[0]); run_ms(&f, 1100);
    assert(start_count == 1 && starts[0].channel == f.gid[1]);
    start_count = 0; etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP);
    chat_service_stop(f.inst); run_ms(&f, 1100); assert(!start_count);
    assert(chat_service_start(f.inst) == 0); assert(!start_count);
    run_ms(&f, 1100); assert(start_count == 4);
    cleanup(&f); puts("PASS: timer failure recovers; channel/stop cancellation and startup retain correct scope");
}

int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
    debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG);
    test_clock_init(1000);
    size_t baseline = u_get_allocated_count();
    test_reinit_isolation(); test_coalesce_and_deadline(); test_sleep_and_cancel();
    test_join_existing_transport(); test_reentrant_cancel(); test_lifecycle_and_timer_failure();
    test_clock_finish();
    assert(u_get_allocated_count() == baseline);
    puts("ALL PASS: targeted chat synchronization scheduling"); return 0;
}