/*
 * commit.c — Commit engine core (ha-commit-engine task 1.2).
 *
 * Single authority for per-entry commit evaluation and committed_lsn
 * advancement (Raft §5.3, umbrella-verbatim commit rule):
 *
 *   entry N is committed when
 *     count(self + active standbys with acked_lsn >= N) > n_peers / 2
 *
 * Evaluated as a watermark median: active (connected, non-catch-up)
 * standby watermarks are scanned into a scratch array, sorted descending,
 * and the candidate is arr[n_peers/2 - 1] — the highest LSN covered by
 * self plus a majority of peers. The candidate is clamped to the highest
 * locally appended LSN and stored with monotonic max() semantics.
 *
 * commit_on_ack is called from the iomgr read path and never blocks
 * beyond the engine mutex (plus the registry slot mutex, which the same
 * iomgr path already takes in repl_on_ack): no I/O, no condvar waits.
 *
 * COMMIT_MODE_WAIT (blocking wait + admission cap) lands in task 2.2.
 *
 * Requirements: 1.2, 2.1, 2.2, 2.3, 5.1
 */
#include <errno.h>
#include <stdatomic.h>
#include <string.h>

#include "commit.h"
#include "mylock.h"
#include "utils.h"

#ifdef XCE_TESTING
#include <assert.h>
#endif

/* ============================================================
 * Internal helpers
 * ============================================================ */

/* Highest locally appended WAL LSN. persist invariant:
 * committed_lsn <= current_lsn - 1 (current_lsn is the NEXT LSN). */
static uint64_t highest_appended_lsn(const persist_t* p)
{
    return (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
}

/* Descending insertion sort — n is bounded by MAX_STANDBYS. */
static void sort_desc(uint64_t* arr, int n)
{
    int      i, j;
    uint64_t v;

    for (i = 1; i < n; i++) {
        v = arr[i];
        for (j = i - 1; j >= 0 && arr[j] < v; j--) {
            arr[j + 1] = arr[j];
        }
        arr[j + 1] = v;
    }
}

/*
 * Compute the highest LSN covered by self + a majority of peers.
 * Caller holds ce->lock.
 *
 * Catch-up standbys are excluded from the evaluation count
 * (umbrella-locked conservative rule); absent peers count as watermark 0.
 */
static uint64_t majority_candidate_locked(commit_engine_t* ce)
{
    replicator_t* r;
    uint64_t      cand;
    uint64_t      cap = highest_appended_lsn(ce->p);
    int           n   = 0;
    int           need, k, i;

    /* Standalone / single-member: self constitutes the majority (Req 6.3). */
    if (ce->n_peers <= 1) {
        return cap;
    }

    /* Bounded scan of active standby watermarks into the scratch array. */
    r = &ce->p->repl;
    mylock_lock(&r->lock);
    for (i = 0; i < MAX_STANDBYS && n < MAX_STANDBYS; i++) {
        standby_conn_t* s = &r->standbys[i];

        if (s->ep == NULL || atomic_load(&s->in_catch_up) != 0) {
            continue; /* empty slot or catch-up: not countable toward majority */
        }
        ce->scratch[n++] = atomic_load(&s->acked_lsn);
    }
    mylock_unlock(&r->lock);

    /* Zero-fill up to the full peer count so missing peers count as 0. */
    need = ce->n_peers - 1;
    if (need > MAX_STANDBYS) {
        need = MAX_STANDBYS;
    }
    while (n < need) {
        ce->scratch[n++] = 0;
    }

    /* Self always covers every appended entry, so a majority needs
     * n_peers/2 standbys at-or-above N; sorted descending, the highest
     * such N is the watermark at index n_peers/2 - 1. */
    k = ce->n_peers / 2 - 1;
    if (k >= n) {
        return 0; /* peer count not representable — no majority possible */
    }
    sort_desc(ce->scratch, n);
    cand = ce->scratch[k];

    /* Contiguous and bounded: never beyond the appended log (Req 2.3). */
    return (cand < cap) ? cand : cap;
}

/*
 * Re-evaluate majority and advance committed_lsn (monotonic max() store).
 * Caller holds ce->lock. Broadcasts the condvar on advancement so future
 * WAIT-mode blockers (task 2.2) re-check their per-LSN predicates.
 */
static void commit_advance_locked(commit_engine_t* ce)
{
    uint64_t cand = majority_candidate_locked(ce);
    uint64_t cur  = atomic_load(&ce->p->committed_lsn);

    if (cand <= cur) {
        return; /* monotonic: committed_lsn never regresses (Req 2.2) */
    }

#ifdef XCE_TESTING
    assert(cand > cur);                              /* monotonic advance only */
    assert(cand <= highest_appended_lsn(ce->p));     /* never beyond the log */
#endif

    atomic_store(&ce->p->committed_lsn, cand);
    TRACE("commit: committed_lsn %lu -> %lu (n_peers=%d)", cur, cand,
          ce->n_peers);
    pthread_cond_broadcast(&ce->cond);
}

/* ============================================================
 * Lifecycle
 * ============================================================ */

int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus)
{
    if (ce == NULL || p == NULL || n_peers < 0 || n_vcpus < 1) {
        return -EINVAL;
    }
    pthread_mutex_init(&ce->lock, NULL);
    pthread_cond_init(&ce->cond, NULL);
    ce->p              = p;
    ce->n_peers        = n_peers;
    ce->n_vcpus        = n_vcpus;
    ce->inflight_waits = 0;
    ce->writable_lsn   = 0; /* no own-term gate until set by promotion */
    memset(ce->scratch, 0, sizeof(ce->scratch));
    return 0;
}

