#include "config.h"

#include <errno.h>
#include <rtti.h>
#include <string.h>
#include <unistd.h>

#include "cluster.h"
#include "dataq.h"
#include "htab.h"
#include "kvmgr.h"
#include "mylock.h"
#include "request.h"
#include "vtaskmgr.h"
#include "xce/cluster_def.h"
#include "zarray.h"

#include "xce/types.h"

#include "jobj.h"
#include "resobj.h"
#include "utils.h"
#include "worker.h"
#include "xce/errnox.h"
#include "xstring.h"

static xce_peerinfo_t from_controller
    = { .name = "controller", .key = "keepSecret" };

static ep_private_t* ep_get_private()
{
    ep_private_t* priv;

    priv = (ep_private_t*)calloc(1, sizeof(*priv));
    return priv;
}

static int handle_client_leave(cluster_t* c) { return 0; }

static void handle_endpoint_error(endpoint_t* ep)
{
    cluster_t*    c    = (cluster_t*)ep->owner;
    ep_private_t* priv = NULL;
    int           role = -1;

    if (c == NULL) {
        ERROR("endpoint %p has no owner cluster", ep);

        return;
    }

    priv = (ep_private_t*)ep->data;

    TRACE("endpoint %p priv data %p", ep, priv);

    if (priv != NULL) {
        role = priv->role;
        switch (role) {
        case HostRole_worker:
            if (priv->u.idx >= 0) {
                handle_worker_leave(c, priv->u.idx);
            } else {
                ERROR("incorrect worker idx %d", priv->u.idx);
            }
            break;
        case HostRole_scheduler:
            handle_scheduler_leave(c);
            break;
        case HostRole_controller:
            break;
        case HostRole_client:
            handle_client_leave(c);
            break;
        case HostRole_apiserver:
            handle_apiserver_leave(c, ep);
            break;
        default:
            ERROR("unknown role %d for endpoint %p", priv->role, ep);
            break;
        }
    }

    if (ep->idx >= 0) {
        /* zarray_del will finalize ep */
        /* TODO P0-8b: replace with deferred cleanup after P0-0a zarray merge is complete */
        zarray_del(&(c->endpoints), ep->idx);
        return;
    }

    endpoint_fini(ep);
    FREEUP(ep);

    return;
}

