/* ctrlstatus.c — controller status query handler (controller-status spec).
 *
 * Implements the read-only `ctrlstatus` opcode (an XCE analog of LSF `lsid`):
 * snapshot the responding node's local HA/config state into a CtrlStatus reply
 * and send it.  Answerable in every role (the dispatch-gate allow-listing for
 * that is task 2.2); this file only builds and sends the reply.
 *
 * Split into a pure snapshot (ctrlstatus_fill) + a thin send wrapper
 * (handle_ctrlstatus), mirroring ha_nop_reply_state_report, so the field
 * sourcing is unit-testable without an endpoint/socket.
 *
 * Requirements: 1.1, 1.2, 1.3, 2.1, 2.2, 2.3, 3.1, 3.2, 4.1, 4.2, 5.1, 5.2, 5.3,
 *               6.1, 6.2, 6.3, 6.4 (controller-status spec, task 2.1).
 */
#include "config.h"

#include <stdatomic.h>
#include <stdlib.h>
#include <string.h>

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

/*
 * ctrlstatus_fill — snapshot local HA/config state into *st.
 *
 * Read-only: every field comes from an atomic load, an immutable config read,
 * or a lock-guarded copy of primary_host/port.  No HA, persist, or config state
 * is modified (Req 1.2).
 *
 * Field sourcing:
 *   responderName / clusterName  — own peerinfo / loaded cluster config
 *   responderRole / term         — ha->role / ha->term atomics (Req 3.2, 6.1)
 *   primaryName                  — ha->primary_host under ha->lock; empty
 *                                  string when no primary is known (Req 1.3,
 *                                  2.2, 2.3 — client renders empty as "unknown")
 *   configVersion                — loaded config_version_tag (Req 5.1)
 *   configChecksum               — lowercase-hex of myconf->config_sha256[32]
 *                                  (64-char Text; Req 5.2 content checksum,
 *                                  5.3 divergence detection) — see NOTE below
 *   committedConfigVersion       — ha->committed_config_tag (Req 5.1)
 *   committedLsn / appliedLsn    — persist->committed_lsn / ha->applied_lsn (6.2)
 *   fenced                       — in_catch_up || recovery_unrecoverable (6.3)
 *   peers / peerCount            — myconf->peers_cfg.candidates (Req 4.1, 4.2)
 *
 * NOTE (configChecksum): the schema field is `configChecksum @6: Text` (it was
 * originally `Data`, but `Data` maps to the project's structured `xce_data_t` —
 * a tagged union over job/worker/resource/resval table payloads, NOT a capnp
 * byte blob, so it cannot carry the raw 32-byte digest).  We therefore carry the
 * full SHA-256 content checksum as a 64-char lowercase-hex `Text` string, which
 * the codec encodes/decodes like any other Text member (clusterName/primaryName)
 * and the client renders as hex anyway.  This gives Req 5.2 (content checksum)
 * and Req 5.3 (same-version divergence detection — identical config bytes on two
 * nodes yield the same hex; differing bytes differ).  configVersion /
 * committedConfigVersion (the 8-byte SHA-256 prefix tag) carry the version-level
 * identity for Req 5.1.  The hex string is heap-allocated here and freed in
 * ctrlstatus_free_fields().
 */
