/* test_ha_durability.c — Unity RED-phase tests for ha-term.bin persist/restore.
 *
 * Tests verify:
 *   Req 9.1: ha_persist_term() writes ha-term.bin with correct magic, term,
 *            voted_for_term, voted_for_name, and valid CRC32.
 *   Req 9.2: ha_restore_term() reads ha-term.bin and restores term and voted_for
 *            into ha_state_t (RED: ha_restore_term does not exist yet).
 *   Req 9.3: Missing ha-term.bin → term=0, voted_for_name empty (safe first-boot).
 *   Req 9.2: CRC mismatch in ha-term.bin → zeroes state safely (no stale data).
 *   Req 9.1: write+rename atomicity — .tmp file cleaned up; ha-term.bin consistent.
 *
 * Functions under test:
 *   ha_persist_term()   — already implemented in ha.c (task 4.2)
 *   ha_restore_term()   — NOT YET IMPLEMENTED; declared below to force linker error
 *
 * RED phase proof: ha_restore_term() does not exist in ha.c. The linker will
 * fail with "undefined reference to ha_restore_term" — this is the expected
 * RED-phase outcome for task 6.1. Task 6.2 adds the implementation.
 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>

#include <zlib.h>
#include <unity/unity.h>

#include "ha.h"
#include "config.h"    /* XCE_MAXNAME_LEN */
#include "xce/cluster_def.h"  /* cluster_t, for bug-1684's raft_on_vote_request test */

/* ── On-disk format (mirrors ha.c internals, duplicated here for white-box tests) ─
 *
 * This struct matches ha_term_file_t in ha.c.  It is intentionally duplicated
 * here (not exported via ha.h) so the tests can construct and inspect raw file
 * contents without coupling ha.h to internal storage details.
 */
#define HA_TERM_MAGIC_TEST 0xDEADBEEFCAFEBABEULL

typedef struct {
    uint64_t magic;
    uint64_t term;
    uint64_t voted_for_term;
    char     voted_for_name[XCE_MAXNAME_LEN];
    uint32_t crc32;
} ha_term_file_test_t;

/* ── RED phase declaration ────────────────────────────────────────────────────
 *
 * ha_restore_term: read {workdir}/ha-term.bin and restore ha->term and
 * ha->voted_for_* if the file exists and passes CRC verification.
 *
 * On missing file:       no-op (term stays 0, voted_for_name stays empty).
 * On CRC mismatch:       log WARN, zero ha->term and ha->voted_for_* for safety.
 * On valid file:         atomically store ha->term; copy voted_for_term/name.
 *
 * Returns 0 on success (including "file not found" and "CRC bad → zeroed"),
 *         -errno on unexpected I/O errors.
 *
 * This declaration exists only to make the tests compile; the linker will
 * fail with "undefined reference to ha_restore_term" until task 6.2 lands.
 */
int ha_restore_term(ha_state_t *ha, const char *workdir);

/* ── Shared minimal ha_config_t ───────────────────────────────────────────── */

static ha_config_t g_durability_cfg = {
    .heartbeat_interval_ms    = 3000,
    .heartbeat_timeout_ms     = 10000,
    .election_timeout_ms      = 2000,
    .probe_connect_timeout_ms = 500,
    .max_promotion_retries    = 5,
};

/* ── Helper: compute CRC32 the same way ha_persist_term() does ──────────────
 *
 * CRC covers all bytes before the crc32 field (i.e. offsetof(…, crc32)).
 */
static uint32_t compute_term_crc(const ha_term_file_test_t *f)
{
    size_t crc_len = offsetof(ha_term_file_test_t, crc32);
    return (uint32_t)crc32(0L, (const Bytef *)f, (uInt)crc_len);
}

/* ── Helper: write an ha_term_file_test_t directly to {dir}/ha-term.bin ──── */

