AriWeb-Suite/hal/jobs.mjs

/**
 * hal/jobs.mjs — HAL 작업 종류(lib/jobs 엔진에 등록).
 *
 * 두 부류가 있다.
 *   1) 포털이 요청하는 작업(docs/HAL-JOBS.md): site.*, domain.*, backup.*  — 대상 서버(또는 외부 Agent)에 Agent 작업을 배정하고 결과를 기다린 뒤 돌려준다.
 *      복합 작업(backup.create 의 원격 저장, site.relocate, backup.restore)은 여러 Agent 작업을 단계로 잇는다. 모든 단계는 ctx.step 으로 기록되어 HAL 재시작 뒤 이어서 실행된다.
 *   2) 운영자 작업(§32·§29): module.deploy / module.rollback / module.health / module.reconcile — 대상 서버마다 자식 작업을 만들고 대상별 결과를 요약한다. 일부 실패는 'partial' 이며 성공으로 보이지 않는다.
 *
 * 대상(target) 해석: 서버 id(srv_…) → 그 서버의 Agent, Agent id(ag_…) → 그 Agent(외부 Agent 포함).
 */
import { JobError } from '../lib/jobs.mjs';
import { nowIso, toJson } from '../lib/db.mjs';
import { audit } from '../lib/audit.mjs';
import { notify } from '../lib/notify.mjs';
import { safeError } from '../lib/redact.mjs';
import { readFileSync, existsSync } from 'node:fs';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import * as core from './core.mjs';
import { descriptor, DEPLOY_ORDER, ARTIFACT_MODULES } from './modules/index.mjs';

const ROOT = join(dirname(fileURLToPath(import.meta.url)), '..');

/** 대상 → Agent id */
function agentFor(d, target) {
  if (!target) throw new JobError('target_required', '대상 서버 또는 Agent 가 필요합니다');
  if (String(target).startsWith('ag_')) { const a = core.getAgent(d, target); if (!a || a.status !== 'active') throw new JobError('agent_unavailable', 'Agent 가 없거나 해지되었습니다'); return a.id; }
  const s = core.getServer(d, target); if (!s) throw new JobError('server_not_found', `서버가 없습니다: ${target}`);
  if (!s.agent_id) throw new JobError('agent_missing', `서버 ${s.name} 에 Agent 가 없습니다`);
  if (s.status === 'isolated') throw new JobError('server_isolated', `서버 ${s.name} 은(는) 격리 상태입니다`);
  return s.agent_id;
}
/** 외부 접속 확인(§11.2 8단계). 실제 응답을 본다. */
async function externalCheck(d, url, { expectMarker = null } = {}) {
  if (!url) return { ok: false, skipped: 'no_url' };
  const t0 = Date.now();
  try { const r = await (d.fetch || fetch)(url, { redirect: 'follow', signal: AbortSignal.timeout(15000), headers: { 'User-Agent': 'Ariweb-HAL-check/0.1' } }); const text = expectMarker ? await r.text() : ''; return { ok: r.ok && (!expectMarker || text.includes(expectMarker)), status: r.status, ms: Date.now() - t0, url, markerFound: expectMarker ? text.includes(expectMarker) : null }; }
  catch (e) { return { ok: false, error: safeError(e), ms: Date.now() - t0, url }; }
}
/** 단일 Agent 작업으로 끝나는 종류를 등록하는 도우미 */
function single(engine, d, kind, { title, agentKind = kind, steps = [], build = (s) => s.payload || {}, after = null, maxAttempts = 1, target = (s) => s.target }) {
  engine.register(kind, { title, maxAttempts, steps: [{ id: 'dispatch', title: '대상 Agent 에 작업 배정' }, { id: 'await', title: '대상 서버에서 실행' }, ...steps],
    async run(job, ctx) {
      const disp = await core.ensureAgentJob(d, ctx, job, { agentId: agentFor(d, target(job.subject)), kind: agentKind, payload: build(job.subject, d) });
      const result = await ctx.step('await', () => core.awaitAgentJob(d, ctx, disp.agentJobId));
      return after ? after(job, ctx, result) : result;
    },
    retryable: (e) => !!e.retryable,
  });
}

