#include <errno.h>
#include <signal.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "jobj.h"
#include "persist.h"
#include "preason.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
#include "xstring.h"

static int mark_job_start(cluster_t*, job_t*, job_pidinfo_t*);
static int mark_job_finish(cluster_t*, job_t*, xce_jobfinish_t*);
static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_update(cluster_t*, job_t*, int);

static job_t* create_job(cluster_t* c, xce_job_t* j)
{
    job_t* job = NULL;

    job = makeJob(c->jobmgr, j);

    return job;
}

uint64_t job_submit(void* c, xce_job_submit_t* submit)
{
    job_t*    job     = NULL;
    uint64_t  base_id = submit->jobid;
    xce_job_t j       = { .submit = submit->submit };
    time_t    now     = time(NULL);
    int       i;

    j.sub_time     = now;
    j.queue_time   = now;
    j.start_time   = 0;
    j.finish_time  = 0;
    j.pidinfo      = NULL;
    j.state        = JOB_STAT_PEND;
    j.allocs       = NULL;
    j.run_time     = 0;
    j.run_usage    = NULL;
    j.exit_info    = NULL;
    j.ckptinfo     = NULL;
    j.pending_ctrl = 0;

    for (i = 0; i < submit->array_size; i++) {
        j.jobid = base_id + i;
        job     = create_job(c, &j);
        if (job == NULL) {
            return -errno;
        }
    }

    return base_id;
}

int load_jobs(void* c)
{
    return 0;
}

/* Reconstruct a job from a WAL_JOB_SUBMIT Cap'n Proto payload.
 * Uses the jobid embedded in the serialized xce_job_t (not the caller's arg).
 * Returns -EEXIST if the job is already present (idempotent replay guard). */
int restore_job(cluster_t* c, uint64_t jobid_unused,
                const void* capnp_payload, size_t len)
{
    (void)jobid_unused;

    if (!capnp_payload || len == 0)
        return -EINVAL;

    struct capn c_ctx;
    capn_init_mem(&c_ctx, (const uint8_t*)capnp_payload, len, 0);
    Job_ptr    jp;
    jp.p = capn_getp(capn_root(&c_ctx), 0, 1);
    struct Job cj;
    xce_job_t  j;
    memset(&j, 0, sizeof(j));
    read_Job(&cj, jp);
    decode_Job(&j, &cj);
    capn_free(&c_ctx);

    /* Safety guard: skip if already present (WAL replay may see duplicates
     * when recovery restarts from a segment boundary). */
    job_t* existing = NULL;
    lock(c->jobmgr, RTTI_LOCK_READ);
    int found = key_find(c->jobmgr, (void*)j.jobid, (void**)&existing);
    unlock(c->jobmgr);
    if (found > 0 && existing != NULL)
        return -EEXIST;

    job_t* job = create_job(c, &j);
    if (!job)
        return -errno;
    return 0;
}

static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
    int i;

    for (i = 0; i < alloc->nallocs; i++) {
        if (strcmp(alloc->allocs[i]->resname, "slots") == 0) {
            return (int)alloc->allocs[i]->value;
        }
    }

    return 1;
}

static void free_launchinfo(xce_launch_t* launch)
{
    int i;

    for (i = 0; i < launch->nhosts; i++) {
        FREEUP(launch->hosts[i]);
    }

    FREEUP(launch->hosts);
}

static int mk_launchinfo(xce_launch_t* launch, xce_job_alloc_t* alloc)
{
    int i;
    int rc = 0;

    memset(launch, 0x0, sizeof(*launch));

    launch->hosts = (xce_host_launch_t**)calloc(alloc->nhosts,
                                                sizeof(xce_host_launch_t*));
    if (launch->hosts == NULL) {
        return -ENOMEM;
    }

    launch->nhosts = alloc->nhosts;
    for (i = 0; i < alloc->nhosts; i++) {
        launch->hosts[i]
            = (xce_host_launch_t*)calloc(1, sizeof(xce_host_launch_t));
        if (launch->hosts[i] == NULL) {
            rc = -ENOMEM;
            break;
        }

        launch->hosts[i]->host   = alloc->allocs[i]->host;
        launch->hosts[i]->nranks = get_nslots_in_alloc(alloc->allocs[i]);
    }

    if (rc < 0) {
        free_launchinfo(launch);
    }

    return rc;
}