static void write_term_file(const char *dir, const ha_term_file_test_t *f)
{
    char path[1024];
    snprintf(path, sizeof(path), "%s/ha-term.bin", dir);
    int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
    TEST_ASSERT_MESSAGE(fd >= 0, "Failed to open ha-term.bin for writing");
    ssize_t written = write(fd, f, sizeof(*f));
    close(fd);
    TEST_ASSERT_EQUAL_size_t(sizeof(*f), (size_t)written);
}

/* ── Test 1: ha_persist_term() writes ha-term.bin with correct content ──────
 *
 * Req 9.1: persist must be called before reporting a vote grant or term increment.
 * We set up an ha_state_t, call ha_persist_term(), then read back and verify:
 *   - file exists
 *   - magic == HA_TERM_MAGIC
 *   - term and voted_for fields match what was in ha_state_t
 *   - CRC32 is valid
 */
static void test_persist_term_writes_file(void)
{
    char tmpdir[] = "/tmp/test_ha_durXXXXXX";
    char *dir = mkdtemp(tmpdir);
    TEST_ASSERT_NOT_NULL_MESSAGE(dir, "mkdtemp failed");

    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    /* Set up some state to persist */
    atomic_store_explicit(&ha.term, 7ULL, memory_order_release);
    ha.voted_for_term = 7;
    strncpy(ha.voted_for_name, "ctrl-node-2", sizeof(ha.voted_for_name) - 1);
    ha.voted_for_name[sizeof(ha.voted_for_name) - 1] = '\0';

    ha_persist_term(&ha, dir);

    /* Verify the file exists */
    char path[1024];
    snprintf(path, sizeof(path), "%s/ha-term.bin", dir);
    struct stat st;
    int stat_rc = stat(path, &st);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat_rc, "ha-term.bin should exist after persist");
    TEST_ASSERT_EQUAL_size_t(sizeof(ha_term_file_test_t), (size_t)st.st_size);

    /* Read back and verify content */
    ha_term_file_test_t f;
    memset(&f, 0, sizeof(f));
    int fd = open(path, O_RDONLY);
    TEST_ASSERT_MESSAGE(fd >= 0, "Cannot open ha-term.bin for reading");
    ssize_t n = read(fd, &f, sizeof(f));
    close(fd);
    TEST_ASSERT_EQUAL_size_t(sizeof(f), (size_t)n);

    TEST_ASSERT_EQUAL_UINT64(HA_TERM_MAGIC_TEST, f.magic);
    TEST_ASSERT_EQUAL_UINT64(7ULL, f.term);
    TEST_ASSERT_EQUAL_UINT64(7ULL, f.voted_for_term);
    TEST_ASSERT_EQUAL_STRING("ctrl-node-2", f.voted_for_name);

    /* CRC must be valid */
    uint32_t expected_crc = compute_term_crc(&f);
    TEST_ASSERT_EQUAL_UINT32(expected_crc, f.crc32);

    /* tmp file must NOT exist (rename succeeded) */
    char tmp_path[1040];
    snprintf(tmp_path, sizeof(tmp_path), "%s/ha-term.bin.tmp", dir);
    TEST_ASSERT_EQUAL_INT(-1, stat(tmp_path, &st));

    ha_fini(&ha);

    /* cleanup */
    unlink(path);
    rmdir(dir);

    printf("test_persist_term_writes_file: PASS\n");
}

/* ── Test 2: ha_restore_term() restores term and voted_for from valid file ──
 *
 * Req 9.2: startup must restore the most recently persisted term and voted_for
 * record before participating in any election.
 *
 * This test writes a valid ha_term_file_test_t directly to disk (bypassing
 * ha_persist_term — this exercises the read path in isolation), then calls
 * ha_restore_term().  Verification: ha->term and ha->voted_for_name match
 * what was written.
 *
 * RED PHASE: this test will FAIL at link time because ha_restore_term()
 * does not exist in ha.c.  This is intentional — linker error proves RED.
 */
