AriWeb-Suite/lib/jobs.mjs

/**
 * lib/jobs.mjs — 내구성 작업 엔진(§2.2·§29·§36).
 *
 * 생각의 틀
 *   배포·복원·이전·백업·주문 처리처럼 "오래 걸리고 중간에 프로세스가 죽을 수 있는" 일은 모두 작업(job)이다.
 *   작업은 DB 행이며 단계(step)마다 결과를 저장한다. 프로세스가 재시작되면 끝난 단계는 건너뛰고 다음 단계부터 이어서 실행한다.
 *   완료되지 않은 작업을 실패로 숨기지 않고, 재시도 횟수·다음 시각·외부 대기 사유·사용자 확인 필요를 상태로 드러낸다.
 *
 * 핵심 규약
 *   - register(kind, def): def = { title, steps:[{id,title}], run(job, ctx), revert?(job, ctx), maxAttempts?, retryable?(err) }
 *   - run() 은 **재진입 가능**해야 한다: 같은 작업으로 여러 번 불릴 수 있으므로 부수 효과가 있는 일은 ctx.step(id, fn) 안에서 한다.
 *     ctx.step 은 이미 done 인 단계면 저장된 결과를 돌려주고 fn 을 실행하지 않는다.
 *   - 같은 idempotencyKey 로 enqueue 하면 새 작업을 만들지 않고 기존 작업을 돌려준다(버튼 연타·새로 고침 §36).
 *   - 일시 오류(err.retryable=true 또는 def.retryable(err)) 는 지수 백오프로 재시도하고 attempt·next_run_at 을 보인다.
 *   - ctx.waitExternal(reason, {resumeAt}) → AWAITING_EXTERNAL. ctx.needsUser(message, actions) → NEEDS_USER. resume(id) 로 다시 QUEUED.
 *   - 되돌리기: revert(id) 는 def.revert 를 실행하며 REVERTING → REVERTED(실패하면 FAILED + error). 되돌리기 전 데이터 영향은 호출자가 안내한다.
 *   - 부모/자식: parentId 로 대상별 작업(여러 서버 모듈 갱신 §29)을 묶고 childrenSummary 로 대상별 성공·실패·대기를 보인다. 일부 실패를 전체 성공으로 보이지 않는다.
 */
import { newId } from './crypto.mjs';
import { redact, safeError } from './redact.mjs';
import { nowIso, toJson, parseJson } from './db.mjs';
import { JOB_STATUS as S, TERMINAL, CANCELLABLE } from './status.mjs';

export const MIGRATIONS = [{ id: 'lib_jobs_0001', sql: `
CREATE TABLE IF NOT EXISTS jobs (
  id TEXT PRIMARY KEY, kind TEXT NOT NULL, status TEXT NOT NULL, label TEXT, subject_json TEXT, steps_json TEXT, attempt INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 1,
  next_run_at TEXT, locked_by TEXT, locked_at TEXT, result_json TEXT, error TEXT, error_code TEXT, wait_reason TEXT, user_message TEXT, actions_json TEXT, eta_at TEXT, progress INTEGER NOT NULL DEFAULT 0,
  idempotency_key TEXT UNIQUE, created_by TEXT, actor_type TEXT, group_id TEXT, subject_type TEXT, subject_id TEXT, parent_id TEXT, external_ref TEXT, cancel_requested INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL, updated_at TEXT NOT NULL, started_at TEXT, ended_at TEXT, log_json TEXT
);
CREATE INDEX IF NOT EXISTS jobs_status ON jobs(status, next_run_at);
CREATE INDEX IF NOT EXISTS jobs_group ON jobs(group_id, created_at);
CREATE INDEX IF NOT EXISTS jobs_subject ON jobs(subject_type, subject_id, created_at);
CREATE INDEX IF NOT EXISTS jobs_parent ON jobs(parent_id);` }];

