/* test_ha_repl_ack.c — Unity tests for Task 6.2: ha_repl_ack_received quorum logic.
 *
 * Tests verify (Req 8.2, 8.3):
 *   Req 8.2: When acks reach quorum-1 threshold, ha_promote_to_primary is called.
 *   Req 8.3: CTRL_PRIMARY opens full traffic (dispatch gate PRIMARY case).
 *
 * Quorum formula: quorum = (n_peers + 2) / 2
 *   n_peers=1 (solo)  → quorum=1 → threshold=0 (quorum-1=0) → first ack promotes
 *   n_peers=2 (3-node)→ quorum=2 → threshold=1 → first ack promotes
 *   n_peers=3 (4-node)→ quorum=2 → threshold=1 → first ack promotes
 *   n_peers=4 (5-node)→ quorum=3 → threshold=2 → second ack promotes (acks >= quorum-1)
 *
 * Boundary: ha.c unit tests (ha_repl_ack_received, ha_dispatch_gate)
 * Requirements: 8.2, 8.3
 */

#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>

#include <unity/unity.h>

#include "ha.h"
#include "xce/cluster_def.h"

/* ha_repl_ack_received declaration (defined in ha.c) */
void ha_repl_ack_received(ha_state_t *ha, cluster_t *c);

/* ha_dispatch_gate declaration (defined in ha.c) */
int ha_dispatch_gate(ha_state_t *ha, int opcode);

/* OPCode constants needed */
#include "controller.capnp.h"

/* ── Shared HA config ──────────────────────────────────────────────────── */

static ha_config_t g_ack_cfg = {
    .heartbeat_interval_ms    = 3000,
    .heartbeat_timeout_ms     = 10000,
    .election_timeout_ms      = 2000,
    .probe_connect_timeout_ms = 500,
    .max_promotion_retries    = 5,
    .election_jitter_max_ms   = 0,
    .election_priority_step_ms = 500,
    .committing_timeout_ms    = 10000,
};

/* ── Helper: build a minimally-zeroed cluster_t with ha_init applied ─── */

static void init_cluster_committing(cluster_t *c, int n_peers)
{
    memset(c, 0, sizeof(*c));
    ha_init(&c->ha, &g_ack_cfg, 0);
    c->ha.n_peers = n_peers;
    /* Simulate being in COMMITTING state (election won, waiting for acks) */
    atomic_store_explicit(&c->ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
                          memory_order_release);
    atomic_store_explicit(&c->ha.repl_ack_count, 0, memory_order_release);
}

/* ── Req 8.2: 2-node cluster (n_peers=2, quorum=2, threshold=1) ──────── */
/* With 2 peers (3 total nodes), quorum=2, threshold=quorum-1=1.
 * First ack reaches threshold → promote. */

static void test_repl_ack_2node_first_ack_promotes(void)
{
    /* n_peers=2 means 3 nodes total (self + 2 peers).
     * quorum = (2+2)/2 = 2; threshold = quorum-1 = 1.
     * First ack (count becomes 1 >= 1) → ha_promote_to_primary called. */
    cluster_t c;
    init_cluster_committing(&c, 2);

    ha_repl_ack_received(&c.ha, &c);

    int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
        "2-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)");

    ha_fini(&c.ha);
    printf("test_repl_ack_2node_first_ack_promotes: PASS\n");
}

/* ── Req 8.2: 2-node cluster — 0 acks → stays COMMITTING ────────────── */

static void test_repl_ack_2node_no_ack_stays_committing(void)
{
    /* With 0 acks, role must remain CTRL_PRIMARY_COMMITTING. */
    cluster_t c;
    init_cluster_committing(&c, 2);

    /* Do NOT call ha_repl_ack_received — verify initial state */
    int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
        "2-node: 0 acks → role must remain CTRL_PRIMARY_COMMITTING (Req 8.2)");

    ha_fini(&c.ha);
    printf("test_repl_ack_2node_no_ack_stays_committing: PASS\n");
}