static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req)
{
    xce_notify_data_t* data = req->body.notifymsg;

    switch (data->event) {
    case NotifyData_data_jobPidInfo:
        on_job_start(c, data->data.pidinfo);
        break;
    case NotifyData_data_jobFailInfo:
        on_job_fail(c, data->data.failinfo);
        break;
    case NotifyData_data_jobFinishInfo:
        on_job_finish(c, data->data.finishinfo);
        break;
    case NotifyData_data_stoppedJob:
        on_job_stop(c, data->data.stopped_job);
        break;
    case NotifyData_data_resumedJob:
        on_job_resume(c, data->data.resumed_job);
        break;
    case NotifyData_data_ckptinfo:
        on_job_ckpt(c, ep, data->data.ckptinfo);
        break;
    case NotifyData_data_postDoneJob:
        on_job_postdone(c, ep, data->data.pdone_job);
        break;
    case NotifyData_data_taskPidInfo:
        on_task_start(c, data->data.taskinfo);
        break;
    case NotifyData_data_taskFailInfo:
        on_task_fail(c, data->data.taskfailinfo);
        break;
    case NotifyData_data_taskFinish:
        on_task_finish(c, data->data.taskfinish);
        break;
    default:
        break;
    }
    return 0;
}
static int myreq_handler(vtask_t* task)
{
    vtask_data_t* data = (vtask_data_t*)task->data;
    endpoint_t*   ep   = data->u.iodata.ep;
    cluster_t*    c    = (cluster_t*)ep->owner;
    int           stop = 0;
    int           rc   = 0;
    xce_request_t request;
    xce_reply_t   reply;

    rc = xce_decode(data->u.iodata.pkt, &request);

    TRACE("%s cpu#%d opcode %d", __func__, task->vcpu, request.opcode);

    init_reply_by_req(&reply, &request);

    switch (request.opcode) {
    case OPCode_shutdown:
        reply.status     = 0;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        stop = 1;
        cbuf_put(&(c->wbuf), &stop);
        break;
    case OPCode_hello:
    case OPCode_nop:
        reply.status     = 0;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);
        break;
    case OPCode_join:
        rc = handle_join(c, ep, &request);
        break;
    case OPCode_jobdispatch:
        rc = handle_jobdispatch(c, ep, &request);
        break;
    case OPCode_jobsub:
        rc = handle_jobsubmit(c, ep, &request);
        break;
    case OPCode_get:
        rc = do_get_op(c, ep, &request);
        break;
    case OPCode_notify:
        rc = do_notify(c, ep, &request);
        break;
    case OPCode_clusterconf:
        rc = handle_clusterconf(c, ep, &request);
        break;
    case OPCode_update:
        rc = do_update_op(c, ep, &request);
        break;
    case OPCode_jobcontrol:
        rc = do_jobcontrol_op(c, ep, &request);
        break;
    case OPCode_queuecontrol:
        rc = do_queuecontrol_op(c, ep, &request);
        break;
    case OPCode_hostcontrol:
        rc = do_hostcontrol_op(c, ep, &request);
        break;
    case OPCode_runtasks:
        rc = do_runtasks(c, ep, &request);
        break;
    case OPCode_setpreason:
        rc = do_set_preason(c, ep, &request);
        break;
    case OPCode_clrpreason:
        rc = do_clr_preason(c, ep, &request);
        break;
    default:
        rc = -E_XCE_BAD_OPCODE;
        ERROR("Unknown opcode: %d", request.opcode);
        reply.status     = rc;
        reply.data_which = RequestReply_data_nodata;
        xce_reply(ep, c->myhost, &reply);

        break;
    }

    free_Request(&request);

    return rc;
}

static int vtask_entry(vtask_t* task)
{
    int rc;

    rc = myreq_handler(task);

    return rc;
}

static int handle_msg(endpoint_t* ep)
{
    cluster_t* c = NULL;

    if (ep == NULL) {
        return -EINVAL;
    }

    c = (cluster_t*)ep->owner;
    if (c == NULL) {
        return -EINVAL;
    }

    return vtask_handle_msg(&(c->vtaskmgr), (vtask_func_t)vtask_entry,
                            (void*)ep);
}

static int handle_connection(struct epoll_event* p)
{
    endpoint_t*   ep = (endpoint_t*)p->data.ptr;
    endpoint_t*   ep2;
    cluster_t*    c     = (cluster_t*)ep->owner;
    iomgr_t*      iomgr = &(c->iomgr);
    int           idx;
    ep_private_t* priv = NULL;

    ep2 = endpoint_get();
    if (ep2 == NULL) {
        PERROR("endpoint_get");

        return -ENOMEM;
    }

    endpoint_accept(ep, ep2);
    if (ep2->fd < 0) {
        PERROR("endpoint_accept");
        return -1;
    }

    TRACE("server sock %d accepted %d\n", ep->fd, ep2->fd);

    priv = ep_get_private();
    if (priv == NULL) {
        PERROR("ep_get_private");

        endpoint_fini(ep2);
        FREEUP(ep2);

        return -ENOMEM;
    }

    /* Assume client as default */
    priv->role  = HostRole_client;
    priv->u.idx = -1;

    idx = zarray_store(&(c->endpoints), ep2);
    if (idx < 0) {
        SET_ERROR(-idx);

        PERROR("zarray_store");

        FREEUP(priv);
        endpoint_fini(ep2);
        FREEUP(ep2);

        return idx;
    }

    ep2->idx   = idx;
    ep2->data  = priv;
    ep2->iomgr = iomgr;

    endpoint_set_owner(ep2, c);
    endpoint_set_peer(ep2, c->myhost);
    endpoint_set_msg_handler(ep2, (msg_handler_t)handle_msg);
    endpoint_set_err_handler(ep2, (err_handler_t)handle_endpoint_error);

    iomgr_monitor(ep2);
    return 0;
}

