utun2/tests/test_chat_poll_network.c

/* Три ядра и UDP: A ↔ B ↔ C, голосование/реакции и возврат C после остановки чата. */
#include "utun_instance.h"
#include "chat/chat_core.h"
#include "chat/chat_poll.h"
#include "chat/chat_feedback.h"
#include "chat/member_sync.h"
#include "chat/merkle_sync.h"
#include "chat/db_sync.h"
#include "routing_layer/topo_group.h"
#include "routing_layer/topo_node_sqlite.h"
#include "transport_layer/node_conn_direct.h"
#include "transport_layer/etcp_connections.h"
#include "transport_layer/etcp.h"
#include "../lib/platform_compat.h"
#include "../lib/socket_compat.h"
#include "../lib/debug_config.h"
#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

static struct UASYNC* ua;
static struct UTUN_INSTANCE* nodes[3];
static char channel[21], message_hex[65];
static uint8_t message_id[32];

/* Порт выбирается ОС, затем сразу используется новым ядром. */
static unsigned free_port(void) {
    socket_t fd = socket(AF_INET, SOCK_DGRAM, 0); assert(fd != SOCKET_INVALID);
    struct sockaddr_in address = {0}; address.sin_family = AF_INET; address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
    assert(bind(fd, (struct sockaddr*)&address, sizeof(address)) == 0); socklen_t size = sizeof(address);
    assert(getsockname(fd, (struct sockaddr*)&address, &size) == 0); unsigned port = ntohs(address.sin_port);
    socket_close_wrapper(fd); return port;
}

static void hex(const uint8_t* data, size_t len, char* out) {
    for (size_t i = 0; i < len; i++) snprintf(out + i * 2, 3, "%02x", data[i]);
}

static struct UTUN_INSTANCE* create_node(unsigned port) {
    struct SC_MYKEYS keys; assert(sc_generate_keypair(&keys) == SC_OK);
    char private_key[65], public_key[65], config[1024]; hex(keys.private_key, 32, private_key); hex(keys.public_key, 32, public_key);
    snprintf(config, sizeof(config),
        "[global]\nmy_private_key=%s\nmy_public_key=%s\nkeepalive_adaptive=0\n"
        "[server: udp]\naddr=127.0.0.1:%u\ntype=public\n[allowed_keys]\nallow_all=1\n"
        "[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n", private_key, public_key, port);
    struct UTUN_INSTANCE* inst = utun_instance_create_from_str(ua, config);
    assert(inst && utun_core_start(inst) == 0 && chat_service_start(inst) == 0); return inst;
}

/* Все три заранее принятых участника имеют настоящие подписи root/join. */
static void seed_members(uint8_t channel_x[32], uint8_t channel_ed[32], uint8_t owner_private[32]) {
    for (int dst = 0; dst < 3; dst++) for (int author = 0; author < 3; author++) {
        uint8_t body[112], join[64], tree[64];
        assert(member_sync_build_join_msg(channel_x, channel_ed, nodes[author]->node_id,
            nodes[author]->my_keys.public_key, 1, body, sizeof(body)) == 112);
        assert(sc_ed25519_sign(nodes[author]->my_ed25519_privkey, body, sizeof(body), join) == SC_OK);
        assert(member_sync_sign_pubkey(owner_private, nodes[author]->my_keys.public_key, tree) == 0);
        assert(member_sync_put(nodes[dst], channel, nodes[author]->node_id, nodes[author]->my_keys.public_key,
            nodes[author]->my_ed25519_pubkey, join, 1, NULL, 0, "{}", NULL, NULL, 0, 0, tree) >= 0);
    }
}

/* Поднимаем транспорт только соседних пар; C получает данные A через B. */
static struct NODE_CONN_DIRECT* connect_node(int from, int to, unsigned port) {
    struct TOPO_ADDR4 address = { .addr = {127, 0, 0, 1}, .port = port, .protocol = TOPO_PROTO_UDP };
    struct TOPO_NODE node = { .node_id = nodes[to]->node_id, .v4_addrs = &address };
    memcpy(node.public_key, nodes[to]->my_keys.public_key, 32); struct NODE_CONN_DIRECT* handle = NULL;
    assert(node_conn_direct_open_node(nodes[from], node.node_id, NULL, NULL, &handle, &node, NULL) != NCD_ERR); return handle;
}