static void test_ha_restore_term_from_file(void)
{
    char tmpdir[] = "/tmp/test_ha_durXXXXXX";
    char *dir = mkdtemp(tmpdir);
    TEST_ASSERT_NOT_NULL_MESSAGE(dir, "mkdtemp failed");

    /* Build a valid ha_term_file_test_t and write it directly */
    ha_term_file_test_t f;
    memset(&f, 0, sizeof(f));
    f.magic          = HA_TERM_MAGIC_TEST;
    f.term           = 42ULL;
    f.voted_for_term = 42ULL;
    strncpy(f.voted_for_name, "ctrl-node-1", sizeof(f.voted_for_name) - 1);
    f.voted_for_name[sizeof(f.voted_for_name) - 1] = '\0';
    f.crc32 = compute_term_crc(&f);
    write_term_file(dir, &f);

    /* Init an ha_state_t (starts at term=0) then restore */
    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    /* Req 9.2: ha_restore_term() should set ha.term and ha.voted_for_* */
    int rc = ha_restore_term(&ha, dir);    /* <-- undefined reference: RED proof */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_restore_term must return 0 on valid file");

    /* After restore, term and voted_for must match the file */
    uint64_t restored_term = atomic_load_explicit(&ha.term, memory_order_acquire);
    TEST_ASSERT_EQUAL_UINT64(42ULL, restored_term);
    TEST_ASSERT_EQUAL_UINT64(42ULL, ha.voted_for_term);
    TEST_ASSERT_EQUAL_STRING("ctrl-node-1", ha.voted_for_name);

    ha_fini(&ha);

    /* cleanup */
    char path[1024];
    snprintf(path, sizeof(path), "%s/ha-term.bin", dir);
    unlink(path);
    rmdir(dir);

    printf("test_ha_restore_term_from_file: PASS\n");
}

/* ── Test 3: Missing ha-term.bin → term=0, voted_for empty (first-boot) ─────
 *
 * Req 9.3: on first boot where no ha-term.bin exists, the system shall
 * initialize the term to zero with no prior vote recorded.
 *
 * This test verifies the "file not found" branch of ha_restore_term().
 * It may compile to a link error (RED) even before we can test the logic.
 */
static void test_ha_restore_term_zero_on_missing_file(void)
{
    /* Use a tmpdir with no ha-term.bin in it */
    char tmpdir[] = "/tmp/test_ha_durXXXXXX";
    char *dir = mkdtemp(tmpdir);
    TEST_ASSERT_NOT_NULL_MESSAGE(dir, "mkdtemp failed");

    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    /* No file written — restore should be a safe no-op */
    int rc = ha_restore_term(&ha, dir);    /* <-- undefined reference: RED proof */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_restore_term must return 0 on missing file");

    /* term must be 0, voted_for_name must be empty */
    uint64_t term = atomic_load_explicit(&ha.term, memory_order_acquire);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(0ULL, term, "term must be 0 on first boot");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, ha.voted_for_name[0],
                                  "voted_for_name must be empty on first boot");

    ha_fini(&ha);
    rmdir(dir);

    printf("test_ha_restore_term_zero_on_missing_file: PASS\n");
}

/* ── Test 4: CRC mismatch → zeroes state safely ─────────────────────────────
 *
 * Req 9.2: a corrupted ha-term.bin must not restore stale state. The system
 * must log WARN and zero ha->term and ha->voted_for_*.
 *
 * We write a file with term=99 but deliberately corrupt the CRC field.
 * After ha_restore_term(), term must be 0 and voted_for_name must be empty.
 */