static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
{
    int                rc = 0;
    job_t*             j  = NULL;
    worker_t*          w  = NULL;
    xce_host_alloc_t** allocs;
    int                nc;
    xce_job_exec_t     einfo;
    xce_launch_t       launchinfo;

    TRACE("%s jobid %ld", __func__, alloc->jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    rc = key_find(c->jobmgr, (void*)alloc->jobid, (void**)&j);
    if (j != NULL) {
        reference(j);
    }
    unlock(c->jobmgr);

    if (rc < 0) {
        return rc;
    }

    if (j == NULL) {
        PERROR("jobmgr_lockjob");

        return -errno;
    }

    nc = alloc->nhosts;
    if (nc <= 0) {
        dereference(j);

        return -EINVAL;
    }

    /* FIXME: change case of nc > 1 just hard coded mpi as
     * the real job type later
     */
    mk_launchinfo(&launchinfo, alloc);

    allocs = (xce_host_alloc_t**)alloc->allocs;
    /* using head node of job */
    lock(c->workermgr, RTTI_LOCK_READ);
    rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w);
    if (w != NULL) {
        reference(w);
    }
    unlock(c->workermgr);
    if (rc < 0) {
        return rc;
    }

    if (w == NULL) {
        free_launchinfo(&launchinfo);
        return -errno;
    }

    lock(w, RTTI_LOCK_READ);
    if ((w->ep == NULL) || (w->host.status != 0)) {
        ERROR("worker %s is not in correct state, status %x ep %p",
              w->host.name, w->host.status, w->ep);
        free_launchinfo(&launchinfo);

        SET_ERROR(E_XCE_BAD_WORKER);

        unlock(w);

        dereference(w);

        return -errno;
    }

    /* FIXME: we may consider to optimize when a job has enter a queue? */
    if (1) {
        char* q_pre_exec;
        char* q_post_exec;

        rc = get_queue_exec(c, j->job.submit->queue, &q_pre_exec, &q_post_exec);
        if (rc < 0) {
            return rc;
        }

        if (j->job.submit->start_info->q_pre_exec != NULL) { }

        if (j->job.submit->start_info->q_post_exec != NULL) { }

        j->job.submit->start_info->q_pre_exec  = q_pre_exec;
        j->job.submit->start_info->q_post_exec = q_post_exec;
    }

    einfo.jobid      = alloc->jobid;
    einfo.start_info = j->job.submit->start_info;
    einfo.launchinfo = &launchinfo;

    rc = xce_startjob_req(w->ep, &einfo);

    unlock(w);

    dereference(w);

    free_launchinfo(&launchinfo);

    if (j->job.allocs != NULL) {
        free_JobAlloc_ptr(&(j->job.allocs));
    }
    j->job.allocs        = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t));
    j->job.allocs->jobid = alloc->jobid;
    j->job.allocs->nhosts = nc;
    MOVE_PTR(j->job.allocs->allocs, allocs);
    alloc->allocs = NULL;

    if (rc < 0) {
        SET_ERROR(-rc);
        PERROR("xce_startjob_req");
        return rc;
    }

    return rc;
}