/* ── Req 8.2: 4-node cluster (n_peers=3, quorum=2, threshold=1) ──────── */
/* quorum = (3+2)/2 = 2; threshold=1. First ack promotes. */

static void test_repl_ack_4node_first_ack_promotes(void)
{
    cluster_t c;
    init_cluster_committing(&c, 3);

    ha_repl_ack_received(&c.ha, &c);

    int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
        "4-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)");

    ha_fini(&c.ha);
    printf("test_repl_ack_4node_first_ack_promotes: PASS\n");
}

/* ── Req 8.2: 5-node cluster (n_peers=4, quorum=3, threshold=2) ──────── */
/* quorum = (4+2)/2 = 3; threshold=2.
 * First ack (count=1) below threshold → stays COMMITTING.
 * Second ack (count=2) reaches threshold → promotes. */

static void test_repl_ack_5node_first_ack_stays_committing(void)
{
    cluster_t c;
    init_cluster_committing(&c, 4);

    ha_repl_ack_received(&c.ha, &c);   /* ack count → 1, threshold=2 → no promote */

    int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
        "5-node: first ack (count=1 < threshold=2) must stay COMMITTING (Req 8.2)");

    ha_fini(&c.ha);
    printf("test_repl_ack_5node_first_ack_stays_committing: PASS\n");
}

static void test_repl_ack_5node_second_ack_promotes(void)
{
    cluster_t c;
    init_cluster_committing(&c, 4);

    ha_repl_ack_received(&c.ha, &c);   /* ack count → 1 */
    /* Reset role back to COMMITTING so we can test from COMMITTING again.
     * (In production ha_promote_to_primary isn't called twice; here we need
     * to test the threshold on the second call.) */
    /* We start clean from count=1 and add one more via a second call. */

    ha_fini(&c.ha);

    /* Fresh cluster with pre-populated ack_count=1 to test the crossing point */
    cluster_t c2;
    init_cluster_committing(&c2, 4);
    atomic_store_explicit(&c2.ha.repl_ack_count, 1, memory_order_release);

    ha_repl_ack_received(&c2.ha, &c2);  /* ack count → 2 >= threshold=2 → promote */

    int role = (int)atomic_load_explicit(&c2.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
        "5-node: second ack (count=2 >= threshold=2) must promote to CTRL_PRIMARY (Req 8.2)");

    ha_fini(&c2.ha);
    printf("test_repl_ack_5node_second_ack_promotes: PASS\n");
}

/* ── Req 8.2: null cluster_t — ha_repl_ack_received must not crash ───── */
/* Design doc: ha_repl_ack_received(ha, c=NULL) increments ack but skips promote. */

static void test_repl_ack_null_cluster_increments_count(void)
{
    ha_state_t ha;
    ha_init(&ha, &g_ack_cfg, 0);
    ha.n_peers = 2;
    atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
                          memory_order_release);
    atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release);

    /* c=NULL: must not crash; quorum check still fires but ha_promote_to_primary
     * is guarded by `if (c)` in ha_repl_ack_received. */
    ha_repl_ack_received(&ha, NULL);

    int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, acks,
        "repl_ack with NULL cluster must still increment repl_ack_count (Req 8.2)");

    /* Role should NOT have changed to PRIMARY (no cluster to promote via) */
    int role = (int)atomic_load_explicit(&ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
        "repl_ack with NULL cluster must NOT promote role (no cluster_t available)");

    ha_fini(&ha);
    printf("test_repl_ack_null_cluster_increments_count: PASS\n");
}

/* ── Req 8.3: after promotion, dispatch gate allows full traffic ──────── */
/* Once ha_repl_ack_received fires ha_promote_to_primary, role = CTRL_PRIMARY.
 * ha_dispatch_gate in CTRL_PRIMARY state must allow job submission. */

