#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <zlib.h>

#include <unity/unity.h>

#include "config.h"  /* WAL_HDR_CRC_SIZE, WAL_CRC_NONE */
#include "persist.h" /* WAL_LEGACY_TERM */
#include "walcompare.h"
#include "wal_types.h"

/* ------------------------------------------------------------------ *
 * Harness: hand-write raw WAL segments (LSN, term, payload all under
 * test control) into per-member temp dirs, then drive walcompare over
 * them. Mirrors test_waldump's writer so the on-disk format is pinned
 * independently of persist.c — and exercises the REAL waldump decoder
 * that walcompare reuses (no second parser).
 * ------------------------------------------------------------------ */

/* Append one entry to `fp`, chaining prev_crc from *last_crc. CRC32 over
 * WAL_HDR_CRC_SIZE header bytes + payload (identical to persist.c append). */
static void put_entry(FILE* fp, uint64_t term, uint64_t lsn, uint32_t type,
                      const void* payload, uint32_t plen, uint32_t* last_crc,
                      int first_in_seg)
{
    wal_entry_hdr_t hdr;
    memset(&hdr, 0, sizeof(hdr));
    hdr.term        = term;
    hdr.lsn         = lsn;
    hdr.timestamp   = 0;
    hdr.type        = type;
    hdr.payload_len = plen;
    hdr.prev_crc    = first_in_seg ? WAL_CRC_NONE : *last_crc;

    uint32_t c = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
    if (plen > 0)
        c = crc32(c, (const Bytef*)payload, plen);
    hdr.crc   = c;
    *last_crc = c;

    fwrite(&hdr, sizeof(hdr), 1, fp);
    if (plen > 0)
        fwrite(payload, plen, 1, fp);
}

/* A single hand-built WAL entry spec for a member. */
typedef struct {
    uint64_t    term;
    uint32_t    type;
    const char* payload; /* NUL-terminated; length is strlen (0 => no payload) */
} wal_spec_t;

/* Mint a fresh temp dir and write a single seg-0001.wal holding `specs`
 * (LSNs 1..n contiguous). Returns the dir path in `dir_out`. */
static void build_member(char* dir_out, size_t dirsz, const wal_spec_t* specs,
                         int n)
{
    snprintf(dir_out, dirsz, "/tmp/xce_walcmp_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(dir_out));

    char path[512];
    snprintf(path, sizeof(path), "%s/seg-%016lx.wal", dir_out, 1UL);
    FILE* fp = fopen(path, "wb");
    TEST_ASSERT_NOT_NULL(fp);

    uint32_t last = WAL_CRC_NONE;
    for (int i = 0; i < n; i++) {
        uint32_t plen =
            specs[i].payload ? (uint32_t)strlen(specs[i].payload) : 0;
        put_entry(fp, specs[i].term, (uint64_t)(i + 1), specs[i].type,
                  specs[i].payload, plen, &last, i == 0);
    }
    fclose(fp);
}

/* ------------------------------------------------------------------ */
/* Identical member WALs → success (exit 0).                            */
/* ------------------------------------------------------------------ */
static void test_identical_members_ok(void)
{
    wal_spec_t specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
        { 2, WAL_JOB_SUBMIT, "jobB" },
    };
    char a[256], b[256];
    build_member(a, sizeof(a), specs, 3);
    build_member(b, sizeof(b), specs, 3);

    walcompare_member_t m[2] = {
        { a, 0 },
        { b, 0 },
    };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc);
}

/* ------------------------------------------------------------------ */
/* Hand-divergent pair: SAME LSN, DIFFERENT term within the committed   */
/* prefix → FAIL with the diverging LSN (R6.2/R6.3).                     */
/* ------------------------------------------------------------------ */
static void test_same_lsn_diff_term_divergent(void)
{
    wal_spec_t a_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
        { 2, WAL_JOB_SUBMIT, "jobB" }, /* LSN 3, term 2 */
    };
    wal_spec_t b_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
        { 3, WAL_JOB_SUBMIT, "jobB" }, /* LSN 3, term 3 — divergent */
    };
    char a[256], b[256];
    build_member(a, sizeof(a), a_specs, 3);
    build_member(b, sizeof(b), b_specs, 3);

    walcompare_member_t m[2] = { { a, 0 }, { b, 0 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc);
    TEST_ASSERT_EQUAL_UINT64(3, div.lsn); /* the diverging LSN */
    TEST_ASSERT_EQUAL_UINT64(2, div.member_term[0]);
    TEST_ASSERT_EQUAL_UINT64(3, div.member_term[1]);
}