int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo)
{
    uint64_t jobid = ctrinfo->jobid;
    job_t*   j     = NULL;
    int      rc    = 0;

    TRACE("%s job_ctrl job %ld", __func__, jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    rc = key_find(c->jobmgr, (void*)jobid, (void**)&j);
    if (j != NULL) {
        reference(j);
    }
    unlock(c->jobmgr);

    if (rc < 0) {
        return rc;
    }

    if (j == NULL) {
        PERROR("jobmgr_lockjob");
        return -ENOENT;
    }

    /* FIXME: eliminate held multiple locks */
    lock(j, RTTI_LOCK_READ);
    if (j->job.allocs != NULL) {
        xce_host_alloc_t** allocs = j->job.allocs->allocs;
        worker_t*          w      = NULL;

        /* job control command will be sent to first node
         *  for parallel jobs
         */

        lock(c->workermgr, RTTI_LOCK_READ);
        rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w);
        if (w != NULL) {
            reference(w);
        }
        unlock(c->workermgr);

        if (rc < 0) {
            return rc;
        }
        if (w != NULL) {
            lock(w, RTTI_LOCK_READ);
            /* FIXME: better method to check host status */
            if ((w->ep != NULL) && (w->host.status == 0)) {
                rc = xce_jobctrl_req(w->ep, ctrinfo);
                if (rc < 0) {
                    ERROR("job ctrl %ld on host %s returns %d", jobid,
                          w->host.name, rc);
                }
            } else {
                ERROR("worker %s is not in correct state, status %x ep %p",
                      w->host.name, w->host.status, w->ep);
                rc = -errno;
            }
            unlock(w);

            dereference(w);
        } else {
            rc = -errno;
        }
    } else {
        /* job has not been started yet */
        job_failinfo_t failinfo = {
            .jobid = jobid,
        };

        xce_use(&failinfo);

        switch (ctrinfo->action) {
        case JobControl_u_siginfo:
            switch (ctrinfo->u.siginfo->signo) {
            case SIGCONT:
                if (j->job.state
                    & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) {
                    int state = j->job.state;

                    /* resume suspend job? */
                    state &= ~(JOB_STAT_PSUSP | JOB_STAT_USUSP);
                    mark_job_update(c, j, state);
                }
                break;
            case SIGSTOP:
            case SIGTSTP:
                if (!(j->job.state
                      & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP))) {
                    int state = j->job.state;

                    state |= JOB_STAT_PSUSP;
                    mark_job_update(c, j, state);
                }
                break;
            case SIGKILL:
            case SIGTERM:
            default:
                if (j->job.state
                    & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
                    break;
                }
                /* FIXME: */
                failinfo.code   = 128 + ctrinfo->u.siginfo->signo;
                failinfo.detail = "Job killed by signal";

                mark_job_fail(c, j, &failinfo);
                break;
            }
            break;
        case JobControl_u_ckptoption:
            if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) {
                ERROR("job <%ld> has been finished, no checkpoint can be done.",
                      jobid);
            } else {
                ERROR("job %ld has not been started, no checkpoint can be done",
                      jobid);
            }
            rc = -EINVAL;
            break;
        default:
            rc = -EINVAL;
            break;
        }
    }

    unlock(j);

    return rc;
}

static void notify_dispatch(cluster_t* c, endpoint_t* ep, uint64_t jobid,
                            int code)
{
    /* clang-format off */
    job_failinfo_t info = {
        .jobid = jobid,
        .code  = code
    };
    xce_notify_data_t data = {
        .event = NotifyData_data_jobDispatchFail,
        .data  = {
             .dispatchfail = &info
	}
    };
    /* clang-format on */

    notify(c, ep, &data);
}

int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int                 rc = 0;
    int                 i;
    xce_job_dispatch_t* dispatchmsg = request->body.dispatchmsg;
    xce_reply_t         reply;

    /* reply job dispatch that we'll handle request */
    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

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

    for (i = 0; i < dispatchmsg->njobs; i++) {
        int              rc1;
        xce_job_alloc_t* alloc = NULL;

        MOVE_PTR(alloc, dispatchmsg->allocs[i]);

        rc1 = dispatch_a_job_(c, ep, alloc);
        if (rc1 < 0) {
            /* notify dispatch fail */
            notify_dispatch(c, ep, alloc->jobid, rc1);
        } else {
            /* notify dispatch ok */
            notify_dispatch(c, ep, alloc->jobid, 0);
        }

        free_JobAlloc_ptr(&(alloc));
    }
    return rc;
}

static int xlist_deref_job_item_(job_t* j, void* arg) { return 0; }

