/* Реальный chat_sync/NCD/SQLite с управляемым транспортом: отказы, корреляция и teardown. */
#include "utun_instance.h"
#include "chat/chat_core.h"
#include "chat/chat_sync.h"
#include "chat/chat_join.h"
#include "chat/db_sync.h"
#include "routing_layer/topo_group.h"
#include "chat/chat_event.h"
#include "chat/chat_join_port.h"
#include "chat/invite_link.h"
#include "transport_layer/etcp.h"
#include "transport_layer/etcp_api.h"
#include "transport_layer/etcp_connections.h"
#include "transport_layer/node_conn_direct.h"
#include "routing_layer/topo_node.h"
#include "../lib/mem.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "join fault line=%d: %s", __LINE__, #x); exit(1); } } while (0)
static struct UTUN_INSTANCE* inst;
static struct UASYNC* ua;
static struct ETCP_CONN* conn;
static char invitation[INVITE_LINK_MAX];
static uint8_t request_wire[512], response_wire[8192], member_wire[8192], ready_wire[32];
static size_t response_size, member_size, ready_size;
static struct UTUN_INSTANCE* server;
static struct ETCP_CONN* server_conn;
static uint64_t channel;
static size_t request_size;
static unsigned allocations, fail_allocation, events[512];
static int results[512], fail_send, reenter, reenter_result;
static const char* fail_timer;
static int fail_runtime, fail_ncd, nested_up;
static unsigned info_sends;
static pthread_mutex_t race_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t race_condition = PTHREAD_COND_INITIALIZER;
static int race_gate, race_waiting, race_release, race_result;
static uint64_t peer;