static int controller_init(controller_t* c, cluster_t* cluster)
{
    c->ep = NULL;

    return 0;
}

static void endpoint_killer(void* p)
{
    endpoint_t* ep = (endpoint_t*)p;
    cluster_t*  c  = (cluster_t*)ep->owner;

    if (ep->idx >= 0) {
        zarray_clear(&(c->endpoints), ep->idx);
        ep->idx = -1;
    }
    endpoint_fini(ep);
    FREEUP(ep);

    return;
}

int cluster_build_config(cluster_t* c)
{
    char valbuf[128];
    int  i;

    snprintf(valbuf, sizeof(valbuf) - 1, "%d",
             c->myconf->cluster_config->heartbeat_interval);
    add_config(c, "heartbeat_interval", valbuf);
    snprintf(valbuf, sizeof(valbuf) - 1, "%d",
             c->myconf->cluster_config->inactive_heartbeat_count);
    add_config(c, "inactive_heartbeat_count", valbuf);
    snprintf(valbuf, sizeof(valbuf) - 1, "%d",
             c->myconf->cluster_config->ncandidates);
    add_config(c, "ncandidates", valbuf);

    for (i = 0; i < c->myconf->cluster_config->ncandidates; i++) {
        char keybuf[128];

        sprintf(keybuf, "candidate_%d_name", i);
        sprintf(valbuf, "%s", c->myconf->cluster_config->candidates[i]->name);

        add_config(c, keybuf, valbuf);

        sprintf(keybuf, "candidate_%d_host", i);
        sprintf(valbuf, "%s", c->myconf->cluster_config->candidates[i]->host);

        add_config(c, keybuf, valbuf);
        sprintf(keybuf, "candidate_%d_port", i);
        sprintf(valbuf, "%d", c->myconf->cluster_config->candidates[i]->port);

        add_config(c, keybuf, valbuf);
    }

    return 0;
}

int cluster_build_resources(cluster_t* c)
{
    int i;

    if (c->myconf == NULL) {
        return -EINVAL;
    }

    for (i = 0; i < c->myconf->cluster_config->nres; i++) {
        xce_resource_t* res;

        res = makeResource(c->resmgr, c->myconf->cluster_config->resources[i]);
        if (res == NULL) {
            return -errno;
        }
    }

    return 0;
}

int cluster_load_jobs(cluster_t* c)
{
    load_jobs(c);

    return 0;
}

int cluster_build_queues(cluster_t* c)
{
    build_queues(c);
    return 0;
}

int cluster_build_groups(cluster_t* c)
{
    /* TODO: implement group building */
    return 0;
}

static int xce_config_init(int type, int situation, void* data, va_list ap)
{
    int        rc = 0;
    cluster_t* c  = va_arg(ap, cluster_t*);

    if (situation == HOOK_BEFORE_EVENT) {
        c->myconf->cluster_config = get_default_conf(c);
        if (c->myconf->cluster_config == NULL) {
            return -EINVAL;
        }
        return 0;
    }

    xce_use(c);

    rc = cluster_build_config(c);
    if (rc < 0) {
        return rc;
    }
    rc = cluster_build_resources(c);
    if (rc < 0) {
        return rc;
    }

    rc = cluster_build_queues(c);
    if (rc < 0) {
        return rc;
    }
    rc = cluster_build_groups(c);
    if (rc < 0) {
        return rc;
    }

    rc = cluster_load_jobs(c);
    if (rc < 0) {
        return rc;
    }

    tlog_info("xce controller is running");

    return rc;
}

static int cluster_hook_fun(int type, int situation, int detail, void* data,
                            va_list ap)
{
    if (detail == HOOK_DETAIL_CLUSTER_INIT) {
        return xce_config_init(type, situation, data, ap);
    }

    return 0;
}