int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    uint64_t          jobid;
    xce_job_submit_t* submit;
    xce_reply_t       reply;
    xlist_t           jlist;
    job_t*            j = NULL;
    int               i;
    int               rc;

    submit = request->body.jobsubmsg;

    init_reply_by_req(&reply, request);

    if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue)
        || (submit->submit->start_info == NULL)
        || NULLSTR(submit->submit->start_info->cmdline)) {
        jobid = 0;
        errno = EINVAL;
    } else {
        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                      HOOK_DETAIL_JOB_NEW, c, submit);

        jobid = job_submit(c, submit);
    }
    if (jobid == 0) {
        reply.status     = -errno;
        reply.data_which = RequestReply_data_nodata;

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

    reply.status     = 0;
    reply.data_which = RequestReply_data_jobid;
    reply.data.jobid = jobid;

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

    xlist_init(&jlist, submit->array_size);

    for (i = 0; i < submit->array_size; i++) {
        /* Run hooks to tell scheduler */
        lock(c->jobmgr, RTTI_LOCK_READ);
        rc = key_find(c->jobmgr, (void*)(jobid + i), (void**)&j);
        if (j != NULL) {
            reference(j);
        }
        unlock(c->jobmgr);

        if ((rc <= 0) || (j == NULL)) {
            char idstring[128];

            fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
            ERROR("%s fail to lock job %s (%lu)", __func__, idstring, jobid);
        } else {
            if (xlist_store(&jlist, j) < 0) {
                PERROR("xlist_store");
            }
        }
    }

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_NEW, c, &jlist);

    xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL);

    xlist_fini(&jlist);

    return 0;
}

static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_START, c, info, job);
    /* update job information */
    job->job.state      = JOB_STAT_RUN;
    job->job.start_time = time(NULL);
    if (job->job.pidinfo == NULL) {
        job->job.pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t));
    }

    job->job.pidinfo->npids  = copy_int_list((int**)&(job->job.pidinfo->pids),
                                             (int*)info->pids, info->npids);
    job->job.pidinfo->npgids = copy_int_list((int**)&(job->job.pidinfo->pgids),
                                             (int*)info->pgids, info->npgids);
    if (info->exdata != NULL) {
        job->job.pidinfo->exdata = makeString(info->exdata);
    } else {
        job->job.pidinfo->exdata = NULL;
    }

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_START, c, info, job);

    return 0;
}

int on_job_start(cluster_t* c, job_pidinfo_t* info)
{
    job_t* job = NULL;
    int    rc;

    TRACE("%s jobid %ld", __func__, info->jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    rc = key_find(c->jobmgr, (void*)info->jobid, (void**)&job);
    if (job != NULL) {
        reference(job);
    }
    unlock(c->jobmgr);

    if ((rc <= 0) || (job == NULL)) {
        ERROR("can not find job %ld", info->jobid);
        return -errno;
    }

    lock(job, RTTI_LOCK_WRITE);
    mark_job_start(c, job, info);
    unlock(job);

    dereference(job);

    return 0;
}

static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_EXIT, c, finish, job);
    /* update job information */
    if (finish->exit_code == 0) {
        job->job.state = JOB_STAT_DONE;
    } else {
        job->job.state = JOB_STAT_EXIT;
    }
    job->job.finish_time = time(NULL);

    /* fill job run time/rusage information */

    if (job->job.exit_info == NULL) {
        job->job.exit_info = (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
    }

    if (job->job.exit_info != NULL) {
        job->job.exit_info->exitcode = finish->exit_code;
    }

    job->job.run_time = job->job.finish_time - job->job.start_time;

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_EXIT, c, finish, job);

    return 0;
}

int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
{
    job_t* job = NULL;

    TRACE("%s job finish jobid %ld exit code %d", __func__, finish->jobid,
          finish->exit_code);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)finish->jobid, (void**)&job);
    if (job != NULL) {
        reference(job);
    }
    unlock(c->jobmgr);

    if (job == NULL) {
        ERROR("can not find job %ld", finish->jobid);
        return -errno;
    }

    lock(job, RTTI_LOCK_WRITE);
    mark_job_finish(c, job, finish);
    unlock(job);

    dereference(job);

    return 0;
}

static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
    /* update job information */

    switch (failinfo->stage) {
    case JOB_EXEC_STAGE_Q_POST_EXEC:
    case JOB_EXEC_STAGE_POST_EXEC:
        if ((job->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
            WARN("job %ld got post exec fail notification but job's exit "
                 "status has not been set",
                 job->job.jobid);
            job->job.state |= JOB_STAT_EXIT;
        }
        job->job.state |= JOB_STAT_PERR;
        break;
    default:
        job->job.state = JOB_STAT_EXIT;
        break;
    }

    job->job.finish_time = time(NULL);

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_FAIL, c, failinfo, job);

    return 0;
}

