#include <argp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <tlog.h>

#include "myconf.h"
#include "utils.h"
#include "xce/cluster.h"

extern int myapp_run(myconf_t*);

/* forward declarations — implemented in startup_validation.c */
int validate_vcpus(int vcpus);
int validate_peer_count(int n_peers);
int validate_cert_files(const char *cert_path, const char *key_path,
                        char *err_buf, size_t err_len);

static char doc[]      = "xce_controller";
static char args_doc[] = "";
/* build-time-versioning task 3.2: no longer a compile-time initializer --
 * argp reads this by pointer when --version/-V is processed, and
 * xce_compose_version() (utils.h) is a runtime function, not a constant
 * expression, so it cannot initialize a file-scope global. main() assigns
 * this to a static buffer, computed via xce_compose_version(), before
 * argp_parse() runs. See utils.h's NOTE FOR TASK 3.2 above MK_VERSION. */
const char* argp_program_version = NULL;
const char* argp_program_bug_address = BUG_EMAIL;

/* Long-option key (above ASCII to avoid collisions with short opts) */
#define OPT_PEERS              256
#define OPT_CONF_KEEP_N        257
#define OPT_CONF_STALE_TIMEOUT 258
#define OPT_VCPUS              259
#define OPT_CERTDIR            260
#define OPT_SYNC_TIMEOUT       261
#define OPT_WAL_SYNC           262
#define OPT_MAX_WAL_RECORDS    263
#define OPT_WAL_SEGMENT_BYTES  264
#define OPT_JOBMGR_NHASH       265
#define OPT_JOBMGR_DELTA       266
#define OPT_WORKERMGR_NHASH    267
#define OPT_WORKERMGR_DELTA    268
#define OPT_QMGR_NHASH         269
#define OPT_QMGR_DELTA         270
#define OPT_RESMGR_NHASH       271
#define OPT_RESMGR_DELTA       272
#define OPT_RESMGR_INST_DELTA  273
#define OPT_XSTRINGS_NHASH     274
#define OPT_XSTRINGS_DELTA     275

struct arguments {
    char* myname;
    char* logfile;
    char* workdir;
    int   verbose;
    char* peers_str;
    int   conf_keep_n;
    long  conf_stale_timeout;
    int   vcpus;
    char* certdir;
    uint32_t sync_timeout_ms;
    uint32_t wal_sync_ms;
    uint32_t max_wal_records;
    uint32_t wal_segment_bytes;
    uint32_t jobmgr_nhash;
    uint32_t jobmgr_delta;
    uint32_t workermgr_nhash;
    uint32_t workermgr_delta;
    uint32_t qmgr_nhash;
    uint32_t qmgr_delta;
    uint32_t resmgr_nhash;
    uint32_t resmgr_delta;
    uint32_t resmgr_inst_delta;
    uint32_t xstrings_nhash;
    uint32_t xstrings_delta;
};

