#include <errno.h>
#include <stdlib.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "request.h"
#include "xce/types.h"
#include "xstring.h"

static int xce_apiserver_push_configs(cluster_t* c, endpoint_t* ep)
{
    xce_configdata_t configdata;
    int              rc = 0;

    memset(&configdata, 0x0, sizeof(configdata));
    configdata.data.clusterconfig = c->myconf->cluster_config;
    configdata.data_which         = ConfigData_data_clusterconfig;

    rc = xce_pushconfig_req(ep, c->myhost, &configdata);

    return rc;
}

static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep)
{
    xce_data_t         data;
    xce_workers_data_t workers
        = { .n_total = 0, .pos = 0, .count = 0, .workers = NULL };
    int     i, rc = 0;
    xlist_t wlist;

    memset(&data, 0x0, sizeof(data));
    xlist_init(&wlist, 8192);

    data.table        = XceData_data_workers;
    data.data.workers = &workers;

    collect_workers(c, (collect_check_fn_t)check_worker_, &wlist);
    workers.n_total = wlist.count;
    workers.workers = (xce_worker_t**)wlist.data;
    workers.count   = workers.n_total;

    rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);

    /* free up collected resource value which used
       passed to api server
    */
    for (i = 0; i < workers.n_total; i++) {
        xce_worker_t* w = workers.workers[i];

        free_name_val_list(&(w->resvals_to_sched), &(w->nresvals_to_sched));
    }

    unlock_collect_workers(&wlist);

    FREEUP(workers.workers);

    return rc;
}

static int check_job_(xce_job_t* j)
{
    if (j == NULL) {
        return 0;
    }

    return 1;
}

static int xce_apiserver_push_jobs(cluster_t* c, endpoint_t* ep)
{
    xce_data_t      data;
    xce_jobs_data_t jobs = { .n_total = 0, .pos = 0, .count = 0, .jobs = NULL };
    int             rc   = 0;
    xlist_t         jlist;

    memset(&data, 0x0, sizeof(data));
    xlist_init(&jlist, 1024 * 1024);

    data.table     = XceData_data_jobs;
    data.data.jobs = &jobs;

    collect_jobs(c, (collect_check_fn_t)check_job_, &jlist);
    jobs.n_total = jlist.count;
    jobs.jobs    = (xce_job_t**)jlist.data;
    jobs.count   = jobs.n_total;

    rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);

    unlock_collect_jobs(&jlist);

    FREEUP(jobs.jobs);

    return rc;
}

static int apiserver_push_joblist(cluster_t* c, xlist_t* l)
{
    int i;
    for (i = 0; i < N_APISERVER_MAX; i++) {
        mylock_lock(&(c->apiservers[i].lock));
        if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
            int             rc;
            xce_data_t      data;
            xce_jobs_data_t jobs = { .n_total = l->count,
                                     .pos     = 0,
                                     .count   = l->count,
                                     .jobs    = (xce_job_t**)l->data };

            memset(&data, 0x0, sizeof(data));
            data.table     = XceData_data_jobs;
            data.data.jobs = &jobs;

            rc = xce_push_data_req(c->apiservers[i].ep, c->myhost, &data,
                                   check_controller, c, 100);
            if (rc < 0) {
                WARN("%s push job list to api server %s has problem, rc %d",
                     __func__, c->apiservers[i].name, rc);
            }
        }
        mylock_unlock(&(c->apiservers[i].lock));
    }

    return 0;
}

/* job-array-single-wal-record task 5.2 (design.md "Compact Scheduler/
 * API-Server Push", File Structure Plan apiserver.c bullet; Req 6.2, 6.4):
 * encodes ONE compact (base jobid, array size, one shared spec) Data.jobArray
 * push per connected API server -- reusing the JobSubmit capnp struct
 * (already added to the Data union in task 1.2, controller.capnp), never N
 * individually-encoded Job structs. (base_jobid, array_size) are derived
 * from the already-constructed xlist_t the HOOK_DETAIL_JOB_NEW fire carries
 * (apply_committed_entry's WAL_JOB_ARRAY_SUBMIT case, task 3.2): the first
 * element's jobid is the array base and l->count is the array size; every
 * element shares the same submit spec by construction (restore_job_array's
 * own construction discipline, job.c), so any one element's submit pointer
 * is the one shared spec to encode.
 *
 * Sent via xce_request() directly (OPCode_push / Request_body_pushmsg)
 * rather than xce_push_data_req(), mirroring sched.c's xcesched_push_
 * jobarray() (task 5.1) for the identical reason: that helper's dispatch
 * switch has explicit cases only for jobs/workers/resources/resvals and
 * hard-rejects (-EINVAL) anything else via its default case -- it cannot
 * carry a single non-list Data.jobArray descriptor. This mirrors xce_push_
 * req()'s (request.c, static) exact shape for a single non-list xce_data_t
 * payload.
 *
 * Fan-out mirrors apiserver_push_joblist()'s existing per-connected-server
 * loop exactly (same lock discipline, same name!=NULL && ep!=NULL guard) --
 * one push sent per connected server, not one push shared/broadcast. */