static int wait_ready(struct TOPO_PEER_REQUEST* handle) {
    uint64_t deadline = get_time_tb() + 15u * 10000u;
    while (get_time_tb() < deadline) {
        if (topo_group_peer_phase(handle) == TOPO_PEER_READY) return 1;
        uasync_poll(ua, 5);
    }
    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "poll network: group readiness timeout phase=%d", topo_group_peer_phase(handle)); return 0;
}

/* C открывает групповой запрос после входящего UP; до него адрес B ещё не получен. */
static int wait_transport(struct NODE_CONN_DIRECT* ab, struct NODE_CONN_DIRECT* bc) {
    struct ETCP_CONN *a = NULL, *b = NULL;
    uint64_t deadline = get_time_tb() + 15u * 10000u;
    while (get_time_tb() < deadline) {
        a = node_conn_direct_get_conn(ab); b = node_conn_direct_get_conn(bc);
        struct ETCP_CONN* c = instance_find_conn(nodes[2], nodes[1]->node_id);
        if (a && b && c && a->initialized && b->initialized && c->initialized
            && a->links_up && b->links_up && c->links_up) return 1;
        uasync_poll(ua, 5);
    }
    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "poll network: bootstrap timeout ab=%p bc=%p", (void*)a, (void*)b); return 0;
}

static sqlite3_int64 field(struct UTUN_INSTANCE* inst, const char* json, const char* path) {
    sqlite3_stmt* st = NULL; assert(sqlite3_prepare_v2(inst->topo_sqlite_db, "SELECT json_extract(?1,?2)", -1, &st, NULL) == SQLITE_OK);
    sqlite3_bind_text(st, 1, json, -1, SQLITE_STATIC); sqlite3_bind_text(st, 2, path, -1, SQLITE_STATIC);
    assert(sqlite3_step(st) == SQLITE_ROW); sqlite3_int64 n = sqlite3_column_int64(st, 0); sqlite3_finalize(st); return n;
}

static int wait_result(int index, int voters, int option_one, int reactions) {
    uint64_t deadline = get_time_tb() + 15u * 10000u;
    while (get_time_tb() < deadline) {
        char* result = NULL;
        if (chat_feedback_snapshot(nodes[index], channel, message_id, &result) == 0) {
            int ok = !field(nodes[index], result, "$.pending") && field(nodes[index], result, "$.poll.voters") == voters
                && field(nodes[index], result, "$.poll.options[1].count") == option_one
                && field(nodes[index], result, "$.reactions[0].count") == reactions;
            u_free(result); if (ok) return 1;
        }
        uasync_poll(ua, 5);
    }
    char* result = NULL; chat_feedback_snapshot(nodes[index], channel, message_id, &result);
    DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "poll network: result timeout node=%d expected voters=%d option1=%d reactions=%d actual=%s",
        index, voters, option_one, reactions, result ? result : "missing"); u_free(result); return 0;
}

static void run_command(int index, const char* command, const char* tail) {
    char request[1024]; snprintf(request, sizeof(request), "{\"cmd\":\"%s\",\"ch\":\"%s\",\"message_id\":\"%s\"%s}",
        command, channel, message_hex, tail);
    char* result = NULL; assert(chat_feedback_command(nodes[index], request, &result) == 0); u_free(result);
}

static void subscribe(int a, int b) {
    assert(member_sync_start(nodes[a], nodes[b]->node_id, channel, NULL, NULL) == 0);
    assert(merkle_sync_start(nodes[a], MT_DATA_FEEDBACK, nodes[b]->node_id, channel, NULL, NULL) == 0);
    db_sync_instance_set_gated(db_sync_instance_find(nodes[a], strtoull(channel, NULL, 10)), 0);
    db_sync_instance_set_gated(db_sync_instance_find(nodes[b], strtoull(channel, NULL, 10)), 0);
    db_sync_reinitiate(db_sync_instance_find(nodes[a], strtoull(channel, NULL, 10)), nodes[b]->node_id);
}