static struct argp_option options[] = {
    { "workdir", 'w', "DIR",   0, "The work directory of xce controller" },
    { "logfile", 'l', "FILE",  0, "The full path of log file" },
    { "verbose", 'v', 0,       0, "Produce verbose logs" },
    { "myname",  'n', "STR",   0, "The name of controller" },
    { "peers",   OPT_PEERS, "PEERS", 0,
      "HA peer list: comma-separated host:port pairs ordered by priority "
      "(e.g. ctrl01:9990,ctrl02:9990). Index 0 = highest priority (primary)." },
    { "conf-keep-n",        OPT_CONF_KEEP_N,        "N",    0,
      "Versioned config files to retain (default: 2)" },
    { "conf-stale-timeout", OPT_CONF_STALE_TIMEOUT,  "SECS", 0,
      "Seconds before saved config is flagged stale (0=disabled, default: 86400)" },
    { "vcpus",              OPT_VCPUS,               "N",    0,
      "Number of virtual CPUs for the task manager (must be >= 1)" },
    { "certdir",            OPT_CERTDIR,  "DIR", 0,
      "Directory containing xce.crt and xce.key (default: <workdir>/certs)" },
    { "sync-timeout-ms",    OPT_SYNC_TIMEOUT, "MS", 0,
      "Synchronous majority-replication commit barrier timeout in ms (0 = default)" },
    { "wal-sync-ms",        OPT_WAL_SYNC, "MS", 0,
      "WAL flush (fsync) cadence in ms (0 = default)" },
    { "max-wal-records",    OPT_MAX_WAL_RECORDS, "N", 0,
      "WAL records since last checkpoint before an async checkpoint is "
      "scheduled (0 = default; persist enforces a minimum of 100)" },
    { "wal-segment-bytes",  OPT_WAL_SEGMENT_BYTES, "N", 0,
      "Physical WAL segment file size in bytes; a segment is only eligible "
      "for trim/delete after it rotates out (0 = default 64 MB; persist "
      "enforces a minimum of 4096)" },
    { "jobmgr-nhash",       OPT_JOBMGR_NHASH,       "N",    0,
      "Hash-table slot count for the job manager (0 = use hardcoded default)" },
    { "jobmgr-delta",       OPT_JOBMGR_DELTA,       "N",    0,
      "Hash-table growth step for the job manager (0 = use hardcoded default)" },
    { "workermgr-nhash",    OPT_WORKERMGR_NHASH,    "N",    0,
      "Hash-table slot count for the worker manager (0 = use hardcoded default)" },
    { "workermgr-delta",    OPT_WORKERMGR_DELTA,    "N",    0,
      "Hash-table growth step for the worker manager (0 = use hardcoded default)" },
    { "qmgr-nhash",         OPT_QMGR_NHASH,         "N",    0,
      "Hash-table slot count for the queue manager (0 = use hardcoded default)" },
    { "qmgr-delta",         OPT_QMGR_DELTA,         "N",    0,
      "Hash-table growth step for the queue manager (0 = use hardcoded default)" },
    { "resmgr-nhash",       OPT_RESMGR_NHASH,       "N",    0,
      "Hash-table slot count for the resource manager (0 = use hardcoded default)" },
    { "resmgr-delta",       OPT_RESMGR_DELTA,       "N",    0,
      "Hash-table growth step for the resource manager (0 = use hardcoded default)" },
    { "resmgr-inst-delta",  OPT_RESMGR_INST_DELTA,  "N",    0,
      "Growth step for the resource manager's shared-instance array "
      "(0 = use hardcoded default)" },
    { "xstrings-nhash",     OPT_XSTRINGS_NHASH,     "N",    0,
      "Hash-table slot count for the interned-string table (0 = use hardcoded default)" },
    { "xstrings-delta",     OPT_XSTRINGS_DELTA,     "N",    0,
      "Hash-table growth step for the interned-string table (0 = use hardcoded default)" },
    { 0 }
};