static int apiserver_push_jobarray(cluster_t* c, xlist_t* l)
{
    xce_job_t*       first;
    xce_job_submit_t job_array;
    int              i;

    if ((c == NULL) || (l == NULL) || (l->count <= 0)) {
        return -EINVAL;
    }

    first = (xce_job_t*)xlist_get(l, 0);
    if ((first == NULL) || (first->submit == NULL)) {
        return -EINVAL;
    }

    memset(&job_array, 0x0, sizeof(job_array));
    job_array.jobid      = first->jobid;
    job_array.array_size = (uint64_t)l->count;
    job_array.submit     = first->submit;

    for (i = 0; i < N_APISERVER_MAX; i++) {
        mylock_lock(&(c->apiservers[i].lock));
        if ((c->apiservers[i].name != NULL) && c->apiservers[i].ep != NULL) {
            xce_data_t    data;
            xce_request_t request;
            int           rc;

            memset(&data, 0x0, sizeof(data));
            data.table          = XceData_data_jobArray;
            data.data.job_array = &job_array;

            memset(&request, 0x0, sizeof(request));
            request.opcode       = OPCode_push;
            request.body_which   = Request_body_pushmsg;
            request.body.pushmsg = &data;

            rc = xce_request(c->apiservers[i].ep, &request);
            if (rc < 0) {
                WARN("%s push job array to api server %s has problem, rc %d",
                     __func__, c->apiservers[i].name, rc);
            }
        }
        mylock_unlock(&(c->apiservers[i].lock));
    }

    return 0;
}

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

    switch (situation) {
    case HOOK_AFTER_EVENT:
        rc = -EINVAL;
        if (l != NULL) {
            /* job-array-single-wal-record task 5.2 (Req 6.2, 6.4):
             * apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case is the
             * ONLY way this hook is ever fired with more than one element
             * (Req 3.4 -- one HOOK_DETAIL_JOB_NEW fire per committed array
             * record, carrying every constructed element). Route that case
             * through the compact descriptor push instead of the per-job
             * push. The ordinary single-job branch (l->count == 1) below is
             * the exact, unmodified call this function always made
             * (Req 8.1). */
            rc = (l->count > 1) ? apiserver_push_jobarray(c, l)
                                : apiserver_push_joblist(c, l);
        }
        break;
    default:
        break;
    }

    return rc;
}