/* ------------------------------------------------------------------ */
/* SAME LSN, SAME term, DIFFERENT payload (pcrc) → FAIL (R6.2: a content */
/* divergence is not masked by equal term + LSN range).                 */
/* ------------------------------------------------------------------ */
static void test_same_lsn_diff_payload_divergent(void)
{
    wal_spec_t a_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" }, /* LSN 2 */
    };
    wal_spec_t b_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobZ" }, /* LSN 2, same term, different bytes */
    };
    char a[256], b[256];
    build_member(a, sizeof(a), a_specs, 2);
    build_member(b, sizeof(b), b_specs, 2);

    walcompare_member_t m[2] = { { a, 0 }, { b, 0 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc);
    TEST_ASSERT_EQUAL_UINT64(2, div.lsn);
    TEST_ASSERT_EQUAL_UINT64(1, div.member_term[0]);
    TEST_ASSERT_EQUAL_UINT64(1, div.member_term[1]);
    TEST_ASSERT_NOT_EQUAL(div.member_pcrc[0], div.member_pcrc[1]);
}

/* ------------------------------------------------------------------ */
/* A member that is AHEAD (longer log) with a DIFFERENT uncommitted tail */
/* is NOT a divergence — the shared committed prefix matches (§6.1).     */
/* ------------------------------------------------------------------ */
static void test_uncommitted_tail_not_flagged(void)
{
    wal_spec_t a_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" }, /* shared committed prefix: LSN 1..2 */
    };
    wal_spec_t b_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
        { 2, WAL_JOB_SUBMIT, "tailX" }, /* uncommitted tail — beyond shared tip */
    };
    char a[256], b[256];
    build_member(a, sizeof(a), a_specs, 2); /* tip 2 */
    build_member(b, sizeof(b), b_specs, 3); /* tip 3 */

    /* Shared boundary = min recovered tip = 2; the LSN-3 tail is not compared. */
    walcompare_member_t m[2] = { { a, 0 }, { b, 0 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc);
}

/* ------------------------------------------------------------------ */
/* An explicit committed boundary SHORTER than a member's log: a         */
/* divergence ABOVE the boundary must NOT be flagged (§6.1 committed     */
/* prefix only).                                                         */
/* ------------------------------------------------------------------ */
static void test_divergence_above_committed_boundary_not_flagged(void)
{
    wal_spec_t a_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" }, /* LSN 2 committed */
        { 2, WAL_JOB_SUBMIT, "uncA" }, /* LSN 3 uncommitted, divergent */
    };
    wal_spec_t b_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
        { 9, WAL_JOB_SUBMIT, "uncB" }, /* LSN 3, different term+bytes */
    };
    char a[256], b[256];
    build_member(a, sizeof(a), a_specs, 3);
    build_member(b, sizeof(b), b_specs, 3);

    /* committed boundary pinned at 2 → the LSN-3 divergence is below scope. */
    walcompare_member_t m[2] = { { a, 2 }, { b, 2 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc);

    /* Same data, but raise the boundary to 3 → now it IS a divergence. */
    walcompare_member_t m3[2] = { { a, 3 }, { b, 3 } };
    rc = walcompare(m3, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc);
    TEST_ASSERT_EQUAL_UINT64(3, div.lsn);
}

/* ------------------------------------------------------------------ */
/* A member MISSING a committed entry (a hole inside the committed       */
/* prefix the boundary covers) is a divergence (R4.2/R6.4).             */
/* ------------------------------------------------------------------ */
static void test_missing_committed_entry_divergent(void)
{
    /* Member A holds 1..3; member B holds only 1..2 but is told its
     * committed boundary is 3 — it is MISSING the committed entry at LSN 3. */
    wal_spec_t a_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
        { 2, WAL_JOB_SUBMIT, "jobB" },
    };
    wal_spec_t b_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
    };
    char a[256], b[256];
    build_member(a, sizeof(a), a_specs, 3);
    build_member(b, sizeof(b), b_specs, 2);

    walcompare_member_t m[2] = { { a, 3 }, { b, 3 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc);
    TEST_ASSERT_EQUAL_UINT64(3, div.lsn);
    TEST_ASSERT_EQUAL_INT(1, div.member_present[0]); /* A holds it */
    TEST_ASSERT_EQUAL_INT(0, div.member_present[1]); /* B is missing it */
}