export function registerKinds(engine, d) {
  // ---------------- 사이트(§11)
  single(engine, d, 'site.deploy', { title: '사이트 배포', steps: [{ id: 'external', title: '외부 접속 확인' }], build: (s) => ({ ...s.payload, siteId: s.payload.siteId }),
    async after(job, ctx, result) {
      const ext = await ctx.step('external', () => externalCheck(d, job.subject.payload.externalCheckUrl || result?.siteUrl, { expectMarker: job.subject.payload.expectMarker }));
      return { ...result, externalCheck: ext };
    } });
  single(engine, d, 'site.rollback', { title: '이전 버전으로 되돌리기' });
  single(engine, d, 'site.remove', { title: '배포 제거' });
  single(engine, d, 'site.check', { title: '사이트 확인', async after(job, ctx, result) { const urls = job.subject.payload?.urls || []; const ext = await ctx.step('external', async () => { const out = []; for (const u of urls) out.push(await externalCheck(d, u, { expectMarker: job.subject.payload?.expectMarker })); return out; }, '외부 확인'); return { ...result, external: ext }; } });
  single(engine, d, 'site.usage', { title: '사용량 수집', maxAttempts: 3 });
  single(engine, d, 'domain.vhost', { title: '도메인 연결 설정' });
  single(engine, d, 'domain.cert', { title: 'HTTPS 인증서 발급', maxAttempts: 2 });
  single(engine, d, 'domain.probe', { title: 'DNS 확인', maxAttempts: 3 });

  // ---------------- 백업(§19·§20). 저장 서버가 운영 서버와 다르면 HAL 전송을 거쳐 옮긴다.
  engine.register('backup.create', { title: '백업 만들기', steps: [{ id: 'create', title: '운영 서버에서 백업 생성' }, { id: 'store', title: '백업 저장소로 복사' }, { id: 'verify', title: '읽을 수 있는지 확인' }],
    async run(job, ctx) {
      const s = job.subject; const p = s.payload || {}; const siteAgent = agentFor(d, s.target); const storeServer = p.storeServerId && p.storeServerId !== s.target ? p.storeServerId : null;
      const created = await ctx.step('create', async () => { const id = core.dispatchAgentJob(d, { halJobId: job.id, agentId: siteAgent, kind: 'backup.create', payload: { ...p, uploadToHal: !!storeServer, transferId: `tr_bk_${job.id}` } }); return core.awaitAgentJob(d, ctx, id); });
      let location = `${s.target}:${created.path}`; let stored = created;
      if (storeServer) stored = await ctx.step('store', async () => { const id = core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, storeServer), kind: 'backup.store', payload: { siteId: p.siteId, backupId: created.backupId, transferId: created.transferId, sha256: created.checksum, sizeBytes: created.sizeBytes } }); const r = await core.awaitAgentJob(d, ctx, id); location = `${storeServer}:${r.path}`; return r; });
      else await ctx.step('store', () => ({ skipped: '같은 서버에 보관(저장 서버 미지정)', warning: '운영 서버와 분리된 저장소를 권장합니다(§19.2)' }));
      const verified = await ctx.step('verify', async () => { const id = core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, storeServer || s.target), kind: 'backup.verify', payload: { location: stored.path, checksum: created.checksum } }); return core.awaitAgentJob(d, ctx, id); });
      if (!verified?.ok) throw new JobError('backup_unreadable', '생성한 백업을 읽을 수 없습니다');
      return { backupId: created.backupId, location, sizeBytes: created.sizeBytes, checksum: created.checksum, verified: true, createdAt: created.createdAt, scope: created.scope };
    } });
  const splitLoc = (loc) => { const i = String(loc).indexOf(':'); return i > 0 ? { serverId: loc.slice(0, i), path: loc.slice(i + 1) } : { serverId: null, path: loc }; };
  single(engine, d, 'backup.verify', { title: '백업 검증', target: (s) => splitLoc(s.payload.location).serverId || s.target, build: (s) => ({ location: splitLoc(s.payload.location).path, checksum: s.payload.checksum }) });
  single(engine, d, 'backup.delete', { title: '백업 삭제', target: (s) => splitLoc(s.payload.location).serverId || s.target, build: (s) => ({ location: splitLoc(s.payload.location).path }) });
  engine.register('backup.restore', { title: '사이트 복원', steps: [{ id: 'fetch', title: '백업 저장소에서 가져오기' }, { id: 'protect', title: '현재 상태 보호' }, { id: 'isolated', title: '격리된 환경에서 백업 확인' }, { id: 'restore', title: '데이터 복원' }, { id: 'external', title: '외부 접속 확인' }],
    async run(job, ctx) {
      const s = job.subject; const p = s.payload || {}; const loc = splitLoc(p.location); const siteAgent = agentFor(d, s.target); const remote = loc.serverId && loc.serverId !== s.target;
      const fetched = await ctx.step('fetch', async () => { if (!remote) return { path: loc.path }; const id = core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, loc.serverId), kind: 'backup.fetch', payload: { location: loc.path, transferId: `tr_rs_${job.id}`, checksum: p.checksum } }); return core.awaitAgentJob(d, ctx, id); });
      const r = await ctx.step('restore', async () => { const id = core.dispatchAgentJob(d, { halJobId: job.id, agentId: siteAgent, kind: 'backup.restore', payload: { ...p, location: remote ? null : fetched.path, transferId: remote ? fetched.transferId : null } }); return core.awaitAgentJob(d, ctx, id); });
      await ctx.step('protect', () => ({ protectPointLocation: r.protectPointLocation })); await ctx.step('isolated', () => ({ ok: r.isolatedVerify }));
      const ext = await ctx.step('external', () => externalCheck(d, p.externalCheckUrl, { expectMarker: p.expectMarker }));
      if (p.externalCheckUrl && !ext.ok) throw new JobError('restore_external_check_failed', '복원 뒤 외부 접속 확인에 실패했습니다 — 보호 시점으로 되돌릴 수 있습니다', { detail: ext });
      return { ...r, externalCheck: ext };
    },
    async revert(job, ctx) { const s = job.subject; const r = job.steps.find((x) => x.id === 'restore')?.result; if (!r?.protectPointLocation) throw new JobError('no_protect_point', '보호 시점이 없어 되돌릴 수 없습니다'); await ctx.step('revert', async () => { const id = core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, s.target), kind: 'backup.restore', payload: { siteId: s.payload.siteId, location: r.protectPointLocation, protectPoint: false, isolatedVerify: true }, dedupeKey: `${job.id}:revert` }); return core.awaitAgentJob(d, ctx, id); }, '보호 시점으로 되돌리기'); } });

  // ---------------- 서버 이전(§21). 두 서버에 동시에 운영 상태가 되지 않도록 verify_new 뒤에만 switch 한다.
  engine.register('site.relocate', { title: '서버 이전', steps: ['precheck', 'prepare', 'precopy', 'verify_copy', 'final_sync', 'verify_new', 'switch', 'external_check', 'cleanup'].map((id) => ({ id, title: { precheck: '조건 확인', prepare: '새 위치 준비', precopy: '데이터 사전 복사', verify_copy: '복사 결과 확인', final_sync: '최종 변경분 반영', verify_new: '새 위치 확인', switch: '연결 전환', external_check: '외부 접속 확인', cleanup: '이전 위치 정리' }[id] })),
    async run(job, ctx) {
      const p = job.subject.payload || {}; const from = agentFor(d, p.fromServerId); const to = agentFor(d, p.toServerId); const dyn = p.kind === 'dynamic';
      await ctx.step('precheck', () => { const t = core.getServer(d, p.toServerId); const f = core.getServer(d, p.fromServerId); if (!t.accepts_new || t.status !== 'ready') throw new JobError('target_not_ready', `도착 서버 ${t.name} 이(가) 신규 배치를 받을 수 없습니다`); if (f.status === 'offline') throw new JobError('source_offline', '출발 서버가 응답하지 않습니다'); const conflict = d.store.get('SELECT id FROM jobs WHERE subject_id=? AND id<>? AND kind IN (\'site.deploy\',\'backup.restore\',\'site.relocate\') AND status IN (\'queued\',\'running\',\'awaiting_external\',\'needs_user\')', p.siteId, job.id); if (conflict) throw new JobError('conflict', `같은 사이트의 다른 작업(${conflict.id})이 진행 중입니다`); return { from: f.name, to: t.name }; });
      await ctx.step('prepare', async () => core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: to, kind: 'site.prepare', payload: { siteId: p.siteId } })));
      const pre = await ctx.step('precopy', async () => core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: from, kind: 'site.export', payload: { siteId: p.siteId, transferId: `tr_rl1_${job.id}`, includeData: true } })));
      await ctx.step('verify_copy', async () => core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: to, kind: 'site.import', payload: { siteId: p.siteId, transferId: pre.transferId, checksum: pre.checksum, stage: 'pre' } })));
      const fin = await ctx.step('final_sync', async () => { if (dyn || p.writeFreeze) await core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: from, kind: 'site.freeze', payload: { siteId: p.siteId, freeze: true } })); const ex = await core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: from, kind: 'site.export', payload: { siteId: p.siteId, transferId: `tr_rl2_${job.id}`, includeData: true }, dedupeKey: `${job.id}:final` })); return core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: to, kind: 'site.import', payload: { siteId: p.siteId, transferId: ex.transferId, checksum: ex.checksum, stage: 'final' }, dedupeKey: `${job.id}:final-import` })); });
      const vn = await ctx.step('verify_new', async () => core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: to, kind: 'site.stage_check', payload: { siteId: p.siteId, expectedChecksum: fin.checksum } })));
      if (!vn?.ok) throw new JobError('verify_new_failed', '새 위치 확인에 실패해 전환하지 않습니다', { detail: vn });
      await ctx.step('switch', async () => { ctx.checkCancel(); const a = await core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: to, kind: 'site.activate', payload: { siteId: p.siteId, domains: p.domains || [], primaryDomain: p.primaryDomain, runtime: p.runtime || 'static' } })); await core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: from, kind: 'site.deactivate', payload: { siteId: p.siteId, retainDays: p.retainDays || 7 } })); return { switchedAt: nowIso(), activate: a }; });
      const ext = await ctx.step('external_check', () => externalCheck(d, p.externalCheckUrl, { expectMarker: p.expectMarker }));
      if (p.externalCheckUrl && !ext.ok) throw new JobError('relocate_external_check_failed', '전환 뒤 외부 접속 확인에 실패했습니다 — 되돌리기를 검토하세요', { detail: ext });
      const cl = await ctx.step('cleanup', () => ({ oldRetainedUntil: new Date(Date.now() + (p.retainDays || 7) * 86400000).toISOString(), note: '이전 위치 사본은 보존 기간 뒤 삭제 작업(site.remove)으로 정리' }));
      return { newServerId: p.toServerId, switchedAt: job.steps?.find((x) => x.id === 'switch')?.result?.switchedAt || nowIso(), oldRetainedUntil: cl.oldRetainedUntil, externalCheck: ext };
    },
    async revert(job, ctx) { const p = job.subject.payload || {}; const sw = (job.steps || []).find((x) => x.id === 'switch'); if (!sw || sw.state !== 'done') { await ctx.step('revert_unfreeze', async () => core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, p.fromServerId), kind: 'site.freeze', payload: { siteId: p.siteId, freeze: false }, dedupeKey: `${job.id}:unfreeze` }))); return; }
      await ctx.step('revert_switch', async () => { await core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, p.fromServerId), kind: 'site.activate', payload: { siteId: p.siteId, domains: p.domains || [], primaryDomain: p.primaryDomain, runtime: p.runtime || 'static', reactivate: true }, dedupeKey: `${job.id}:revert-act` })); await core.awaitAgentJob(d, ctx, core.dispatchAgentJob(d, { halJobId: job.id, agentId: agentFor(d, p.toServerId), kind: 'site.deactivate', payload: { siteId: p.siteId, retainDays: 1 }, dedupeKey: `${job.id}:revert-deact` })); }, '이전 위치로 연결 되돌리기'); } });

  // ---------------- 모듈 배포(§32)
  const unitTemplate = (name) => { const f = name ? join(ROOT, 'deploy', 'units', name) : null; return f && existsSync(f) ? readFileSync(f, 'utf8') : null; };
  /** 배포 전제: 릴리스 승인, artifact 존재, HAL 먼저(다른 모듈은 HAL 이 그 릴리스로 실행 중이어야), 부트스트랩 완료 */
  function precheckModule(module, version) {
    const rel = core.getRelease(d, version); if (!rel) throw new JobError('release_not_found', `릴리스 ${version} 이(가) 없습니다`);
    if (rel.status !== 'approved') throw new JobError('release_not_approved', `릴리스 ${version} 은(는) 승인되지 않았습니다`);
    if (ARTIFACT_MODULES.includes(module) && !core.artifactPath(d, version, module)) throw new JobError('artifact_missing', `릴리스 ${version} 의 ${module} artifact 가 없습니다`);
    const health = core.selfHealth(d); if (!health.ok) throw new JobError('hal_unhealthy', 'HAL 자체 건강 확인에 실패해 배포하지 않습니다');
    if (!core.bootstrapComplete(d)) throw new JobError('bootstrap_incomplete', '부트스트랩(HAL 설치·첫 Agent 등록)이 끝나지 않았습니다');
    if (module !== 'hal' && health.bootstrap.halRunningVersion !== version) throw new JobError('hal_first', `HAL 이 아직 ${version} 으로 실행되고 있지 않습니다(현재 ${health.bootstrap.halRunningVersion}). HAL 을 먼저 배포하세요`);
    return rel;
  }
  engine.register('module.deploy.target', { title: '모듈 배포(대상)', steps: [{ id: 'dispatch', title: 'Agent 에 설치 지시' }, { id: 'await', title: '설치·재시작' }, { id: 'verify', title: '실제 버전·건강 확인' }],
    async run(job, ctx) {
      const { module, version, serverId, action = 'install' } = job.subject; const desc = descriptor(module); const srv = core.getServer(d, serverId); if (!srv) throw new JobError('server_not_found', '서버가 없습니다');
      const art = core.artifactPath(d, version, module);
      d.store.run('INSERT INTO module_targets(module, server_id, target_version, last_job_id, updated_at) VALUES(?,?,?,?,?) ON CONFLICT(module, server_id) DO UPDATE SET target_version=excluded.target_version, last_job_id=excluded.last_job_id, updated_at=excluded.updated_at', module, serverId, version, job.id, nowIso());
      const facts = srv.facts || {}; const unitName = (facts.units && facts.units[module]) || desc.unit;
      const disp = await core.ensureAgentJob(d, ctx, job, { agentId: agentFor(d, serverId), kind: action === 'rollback' ? 'module.rollback' : 'module.install', title: 'Agent 에 설치 지시', payload: { module, version, action, artifactUrl: art ? `/agent/v1/releases/${version}/artifact?module=${module}` : null, sha256: art?.sha256, size: art?.size, descriptor: { ...desc, unit: unitName }, unitTemplate: unitTemplate(desc.unitTemplate), appDir: (facts.appDirs && facts.appDirs[module]) || (desc.baseDir || `/srv/ariweb/${module}`), toVersion: job.subject.toVersion || null } });
      const r = await ctx.step('await', () => core.awaitAgentJob(d, ctx, disp.agentJobId));
      const v = await ctx.step('verify', () => {
        const expected = action === 'rollback' ? (job.subject.toVersion || r.actualVersion) : version; const ok = r.actualVersion === expected && r.health?.ok !== false;
        d.store.run('UPDATE module_targets SET actual_version=?, health=?, active=?, last_check_at=?, last_error=?, updated_at=? WHERE module=? AND server_id=?', r.actualVersion || null, r.health?.ok === false ? 'down' : 'ok', r.active ? 1 : 0, nowIso(), ok ? null : `기대 ${expected} 실제 ${r.actualVersion}`, nowIso(), module, serverId);
        if (!ok) throw new JobError('version_mismatch', `${srv.name}: 설치 뒤 실제 버전(${r.actualVersion})·건강(${r.health?.ok})이 목표와 다릅니다`, { detail: r });
        if (module === 'hal') { d.store.run('UPDATE releases SET hal_deployed_at=? WHERE version=?', nowIso(), version); core.setBootstrap(d, { hal_running_version: expected, release_version: expected }); }
        return { actualVersion: r.actualVersion, health: r.health, previousVersion: r.previousVersion };
      });
      return v;
    } });
  /** 부모: 대상 서버마다 자식 작업 → 대상별 결과 요약. 일부 실패는 실패로 끝낸다. */
  async function fanout(job, ctx, { kind, mk, serverIds }) {
    const ids = await ctx.step('fanout', () => serverIds.map((serverId) => engine.enqueue({ kind, subject: mk(serverId), parentId: job.id, groupId: null, subjectType: 'server', subjectId: serverId, idempotencyKey: `${job.id}:${serverId}`, createdBy: job.created_by, actorType: job.actor_type, label: `${job.label} — ${core.getServer(d, serverId)?.name || serverId}` }).job.id), '대상별 작업 만들기');
    const t0 = Date.now(); while (true) { if (Date.now() - t0 > (d.config.fanoutTimeoutMs || 12 * 3600 * 1000)) throw new JobError('fanout_timeout', '대상별 작업이 제한 시간 안에 끝나지 않았습니다'); const sum = engine.childrenSummary(job.id); ctx.progress(Math.round(100 * (sum.byStatus.done || 0) / Math.max(1, sum.total))); if (['done', 'partial', 'failed'].includes(sum.overall) && !sum.children.some((c) => ['queued', 'running', 'reverting'].includes(c.status))) { if (sum.overall !== 'done') throw new JobError(sum.overall === 'partial' ? 'partial_failure' : 'all_failed', `${sum.total}대 중 ${sum.children.filter((c) => c.status !== 'done').length}대 실패·대기 — 대상별 결과를 확인하세요`, { detail: sum }); return sum; } ctx.checkCancel(); await new Promise((r) => setTimeout(r, d.config.agentPollMs || 1000)); }
  }
  engine.register('module.deploy', { title: '모듈 배포', steps: [{ id: 'precheck', title: '릴리스·HAL·순서 확인' }, { id: 'fanout', title: '대상별 작업 만들기' }, { id: 'collect', title: '대상별 결과 확인' }],
    async run(job, ctx) {
      const { module, version, serverIds } = job.subject; await ctx.step('precheck', () => { precheckModule(module, version); if (!serverIds?.length) throw new JobError('no_targets', '대상 서버가 없습니다'); return { order: DEPLOY_ORDER.indexOf(module) }; });
      d.store.run('INSERT INTO modules(name, target_version, updated_at) VALUES(?,?,?) ON CONFLICT(name) DO UPDATE SET target_version=excluded.target_version, updated_at=excluded.updated_at', module, version, nowIso());
      const sum = await ctx.step('collect', () => fanout(job, ctx, { kind: 'module.deploy.target', serverIds, mk: (serverId) => ({ module, version, serverId, action: 'install' }) }));
      audit(d.store, { actorType: job.actor_type, actorId: job.created_by, action: 'deploy.module', subjectType: 'module', subjectId: module, detail: { version, serverIds, overall: sum.overall } });
      return sum;
    } });
  engine.register('module.rollback', { title: '모듈 되돌리기', steps: [{ id: 'fanout', title: '대상별 작업 만들기' }, { id: 'collect', title: '대상별 결과 확인' }],
    async run(job, ctx) { const { module, toVersion, serverIds } = job.subject; const sum = await ctx.step('collect', () => fanout(job, ctx, { kind: 'module.deploy.target', serverIds, mk: (serverId) => ({ module, version: toVersion || core.getRelease(d, d.store.get('SELECT target_version FROM module_targets WHERE module=? AND server_id=?', module, serverId)?.target_version)?.version, serverId, action: 'rollback', toVersion }) })); audit(d.store, { actorType: job.actor_type, actorId: job.created_by, action: 'deploy.module_rollback', subjectType: 'module', subjectId: module, detail: { toVersion, serverIds, overall: sum.overall } }); return sum; } });
  single(engine, d, 'module.health.target', { title: '모듈 건강 확인(대상)', agentKind: 'module.health', target: (s) => s.serverId, build: (s) => ({ module: s.module, descriptor: descriptor(s.module), unit: (core.getServer(d, s.serverId)?.facts?.units || {})[s.module] || descriptor(s.module).unit }),
    after(job, ctx, r) { d.store.run('UPDATE module_targets SET actual_version=COALESCE(?, actual_version), health=?, active=?, last_check_at=?, last_error=?, updated_at=? WHERE module=? AND server_id=?', r.actualVersion || null, r.health?.ok ? 'ok' : 'down', r.active ? 1 : 0, nowIso(), r.health?.ok ? null : (r.health?.error || '건강 확인 실패'), nowIso(), job.subject.module, job.subject.serverId); return r; } });
  engine.register('module.health', { title: '모듈 건강 확인', steps: [{ id: 'fanout', title: '대상별 작업 만들기' }, { id: 'collect', title: '대상별 결과 확인' }], async run(job, ctx) { const { module, serverIds } = job.subject; return ctx.step('collect', () => fanout(job, ctx, { kind: 'module.health.target', serverIds: serverIds || d.store.all('SELECT server_id FROM module_targets WHERE module=?', module).map((r) => r.server_id), mk: (serverId) => ({ module, serverId }) })); } });
  /** 다시 맞추기(§29·§34): 목표 버전과 실제 버전이 다른 대상만 다시 설치 */
  engine.register('module.reconcile', { title: '목표·실제 상태 맞추기', steps: [{ id: 'diff', title: '차이 확인' }, { id: 'fanout', title: '대상별 작업 만들기' }, { id: 'collect', title: '대상별 결과 확인' }],
    async run(job, ctx) { const diff = await ctx.step('diff', () => d.store.all('SELECT module, server_id, target_version, actual_version FROM module_targets WHERE target_version IS NOT NULL AND (actual_version IS NULL OR actual_version<>target_version)' + (job.subject.module ? ' AND module=?' : ''), ...(job.subject.module ? [job.subject.module] : []))); if (!diff.length) return { drift: [], overall: 'done' }; const sum = await ctx.step('collect', () => fanout(job, ctx, { kind: 'module.deploy.target', serverIds: diff.map((x) => x.server_id), mk: (serverId) => { const x = diff.find((y) => y.server_id === serverId); return { module: x.module, version: x.target_version, serverId, action: 'install' }; } })); return { drift: diff, ...sum }; } });

  // ---------------- 포털 이벤트: 포털이 만든 작업(subject._portal)의 상태 변화를 outbox 에 넣는다
  engine.onChange((job) => { if (job.subject && job.subject._portal) core.queuePortalEvent(d, { halJobId: job.id, kind: job.kind, status: job.status, steps: job.steps, progress: job.progress, result: job.result, error: job.error, errorCode: job.error_code, userMessage: job.user_message, updatedAt: job.updated_at, subjectType: job.subject_type, subjectId: job.subject_id }); });
}