static error_t parse_opt(int key, char* arg, struct argp_state* state)
{
    struct arguments* arguments = state->input;

    switch (key) {
    case 'l':
        arguments->logfile = arg;
        break;
    case 'v':
        arguments->verbose++;
        break;
    case 'w':
        arguments->workdir = arg;
        break;
    case 'n':
        arguments->myname = arg;
        break;
    case OPT_PEERS:
        arguments->peers_str = arg;
        break;
    case OPT_CONF_KEEP_N:
        arguments->conf_keep_n = atoi(arg);
        if (arguments->conf_keep_n < 1)
            argp_error(state, "--conf-keep-n must be >= 1 (got: %s)", arg);
        break;
    case OPT_CONF_STALE_TIMEOUT:
        arguments->conf_stale_timeout = atol(arg);
        if (arguments->conf_stale_timeout < 0)
            argp_error(state, "--conf-stale-timeout must be >= 0 (got: %s)", arg);
        break;
    case OPT_VCPUS:
        arguments->vcpus = atoi(arg);
        if (validate_vcpus(arguments->vcpus) != 0)
            argp_error(state, "--vcpus must be >= 1 (got: %s)", arg);
        break;
    case OPT_CERTDIR:
        arguments->certdir = arg;
        break;
    case OPT_SYNC_TIMEOUT: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--sync-timeout-ms must be >= 0 (got: %s)", arg);
        arguments->sync_timeout_ms = (uint32_t)v;
        break;
    }
    case OPT_WAL_SYNC: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--wal-sync-ms must be >= 0 (got: %s)", arg);
        arguments->wal_sync_ms = (uint32_t)v;
        break;
    }
    case OPT_MAX_WAL_RECORDS: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--max-wal-records must be >= 0 (got: %s)", arg);
        arguments->max_wal_records = (uint32_t)v;
        break;
    }
    case OPT_WAL_SEGMENT_BYTES: {
        long v = atol(arg);
        if (v < 0)
            argp_error(state, "--wal-segment-bytes must be >= 0 (got: %s)", arg);
        arguments->wal_segment_bytes = (uint32_t)v;
        break;
    }
    case OPT_JOBMGR_NHASH: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--jobmgr-nhash must be >= 0 (got: %s)", arg);
        arguments->jobmgr_nhash = (uint32_t)v;
        break;
    }
    case OPT_JOBMGR_DELTA: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--jobmgr-delta must be >= 0 (got: %s)", arg);
        arguments->jobmgr_delta = (uint32_t)v;
        break;
    }
    case OPT_WORKERMGR_NHASH: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--workermgr-nhash must be >= 0 (got: %s)", arg);
        arguments->workermgr_nhash = (uint32_t)v;
        break;
    }
    case OPT_WORKERMGR_DELTA: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--workermgr-delta must be >= 0 (got: %s)", arg);
        arguments->workermgr_delta = (uint32_t)v;
        break;
    }
    case OPT_QMGR_NHASH: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--qmgr-nhash must be >= 0 (got: %s)", arg);
        arguments->qmgr_nhash = (uint32_t)v;
        break;
    }
    case OPT_QMGR_DELTA: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--qmgr-delta must be >= 0 (got: %s)", arg);
        arguments->qmgr_delta = (uint32_t)v;
        break;
    }
    case OPT_RESMGR_NHASH: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--resmgr-nhash must be >= 0 (got: %s)", arg);
        arguments->resmgr_nhash = (uint32_t)v;
        break;
    }
    case OPT_RESMGR_DELTA: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--resmgr-delta must be >= 0 (got: %s)", arg);
        arguments->resmgr_delta = (uint32_t)v;
        break;
    }
    case OPT_RESMGR_INST_DELTA: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--resmgr-inst-delta must be >= 0 (got: %s)", arg);
        arguments->resmgr_inst_delta = (uint32_t)v;
        break;
    }
    case OPT_XSTRINGS_NHASH: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--xstrings-nhash must be >= 0 (got: %s)", arg);
        arguments->xstrings_nhash = (uint32_t)v;
        break;
    }
    case OPT_XSTRINGS_DELTA: {
        int v = atoi(arg);
        if (v < 0)
            argp_error(state, "--xstrings-delta must be >= 0 (got: %s)", arg);
        arguments->xstrings_delta = (uint32_t)v;
        break;
    }
    default:
        return ARGP_ERR_UNKNOWN;
    }

    return 0;
}

static struct argp argp = { options, parse_opt, args_doc, doc };

/*
 * Saved argv for self-restart (ha-convergence-recovery Req 1.3).
 *
 * main() captures the original, NULL-terminated argv pointer before argp_parse
 * mutates the program state, so ha_self_restart can execv() the controller back
 * into existence after a demotion/rejoin convergence truncates the divergent WAL
 * suffix.  argv supplied to main() is already NULL-terminated per C11 §5.1.2.2.1
 * (argv[argc] == NULL) and lives for the whole program lifetime, so storing the
 * pointer is sufficient — no copy needed.
 */
static char **g_saved_argv = NULL;

/*
 * controller_saved_argv — accessor declared in ha.h, consumed by ha_self_restart.
 * Returns the NULL-terminated argv captured by main(), or NULL if never saved
 * (e.g. unit-test binaries that do not run main()).
 */
char **controller_saved_argv(void)
{
    return g_saved_argv;
}