static int apiserver_job_start_fun(int type, int situation, void* data,
                                   va_list ap)
{
    cluster_t*     c    = va_arg(ap, cluster_t*);
    job_pidinfo_t* info = va_arg(ap, job_pidinfo_t*);
    int            i;

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata = { .event = NotifyData_data_jobPidInfo,
                                            .data  = { .pidinfo = info } };

                /* FIXME: add job pid to API server? */
                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN("%s notify job start to api server %s fail, rc %d",
                         __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

static int apiserver_job_exit_fun(int type, int situation, void* data,
                                  va_list ap)
{
    cluster_t*       c    = va_arg(ap, cluster_t*);
    xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*);
    int              i;

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata
                    = { .event = NotifyData_data_jobFinishInfo,
                        .data  = { .finishinfo = info } };

                /* FIXME: we need to send exit code of job to API server */
                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN("%s notify job exit to api server %s fail, rc %d",
                         __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

static int apiserver_job_fail_fun(int type, int situation, void* data,
                                  va_list ap)
{
    cluster_t*      c    = va_arg(ap, cluster_t*);
    job_failinfo_t* info = va_arg(ap, job_failinfo_t*);
    int             i;

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata
                    = { .event = NotifyData_data_jobFailInfo,
                        .data  = { .failinfo = info } };

                /* FIXME: we need to add fail reason of job to api server */
                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN("%s notify job fail to api server %s fail, rc %d",
                         __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

static int apiserver_job_update_fun(int type, int situation, void* data,
                                    va_list ap)
{
    cluster_t* c   = va_arg(ap, cluster_t*);
    job_t*     job = va_arg(ap, job_t*);
    int        i;
    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata
                    = { .event = NotifyData_data_updateJob,
                        .data  = { .update_job = &(job->job) } };

                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN("%s notify job update to api server %s fail, rc %d",
                         __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

static int apiserver_job_rusage_update_fun(int type, int situation, void* data,
                                           va_list ap)
{
    cluster_t*         c       = va_arg(ap, cluster_t*);
    xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*);
    int                i;

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata
                    = { .event = NotifyData_data_schedJobRusages,
                        .data  = { .sched_job_rusages = rusages } };
                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN("%s notify rusage update to api server %s fail, rc %d",
                         __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

static int apiserver_job_hook_fun(int type, int situation, int detail,
                                  void* data, va_list ap)
{
    switch (detail) {
    case HOOK_DETAIL_JOB_NEW:
        return apiserver_job_new_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_START:
        return apiserver_job_start_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_EXIT:
        return apiserver_job_exit_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_UPDATE:
        return apiserver_job_update_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_FAIL:
        return apiserver_job_fail_fun(type, situation, data, ap);
    case HOOK_DETAIL_JOB_RUSAGE_UPDATE:
        return apiserver_job_rusage_update_fun(type, situation, data, ap);
    }

    return 0;
}

static int apiserver_push_workerlist(cluster_t* c, endpoint_t* ep, xlist_t* l)
{
    xce_data_t         data;
    xce_workers_data_t workers = { .n_total = l->count,
                                   .pos     = 0,
                                   .count   = l->count,
                                   .workers = (xce_worker_t**)l->data };
    int                rc      = 0;

    memset(&data, 0x0, sizeof(data));

    data.table        = XceData_data_workers;
    data.data.workers = &workers;

    rc = xce_push_data_req(ep, c->myhost, &data, check_controller, c, 100);
    return rc;
}

static int apiserver_worker_change(int type, int situation, void* data,
                                   cluster_t* c, worker_t* w)
{
    int     i;
    xlist_t resvalist;

    TRACE("%s worker %s idx %d situation %d", __func__, w->host.name, w->idx,
          situation);

    xlist_init(&resvalist, 256);
    collect_resval(c->resmgr, &(w->host.resvals), &resvalist);
    w->host.nresvals_to_sched = resvalist.count;
    w->host.resvals_to_sched  = (xce_name_val_t**)resvalist.data;

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                xlist_t wlist;
                int     rc;

                xlist_init(&wlist, 2);
                xlist_store(&wlist, w);
                rc = apiserver_push_workerlist(c, c->apiservers[i].ep, &wlist);
                if (rc < 0) {
                    WARN("%s notify worker change failed to api server %s, rc "
                         "%d",
                         __func__, c->apiservers[i].name, rc);
                }
                xlist_fini(&wlist);
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    w->host.nresvals_to_sched = 0;
    w->host.resvals_to_sched  = NULL;
    xlist_fini(&resvalist);

    return 0;
}

static int apiserver_worker_join(int type, int situation, void* data,
                                 va_list ap)
{
    /* Unified JOIN signature: (cluster_t*, char* name, worker_t*)
     * w is NULL for BEFORE_EVENT (worker not yet created). */
    cluster_t* c    = va_arg(ap, cluster_t*);
    char*      name = va_arg(ap, char*);
    worker_t*  w    = va_arg(ap, worker_t*);
    (void)name;

    if (w == NULL)
        return 0;

    return apiserver_worker_change(type, situation, data, c, w);
}

static int apiserver_worker_dead(int type, int situation, void* data,
                                 va_list ap)
{
    cluster_t* c = va_arg(ap, cluster_t*);
    worker_t*  w = va_arg(ap, worker_t*);

    /* FIXME: need to handle job status in this worker */
    if (situation == HOOK_BEFORE_EVENT) {
        return 0;
    }

    return apiserver_worker_change(type, situation, data, c, w);
}

static int apiserver_worker_update(int type, int situation, void* data,
                                   va_list ap)
{
    cluster_t* c = va_arg(ap, cluster_t*);
    worker_t*  w = va_arg(ap, worker_t*);

    if (situation == HOOK_BEFORE_EVENT) {
        return 0;
    }

    return apiserver_worker_change(type, situation, data, c, w);
}

static int apiserver_worker_hook_fun(int type, int situation, int detail,
                                     void* data, va_list ap)
{
    switch (detail) {
    case HOOK_DETAIL_WORKER_JOIN:
        return apiserver_worker_join(type, situation, data, ap);
    case HOOK_DETAIL_WORKER_DEAD:
        return apiserver_worker_dead(type, situation, data, ap);
    case HOOK_DETAIL_WORKER_UPDATE:
        return apiserver_worker_update(type, situation, data, ap);
    }

    return 0;
}

static int apiserver_resource_hook_fun(int type, int situation, int detail,
                                       void* data, va_list ap)
{
    cluster_t*         c       = va_arg(ap, cluster_t*);
    xce_resval_list_t* resvals = va_arg(ap, xce_resval_list_t*);
    int                i;

    if (detail != HOOK_DETAIL_RESOURCE_UPDATE) {
        return 0;
    }

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata
                    = { .event = NotifyData_data_schedUpdateResvals,
                        .data  = { .sched_update_resvals = resvals } };

                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN("%s notify api server %s resource update fail, rc %d",
                         __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

static int apiserver_queue_hook_fun(int type, int situation, int detail,
                                    void* data, va_list ap)
{
    cluster_t* c = va_arg(ap, cluster_t*);
    queue_t*   q = va_arg(ap, queue_t*);
    int        i;

    switch (situation) {
    case HOOK_AFTER_EVENT:
        for (i = 0; i < N_APISERVER_MAX; i++) {
            mylock_lock(&(c->apiservers[i].lock));
            if ((c->apiservers[i].name != NULL)
                && (c->apiservers[i].ep != NULL)) {
                int               rc;
                xce_notify_data_t xdata
                    = { .event = NotifyData_data_schedUpdateQueue,
                        .data  = { .sched_update_queue = &(q->queue) } };

                rc = notify(c, c->apiservers[i].ep, &xdata);
                if (rc < 0) {
                    WARN(
                        "%s notify queue change to api server %s failed, rc %d",
                        __func__, c->apiservers[i].name, rc);
                }
            }
            mylock_unlock(&(c->apiservers[i].lock));
        }
        break;
    default:
        break;
    }

    return 0;
}

int apiserver_init(cluster_t* c)
{
    int i;

    for (i = 0; i < N_APISERVER_MAX; i++) {
        mylock_init(&(c->apiservers[i].lock));
        c->apiservers[i].name = NULL;
        c->apiservers[i].ep   = NULL;
    }

    xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
                 (hook_fn_t)apiserver_job_hook_fun, NULL);
    xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER,
                 (hook_fn_t)apiserver_worker_hook_fun, NULL);
    xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_RESOURCE,
                 (hook_fn_t)apiserver_resource_hook_fun, NULL);
    xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE,
                 (hook_fn_t)apiserver_queue_hook_fun, NULL);

    return 0;
}

int on_apiserver_join(cluster_t* c, endpoint_t* ep)
{
    xce_apiserver_push_configs(c, ep);
    xce_apiserver_push_workers(c, ep);
    xce_apiserver_push_jobs(c, ep);

    return 0;
}

int handle_apiserver_leave(cluster_t* c, endpoint_t* ep)
{
    ep_private_t* priv;
    int           i;

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

    priv = (ep_private_t*)ep->data;
    if (priv == NULL) {
        return -EINVAL;
    }

    i = priv->u.idx;
    if (i < 0) {
        SET_ERROR(EINVAL);
        return -errno;
    }

    if (c->apiservers[i].name != NULL) {
        c->apiservers[i].ep = NULL;
    }

    /* FIXME: kill ep? */
    return 0;
}

int add_apiserver(cluster_t* c, char* name, endpoint_t* ep)
{
    int i;

    for (i = 0; i < N_APISERVER_MAX; i++) {
        if ((c->apiservers[i].name != NULL)
            && (strcmp(c->apiservers[i].name, name) == 0)) {
            if (c->apiservers[i].ep != NULL) {
                /* FIXME: we need to delete 1 */
            }
            c->apiservers[i].ep = ep;
            return i;
        }
    }

    for (i = 0; i < N_APISERVER_MAX; i++) {
        if (c->apiservers[i].name == NULL) {
            c->apiservers[i].name = makeString(name);
            c->apiservers[i].ep   = ep;
            return i;
        }
    }

    /* better error code */
    return -1;
}