static void test_ha_restore_crc_mismatch_zeroes_state(void)
{
    char tmpdir[] = "/tmp/test_ha_durXXXXXX";
    char *dir = mkdtemp(tmpdir);
    TEST_ASSERT_NOT_NULL_MESSAGE(dir, "mkdtemp failed");

    /* Build a file with valid structure but wrong CRC */
    ha_term_file_test_t f;
    memset(&f, 0, sizeof(f));
    f.magic          = HA_TERM_MAGIC_TEST;
    f.term           = 99ULL;
    f.voted_for_term = 99ULL;
    strncpy(f.voted_for_name, "ctrl-node-3", sizeof(f.voted_for_name) - 1);
    f.voted_for_name[sizeof(f.voted_for_name) - 1] = '\0';
    f.crc32 = 0xDEAD1234U; /* deliberately wrong CRC */
    write_term_file(dir, &f);

    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    /* ha_restore_term must detect CRC mismatch and zero state */
    int rc = ha_restore_term(&ha, dir);    /* <-- undefined reference: RED proof */
    /* Return value 0 is acceptable: "handled gracefully, no fatal error" */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_restore_term must return 0 on CRC mismatch");

    /* State must be zeroed — no stale data restored */
    uint64_t term = atomic_load_explicit(&ha.term, memory_order_acquire);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(0ULL, term,
                                     "term must be 0 after CRC mismatch");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, ha.voted_for_name[0],
                                  "voted_for_name must be empty after CRC mismatch");

    ha_fini(&ha);

    char path[1024];
    snprintf(path, sizeof(path), "%s/ha-term.bin", dir);
    unlink(path);
    rmdir(dir);

    printf("test_ha_restore_crc_mismatch_zeroes_state: PASS\n");
}

/* ── Test 5: write+rename atomicity — .tmp cleaned up; final file consistent ─
 *
 * Req 9.1: ha_persist_term uses write-to-tmp + rename for atomic durability.
 * Verify:
 *   a) After writing term=1, ha-term.bin contains term=1.
 *   b) After writing term=2 over it, ha-term.bin contains term=2.
 *   c) The .tmp file does not exist after either call (rename succeeded).
 *
 * This test does NOT simulate a partial write (that would require filesystem
 * hooks not available in unit tests). Instead it verifies the postcondition:
 * the final ha-term.bin is consistent and no .tmp artifact is left behind.
 */
static void test_persist_atomicity_tmp_cleaned_up(void)
{
    char tmpdir[] = "/tmp/test_ha_durXXXXXX";
    char *dir = mkdtemp(tmpdir);
    TEST_ASSERT_NOT_NULL_MESSAGE(dir, "mkdtemp failed");

    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    /* -- Round 1: persist term=1 -- */
    atomic_store_explicit(&ha.term, 1ULL, memory_order_release);
    ha.voted_for_term = 1;
    strncpy(ha.voted_for_name, "ctrl-node-a", sizeof(ha.voted_for_name) - 1);
    ha.voted_for_name[sizeof(ha.voted_for_name) - 1] = '\0';
    ha_persist_term(&ha, dir);

    char path[1024];
    snprintf(path, sizeof(path), "%s/ha-term.bin", dir);
    char tmp_path[1040];
    snprintf(tmp_path, sizeof(tmp_path), "%s/ha-term.bin.tmp", dir);

    /* ha-term.bin must exist and contain term=1 */
    ha_term_file_test_t f1;
    int fd = open(path, O_RDONLY);
    TEST_ASSERT_MESSAGE(fd >= 0, "ha-term.bin should exist after first persist");
    read(fd, &f1, sizeof(f1));
    close(fd);
    TEST_ASSERT_EQUAL_UINT64(1ULL, f1.term);
    TEST_ASSERT_EQUAL_UINT32(compute_term_crc(&f1), f1.crc32);

    /* .tmp must not exist */
    struct stat st;
    TEST_ASSERT_EQUAL_INT_MESSAGE(-1, stat(tmp_path, &st),
                                  "ha-term.bin.tmp must not exist after successful persist");

    /* -- Round 2: persist term=2 over existing file -- */
    atomic_store_explicit(&ha.term, 2ULL, memory_order_release);
    ha.voted_for_term = 2;
    strncpy(ha.voted_for_name, "ctrl-node-b", sizeof(ha.voted_for_name) - 1);
    ha.voted_for_name[sizeof(ha.voted_for_name) - 1] = '\0';
    ha_persist_term(&ha, dir);

    /* ha-term.bin must now contain term=2 */
    ha_term_file_test_t f2;
    fd = open(path, O_RDONLY);
    TEST_ASSERT_MESSAGE(fd >= 0, "ha-term.bin should still exist after second persist");
    read(fd, &f2, sizeof(f2));
    close(fd);
    TEST_ASSERT_EQUAL_UINT64(2ULL, f2.term);
    TEST_ASSERT_EQUAL_STRING("ctrl-node-b", f2.voted_for_name);
    TEST_ASSERT_EQUAL_UINT32(compute_term_crc(&f2), f2.crc32);

    /* .tmp still must not exist */
    TEST_ASSERT_EQUAL_INT_MESSAGE(-1, stat(tmp_path, &st),
                                  "ha-term.bin.tmp must not exist after second persist");

    ha_fini(&ha);
    unlink(path);
    rmdir(dir);

    printf("test_persist_atomicity_tmp_cleaned_up: PASS\n");
}

