/* Реальный UDP/JOIN: авторизация сервера, несколько каналов и подтверждение именно сервером. */
#include "../lib/platform_compat.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../src/utun_instance.h"
#include "../src/chat/chat_core.h"
#include "../src/chat/chat_event.h"
#include "../src/chat/chat_sync.h"
#include "../src/chat/invite_build.h"
#include "../src/chat/invite_link.h"
#include "../src/routing_layer/topo_node_sqlite.h"
#include "../src/transport_layer/etcp.h"
#include "../src/transport_layer/etcp_api.h"
#include "test_clock.h"
#include <stdlib.h>
#include <string.h>

#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "server join test line=%d: %s", __LINE__, #x); abort(); } } while (0)
static struct UTUN_INSTANCE *client, *server;
static unsigned events[32], info_requests;
static int results[32], hold_ready, hold_result, hold_request;
static uint8_t ready_wire[64], result_wire[64], request_wire[INVITE_LINK_MAX + 64];
static size_t ready_size, result_size, request_size;

int __real_etcp_send(struct ETCP_CONN*, struct ll_entry*);
int __wrap_etcp_send(struct ETCP_CONN* conn, struct ll_entry* entry) {
    if (entry->len >= 18 && entry->dgram[0] == ETCP_ID_CHAT_SYNC) {
        uint8_t type = entry->dgram[9];
        if (type == CS_MSG_JOIN_INFO_REQ && conn->instance == server) info_requests++;
        uint8_t* buffer = NULL;
        if (type == CS_MSG_SERVER_JOIN_REQ) {
            CHECK(entry->len <= sizeof(request_wire)); request_size = entry->len;
            memcpy(request_wire, entry->dgram, request_size);
            if (hold_request) buffer = request_wire;
        } else if (type == CS_MSG_JOIN_READY && hold_ready && conn->instance == client) {
            CHECK(entry->len <= sizeof(ready_wire)); ready_size = entry->len; buffer = ready_wire;
        } else if (type == CS_MSG_SERVER_JOIN_RESULT && hold_result) {
            CHECK(entry->len <= sizeof(result_wire)); result_size = entry->len; buffer = result_wire;
        }
        if (buffer) {
            memcpy(buffer, entry->dgram, entry->len);
            queue_dgram_free(entry); queue_entry_free(entry); return 0;
        }
    }
    return __real_etcp_send(conn, entry);
}

static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len) {
    if (inst != client || type != CHAT_EVT_CONNECT_RESULT) return;
    CHECK(len == 28);
    uint64_t request; int32_t result;
    memcpy(&request, data + 20, 8); memcpy(&result, data + 8, 4);
    CHECK(request < 32); events[request]++; results[request] = result;
}

static void deliver(struct UTUN_INSTANCE* target, struct UTUN_INSTANCE* source, const uint8_t* bytes, size_t size) {
    struct ETCP_CONN* conn = instance_find_conn(target, source->node_id); CHECK(conn);
    struct ll_entry* entry = ll_alloc_lldgram(size); CHECK(entry);
    memcpy(entry->dgram, bytes, size); entry->len = (uint16_t)size;
    target->api_bindings.callbacks[ETCP_ID_CHAT_SYNC](conn, entry);
}