static int xce_install_hooks(cluster_t* c)
{
    xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
                 (hook_fn_t)cluster_hook_fun, NULL);
    return 0;
}

static int run_delay_tasks(delay_task_entry_fn_t entry, void* arg)
{
    return entry(arg);
}

static void res_obj_killer(xce_resource_t* r)
{
    if (r == NULL) {
        return;
    }
}

static void job_obj_killer(void* arg)
{
    if (arg == NULL) {
        return;
    }
}

static void worker_obj_killer(void* arg)
{
    if (arg == NULL) {
        return;
    }
}

static void queue_obj_killer(void* arg)
{
    if (arg == NULL) {
        return;
    }
}

static void resval_obj_killer(void* arg)
{
    if (arg == NULL) {
        return;
    }
}

static void config_obj_killer(void* arg)
{
    if (arg == NULL) {
        return;
    }
}

int cluster_init(cluster_t* c)
{
    int            i, rc;
    SSL_CTX*       ssl_ctx = NULL;
    kvmgr_params_t config_mgr_params
        = { .key_type   = KVMGR_KEY_STRING,
            .nslots     = 128,
            .delta      = 128,
            .datakiller = (destroy_fn_t)config_obj_killer };
    kvmgr_params_t resmgr_params
        = { .key_type   = KVMGR_KEY_STRING,
            .datakiller = (destroy_fn_t)res_obj_killer };
    kvmgr_params_t jobmgr_params
        = { .key_type   = KVMGR_KEY_INT,
            .datakiller = (destroy_fn_t)job_obj_killer };
    kvmgr_params_t workermgr_params
        = { .key_type   = KVMGR_KEY_STRING,
            .datakiller = (destroy_fn_t)worker_obj_killer };
    kvmgr_params_t qmgr_params
        = { .key_type   = KVMGR_KEY_STRING,
            .datakiller = (destroy_fn_t)queue_obj_killer };

    if (c == NULL) {
        return -EINVAL;
    }

    c->myhost = &from_controller;
    /* FIXME: */
    c->ncontrollers = 1;
    c->myidx        = 0;

    xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX);

    xce_install_hooks(c);

    rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER,
                       HOOK_BEFORE_EVENT, HOOK_DETAIL_CLUSTER_INIT, c);
    if (rc < 0) {
        return rc;
    }

    mylock_init(&(c->sched.lock));
    c->sched.ep = NULL;

    initialize_openssl();
    ssl_ctx = create_ssl_context(c->myconf->certfile, c->myconf->keyfile);
    if (ssl_ctx == NULL) {
        return -errno;
    }

    resmgr_params.nslots    = c->myconf->cluster_config->resmgr->nhash_slots;
    resmgr_params.delta     = c->myconf->cluster_config->resmgr->delta;
    jobmgr_params.nslots    = c->myconf->cluster_config->jobmgr->nhash_slots;
    jobmgr_params.delta     = c->myconf->cluster_config->jobmgr->delta;
    workermgr_params.nslots = c->myconf->cluster_config->workermgr->nhash_slots;
    workermgr_params.delta  = c->myconf->cluster_config->workermgr->delta;
    qmgr_params.nslots      = c->myconf->cluster_config->qmgr->nhash_slots;
    qmgr_params.delta       = c->myconf->cluster_config->qmgr->delta;

    c->configmgr = makeInstance(&xkvmgr_type, &config_mgr_params);

    c->jobmgr = makeInstance(&xkvmgr_type, &jobmgr_params);
    c->resmgr = makeInstance(&xkvmgr_type, &resmgr_params);
    zarray_init(&(c->sharedinst), c->myconf->cluster_config->resmgr->inst_delta,
                (zarray_item_killer_t)resval_obj_killer);
    c->workermgr = makeInstance(&xkvmgr_type, &workermgr_params);
    c->qmgr      = makeInstance(&xkvmgr_type, &qmgr_params);
    iomgr_init(&(c->iomgr));

    iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);

    xstring_table_init(c->myconf->cluster_config->xstrings->nhash_slots);
    job_obj_init(5381, 5381);

    zarray_init(&(c->endpoints), 16, endpoint_killer);

    rc = endpoint_init(&(c->ep), AF_INET, SOCK_STREAM,
                       c->myconf->cluster_config->port, 5);
    if (rc < 0) {
        cleanup_openssl();
        return rc;
    }

    endpoint_set_peer(&(c->ep), c->myhost);

    for (i = 0; i < c->ncontrollers; i++) {
        controller_init(&(c->controllers[i]), c);
    }

    cbuf_init(&(c->wbuf), 1, sizeof(int));

    zarray_init(&(c->configs), 64, NULL /* FIXME */);

    vtaskmgr_init(&(c->vtaskmgr), c->myconf->cluster_config->vcpus);

    delaymgr_init(&(c->delaymgr), 200 /* precision_ms */, run_delay_tasks);

    apiserver_init(c);

    /* Persistence: derive paths, init, and recover from WAL/snapshot */
    {
        const char*      workdir = c->myconf->workdir ? c->myconf->workdir
                                                      : DEFAULT_WORKDIR;
        const char*      name    = c->myconf->name    ? c->myconf->name
                                                      : "controller";
        persist_config_t pcfg    = persist_config_defaults();

        persist_set_workdir(&c->persist, workdir, name);
        rc = persist_init(&c->persist, &pcfg, c);
        if (rc != 0) {
            ERROR("persist_init failed: %d", rc);
            return rc;
        }
        rc = persist_recover(&c->persist);
        if (rc != 0) {
            ERROR("persist_recover failed: %d", rc);
            persist_fini(&c->persist);
            return rc;
        }
    }

    /* HA initialisation — P1-5 */
    {
        ha_config_t ha_cfg = {
            .heartbeat_interval_ms    = 3000,
            .heartbeat_timeout_ms     = 10000,
            .election_timeout_ms      = 2000,
            .probe_connect_timeout_ms = 500,
            .max_promotion_retries    = 5,
        };
        /* my_priority: single-node P1 always uses 0 (highest priority) */
        rc = ha_init(&c->ha, &ha_cfg, 0);
        if (rc != 0) {
            ERROR("ha_init failed: %d", rc);
            persist_fini(&c->persist);
            return rc;
        }
        ha_install_sigusr1(&c->ha);

        /* Determine initial role: priority-0 node starts as primary */
        if (am_i_primary(c))
            ha_promote_to_primary(c);
        else
            atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY,
                                  memory_order_release);
    }

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_CLUSTER_INIT, c);

    return 0;
}