void* __real_u_calloc_impl(uint32_t, uint32_t, const char*);
void* __real_u_malloc_impl(uint32_t, const char*);
static int allocation_failed(const char* site) {
    if (fail_runtime == 1 && strstr(site, "db_sync.c")) { fail_runtime = 0; return 1; }
    if (!strstr(site, "chat_sync.c") && !strstr(site, "chat_join_port.c")) return 0;
    allocations++;
    return fail_allocation && allocations == fail_allocation;
}
void* __wrap_u_calloc_impl(uint32_t count, uint32_t size, const char* site) {
    if (allocation_failed(site)) return NULL;
    void* data = __real_u_calloc_impl(count, size, site);
    if (race_gate && strstr(site, "chat_join_port.c")) {
        pthread_mutex_lock(&race_lock); race_waiting = 1; pthread_cond_signal(&race_condition);
        while (!race_release) pthread_cond_wait(&race_condition, &race_lock);
        pthread_mutex_unlock(&race_lock);
    }
    return data;
}
void* __wrap_u_malloc_impl(uint32_t size, const char* site) {
    if (allocation_failed(site)) return NULL;
    return __real_u_malloc_impl(size, site);
}
void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
void* __wrap_uasync_set_timeout(struct UASYNC* loop, int delay, void* arg, timeout_callback_t fn, const char* name) {
    if (fail_timer && !strcmp(name, fail_timer)) return NULL;
    return __real_uasync_set_timeout(loop, delay, arg, fn, name);
}
struct TOPO_GROUP* __real_topo_groups_create_group(struct TOPO_GROUPS*, uint64_t, uint8_t, const char*);
struct TOPO_GROUP* __wrap_topo_groups_create_group(struct TOPO_GROUPS* groups, uint64_t id, uint8_t type, const char* name) {
    if (fail_runtime == 2) { fail_runtime = 0; return NULL; }
    return __real_topo_groups_create_group(groups, id, type, name);
}
void __real_topo_group_add_node_cbk(struct TOPO_GROUP*, topo_node_event_fn, void*);
void __wrap_topo_group_add_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg) {
    if (fail_runtime == 3) { fail_runtime = 0; return; }
    __real_topo_group_add_node_cbk(group, fn, arg);
}
int __real_node_conn_direct_open_node(struct UTUN_INSTANCE*, uint64_t, ncd_callback, void*, struct NODE_CONN_DIRECT**, struct TOPO_NODE*, struct ETCP_SOCKET*);
int __wrap_node_conn_direct_open_node(struct UTUN_INSTANCE* owner, uint64_t id, ncd_callback cb, void* arg,
    struct NODE_CONN_DIRECT** handle, struct TOPO_NODE* node, struct ETCP_SOCKET* socket) {
    if (fail_ncd) { *handle = NULL; return NCD_ERR; }
    if (nested_up && owner == inst) { nested_up = 0; etcp_fire_conn_status(conn, ETCP_CONN_STATUS_UP); }
    return __real_node_conn_direct_open_node(owner, id, cb, arg, handle, node, socket);
}
int __real_etcp_send(struct ETCP_CONN*, struct ll_entry*);
int __wrap_etcp_send(struct ETCP_CONN* c, struct ll_entry* entry) {
    if (entry->len >= 18 && entry->dgram[0] == ETCP_ID_CHAT_SYNC) {
        if (entry->dgram[9] == CS_MSG_JOIN_INFO_REQ) {
            if (c->instance == inst) info_sends++;
            CHECK(entry->len <= sizeof(request_wire)); request_size = entry->len;
            memcpy(request_wire, entry->dgram, request_size);
            if (fail_send) return -1;
        }
        if (entry->dgram[9] == CS_MSG_JOIN_INFO_RESP) { response_size = entry->len; CHECK(response_size <= sizeof(response_wire)); memcpy(response_wire, entry->dgram, response_size); }
        if (entry->dgram[9] == CS_MSG_JOIN_REQUEST) { member_size = entry->len; CHECK(member_size <= sizeof(member_wire)); memcpy(member_wire, entry->dgram, member_size); }
        if (entry->dgram[9] == CS_MSG_JOIN_READY) { ready_size = entry->len; CHECK(ready_size <= sizeof(ready_wire)); memcpy(ready_wire, entry->dgram, ready_size); }
        queue_dgram_free(entry); queue_entry_free(entry); return 0;
    }
    return __real_etcp_send(c, entry);
}
static void event(struct UTUN_INSTANCE* owner, int type, const uint8_t* data, int len) {
    if (type != CHAT_EVT_CONNECT_RESULT) return;
    CHECK(owner == inst && len == 28);
    uint64_t request; int32_t result;
    memcpy(&request, data + 20, 8); memcpy(&result, data + 8, 4);
    CHECK(request < 512); events[request]++; results[request] = result;
    if (reenter && request == 30) {
        reenter = 0; reenter_result = chat_sync_join_link(inst, 31, invitation, strlen(invitation));
    }
}
static void deliver_error(const uint8_t token[8]) {
    struct ll_entry* entry = ll_alloc_lldgram(18); CHECK(entry);
    entry->len = 18; entry->dgram[0] = ETCP_ID_CHAT_SYNC;
    uint64_t group = htobe64(channel); memcpy(entry->dgram + 1, &group, 8);
    entry->dgram[9] = CS_MSG_ERROR; memcpy(entry->dgram + 10, token, 8);
    inst->api_bindings.callbacks[ETCP_ID_CHAT_SYNC](conn, entry);
}
static void deliver(struct UTUN_INSTANCE* target, struct ETCP_CONN* connection, const uint8_t* bytes, size_t size) {
    struct ll_entry* entry = ll_alloc_lldgram(size); CHECK(entry); entry->len = size; memcpy(entry->dgram, bytes, size);
    target->api_bindings.callbacks[ETCP_ID_CHAT_SYNC](connection, entry);
}
static void description(void) {
    response_size = 0; deliver(server, server_conn, request_wire, request_size); CHECK(response_size > 200);
}
static int channels(void) {
    sqlite3_stmt* st = NULL; CHECK(sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT count(*) FROM channels", -1, &st, NULL) == SQLITE_OK);
    CHECK(sqlite3_step(st) == SQLITE_ROW); int value = sqlite3_column_int(st, 0); sqlite3_finalize(st); return value;
}
static void* producer(void* arg) {
    struct chat_join_port* port = arg;
    CHECK(!chat_join_port_submit(port, 60, invitation, strlen(invitation)));
    return NULL;
}
static void* stopped_producer(void* arg) {
    race_result = chat_join_port_submit(arg, 71, invitation, strlen(invitation)); return NULL;
}
static void stop_in_batch(void* arg) { chat_service_stop(arg); }
int main(void) {
    size_t before = u_get_allocated_count();
    debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
    ua = uasync_create(); CHECK(ua);
    inst = utun_instance_create_from_str(ua, "[global]\nauto_sockets=no\n[ntp]\nenabled=no\n"
        "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n");
    CHECK(inst && !utun_core_start(inst) && !chat_service_start(inst));
    chat_event_set_handler(inst, event);
    server = utun_instance_create_from_str(ua, "[global]\nauto_sockets=no\n[ntp]\nenabled=no\n"
        "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n");
    CHECK(server && !utun_core_start(server) && !chat_service_start(server));
    chat_core_create_channel_auto(server, "fault channel");
    CHECK(server->topo_groups->group_list->head);
    channel = ((struct TOPO_GROUP*)server->topo_groups->group_list->head)->group_id;
    CHECK(!chat_join_store_local_key(server, channel, 123));
    struct InviteData invite = { .channelId = channel, .join_key = 123, .addrCount = 4 };
    memcpy(invite.pubkey, server->my_keys.public_key, 32); peer = server->node_id;
    for (unsigned i = 0; i < invite.addrCount; i++) {
        invite.addrs[i].family = i & 1 ? 6 : 4; invite.addrs[i].port = 12345;
        invite.addrs[i].proto = 1; invite.addrs[i].address[0] = 127; invite.addrs[i].address[3] = 1;
    }
    invite.reality.has = 1; memcpy(invite.reality.server_pubkey, server->my_keys.public_key, 32);
    snprintf(invite.reality.server_name, sizeof(invite.reality.server_name), "example.org"); invite.addrs[3].proto = 6;
    CHECK(invite_link_encode(&invite, NULL, invitation, sizeof(invitation)) > 0);
    conn = etcp_connection_create(inst, "join fault transport"); CHECK(conn);
    conn->peer_node_id = peer; CHECK(!etcp_conn_index_peer(conn));
    CHECK(sc_init_ctx(&conn->crypto_ctx, &inst->my_keys) == SC_OK);
    CHECK(sc_set_peer_public_key(&conn->crypto_ctx, invite.pubkey, SC_PEER_PUBKEY_BIN) == SC_OK);
    conn->links_up = 1; etcp_conn_ready(conn); conn->initialized = 1;
    struct NODE_CONN_DIRECT* other_owner = NULL;
    CHECK(node_conn_direct_open(inst, peer, NULL, NULL, &other_owner, NULL) != NCD_ERR);
    uasync_drain_immediate(ua);
    server_conn = etcp_connection_create(server, "join server fixture"); CHECK(server_conn);
    server_conn->peer_node_id = inst->node_id; CHECK(!etcp_conn_index_peer(server_conn));
    CHECK(sc_init_ctx(&server_conn->crypto_ctx, &server->my_keys) == SC_OK);
    CHECK(sc_set_peer_public_key(&server_conn->crypto_ctx, inst->my_keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK);
    server_conn->links_up = 1; etcp_conn_ready(server_conn); server_conn->initialized = 1;
    CHECK(!channels());
    CHECK(chat_sync_join_link(inst, 1, "utun://", 7) < 0 && !channels() && !events[1]);

    CHECK(!chat_sync_join_link(inst, 10, invitation, strlen(invitation)) && request_size > 18);
    uint8_t old[8]; memcpy(old, request_wire + 10, 8);
    CHECK(chat_sync_join_link(inst, 11, invitation, strlen(invitation)) == CHAT_CONNECT_BUSY && !events[11]);
    chat_sync_cancel_join(inst, 999); CHECK(!events[10]);
    chat_sync_cancel_join(inst, 10); CHECK(events[10] == 1 && results[10] == CHAT_CONNECT_CANCELLED);
    CHECK(!chat_sync_join_link(inst, 12, invitation, strlen(invitation)));
    uint8_t current[8]; memcpy(current, request_wire + 10, 8); CHECK(memcmp(old, current, 8));
    deliver_error(old); CHECK(!events[12]);
    deliver_error(current); CHECK(events[12] == 1 && results[12] == CHAT_CONNECT_ERROR);
    deliver_error(current); CHECK(events[12] == 1);
    CHECK(node_conn_direct_get_conn(other_owner) == conn && !conn->close_requested);
    CHECK(!channels());

    reenter = 1; CHECK(!chat_sync_join_link(inst, 30, invitation, strlen(invitation)));
    chat_sync_cancel_join(inst, 30); CHECK(!reenter_result && events[30] == 1);
    chat_sync_cancel_join(inst, 31); CHECK(events[31] == 1);

    /* Сначала измерить все реальные точки выделения этого пути, затем отказать в каждой. */
    allocations = 0; CHECK(!chat_sync_join_link(inst, 40, invitation, strlen(invitation)));
    unsigned total = allocations; CHECK(total >= 5); chat_sync_cancel_join(inst, 40);
    for (unsigned i = 1; i <= total; i++) {
        allocations = 0; fail_allocation = i;
        int rc = chat_sync_join_link(inst, 41, invitation, strlen(invitation));
        fail_allocation = 0;
        CHECK(rc < 0 && !events[41] && !channels());
        CHECK(node_conn_direct_get_conn(other_owner) == conn && !conn->close_requested);
        CHECK(!chat_sync_join_link(inst, 42, invitation, strlen(invitation))); chat_sync_cancel_join(inst, 42);
    }
    for (unsigned i = 0; i < 2; i++) {
        fail_timer = i ? "cs_info_req" : "join_transport_timeout";
        CHECK(chat_sync_join_link(inst, 43, invitation, strlen(invitation)) < 0 && !events[43]); fail_timer = NULL;
        CHECK(!chat_sync_join_link(inst, 44, invitation, strlen(invitation))); chat_sync_cancel_join(inst, 44);
    }
    fail_send = 1; CHECK(chat_sync_join_link(inst, 45, invitation, strlen(invitation)) < 0 && !events[45]); fail_send = 0;
    CHECK(!chat_sync_join_link(inst, 46, invitation, strlen(invitation))); chat_sync_cancel_join(inst, 46);

    fail_ncd = 1; CHECK(chat_sync_join_link(inst, 47, invitation, strlen(invitation)) < 0 && !events[47]); fail_ncd = 0;
    CHECK(node_conn_direct_get_conn(other_owner) == conn && !conn->close_requested);
    unsigned previous_sends = info_sends; nested_up = 1;
    CHECK(!chat_sync_join_link(inst, 48, invitation, strlen(invitation)) && !nested_up && info_sends == previous_sends + 1);
    chat_sync_cancel_join(inst, 48); CHECK(events[48] == 1);

    /* Сервер обслуживает чужой JOIN, пока сам ждёт завершения локального JOIN. */
    struct InviteData reverse = invite;
    memcpy(reverse.pubkey, inst->my_keys.public_key, 32);
    char reverse_link[INVITE_LINK_MAX]; CHECK(invite_link_encode(&reverse, NULL, reverse_link, sizeof(reverse_link)) > 0);
    CHECK(!chat_sync_join_link(server, 90, reverse_link, strlen(reverse_link)));
    CHECK(!sqlite3_exec(inst->topo_sqlite_db, "CREATE TEMP TRIGGER reject_join BEFORE INSERT ON channels "
        "BEGIN SELECT RAISE(ABORT,'join fault'); END", NULL, NULL, NULL));
    CHECK(!chat_sync_join_link(inst, 80, invitation, strlen(invitation))); description();
    deliver(inst, conn, response_wire, response_size);
    CHECK(events[80] == 1 && results[80] == CHAT_CONNECT_ERROR && !channels() && sqlite3_get_autocommit(inst->topo_sqlite_db));
    CHECK(!sqlite3_exec(inst->topo_sqlite_db, "DROP TRIGGER reject_join", NULL, NULL, NULL));
    CHECK(!chat_sync_join_link(inst, 81, invitation, strlen(invitation))); description();
    size_t signature_offset = 19 + response_wire[18] + 8 + 32 + 32 + 64 + 8 + 32 + 32 + 64 + 8;
    CHECK(signature_offset + 64 < response_size); response_wire[signature_offset] ^= 1;
    deliver(inst, conn, response_wire, response_size);
    CHECK(events[81] == 1 && results[81] == CHAT_CONNECT_ERROR && !channels() && sqlite3_get_autocommit(inst->topo_sqlite_db));
    CHECK(!topo_groups_find(inst->topo_groups, channel));
    for (unsigned failure = 1; failure <= 3; failure++) {
        uint64_t request = 85 + failure;
        CHECK(!chat_sync_join_link(inst, request, invitation, strlen(invitation))); description();
        fail_runtime = (int)failure; deliver(inst, conn, response_wire, response_size); CHECK(!fail_runtime);
        CHECK(events[request] == 1 && results[request] == CHAT_CONNECT_ERROR && channels() == 1);
        CHECK(!topo_groups_find(inst->topo_groups, channel) && !db_sync_instance_find(inst, channel));
    }
    CHECK(!chat_sync_join_link(inst, 82, invitation, strlen(invitation))); description();
    fail_timer = "cs_join"; deliver(inst, conn, response_wire, response_size); fail_timer = NULL;
    CHECK(events[82] == 1 && results[82] == CHAT_CONNECT_ERROR && channels() == 1);
    CHECK(!chat_sync_join_link(inst, 83, invitation, strlen(invitation))); description();
    member_size = 0; deliver(inst, conn, response_wire, response_size); CHECK(member_size > 200 && !events[83]);
    ready_size = 0; deliver(server, server_conn, member_wire, member_size); CHECK(ready_size == 26);
    deliver(inst, conn, ready_wire, ready_size); CHECK(events[83] == 1 && !results[83]);
    deliver(inst, conn, ready_wire, ready_size); CHECK(events[83] == 1);
    uint8_t previous_ready[32]; memcpy(previous_ready, ready_wire, ready_size);
    for (unsigned failure = 0; failure < 2; failure++) {
        uint64_t request = 84 + failure;
        CHECK(!chat_sync_join_link(inst, request, invitation, strlen(invitation))); description();
        deliver(inst, conn, response_wire, response_size); CHECK(!events[request]);
        deliver(inst, conn, previous_ready, 26); CHECK(!events[request]);
        deliver(server, server_conn, member_wire, member_size); CHECK(ready_size == 26);
        if (failure) { memset(ready_wire + 18, 0, 8); deliver(inst, conn, ready_wire, 26); }
        else deliver(inst, conn, ready_wire, 25);
        CHECK(events[request] == 1 && results[request] == CHAT_CONNECT_ERROR);
    }
    chat_sync_cancel_join(server, 90);

    struct chat_join_port* port = chat_sync_join_port(inst); CHECK(port);
    allocations = 0; fail_allocation = 1;
    CHECK(chat_join_port_submit(port, 50, invitation, strlen(invitation)) < 0 && !events[50]); fail_allocation = 0;
    pthread_t thread; CHECK(!pthread_create(&thread, NULL, producer, port)); CHECK(!pthread_join(thread, NULL));
    CHECK(chat_join_port_submit(port, 60, invitation, strlen(invitation)) == CHAT_CONNECT_BUSY);
    chat_service_stop(inst); CHECK(events[60] == 1 && results[60] == CHAT_CONNECT_CANCELLED);
    CHECK(chat_join_port_submit(port, 61, invitation, strlen(invitation)) == CHAT_CONNECT_CANCELLED);
    CHECK(!chat_service_start(inst)); CHECK(chat_join_port_submit(port, 62, invitation, strlen(invitation)) == CHAT_CONNECT_CANCELLED);
    chat_join_port_release(port);

    port = chat_sync_join_port(inst); CHECK(port);
    CHECK(!chat_join_port_submit(port, 65, invitation, strlen(invitation))); uasync_poll(ua, 0);
    CHECK(!events[65] && chat_join_port_submit(port, 65, invitation, strlen(invitation)) == CHAT_CONNECT_BUSY);
    CHECK(!chat_join_port_submit(port, 66, invitation, strlen(invitation))); uasync_poll(ua, 0);
    CHECK(events[66] == 1 && results[66] == CHAT_CONNECT_BUSY && !events[65]);
    chat_sync_cancel_join(inst, 65); CHECK(events[65] == 1 && results[65] == CHAT_CONNECT_CANCELLED);
    /* Детерминированная гонка: stop между копированием аргументов producer и публикацией post. */
    race_gate = 1; CHECK(!pthread_create(&thread, NULL, stopped_producer, port));
    pthread_mutex_lock(&race_lock);
    while (!race_waiting) pthread_cond_wait(&race_condition, &race_lock);
    pthread_mutex_unlock(&race_lock);
    chat_service_stop(inst);
    pthread_mutex_lock(&race_lock); race_release = 1; pthread_cond_signal(&race_condition); pthread_mutex_unlock(&race_lock);
    CHECK(!pthread_join(thread, NULL)); race_gate = 0;
    CHECK(race_result == CHAT_CONNECT_CANCELLED && !events[71]); chat_join_port_release(port);
    CHECK(!chat_service_start(inst));

    /* close после извлечения списка post: отмена не находит task, пустой callback обязан освободить его. */
    port = chat_sync_join_port(inst); CHECK(port);
    struct posted_task* stop = u_calloc(1, sizeof(*stop)); CHECK(stop);
    stop->callback = stop_in_batch; stop->arg = inst; uasync_post_reserved(ua, stop);
    CHECK(!chat_join_port_submit(port, 70, invitation, strlen(invitation)));
    uasync_poll(ua, 0); CHECK(events[70] == 1 && results[70] == CHAT_CONNECT_CANCELLED);
    chat_join_port_release(port);
    node_conn_direct_close(other_owner);
    utun_instance_destroy(server); utun_instance_destroy(inst); uasync_poll(ua, 0); uasync_destroy(ua, 0);
    CHECK(u_get_allocated_count() == before);
    printf("join faults: %u allocation failures, deadlines, send, stale replies, reentry, ownership, thread and restart PASS\n", total);
    return 0;
}