int main(void) {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
    debug_set_category_level(DEBUG_CATEGORY_CHAT_SYNC, DEBUG_LEVEL_DEBUG);
    debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_INFO);
    ua = uasync_create(); assert(ua); unsigned ports[3];
    for (int i = 0; i < 3; i++) { ports[i] = free_port(); nodes[i] = create_node(ports[i]); }
    chat_core_create_channel_auto(nodes[0], "Poll network");
    sqlite3_stmt* st = NULL;
    assert(sqlite3_prepare_v2(nodes[0]->topo_sqlite_db, "SELECT channel_id FROM channels", -1, &st, NULL) == SQLITE_OK);
    assert(sqlite3_step(st) == SQLITE_ROW); snprintf(channel, sizeof(channel), "%s", sqlite3_column_text(st, 0)); sqlite3_finalize(st);
    uint8_t x[32], ed[32], signature[64], owner_private[32];
    assert(topo_node_sqlite_channel_get(nodes[0]->topo_sqlite_db, channel, NULL, 0, NULL, x, ed, signature) == 0);
    assert(topo_node_sqlite_channel_get_priv(nodes[0]->topo_sqlite_db, channel, owner_private) == 0);
    for (int i = 1; i < 3; i++) {
        assert(topo_node_sqlite_channel_put(nodes[i]->topo_sqlite_db, channel, "Poll network", nodes[0]->node_id,
            x, NULL, ed, NULL, signature) == 0); chat_core_ensure_channel_ready(nodes[i], channel);
    }
    seed_members(x, ed, owner_private);
    struct NODE_CONN_DIRECT* ab = connect_node(0, 1, ports[1]), *bc = connect_node(1, 2, ports[2]);
    assert(wait_transport(ab, bc));
    struct TOPO_PEER_REQUEST *groups[3] = {0}; uint64_t gid = strtoull(channel, NULL, 10);
    assert(topo_group_peer_open(topo_groups_find(nodes[0]->topo_groups, gid), nodes[1]->node_id, &groups[0]) == 0);
    assert(topo_group_peer_open(topo_groups_find(nodes[1]->topo_groups, gid), nodes[2]->node_id, &groups[1]) == 0);
    assert(topo_group_peer_open(topo_groups_find(nodes[2]->topo_groups, gid), nodes[1]->node_id, &groups[2]) == 0);
    for (int i = 0; i < 3; i++) assert(wait_ready(groups[i]));
    subscribe(0, 1); subscribe(1, 2);
    struct chat_poll poll = { .mode = CHAT_POLL_MULTIPLE, .allow_custom_answer = 1, .option_count = 2 };
    strcpy(poll.question, "Сетевой опрос"); strcpy(poll.options[0], "Первый"); strcpy(poll.options[1], "Второй");
    assert(chat_poll_create(nodes[0], channel, &poll, message_id) == 0); hex(message_id, 32, message_hex);
    assert(wait_result(2, 0, 0, 0));
    run_command(0, "vote", ",\"choices\":[0]"); run_command(1, "vote", ",\"choices\":[1]");
    run_command(2, "vote", ",\"custom\":\"Ответ C 👍\""); run_command(0, "react", ",\"emoji\":0,\"on\":true");
    assert(wait_result(0, 3, 1, 1) && wait_result(2, 3, 1, 1));
    puts("PASS: signed poll, votes, custom Unicode answer and reactions converge through the intermediate node");
    topo_group_peer_close(groups[2]); groups[2] = NULL; chat_service_stop(nodes[2]);
    run_command(0, "vote", ",\"choices\":[1]"); run_command(1, "vote", ",\"choices\":[]");
    run_command(0, "react", ",\"clear\":true"); assert(wait_result(1, 2, 1, 0));
    assert(chat_service_start(nodes[2]) == 0);
    assert(topo_group_peer_open(topo_groups_find(nodes[2]->topo_groups, gid), nodes[1]->node_id, &groups[2]) == 0);
    assert(wait_ready(groups[2])); subscribe(1, 2);
    assert(wait_result(2, 2, 1, 0) && wait_result(0, 2, 1, 0));
    struct chat_feedback_record old;
    assert(chat_feedback_get(nodes[2], channel, message_id, nodes[0]->node_id, CHAT_FEEDBACK_REACTION, &old) == 1);
    assert(old.ver == 2 && old.data_len == 0);
    puts("PASS: returning peer receives new versions; withdrawn votes and reactions stay withdrawn");
    for (int i = 0; i < 3; i++) topo_group_peer_close(groups[i]);
    node_conn_direct_close(ab); node_conn_direct_close(bc);
    for (int i = 0; i < 3; i++) utun_instance_destroy(nodes[i]);
    uasync_poll(ua, 0); assert(ua->timer_alloc_count == ua->timer_free_count && ua->socket_alloc_count == ua->socket_free_count);
    uasync_destroy(ua, 0); puts("ALL PASS: network polls and offline recovery"); return 0;
}