/* ── Test 6: ha_persist_term() returns a real error on a genuine write
 * failure (bug-1684) ─────────────────────────────────────────────────────
 *
 * Before this fix, ha_persist_term() had a void return -- every open/write/
 * fsync/rename failure was silently swallowed ("best-effort; caller must
 * not rely on this for correctness"). Its sole vote-granting caller,
 * raft_cb_persist(), always returned 0 to the raft library regardless, so a
 * vote was granted even when the term/voted_for record never reached disk.
 *
 * Points workdir at a path whose PARENT directory does not exist, so
 * open(O_CREAT) on the .tmp file deterministically fails ENOENT -- no
 * permission-bit tricks needed (those misbehave when running as root,
 * common in sandboxes), and no real filesystem quota/full-disk simulation
 * required either.
 */
static void test_persist_term_returns_error_on_write_failure(void)
{
    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    atomic_store_explicit(&ha.term, 3ULL, memory_order_release);
    ha.voted_for_term = 3;
    strncpy(ha.voted_for_name, "ctrl-node-9", sizeof(ha.voted_for_name) - 1);

    const char *bad_workdir =
        "/tmp/xce_ha_durability_test_nonexistent_parent_xyz/subdir";
    int rc = ha_persist_term(&ha, bad_workdir);
    TEST_ASSERT_TRUE_MESSAGE(
        rc < 0,
        "ha_persist_term must return a negative errno when the write "
        "genuinely fails (bug-1684)");

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

/* ── Test 7: NULL workdir stays a no-op SUCCESS, not folded into the new
 * error path (bug-1684 regression guard) ──────────────────────────────── */
static void test_persist_term_null_workdir_returns_success(void)
{
    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);

    int rc = ha_persist_term(&ha, NULL);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "ha_persist_term(..., NULL) must return 0 -- test-mode skip is not "
        "a failure");

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

/* ── Test 8: a vote denied by a real persist failure leaves ha's in-memory
 * voted_for state UNCHANGED (bug-1684) ─────────────────────────────────────
 *
 * Drives the REAL production path: ha_init_raft() wires raft_cb_persist as
 * the raft library's on_persist callback (s_real_cbs, ha.c); the function
 * under test, raft_on_vote_request() (raft.c, the pure library), is the
 * exact function cluster.c's production dispatch calls. ha_init_raft(ha,
 * NULL, n_peers) avoids needing a real persist_t (only used for
 * initial_log_len/initial_commit_index when c is non-NULL) -- ha->raft_c is
 * then set directly afterward, the same test-shim pattern ha.h's own
 * raft_cb_become_leader_for_test/raft_cb_become_follower_for_test comments
 * document ("this shim sets ha->raft_c = c first").
 *
 * The fixture's myconf populates ONE real candidate name (index 1) so
 * raft_cb_persist's voted_for_name lookup resolves to a genuine non-empty
 * name -- not the self/empty fallback -- making the rollback assertion
 * meaningful: before this fix, ha.voted_for_name/voted_for_term would show
 * that candidate's vote despite the vote being denied and never durably
 * recorded.
 */