void ctrlstatus_fill(cluster_t* c, xce_ctrlstatus_t* st)
{
    ha_state_t* ha = &c->ha;
    myconf_t*   mc = c->myconf;

    memset(st, 0, sizeof(*st));

    /* responder identity + cluster name */
    st->responderName = (c->myhost && c->myhost->name[0] != '\0')
                            ? c->myhost->name
                        : (mc && mc->name) ? mc->name
                                           : "";
    st->clusterName = (mc && mc->cluster_config && mc->cluster_config->clustername)
                          ? mc->cluster_config->clustername
                          : "";

    /* role / term (atomics) */
    st->responderRole = (uint8_t)atomic_load_explicit(&ha->role,
                                                      memory_order_acquire);
    st->term = atomic_load_explicit(&ha->term, memory_order_acquire);

    /* primary identity — guarded by ha->lock (Req 2.2, 2.3) */
    pthread_mutex_lock(&ha->lock);
    st->primaryName = (ha->primary_host[0] != '\0') ? strdup(ha->primary_host)
                                                     : strdup("");
    pthread_mutex_unlock(&ha->lock);

    /* loaded config version (+ committed version) */
    st->configVersion          = mc ? mc->config_version_tag : 0;
    st->committedConfigVersion = atomic_load_explicit(&ha->committed_config_tag,
                                                      memory_order_acquire);

    /* configChecksum: 64-char lowercase-hex of the loaded config_sha256[32]
     * (Req 5.2 content checksum, 5.3 divergence detection — see NOTE above).
     * Heap-allocated; freed in ctrlstatus_free_fields(). */
    st->configChecksum = NULL;
    if (mc) {
        static const char hexdig[] = "0123456789abcdef";
        char* hex = (char*)malloc(2u * sizeof(mc->config_sha256) + 1u);
        if (hex) {
            for (size_t i = 0; i < sizeof(mc->config_sha256); i++) {
                hex[2 * i]     = hexdig[(mc->config_sha256[i] >> 4) & 0x0F];
                hex[2 * i + 1] = hexdig[mc->config_sha256[i] & 0x0F];
            }
            hex[2 * sizeof(mc->config_sha256)] = '\0';
            st->configChecksum                 = hex;
        }
    }

    /* commit / apply progress (Req 6.2) */
    st->committedLsn = atomic_load_explicit(&c->persist.committed_lsn,
                                            memory_order_acquire);
    st->appliedLsn = atomic_load_explicit(&ha->applied_lsn,
                                          memory_order_acquire);

    /* fenced: catch-up OR recovery-unrecoverable (Req 6.3) */
    int in_catch_up = atomic_load_explicit(&ha->in_catch_up,
                                           memory_order_acquire);
    st->fenced = (in_catch_up || persist_recovery_unrecoverable(&c->persist))
                     ? 1u
                     : 0u;

    /* configured controllers (Req 4.1, 4.2) */
    int n = 0;
    if (mc) {
        n = mc->peers_cfg.candidates_count;
        if (n < 0)
            n = 0;
    }
    st->peerCount   = (uint32_t)n;
    st->peers.count = n;
    st->peers.list  = NULL;
    if (n > 0) {
        st->peers.list = (xce_ctrlpeer_t**)calloc((size_t)n,
                                                  sizeof(xce_ctrlpeer_t*));
        for (int i = 0; i < n; i++) {
            xce_ctrlpeer_t* p = (xce_ctrlpeer_t*)calloc(1, sizeof(*p));
            p->name           = mc->peers_cfg.candidates[i].name;
            p->host           = mc->peers_cfg.candidates[i].hostname;
            p->port           = (uint32_t)mc->peers_cfg.candidates[i].port;
            st->peers.list[i] = p;
        }
    }
}

/*
 * ctrlstatus_free_fields — release the heap that ctrlstatus_fill allocated.
 * Frees the per-peer structs + the pointer array and the primaryName copy.
 * The peer name/host strings are NOT owned (they point into myconf) and are
 * not freed.  Safe to call once after the reply has been serialised.
 */
void ctrlstatus_free_fields(xce_ctrlstatus_t* st)
{
    if (st->peers.list) {
        for (int i = 0; i < st->peers.count; i++)
            free(st->peers.list[i]);
        free(st->peers.list);
        st->peers.list  = NULL;
        st->peers.count = 0;
    }
    free(st->primaryName);
    st->primaryName = NULL;
    free(st->configChecksum);
    st->configChecksum = NULL;
}

/*
 * handle_ctrlstatus — dispatch entry point for OPCode_ctrlstatus.
 * Builds a CtrlStatus from local state, replies, then frees.  Mirrors
 * handle_clusterconf (reply.c): stack reply, init_reply_by_req, point the union
 * at the payload, xce_reply, then free what was allocated.
 */
int handle_ctrlstatus(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    xce_reply_t      reply;
    xce_ctrlstatus_t status;

    ctrlstatus_fill(c, &status);

    init_reply_by_req(&reply, request);
    reply.status         = 0;
    reply.data_which     = RequestReply_data_ctrlstatus;
    reply.data.ctrlstatus = &status;

    xce_reply(ep, c->myhost, &reply);

    ctrlstatus_free_fields(&status);
    return 0;
}