/** run() 안에서 흐름을 바꾸는 신호(예외로 전달) */
class Signal extends Error { constructor(type, payload) { super(type); this.signal = type; this.payload = payload || {}; } }
export class JobError extends Error { constructor(code, message, { retryable = false, detail, status = 400 } = {}) { super(message || code); this.code = code; this.retryable = retryable; this.detail = detail; this.status = status; /* API 로 새면 400 계열 — 500 으로 숨기지 않는다 */ } }

const BACKOFF_BASE_MS = 30 * 1000; const BACKOFF_MAX_MS = 60 * 60 * 1000; const STALE_LOCK_MS = 10 * 60 * 1000;

/**
 * 작업 엔진 만들기.
 * @param {object} store lib/db
 * @param {{workerId?:string, pollMs?:number, log?:Function}} opts
 */
export function createJobEngine(store, opts = {}) {
  const kinds = new Map(); const listeners = new Set();
  const workerId = opts.workerId || `${process.pid}@${(await0() || 'host')}`; const log = opts.log || (() => {});
  let timer = null; let running = 0; let stopped = true; const concurrency = opts.concurrency || 4;

  function row2job(r) { if (!r) return null; return { ...r, subject: parseJson(r.subject_json, {}), steps: parseJson(r.steps_json, []), result: parseJson(r.result_json, null), actions: parseJson(r.actions_json, null), log: parseJson(r.log_json, []) }; }
  function emit(job) { for (const fn of listeners) { try { fn(job); } catch (e) { log('[jobs] listener error', e.message); } } }
  function save(id, patch) {
    const sets = []; const vals = [];
    for (const [k, v] of Object.entries(patch)) { sets.push(`${k}=?`); vals.push(v && typeof v === 'object' ? toJson(v) : v); }
    sets.push('updated_at=?'); vals.push(nowIso());
    store.run(`UPDATE jobs SET ${sets.join(', ')} WHERE id=?`, ...vals, id);
    const j = engine.get(id); emit(j); return j;
  }
  /** 실행 가능한 작업 하나를 잠근다(큐 → 실행 중). 없으면 null */
  function claim() {
    if (running >= concurrency) return null;
    const now = nowIso();
    const cand = store.get('SELECT id FROM jobs WHERE status=? AND (next_run_at IS NULL OR next_run_at<=?) ORDER BY next_run_at ASC, created_at ASC LIMIT 1', S.QUEUED, now);
    if (!cand) return null;
    const got = store.run('UPDATE jobs SET status=?, locked_by=?, locked_at=?, started_at=COALESCE(started_at, ?), attempt=attempt+1, updated_at=? WHERE id=? AND status=?', S.RUNNING, workerId, now, now, now, cand.id, S.QUEUED).changes;
    return got ? cand.id : null;
  }
  const engine = {
    workerId,
    register(kind, def) { kinds.set(kind, { maxAttempts: 1, steps: [], ...def }); return engine; },
    kinds: () => [...kinds.keys()],
    onChange(fn) { listeners.add(fn); return () => listeners.delete(fn); },
    /**
     * 작업 넣기. idempotencyKey 가 같은 작업이 있으면 그것을 돌려준다(created=false).
     * @returns {{job:object, created:boolean}}
     */
    enqueue({ kind, subject = {}, idempotencyKey = null, createdBy = null, actorType = 'user', groupId = null, subjectType = null, subjectId = null, label = null, parentId = null, runAt = null, maxAttempts = null, initialStatus = null }) {
      const def = kinds.get(kind); if (!def) throw new JobError('unknown_job_kind', `알 수 없는 작업 종류: ${kind}`);
      return store.tx(() => {
        if (idempotencyKey) { const ex = store.get('SELECT * FROM jobs WHERE idempotency_key=?', idempotencyKey); if (ex) return { job: row2job(ex), created: false }; }
        const id = newId('j_'); const now = nowIso();
        const steps = (def.steps || []).map((s) => ({ id: s.id, title: s.title, state: 'pending' }));
        store.run('INSERT INTO jobs(id, kind, status, label, subject_json, steps_json, attempt, max_attempts, next_run_at, idempotency_key, created_by, actor_type, group_id, subject_type, subject_id, parent_id, created_at, updated_at, log_json) VALUES(?,?,?,?,?,?,0,?,?,?,?,?,?,?,?,?,?,?,?)',
          id, kind, initialStatus || S.QUEUED, label || def.title || kind, toJson(subject), toJson(steps), maxAttempts || def.maxAttempts || 1, runAt || now, idempotencyKey, createdBy, actorType, groupId, subjectType, subjectId, parentId, now, now, '[]');
        const job = engine.get(id); emit(job); return { job, created: true };
      });
    },
    get: (id) => row2job(store.get('SELECT * FROM jobs WHERE id=?', id)),
    list({ groupId, status, kind, subjectType, subjectId, parentId, limit = 100 } = {}) {
      const w = []; const p = [];
      if (groupId) { w.push('group_id=?'); p.push(groupId); } if (status) { w.push(Array.isArray(status) ? `status IN (${status.map(() => '?').join(',')})` : 'status=?'); p.push(...[].concat(status)); }
      if (kind) { w.push('kind=?'); p.push(kind); } if (subjectType) { w.push('subject_type=?'); p.push(subjectType); } if (subjectId) { w.push('subject_id=?'); p.push(subjectId); } if (parentId) { w.push('parent_id=?'); p.push(parentId); }
      return store.all(`SELECT * FROM jobs ${w.length ? 'WHERE ' + w.join(' AND ') : ''} ORDER BY created_at DESC LIMIT ?`, ...p, Math.min(limit, 1000)).map(row2job);
    },
    /** 자식 작업 요약(대상별 성공·실패·대기). 일부 실패는 'partial' */
    childrenSummary(parentId) {
      const kids = engine.list({ parentId, limit: 1000 }); const by = {}; for (const k of kids) by[k.status] = (by[k.status] || 0) + 1;
      const failed = kids.filter((k) => ['failed', 'needs_user'].includes(k.status)).length; const done = kids.filter((k) => k.status === 'done').length;
      return { total: kids.length, byStatus: by, overall: kids.length === 0 ? 'none' : done === kids.length ? 'done' : failed && done ? 'partial' : failed === kids.length ? 'failed' : 'running', children: kids.map((k) => ({ id: k.id, label: k.label, status: k.status, subject: k.subject, error: k.error, attempt: k.attempt, next_run_at: k.next_run_at })) };
    },
    /** 외부 대기·사용자 확인·일시 중지 상태의 작업을 다시 큐에 넣는다 */
    resume(id, by = null) { const j = engine.get(id); if (!j) throw new JobError('job_not_found', '작업이 없습니다'); if (![S.AWAITING_EXTERNAL, S.NEEDS_USER, S.PAUSED, S.AWAITING_APPROVAL, S.AWAITING_PAYMENT, S.NOT_READY, S.NEEDS_SETUP].includes(j.status)) throw new JobError('job_not_resumable', `지금 상태(${j.status})에서는 이어서 할 수 없습니다`); return save(id, { status: S.QUEUED, next_run_at: nowIso(), wait_reason: null, user_message: null, actions_json: null, log_json: appendLog(j, `재개 (${by || 'system'})`) }); },
    /** 실패한 작업 재시도(사용자·운영자 명시) */
    retry(id, by = null) { const j = engine.get(id); if (!j) throw new JobError('job_not_found', '작업이 없습니다'); if (![S.FAILED, S.NEEDS_USER].includes(j.status)) throw new JobError('job_not_retryable', `지금 상태(${j.status})에서는 재시도할 수 없습니다`); return save(id, { status: S.QUEUED, next_run_at: nowIso(), error: null, error_code: null, max_attempts: j.attempt + 1, log_json: appendLog(j, `재시도 요청 (${by || 'system'})`) }); },
    pause(id, by = null) { const j = engine.get(id); if (!j || ![S.QUEUED, S.AWAITING_EXTERNAL].includes(j.status)) throw new JobError('job_not_pausable', '지금 상태에서는 일시 중지할 수 없습니다'); return save(id, { status: S.PAUSED, log_json: appendLog(j, `일시 중지 (${by || 'system'})`) }); },
    /** 취소: 대기 상태면 즉시, 실행 중이면 취소 요청 표시(ctx.checkCancel 에서 멈춤) */
    cancel(id, by = null) {
      const j = engine.get(id); if (!j) throw new JobError('job_not_found', '작업이 없습니다');
      if (CANCELLABLE.has(j.status)) return save(id, { status: S.CANCELLED, ended_at: nowIso(), log_json: appendLog(j, `취소 (${by || 'system'})`) });
      if (j.status === S.RUNNING) return save(id, { cancel_requested: 1, log_json: appendLog(j, `취소 요청 (${by || 'system'})`) });
      throw new JobError('job_not_cancellable', `지금 상태(${j.status})에서는 취소할 수 없습니다`);
    },
    /** 되돌리기 */
    async revert(id, by = null) {
      const j = engine.get(id); if (!j) throw new JobError('job_not_found', '작업이 없습니다');
      const def = kinds.get(j.kind); if (!def || !def.revert) throw new JobError('job_not_revertible', '이 작업은 되돌리기를 지원하지 않습니다');
      if (![S.DONE, S.FAILED, S.NEEDS_USER].includes(j.status)) throw new JobError('job_not_revertible', `지금 상태(${j.status})에서는 되돌릴 수 없습니다`);
      save(id, { status: S.REVERTING, log_json: appendLog(j, `되돌리기 시작 (${by || 'system'})`) });
      const ctx = makeCtx(engine.get(id), { reverting: true });
      try { await def.revert(engine.get(id), ctx); return save(id, { status: S.REVERTED, ended_at: nowIso(), steps_json: ctx.steps(), log_json: ctx.logs() }); }
      catch (e) { return save(id, { status: S.FAILED, error: safeError(e), error_code: e.code || 'revert_failed', steps_json: ctx.steps(), log_json: ctx.logs(), ended_at: nowIso() }); }
    },
    /** 외부(HAL 등)에서 진행 상태를 밀어 넣을 때 */
    update(id, patch) { return save(id, patch); },
    /** 프로세스 시작 시: 이전 실행에서 RUNNING 으로 남은 작업(잠금 소유자 불문, 단일 프로세스 운영)을 QUEUED 로 되돌려 이어서 실행한다 */
    recover() {
      const stale = store.all('SELECT * FROM jobs WHERE status=?', S.RUNNING);
      for (const r of stale) { const j = row2job(r); save(j.id, { status: S.QUEUED, locked_by: null, locked_at: null, next_run_at: nowIso(), log_json: appendLog(j, '프로세스 재시작 — 끝난 단계 다음부터 이어서 실행') }); }
      return stale.length;
    },
    /** 실행 가능한 작업 하나를 집어 실행(테스트에서도 직접 부름). 돌아온 값: 실행한 작업 또는 null */
    async tick() { const id = claim(); if (!id) return null; running++; try { return await execute(id); } finally { running--; } },
    /** 루프용: 작업을 집어 **기다리지 않고** 실행을 시작한다(느린 작업이 다른 작업을 막지 않게, 부모/자식 작업 교착 방지). 집은 작업 id 또는 null */
    dispatch() { const id = claim(); if (!id) return null; running++; execute(id).catch((e) => log('[jobs] execute error', safeError(e))).finally(() => { running--; }); return id; },
    /** 모든 큐 작업을 비울 때까지 tick(테스트용) */
    async drain(max = 100) { let n = 0; while (n < max) { const r = await engine.tick(); if (!r) break; n++; } return n; },
    start() { stopped = false; engine.recover(); const loop = () => { if (stopped) return; try { let n = 0; while (running < concurrency && n < concurrency && engine.dispatch()) n++; } catch (e) { log('[jobs] dispatch error', safeError(e)); } timer = setTimeout(loop, opts.pollMs || 1000); }; loop(); return engine; },
    stop() { stopped = true; if (timer) clearTimeout(timer); },
  };
  function appendLog(job, line) { const l = (job.log || []).slice(-200); l.push({ at: nowIso(), line: String(line).slice(0, 500) }); return l; }
  function makeCtx(job, { reverting = false } = {}) {
    const steps = (job.steps || []).map((s) => ({ ...s })); const logs = (job.log || []).slice(-200);
    const ctx = {
      job, reverting, workerId,
      /** 단계 실행. 이미 done 이면 저장된 결과를 돌려준다. 결과는 JSON 직렬화 가능해야 한다. */
      async step(id, fn, title) {
        let s = steps.find((x) => x.id === id); if (!s) { s = { id, title: title || id, state: 'pending' }; steps.push(s); }
        if (s.state === 'done' && !reverting) return s.result;
        s.state = 'running'; s.startedAt = nowIso(); delete s.error; ctx.flush();
        try { const result = await fn(); s.state = 'done'; s.endedAt = nowIso(); s.result = result === undefined ? null : redact(result); ctx.flush(); return result; }
        catch (e) { if (e instanceof Signal) { s.state = e.signal === 'wait' ? 'waiting' : e.signal === 'needs_user' ? 'needs_user' : 'pending'; s.detail = e.payload.reason || e.payload.message || ''; ctx.flush(); throw e; } s.state = 'failed'; s.endedAt = nowIso(); s.error = safeError(e); ctx.flush(); throw e; }
      },
      log(line) { logs.push({ at: nowIso(), line: String(line).slice(0, 500) }); ctx.flush(); },
      progress(pct) { store.run('UPDATE jobs SET progress=?, updated_at=? WHERE id=?', Math.max(0, Math.min(100, Math.round(pct))), nowIso(), job.id); },
      eta(at) { store.run('UPDATE jobs SET eta_at=?, updated_at=? WHERE id=?', at || null, nowIso(), job.id); },
      /** 외부 작업(사용자 DNS 변경·운영자 승인 등)을 기다린다. run() 은 여기서 끝나고 resume() 뒤 다시 불린다. */
      waitExternal(reason, { resumeAt = null } = {}) { throw new Signal('wait', { reason, resumeAt }); },
      /** 사용자 확인이 필요해 멈춘다. actions: [{id,label,description}] */
      needsUser(message, actions = null) { throw new Signal('needs_user', { message, actions }); },
      /** 일시 오류로 나중에 다시(지수 백오프 대신 지정 시각) */
      retryLater(ms, reason) { throw new Signal('retry', { ms, reason }); },
      /** 취소 요청이 들어왔으면 멈춘다 */
      checkCancel() { const r = store.get('SELECT cancel_requested FROM jobs WHERE id=?', job.id); if (r && r.cancel_requested) throw new Signal('cancel', {}); },
      steps: () => steps, logs: () => logs,
      flush() { store.run('UPDATE jobs SET steps_json=?, log_json=?, updated_at=? WHERE id=?', toJson(steps), toJson(logs), nowIso(), job.id); },
    };
    return ctx;
  }
  async function execute(id) {
    const job = engine.get(id); const def = kinds.get(job.kind);
    if (!def) return save(id, { status: S.FAILED, error: `알 수 없는 작업 종류: ${job.kind}`, error_code: 'unknown_job_kind', ended_at: nowIso() });
    const ctx = makeCtx(job);
    try {
      const result = await def.run(job, ctx);
      return save(id, { status: S.DONE, result_json: redact(result === undefined ? null : result), progress: 100, steps_json: ctx.steps(), log_json: ctx.logs(), ended_at: nowIso(), locked_by: null, error: null, error_code: null });
    } catch (e) {
      if (e instanceof Signal) {
        // 단계 밖에서 대기·확인 신호가 나면 다음 미완료 단계에 그 상태를 표시해 화면이 "현재 단계" 를 보일 수 있게 한다
        if (['wait', 'needs_user'].includes(e.signal)) { const st = ctx.steps(); if (!st.some((x) => ['waiting', 'needs_user'].includes(x.state))) { const nx = st.find((x) => x.state === 'pending'); if (nx) { nx.state = e.signal === 'wait' ? 'waiting' : 'needs_user'; nx.detail = e.payload.reason || e.payload.message || ''; } } }
        if (e.signal === 'wait') return save(id, { status: S.AWAITING_EXTERNAL, wait_reason: e.payload.reason || null, eta_at: e.payload.resumeAt || null, next_run_at: e.payload.resumeAt || null, steps_json: ctx.steps(), log_json: ctx.logs(), locked_by: null });
        if (e.signal === 'needs_user') return save(id, { status: S.NEEDS_USER, user_message: e.payload.message || null, actions_json: e.payload.actions || null, steps_json: ctx.steps(), log_json: ctx.logs(), locked_by: null });
        if (e.signal === 'cancel') return save(id, { status: S.CANCELLED, ended_at: nowIso(), steps_json: ctx.steps(), log_json: ctx.logs(), locked_by: null, cancel_requested: 0 });
        if (e.signal === 'retry') { const at = new Date(Date.now() + (e.payload.ms || BACKOFF_BASE_MS)).toISOString(); return save(id, { status: S.QUEUED, next_run_at: at, wait_reason: e.payload.reason || null, max_attempts: Math.max(job.max_attempts, job.attempt + 1), steps_json: ctx.steps(), log_json: ctx.logs(), locked_by: null }); }
      }
      const retryable = !!(e && e.retryable) || (def.retryable ? !!def.retryable(e) : false);
      const fresh = engine.get(id);
      if (retryable && fresh.attempt < fresh.max_attempts) {
        const backoff = Math.min(BACKOFF_MAX_MS, BACKOFF_BASE_MS * 2 ** (fresh.attempt - 1));
        return save(id, { status: S.QUEUED, next_run_at: new Date(Date.now() + backoff).toISOString(), error: safeError(e), error_code: e.code || 'transient', steps_json: ctx.steps(), log_json: ctx.logs(), locked_by: null });
      }
      log(`[jobs] ${job.kind} ${id} 실패: ${safeError(e)}`);
      return save(id, { status: S.FAILED, error: safeError(e), error_code: e.code || 'failed', steps_json: ctx.steps(), log_json: ctx.logs(), ended_at: nowIso(), locked_by: null });
    }
  }
  return engine;
}
function await0() { try { return (process.env.HOSTNAME || '').slice(0, 20); } catch (_) { return ''; } }
/** 화면용 요약: 현재 단계·완료 단계·다음 단계·재시도 정보(§2.2) */
export function describe(job) {
  if (!job) return null;
  const steps = job.steps || []; const doneSteps = steps.filter((s) => s.state === 'done'); const current = steps.find((s) => ['running', 'waiting', 'needs_user', 'failed'].includes(s.state)) || steps.find((s) => s.state === 'pending') || null;
  const idx = current ? steps.indexOf(current) : steps.length; const next = steps[idx + 1] || null;
  return {
    id: job.id, kind: job.kind, label: job.label, status: job.status, progress: job.progress, attempt: job.attempt, maxAttempts: job.max_attempts, nextRunAt: job.next_run_at, etaAt: job.eta_at || null,
    etaUnknownReason: job.eta_at ? null : '외부 반영 시간에 따라 달라져 산정할 수 없습니다', updatedAt: job.updated_at, startedAt: job.started_at, endedAt: job.ended_at,
    currentStep: current ? { id: current.id, title: current.title, state: current.state, detail: current.detail || current.error || null } : null,
    completedSteps: doneSteps.map((s) => ({ id: s.id, title: s.title })), nextStep: next ? { id: next.id, title: next.title } : null, steps,
    waitReason: job.wait_reason, userMessage: job.user_message, actions: job.actions, error: job.error, errorCode: job.error_code,
    canCancel: CANCELLABLE.has(job.status) || job.status === 'running', canRetry: ['failed', 'needs_user'].includes(job.status), canResume: ['awaiting_external', 'needs_user', 'paused'].includes(job.status), terminal: TERMINAL.has(job.status),
    stale: job.status === 'running' && Date.now() - Date.parse(job.updated_at) > 10 * 60 * 1000,
  };
}