static void test_vote_denied_and_state_rolled_back_on_persist_failure(void)
{
    ha_state_t ha;
    ha_init(&ha, &g_durability_cfg, 0);
    ha_init_raft(&ha, NULL, 2);

    myconf_t test_myconf;
    memset(&test_myconf, 0, sizeof(test_myconf));
    test_myconf.workdir =
        (char *)"/tmp/xce_ha_durability_test_nonexistent_parent_xyz/subdir";
    test_myconf.name = (char *)"self-node";
    test_myconf.peers_cfg.candidates_count = 2;
    strncpy(test_myconf.peers_cfg.candidates[1].name, "candidate-1",
            sizeof(test_myconf.peers_cfg.candidates[1].name) - 1);

    cluster_t test_c;
    memset(&test_c, 0, sizeof(test_c));
    test_c.myconf = &test_myconf;

    /* Test shim: same pattern ha.h documents for the other raft_cb_*_for_test
     * helpers -- set ha->raft_c directly rather than driving a full
     * persist_t through ha_init_raft(). */
    ha.raft_c = &test_c;

    /* Baseline: no prior vote. */
    TEST_ASSERT_EQUAL_UINT64(0ULL, ha.voted_for_term);
    TEST_ASSERT_EQUAL_STRING("", ha.voted_for_name);

    uint64_t reply_term = 0;
    int result = raft_on_vote_request(&ha.raft, /*from_idx=*/1, /*term=*/5,
                                      /*cand_last_log_term=*/0,
                                      /*cand_last_log_index=*/0, &reply_term);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        RAFT_DENY, result,
        "vote must be DENIED when the term/vote persist genuinely fails "
        "(bug-1684)");

    /* voted_for state must be UNCHANGED -- the vote was never durably
     * recorded, so this node must not believe it already voted. */
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        0ULL, ha.voted_for_term,
        "voted_for_term must roll back to its pre-call value on persist "
        "failure (bug-1684)");
    TEST_ASSERT_EQUAL_STRING_MESSAGE(
        "", ha.voted_for_name,
        "voted_for_name must roll back to its pre-call value on persist "
        "failure (bug-1684)");

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

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

void test_ha_durability(void)
{
    UnitySetTestFile(__FILE__);
    /* Req 9.1: persist writes a correct, CRC-valid file */
    RUN_TEST(test_persist_term_writes_file);

    /* Req 9.2: restore reads back term and voted_for — RED: linker error expected */
    RUN_TEST(test_ha_restore_term_from_file);

    /* Req 9.3: missing file → safe zero init */
    RUN_TEST(test_ha_restore_term_zero_on_missing_file);

    /* Req 9.2: CRC mismatch → safe zero init (no stale state) */
    RUN_TEST(test_ha_restore_crc_mismatch_zeroes_state);

    /* Req 9.1: write+rename atomicity — no .tmp artifact left */
    RUN_TEST(test_persist_atomicity_tmp_cleaned_up);

    /* bug-1684: ha_persist_term() must return a real error on write failure,
     * NULL-workdir must stay a no-op success, and raft_cb_persist() must
     * deny the vote AND roll back in-memory voted_for state on failure. */
    RUN_TEST(test_persist_term_returns_error_on_write_failure);
    RUN_TEST(test_persist_term_null_workdir_returns_success);
    RUN_TEST(test_vote_denied_and_state_rolled_back_on_persist_failure);
}