static int mark_job_update(cluster_t* c, job_t* job, int state)
{
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, job);
    /* update job state */
    job->job.state = state;

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, job);

    return 0;
}

int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
{
    job_t* job = NULL;

    TRACE("%s job fail jobid %ld code %d", __func__, failinfo->jobid,
          failinfo->code);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)failinfo->jobid, (void**)&job);
    if (job != NULL) {
        reference(job);
    }
    unlock(c->jobmgr);

    if (job == NULL) {
        ERROR("can not find job %ld", failinfo->jobid);
        return -errno;
    }

    lock(job, RTTI_LOCK_WRITE);
    mark_job_fail(c, job, failinfo);
    unlock(job);

    dereference(job);

    return 0;
}

int on_job_stop(cluster_t* c, uint64_t jobid)
{
    job_t* job = NULL;
    int    state;

    TRACE("%s job stop jobid %ld", __func__, jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)jobid, (void**)&job);
    if (job != NULL) {
        reference(job);
    }
    unlock(c->jobmgr);

    if (job == NULL) {
        ERROR("can not find job %ld", jobid);
        return -errno;
    }

    lock(job, RTTI_LOCK_WRITE);
    state = job->job.state;

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_STOP, c, job);
    state |= JOB_STAT_USUSP;
    job->job.state = state;
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_STOP, c, job);
    unlock(job);

    dereference(job);

    return 0;
}

int on_job_resume(cluster_t* c, uint64_t jobid)
{
    job_t* job = NULL;
    int    state;

    TRACE("%s job resume jobid %ld", __func__, jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)jobid, (void**)&job);
    if (job != NULL) {
        reference(job);
    }
    unlock(c->jobmgr);

    if (job == NULL) {
        ERROR("can not find job %ld", jobid);
        return -errno;
    }

    lock(job, RTTI_LOCK_WRITE);
    state = job->job.state;
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_RESUME, c, job);
    state &= ~JOB_STAT_USUSP;
    job->job.state = state;
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_RESUME, c, job);
    unlock(job);

    dereference(job);

    return 0;
}

int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid)
{
    job_t* job = NULL;
    int    state;

    TRACE("%s job post done jobid %ld", __func__, jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)jobid, (void**)&job);
    if (job != NULL) {
        reference(job);
    }
    unlock(c->jobmgr);

    if (job == NULL) {
        ERROR("can not find job %ld", jobid);
        return -errno;
    }

    lock(job, RTTI_LOCK_WRITE);
    state = job->job.state;
    if ((state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) {
        WARN("job %ld got post exec fail notification but job's exit status "
             "has not been set",
             job->job.jobid);
        state |= JOB_STAT_EXIT;
        /* ensure that job finish time has been set */
        job->job.finish_time = time(NULL);
    }

    state |= JOB_STAT_PDONE;
    mark_job_update(c, job, state);
    unlock(job);

    dereference(job);

    return 0;
}

int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo)
{
    uint64_t jobid;
    char*    info = NULL;
    job_t*   j    = NULL;

    if (ckptinfo == NULL) {
        ERROR("%s ckptinfo should not be NULL", __func__);
        return -EINVAL;
    }

    jobid = ckptinfo->jobid;
    info  = ckptinfo->ckptinfo;

    TRACE("%s job %ld ckptinfo %s", __func__, jobid, info);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)jobid, (void**)&j);
    if (j != NULL) {
        reference(j);
    }
    unlock(c->jobmgr);

    if (j == NULL) {
        ERROR("%s can not find job %ld", __func__, jobid);
        return -errno;
    }

    lock(j, RTTI_LOCK_WRITE);
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_CKPT, c, j);
    if (j->job.ckptinfo != NULL) {
        dereference(j->job.ckptinfo);
    }

    j->job.ckptinfo = makeString(info);

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_CKPT, c, j);
    unlock(j);

    dereference(j);

    return 0;
}