static int make_heartbeat(cluster_t* c, endpoint_t* ep)
{
    return xce_nop_req(ep);
}

static int check_worker_timeout_(cluster_t* c, worker_t* w)
{
    worker_t*         w1 = NULL;
    endpoint_t*       ep;
    time_t            now = time(NULL);
    int               timeout;
    cluster_config_t* config;

    if (w == NULL) {
        return 0;
    }

    if (c == NULL) {
        return -EINVAL;
    }

    ep = w->ep;
    if (ep == NULL) {
        return 0;
    }

    /* grab lock on worker */

    config  = c->myconf->cluster_config;
    timeout = config->heartbeat_interval * config->inactive_heartbeat_count;

    if (now - ep->ts_last_in > timeout) {
        TRACE("%s worker %s becomes UNAVAIL as heartbeat timeout", __func__,
              w->host.name);

        mark_worker_unavail(c, w);

        return 0;
    } else {
        if (IS_WORKER_UNAVAIL(w->host.status)) {
            mark_worker_back(c, w);
        }
    }

    if (now - ep->ts_last_out >= config->heartbeat_interval) {
        if (DATAQ_EMPTY(&(ep->outq))) {
            int rc = 0;

            TRACE("%s send heartbeat to %s", __func__, w->host.name);

            rc = make_heartbeat(c, ep);

            if (rc < 0) {
                ERROR("%s make_heartbeat to %s failed", __func__, w->host.name);

                return rc;
            }
        }
    }

    return 0;
}