/* ------------------------------------------------------------------ */
/* Two legacy-0-term entries at the same LSN are NOT divergent: legacy 0 */
/* maps to WAL_LEGACY_TERM for both (R1.8).                              */
/* ------------------------------------------------------------------ */
static void test_legacy_zero_term_not_divergent(void)
{
    wal_spec_t specs[] = {
        { 0, WAL_NOOP, NULL },          /* legacy-0 term */
        { 0, WAL_JOB_SUBMIT, "jobA" },  /* legacy-0 term, same payload */
    };
    char a[256], b[256];
    build_member(a, sizeof(a), specs, 2);
    build_member(b, sizeof(b), specs, 2);

    walcompare_member_t m[2] = { { a, 0 }, { b, 0 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc);
}

/* ------------------------------------------------------------------ */
/* Three members, all identical → success (the live regression shape).  */
/* ------------------------------------------------------------------ */
static void test_three_members_identical_ok(void)
{
    wal_spec_t specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "j1" },
        { 1, WAL_CONFIG_MARKER, "cfgref" }, /* a config marker in the log */
        { 2, WAL_JOB_SUBMIT, "j2" },
    };
    char a[256], b[256], c[256];
    build_member(a, sizeof(a), specs, 4);
    build_member(b, sizeof(b), specs, 4);
    build_member(c, sizeof(c), specs, 4);

    walcompare_member_t m[3] = { { a, 0 }, { b, 0 }, { c, 0 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 3, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc);
}

/* ------------------------------------------------------------------ */
/* Three members, one with a divergent committed config marker → FAIL    */
/* (R6.6: config markers reconcile by (term,lsn) like job entries).      */
/* ------------------------------------------------------------------ */
static void test_three_members_divergent_config_marker(void)
{
    wal_spec_t ok_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "j1" },
        { 1, WAL_CONFIG_MARKER, "cfgX" }, /* LSN 3 config ref */
    };
    wal_spec_t bad_specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "j1" },
        { 1, WAL_CONFIG_MARKER, "cfgY" }, /* LSN 3, different config ref */
    };
    char a[256], b[256], c[256];
    build_member(a, sizeof(a), ok_specs, 3);
    build_member(b, sizeof(b), ok_specs, 3);
    build_member(c, sizeof(c), bad_specs, 3);

    walcompare_member_t m[3] = { { a, 0 }, { b, 0 }, { c, 0 } };
    walcompare_divergence_t div;
    int rc = walcompare(m, 3, stdout, &div);
    TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc);
    TEST_ASSERT_EQUAL_UINT64(3, div.lsn);
    /* A and B agree; C diverges. */
    TEST_ASSERT_EQUAL_UINT32(div.member_pcrc[0], div.member_pcrc[1]);
    TEST_ASSERT_NOT_EQUAL(div.member_pcrc[0], div.member_pcrc[2]);
}

/* ------------------------------------------------------------------ */
/* A missing WAL source is a hard (negative) error, not a false OK.      */
/* ------------------------------------------------------------------ */
static void test_missing_member_errors(void)
{
    wal_spec_t specs[] = {
        { 1, WAL_NOOP, NULL },
        { 1, WAL_JOB_SUBMIT, "jobA" },
    };
    char a[256];
    build_member(a, sizeof(a), specs, 2);

    walcompare_member_t m[2] = {
        { a, 0 },
        { "/tmp/xce_walcmp_does_not_exist_zzz", 0 },
    };
    walcompare_divergence_t div;
    int rc = walcompare(m, 2, stdout, &div);
    TEST_ASSERT_LESS_THAN_INT(0, rc); /* negative errno, not OK/DIVERGENT */
}

/* ------------------------------------------------------------------ */
/* Fewer than two members is a usage error.                             */
/* ------------------------------------------------------------------ */
static void test_too_few_members_errors(void)
{
    wal_spec_t specs[] = { { 1, WAL_NOOP, NULL } };
    char a[256];
    build_member(a, sizeof(a), specs, 1);
    walcompare_member_t m[1] = { { a, 0 } };
    int rc = walcompare(m, 1, stdout, NULL);
    TEST_ASSERT_LESS_THAN_INT(0, rc);
}

void test_walcompare(void)
{
    RUN_TEST(test_identical_members_ok);
    RUN_TEST(test_same_lsn_diff_term_divergent);
    RUN_TEST(test_same_lsn_diff_payload_divergent);
    RUN_TEST(test_uncommitted_tail_not_flagged);
    RUN_TEST(test_divergence_above_committed_boundary_not_flagged);
    RUN_TEST(test_missing_committed_entry_divergent);
    RUN_TEST(test_legacy_zero_term_not_divergent);
    RUN_TEST(test_three_members_identical_ok);
    RUN_TEST(test_three_members_divergent_config_marker);
    RUN_TEST(test_missing_member_errors);
    RUN_TEST(test_too_few_members_errors);
}