int main(int argc, char* argv[])
{
    struct arguments arguments;
    myconf_t         myconf;
    char*            logfile  = "/tmp/xce_controller.log";
    char*            loglevel = "info";
    int              logsize  = 64 * 1024 * 1024;
    int              logcount = 10;
    char             key_file[XCE_MAXPATH_LEN + 1];
    char             cert_file[XCE_MAXPATH_LEN + 1];
    char             myname[XCE_MAXNAME_LEN + 1];
    /* build-time-versioning task 3.2: composed once, read by both
     * argp_program_version (--version/-V) and the startup INFO log below --
     * never recomputed, so the two can never disagree (Req 2.2). static so
     * it outlives this point in main() (contract documented on
     * xce_compose_version() in utils.c). */
    static char      version_buf[128];
    /* dirty-content-fingerprint task 3.2: SEPARATE buffer/composition, used
     * ONLY for argp_program_version (--version/-V). version_buf above is
     * left completely untouched -- its own xce_compose_version() call and
     * the startup INFO log line reading it (below) are unaffected by this
     * feature, per Requirement 3.2. This divergence (the two call sites now
     * showing different strings) is deliberate, not an oversight. */
    static char      version_display_buf[128];

    /* Capture the NULL-terminated argv for self-restart BEFORE argp_parse so a
     * later demotion/rejoin convergence can execv() the same image (Req 1.3). */
    g_saved_argv = argv;

    /* Must run before argp_parse(): argp reads argp_program_version by
     * pointer when --version/-V is processed. */
    xce_compose_version(version_buf, sizeof(version_buf), "xce v0.0.1-",
                        STR(VERSION_COMMIT_ID), STR(VERSION_BUILD_TIME));
    /* dirty-content-fingerprint task 3.2: --version-only composition --
     * appends the dirty_fingerprint suffix (if any) on top of the same
     * commit/build_time inputs. Do NOT fold this into version_buf above; the
     * startup INFO log must keep showing the original, fingerprint-less
     * format (Req 3.2). */
    xce_compose_version_ex(version_display_buf, sizeof(version_display_buf),
                           "xce v0.0.1-", STR(VERSION_COMMIT_ID),
                           STR(VERSION_BUILD_TIME),
                           STR(VERSION_DIRTY_FINGERPRINT));
    argp_program_version = version_display_buf;

    memset(&arguments, 0x0, sizeof(arguments));
    arguments.conf_keep_n        = 2;
    arguments.conf_stale_timeout = 86400;
    arguments.vcpus              = 0; /* 0 = not set; use cluster config default */
    arguments.sync_timeout_ms    = 0; /* 0 = not set; use persist config default */
    arguments.wal_sync_ms        = 0; /* 0 = not set; use persist config default */
    arguments.max_wal_records    = 0; /* 0 = not set; use persist config default */
    arguments.wal_segment_bytes  = 0; /* 0 = not set; use persist config default */
    arguments.jobmgr_nhash       = 0; /* 0 = not set; use cluster tunables default */
    arguments.jobmgr_delta       = 0; /* 0 = not set; use cluster tunables default */
    arguments.workermgr_nhash    = 0; /* 0 = not set; use cluster tunables default */
    arguments.workermgr_delta    = 0; /* 0 = not set; use cluster tunables default */
    arguments.qmgr_nhash         = 0; /* 0 = not set; use cluster tunables default */
    arguments.qmgr_delta         = 0; /* 0 = not set; use cluster tunables default */
    arguments.resmgr_nhash       = 0; /* 0 = not set; use cluster tunables default */
    arguments.resmgr_delta       = 0; /* 0 = not set; use cluster tunables default */
    arguments.resmgr_inst_delta  = 0; /* 0 = not set; use cluster tunables default */
    arguments.xstrings_nhash     = 0; /* 0 = not set; use cluster tunables default */
    arguments.xstrings_delta     = 0; /* 0 = not set; use cluster tunables default */

    argp_parse(&argp, argc, argv, 0, 0, &arguments);

    if (arguments.logfile != NULL) {
        logfile = arguments.logfile;
    }

    if (arguments.verbose != 0) {
        loglevel = verbose_to_loglevel(arguments.verbose);
    }

    mylog_init(logfile, loglevel, logsize, logcount);

    if (arguments.workdir == NULL) {
        INFO("no cluster work directory specified. The default work directory "
             "`" DEFAULT_WORKDIR "' will be used.");
        arguments.workdir = DEFAULT_WORKDIR;
    }

    if (arguments.myname == NULL) {
        if (get_myname(myname, sizeof(myname) - 1) < 0) {
            ERROR("get_myname");
            return -1;
        }
        arguments.myname = myname;
    }

    INFO("My name is %s, version %s", arguments.myname, version_buf);

    const char *certdir = arguments.certdir;
    char certdir_buf[XCE_MAXPATH_LEN + 1];
    if (!certdir) {
        snprintf(certdir_buf, sizeof(certdir_buf),
                 "%s/certs", arguments.workdir);
        certdir = certdir_buf;
    }
    snprintf(key_file, XCE_MAXPATH_LEN, "%s/xce.key", certdir);
    snprintf(cert_file, XCE_MAXPATH_LEN, "%s/xce.crt", certdir);

    myconf.cluster_config    = NULL;
    myconf.conffile          = NULL; /* set by load_primary_config() or fetch_config_from_primary() */
    myconf.certfile          = cert_file;
    myconf.keyfile           = key_file;
    myconf.workdir           = arguments.workdir;
    myconf.name              = arguments.myname;
    myconf.conf_keep_n       = arguments.conf_keep_n;
    myconf.conf_stale_timeout = (time_t)arguments.conf_stale_timeout;
    myconf.vcpus             = arguments.vcpus;
    myconf.sync_timeout_ms   = arguments.sync_timeout_ms;
    myconf.wal_sync_ms       = arguments.wal_sync_ms;
    myconf.max_wal_records   = arguments.max_wal_records;
    myconf.wal_segment_bytes = arguments.wal_segment_bytes;
    myconf.jobmgr_nhash        = arguments.jobmgr_nhash;
    myconf.jobmgr_delta        = arguments.jobmgr_delta;
    myconf.workermgr_nhash     = arguments.workermgr_nhash;
    myconf.workermgr_delta     = arguments.workermgr_delta;
    myconf.qmgr_nhash          = arguments.qmgr_nhash;
    myconf.qmgr_delta          = arguments.qmgr_delta;
    myconf.resmgr_nhash        = arguments.resmgr_nhash;
    myconf.resmgr_delta        = arguments.resmgr_delta;
    myconf.resmgr_inst_delta   = arguments.resmgr_inst_delta;
    myconf.xstrings_nhash      = arguments.xstrings_nhash;
    myconf.xstrings_delta      = arguments.xstrings_delta;
    memset(&myconf.peers_cfg, 0, sizeof(myconf.peers_cfg));

    /* bug-1476: myconf.lock serializes load_primary_config() against
     * concurrent invocations for this myconf_t — must be initialized before
     * cluster_run/myapp_run can ever call it. */
    if (mylock_init(&myconf.lock) != 0) {
        ERROR("Failed to initialize myconf lock");
        return -1;
    }

    if (arguments.peers_str) {
        if (xce_parse_peers(arguments.peers_str, &myconf.peers_cfg) != 0) {
            fprintf(stderr,
                    "xce_controller: --peers parse error: expected "
                    "host:port,... got: %s\n",
                    arguments.peers_str);
            return 1;
        }

        /* Req 4.1: reject even-N peer counts (cannot form safe majority) */
        if (validate_peer_count(myconf.peers_cfg.candidates_count) != 0) {
            fprintf(stderr,
                    "xce_controller: --peers: even peer count (%d) is not "
                    "allowed; Raft quorum requires an odd number of nodes "
                    "(1, 3, 5, ...)\n",
                    myconf.peers_cfg.candidates_count);
            return 1;
        }

        strncpy(myconf.peers_cfg.identity, arguments.myname,
                sizeof(myconf.peers_cfg.identity) - 1);
        INFO("HA mode: %d controller candidates, identity=%s",
             myconf.peers_cfg.candidates_count, myconf.peers_cfg.identity);
        for (int i = 0; i < myconf.peers_cfg.candidates_count; i++) {
            if (myconf.peers_cfg.candidates[i].hostname[0] == '\0' ||
                myconf.peers_cfg.candidates[i].port == 0) {
                WARN("--peers: candidate %d has empty host or port 0 — "
                     "check peer string", i);
            }
        }
    } else {
        INFO("No --peers flag; running as single-node primary");
    }

    /* Req 4.2: cert pre-flight — fail early with a clear path in the message */
    {
        char cert_err[XCE_MAXPATH_LEN + 64];
        if (validate_cert_files(cert_file, key_file,
                                cert_err, sizeof(cert_err)) != 0) {
            fprintf(stderr, "xce_controller: %s\n", cert_err);
            return 1;
        }
    }

    myapp_run(&myconf);

    mylog_fini();

    return 0;
}