static int check_workers_timeout(void* args)
{
    iterator_t it;
    cluster_t* c  = (cluster_t*)args;
    int        rc = 0;
    xlist_t    l;
    int        i;

    xlist_init(&l, 1024);

    lock(c->workermgr, RTTI_LOCK_READ);
    rc = iter_make(&it, c->workermgr);
    if (rc < 0) {
        unlock(c->workermgr);
        xlist_fini(&l);

        return rc;
    }

    for (;;) {
        worker_t* w = NULL;

        rc = iter_next(&it, (void**)&w);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        /* check_worker_timeout_(c, w); */
        if (w != NULL) {
            reference(w);

            xlist_store(&l, w);
        }
    }

    if (rc == RTTI_ITER_NO_DATA) {
        rc = 0;
    }

    iter_destroy(&it);
    unlock(c->workermgr);

    for (i = 0; i < l.count; i++) {
        worker_t* w = xlist_get(&l, i);

        if (w == NULL) {
            continue;
        }

        lock(w, RTTI_LOCK_WRITE);
        check_worker_timeout_(c, w);
        unlock(w);

        dereference(w);
    }

    xlist_fini(&l);

    return rc;
}

int cluster_start(cluster_t* c)
{
    int rc;

    endpoint_set_owner(&(c->ep), c);
    endpoint_set_conn_handler(&(c->ep), (conn_handler_t)handle_connection);
    c->ep.iomgr = &(c->iomgr);

    rc = iomgr_monitor(&(c->ep));

    if (rc < 0) {
        return rc;
    }

    iomgr_start(&(c->iomgr));
    vtaskmgr_start(&(c->vtaskmgr));
    delaymgr_start(&(c->delaymgr));

    /* Start persistence fsync thread and register WAL hooks */
    rc = persist_start(&c->persist);
    if (rc != 0) {
        ERROR("persist_start failed: %d", rc);
        return rc;
    }

    /* Start HA heartbeat watchdog thread — P1-5 */
    rc = ha_start_watchdog(&c->ha);
    if (rc != 0) {
        ERROR("ha_start_watchdog failed: %d", rc);
        return rc;
    }

    delay_task(&(c->delaymgr), c->myconf->cluster_config->heartbeat_interval,
               -1, (delay_task_entry_fn_t)check_workers_timeout, (void*)c);

    c->running = 1;
    return 0;
}

int cluster_run(cluster_t* c)
{
    for (;;) {
        int stop;

        cbuf_get(&(c->wbuf), &stop);
        if (stop) {
            break;
        }
    }

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_CLUSTER_SHUTDOWN, c);
    c->running = 0;

    delaymgr_stop(&(c->delaymgr));

    iomgr_unmonitor(&(c->iomgr), &(c->ep));
    iomgr_stop(&(c->iomgr));

    return 0;
}

void cluster_fini(cluster_t* c)
{
    /* Stop HA watchdog thread before tearing down other subsystems — P1-5 */
    ha_fini(&c->ha);

    /* Stop persistence before tearing down the rest */
    persist_stop(&c->persist);
    persist_fini(&c->persist);

    delaymgr_fini(&(c->delaymgr));
    vtaskmgr_fini(&(c->vtaskmgr));
    endpoint_fini(&(c->ep));

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_CLUSTER_SHUTDOWN, c);
    cbuf_fini(&(c->wbuf));
    iomgr_fini(&(c->iomgr));

    destroy(c->jobmgr);
    destroy(c->workermgr);
    destroy(c->qmgr);
    destroy(c->resmgr);
    destroy(c->configmgr);
    mylock_fini(&(c->sched.lock));

    zarray_fini(&(c->endpoints));
    zarray_fini(&(c->configs));

    xce_hook_fini(&(c->hooksys));

    cleanup_openssl();

    job_obj_fini();

    xstring_table_fini();

    return;
}