int do_jobcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int         rc = 0;
    xce_reply_t reply;

    init_reply_by_req(&reply, request);

    rc               = job_ctrl(c, request->body.jobctrlmsg);
    reply.status     = rc;
    reply.data_which = RequestReply_data_nodata;

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

int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    xce_job_rusages_t* rusages = request->body.updatemsg->data.jobrusages;
    job_t*             j       = NULL;
    worker_t*          w       = NULL;
    char*              host    = NULL;

    TRACE("%s trying to update job(%ld) rusage", __func__, rusages->jobid);

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)rusages->jobid, (void**)&j);
    if (j != NULL) {
        reference(j);
    }
    unlock(c->jobmgr);

    if (j == NULL) {
        PERROR("jobmgr_lockjob");
        return -errno;
    }

    w = get_ep_worker(c, ep);
    if (w == NULL) {
        return errno;
    }

    host = makeString(w->host.name);
    if (host == NULL) {
        dereference(j);

        return -ENOMEM;
    }

    TRACE("%s update job (%ld) rusage for host %s", __func__, rusages->jobid,
          host);

    lock(j, RTTI_LOCK_WRITE);
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, j);
    update_host_runusage(&(j->job.run_usage), host, rusages->count,
                         rusages->usages);

    update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids);

    update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids);

    /* FIXME
     * update other fields of job rusage later
     */
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_UPDATE, c, j);

    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages);
    unlock(j);

    dereference(j);

    return 0;
}

static int set_preason(cluster_t* c, xce_pendreason_t* reasons)
{
    job_t*     j   = NULL;
    xce_job_t* job = NULL;

    TRACE("%s jobid %ld", __func__, reasons->jobid);

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)reasons->jobid, (void**)&j);
    if (j != NULL) {
        reference(j);
    }
    unlock(c->jobmgr);

    if (j == NULL) {
        PERROR("jobmgr_lockjob");
        return -ENOENT;
    }

    lock(j, RTTI_LOCK_WRITE);
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                  HOOK_DETAIL_SETPREASON, c, j);
    job = &(j->job);
    if (job->pend_reasons != NULL) {
        erase_job_preason(job->pend_reasons);
        FREEUP(job->pend_reasons);
    }

    MOVE_PTR(job->pend_reasons, reasons->reasons);
    xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                  HOOK_DETAIL_SETPREASON, c, j);
    unlock(j);

    dereference(j);

    return 0;
}

int do_set_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int                i;
    xce_preason_set_t* reasons = request->body.preasonmsg;
    xce_reply_t        reply;

    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

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

    for (i = 0; i < reasons->nreasons; i++) {
        int rc = 0;

        if (reasons->reasons[i] == NULL) {
            continue;
        }

        rc = set_preason(c, reasons->reasons[i]);
        if (rc < 0) {
            PERROR("set_preason");
        }
    }

    return 0;
}

int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int                 i;
    xce_reply_t         reply;
    xce_clear_reason_t* clreason = request->body.clrpreasonmsg;

    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;

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

    for (i = 0; i < clreason->njobs; i++) {
        uint64_t   jobid = clreason->jobs[i];
        job_t*     j;
        xce_job_t* job;

        lock(c->jobmgr, RTTI_LOCK_READ);
        key_find(c->jobmgr, (void*)jobid, (void**)&j);
        if (j != NULL) {
            reference(j);
        }
        unlock(c->jobmgr);

        if (j == NULL) {
            PERROR("jobmgr_lockjob");
            continue;
        }

        lock(j, RTTI_LOCK_WRITE);
        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
                      HOOK_DETAIL_CLRPREASON, c, j);

        job = &(j->job);
        if (job->pend_reasons != NULL) {
            erase_job_preason(job->pend_reasons);
            FREEUP(job->pend_reasons);
        }

        xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
                      HOOK_DETAIL_CLRPREASON, c, j);
        unlock(j);

        dereference(j);
    }

    return 0;
}

/* job_is_terminal and job_should_retain live in persist.c —
 * they only depend on xce_job_t fields and persist_config_t,
 * and must be accessible from the unit test binary which links persist.c. */