static void wait_value(struct UASYNC* ua, const unsigned* value) {
    uint64_t deadline = get_time_tb() + 400000;
    while (!*value && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(*value);
}

static int has_channel(uint64_t channel) {
    char name[64]; snprintf(name, sizeof(name), "%llu", (unsigned long long)channel);
    return topo_node_sqlite_channel_get(server->topo_sqlite_db, name, NULL, 0, NULL, NULL, NULL, NULL) == 0;
}

static void wait_packet(struct UASYNC* ua, const size_t* size) {
    uint64_t deadline = get_time_tb() + 100000;
    while (!*size && get_time_tb() < deadline) test_clock_poll(ua, 100);
    CHECK(*size);
}

int main(void) {
    debug_config_init(); debug_set_level(getenv("UTUN_TEST_DEBUG") ? DEBUG_LEVEL_DEBUG : DEBUG_LEVEL_WARN);
    debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO);
    debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_DEBUG);
    test_clock_init(1000);
    size_t baseline = u_get_allocated_count();
    struct UASYNC* ua = uasync_create(); CHECK(ua); utun_instance_set_tun_init_enabled(0);
    const char* base = "[global]\nmy_node_name=server-join-client\n[ntp]\nenabled=no\n"
        "[server: udp]\naddr=127.0.0.1:0\ntype=public\n[allowed_keys]\nallow_all=yes\n"
        "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n";
    client = utun_instance_create_from_str(ua, base);
    char config[1024]; snprintf(config, sizeof(config), "%sadd_key=%s\n", base,
        "5757575757575757575757575757575757575757575757575757575757575757");
    server = utun_instance_create_from_str(ua, config); CHECK(client && server);
    CHECK(!utun_core_start(client) && !chat_service_start(client));
    CHECK(!utun_core_start(server) && !chat_service_start(server));
    chat_event_set_handler(client, event);
    uint64_t first = chat_core_create_channel_auto(client, "first"); CHECK(first);
    uint64_t second = chat_core_create_channel_auto(client, "second"); CHECK(second && second != first);
    uint64_t third = chat_core_create_channel_auto(client, "third"); CHECK(third);
    char link[INVITE_LINK_MAX], wrong[INVITE_LINK_MAX]; struct InviteData target;
    CHECK(!chat_server_link_build(server, link, sizeof(link)));
    CHECK(!invite_link_decode(link, strlen(link), &target, NULL, 0));
    CHECK(target.kind == INVITE_KIND_SERVER && target.addrCount && target.addrs[0].port);
    CHECK(chat_sync_join_link(client, 1, link, strlen(link)) == CHAT_CONNECT_INVALID && !events[1]);
    target.add_key[0] ^= 1; CHECK(invite_link_encode(&target, NULL, wrong, sizeof(wrong)) > 0);
    CHECK(!chat_sync_offer_link(client, 2, first, wrong, strlen(wrong))); wait_value(ua, &events[2]);
    CHECK(events[2] == 1 && results[2] == CHAT_CONNECT_AUTH && !has_channel(first) && !info_requests);

    hold_ready = hold_result = 1;
    CHECK(!chat_sync_offer_link(client, 3, first, link, strlen(link)));
    wait_packet(ua, &ready_size); CHECK(!events[3] && !result_size && has_channel(first) && info_requests);
    /* Инвайтер уже сохранил S, но S ещё не получил JOIN_READY. */
    hold_ready = 0; deliver(server, client, ready_wire, ready_size);
    wait_packet(ua, &result_size); CHECK(!events[3]);
    uint8_t stale[64]; memcpy(stale, result_wire, result_size); stale[10] ^= 1;
    deliver(client, server, stale, result_size); CHECK(!events[3]);
    hold_result = 0; deliver(client, server, result_wire, result_size);
    CHECK(events[3] == 1 && !results[3]);
    deliver(client, server, result_wire, result_size); CHECK(events[3] == 1);

    CHECK(!chat_sync_offer_link(client, 4, second, link, strlen(link))); wait_value(ua, &events[4]);
    CHECK(events[4] == 1 && !results[4] && has_channel(first) && has_channel(second));
    CHECK(!chat_sync_offer_link(client, 5, first, link, strlen(link))); wait_value(ua, &events[5]);
    CHECK(events[5] == 1 && !results[5]);

    /* Ротация: старая ссылка не добавляет новый канал; существующие сохраняются. */
    server->config->global.chatserver_add_key[0] ^= 1;
    CHECK(!chat_sync_offer_link(client, 6, third, link, strlen(link))); wait_value(ua, &events[6]);
    CHECK(results[6] == CHAT_CONNECT_AUTH && !has_channel(third) && has_channel(first) && has_channel(second));
    CHECK(!chat_server_link_build(server, link, sizeof(link)));
    hold_request = 1; request_size = 0;
    CHECK(!chat_sync_offer_link(client, 7, third, link, strlen(link))); wait_packet(ua, &request_size);
    /* Каждый усечённый запрос должен закончиться отказом до создания канала. */
    for (size_t cut = 18; cut < request_size; cut++) deliver(server, client, request_wire, cut);
    CHECK(!has_channel(third));
    hold_request = 0; deliver(server, client, request_wire, request_size); wait_value(ua, &events[7]);
    /* Первый отказ от усечённого пакета завершил исходный request ровно один раз. */
    CHECK(events[7] == 1 && (results[7] == CHAT_CONNECT_AUTH || results[7] == CHAT_CONNECT_INVALID));
    test_clock_run_for(ua, 20000);
    CHECK(!chat_sync_offer_link(client, 8, third, link, strlen(link))); wait_value(ua, &events[8]);
    CHECK(events[8] == 1 && !results[8] && has_channel(third));

    /* Отмена ожидания после отправки не откатывает уже выполненный JOIN. */
    hold_request = 1; request_size = 0;
    CHECK(!chat_sync_offer_link(client, 9, first, link, strlen(link))); wait_packet(ua, &request_size);
    chat_sync_cancel_join(client, 9); CHECK(events[9] == 1 && results[9] == CHAT_CONNECT_CANCELLED);
    hold_request = 0;
    CHECK(!chat_sync_offer_link(client, 10, first, link, strlen(link))); wait_value(ua, &events[10]);
    CHECK(events[10] == 1 && !results[10]);

    /* Отмена до callback регистрации освобождает builder; следующий запрос работает. */
    CHECK(!chat_sync_offer_link(client, 11, first, link, strlen(link)));
    chat_sync_cancel_join(client, 11); CHECK(events[11] == 1 && results[11] == CHAT_CONNECT_CANCELLED);
    test_clock_run_for(ua, 1000); CHECK(events[11] == 1);

    /* Молчащий сервер ограничен общим deadline; повтор может использовать тот же транспорт. */
    hold_request = 1; request_size = 0;
    CHECK(!chat_sync_offer_link(client, 12, first, link, strlen(link))); wait_packet(ua, &request_size);
    test_clock_run_for(ua, 310000); CHECK(events[12] == 1 && results[12] == CHAT_CONNECT_TIMEOUT);
    hold_request = 0;
    CHECK(!chat_sync_offer_link(client, 13, first, link, strlen(link))); wait_value(ua, &events[13]);
    CHECK(events[13] == 1 && !results[13]);

    /* Остановка сервиса отменяет ожидающую команду ровно один раз. */
    hold_request = 1; request_size = 0;
    CHECK(!chat_sync_offer_link(client, 14, first, link, strlen(link))); wait_packet(ua, &request_size);
    chat_service_stop(client); CHECK(events[14] == 1 && results[14] == CHAT_CONNECT_CANCELLED);
    hold_request = 0;
    CHECK(!chat_service_start(client));
    CHECK(!chat_sync_offer_link(client, 15, first, link, strlen(link))); wait_value(ua, &events[15]);
    CHECK(events[15] == 1 && !results[15]);

    utun_instance_destroy(server); utun_instance_destroy(client); uasync_poll(ua, 0); uasync_destroy(ua, 0);
    CHECK(u_get_allocated_count() == baseline); test_clock_finish();
    DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "PASS server join: auth, ordinary JOIN, confirmation, channels, rotation, framing, replay, cancel, timeout, restart");
    return 0;
}