static void test_repl_ack_after_promote_gate_allows_jobsub(void)
{
    cluster_t c;
    init_cluster_committing(&c, 2);

    /* Trigger promotion via ack */
    ha_repl_ack_received(&c.ha, &c);

    /* Verify role is now PRIMARY */
    int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
        "after ack promote: role must be CTRL_PRIMARY (Req 8.2)");

    /* Dispatch gate in CTRL_PRIMARY must allow OPCode_jobsub (Req 8.3) */
    int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
        "CTRL_PRIMARY dispatch gate must allow OPCode_jobsub (Req 8.3)");

    ha_fini(&c.ha);
    printf("test_repl_ack_after_promote_gate_allows_jobsub: PASS\n");
}

/* ── Req 8.3: before promotion, gate still blocks jobsub ─────────────── */
/* In COMMITTING state, before threshold is reached, job ops are blocked. */

static void test_repl_ack_committing_gate_blocks_jobsub_until_promote(void)
{
    /* 5-node cluster: threshold=2. First ack leaves us in COMMITTING. */
    cluster_t c;
    init_cluster_committing(&c, 4);

    ha_repl_ack_received(&c.ha, &c);  /* count=1, threshold=2 → still COMMITTING */

    int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
        "5-node after 1 ack: must remain COMMITTING before threshold (Req 8.2)");

    /* Gate must block job submission while still COMMITTING */
    int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub);
    TEST_ASSERT_NOT_EQUAL_MESSAGE(0, rc,
        "COMMITTING gate must block OPCode_jobsub (Req 8.3, 5.1)");

    ha_fini(&c.ha);
    printf("test_repl_ack_committing_gate_blocks_jobsub_until_promote: PASS\n");
}

/* ── Req 8.2: repl_ack_count accumulates correctly across calls ──────── */

static void test_repl_ack_count_increments_monotonically(void)
{
    ha_state_t ha;
    ha_init(&ha, &g_ack_cfg, 0);
    ha.n_peers = 10; /* large cluster so quorum > 3 calls */
    /* n_peers=10 → quorum=(10+2)/2=6 → threshold=5 */
    atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
                          memory_order_release);
    atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release);

    ha_repl_ack_received(&ha, NULL); /* count → 1 */
    ha_repl_ack_received(&ha, NULL); /* count → 2 */
    ha_repl_ack_received(&ha, NULL); /* count → 3 */

    int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire);
    TEST_ASSERT_EQUAL_INT_MESSAGE(3, acks,
        "repl_ack_count must monotonically increment with each ack call (Req 8.2)");

    ha_fini(&ha);
    printf("test_repl_ack_count_increments_monotonically: PASS\n");
}

/* ── Req 8.2: null ha pointer — must not crash ────────────────────────── */

static void test_repl_ack_null_ha_is_noop(void)
{
    /* ha_repl_ack_received(NULL, ...) must return without crashing */
    ha_repl_ack_received(NULL, NULL);
    printf("test_repl_ack_null_ha_is_noop: PASS\n");
}

/* ── Test suite entry point ───────────────────────────────────────────── */

void test_ha_repl_ack(void)
{
    /* Req 8.2 — quorum threshold: 2-node (3 total nodes) */
    RUN_TEST(test_repl_ack_2node_first_ack_promotes);
    RUN_TEST(test_repl_ack_2node_no_ack_stays_committing);

    /* Req 8.2 — quorum threshold: 4-node (n_peers=3) */
    RUN_TEST(test_repl_ack_4node_first_ack_promotes);

    /* Req 8.2 — quorum threshold: 5-node (n_peers=4, threshold=2) */
    RUN_TEST(test_repl_ack_5node_first_ack_stays_committing);
    RUN_TEST(test_repl_ack_5node_second_ack_promotes);

    /* Req 8.2 — null cluster guard and ack count accumulation */
    RUN_TEST(test_repl_ack_null_cluster_increments_count);
    RUN_TEST(test_repl_ack_count_increments_monotonically);
    RUN_TEST(test_repl_ack_null_ha_is_noop);

    /* Req 8.3 — dispatch gate reflects promotion state */
    RUN_TEST(test_repl_ack_after_promote_gate_allows_jobsub);
    RUN_TEST(test_repl_ack_committing_gate_blocks_jobsub_until_promote);
}