void commit_engine_fini(commit_engine_t* ce)
{
    if (ce == NULL) {
        return;
    }
    pthread_cond_destroy(&ce->cond);
    pthread_mutex_destroy(&ce->lock);
    ce->p = NULL;
}

/* ============================================================
 * Entry registration
 * ============================================================ */

int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode,
                  uint32_t timeout_ms)
{
    (void)timeout_ms; /* RETAIN never honors the timeout; WAIT uses it (2.2) */

    if (ce == NULL || ce->p == NULL || lsn == 0) {
        return -EINVAL;
    }

    if (mode == COMMIT_MODE_RETAIN) {
        /* Commit state is fully derived from the registry watermarks at
         * ACK time, so RETAIN registration has nothing to record per
         * entry: return immediately, never block; the entry commits later
         * via commit_on_ack (Req 2.1). */
        return 0;
    }

    /* COMMIT_MODE_WAIT: blocking wait + admission cap land in task 2.2. */
    return -ENOSYS;
}

/* ============================================================
 * ACK and membership events
 * ============================================================ */

void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn)
{
    if (ce == NULL || ce->p == NULL || standby_idx < 0 ||
        standby_idx >= MAX_STANDBYS) {
        return;
    }

    TRACE("commit: ack standby_idx=%d acked_lsn=%lu", standby_idx, acked_lsn);

    /* The authoritative watermark is read from the registry during
     * evaluation (repl_on_ack stored it before this call); the args
     * identify the event for tracing only. */
    pthread_mutex_lock(&ce->lock);
    commit_advance_locked(ce);
    pthread_mutex_unlock(&ce->lock);
}

void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn)
{
    if (ce == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    ce->writable_lsn = noop_lsn;
    /* Gate change is part of the WAIT completion predicate: wake blockers
     * so they re-evaluate (no successful completion until
     * committed_lsn >= writable_lsn). */
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);
}

void commit_on_membership_change(commit_engine_t* ce)
{
    if (ce == NULL || ce->p == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    /* Catch-up transitions / standby removal can newly satisfy a pending
     * majority: re-evaluate (monotonic, so shrink never regresses) and
     * wake all waiters to re-check their predicates. */
    commit_advance_locked(ce);
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);
}