exportfunctionbootstrapState(d){letr=d.store.get('SELECT * FROM bootstrap_state WHERE id=1');if(!r){d.store.run('INSERT INTO bootstrap_state(id, updated_at) VALUES(1, ?)',nowIso());r=d.store.get('SELECT * FROM bootstrap_state WHERE id=1');}returnr;}
/** [실제 환경 결함 수정] 행(id=1)이 아직 없으면(HAL 기동 전 CLI bootstrap-release) UPDATE 가 아무 행도 바꾸지 않아 릴리스 정보가 비었다 — 먼저 행을 만든다. */
exportfunctionsetBootstrap(d,patch){bootstrapState(d);constsets=Object.keys(patch).map((k)=>`${k}=?`);d.store.run(`UPDATE bootstrap_state SET ${sets.join(', ')}, updated_at=? WHERE id=1`,...Object.values(patch),nowIso());returnbootstrapState(d);}
exportfunctionrenewLease(d,instanceId){d.store.run('UPDATE bootstrap_state SET live_lease_at=? WHERE id=1 AND live_instance_id=?',nowIso(),instanceId);}
exportfunctiongetServer(d,id){consts=d.store.get('SELECT * FROM servers WHERE id=?',id);returns?{...s,metrics:parseJson(s.metrics_json,null),capacity:parseJson(s.capacity_json,null),facts:parseJson(s.facts_json,{})}:null;}
exportfunctionlistServers(d,{role=null}={}){returnd.store.all(`SELECT * FROM servers ${role?'WHERE role=?':''} ORDER BY created_at`,...(role?[role]:[])).map((s)=>publicServer(d,s));}
exportfunctionapproveServer(d,id,by){consts=getServer(d,id);if(!s)thrownewJobError('server_not_found','서버가 없습니다');if(!s.agent_id)thrownewJobError('agent_missing','Agent 가 아직 연결되지 않았습니다');updateServer(d,id,{status:'ready',accepts_new:1},by);d.store.run('UPDATE servers SET approved_at=?, approved_by=? WHERE id=?',nowIso(),by,id);maybeCompleteBootstrap(d,s.agent_id);returngetServer(d,id);}
if(serverId)d.store.run('UPDATE servers SET agent_id=?, last_seen_at=?, agent_version=?, facts_json=?, updated_at=? WHERE id=?',agentId,now,version,toJson(redact(facts)),now,serverId);
exportfunctiongetAgent(d,id){returnd.store.get('SELECT * FROM agents WHERE id=?',id)||null;}
exportfunctionrevokeAgent(d,id,by=null){d.store.run('UPDATE agents SET status=\'revoked\', revoked_at=? WHERE id=?',nowIso(),id);d.vault.remove(`service:agent:${id}`);d.store.run('UPDATE agent_jobs SET status=\'cancelled\', ended_at=? WHERE agent_id=? AND status IN (\'queued\',\'running\')',nowIso(),id);audit(d.store,{actorType:by?'operator':'system',actorId:by,action:'agent.revoke',subjectType:'agent',subjectId:id,immutable:true});}
constlast=d.store.get('SELECT * FROM agent_jobs WHERE agent_id=? AND kind=\'site.deploy\' ORDER BY created_at DESC LIMIT 1',id);
constlastCheck=d.store.get('SELECT * FROM agent_jobs WHERE agent_id=? AND kind=\'site.check\' AND status=\'done\' ORDER BY created_at DESC LIMIT 1',id);
if(!scanId||scanId.length>100||!Number.isInteger(offset)||offset<0||!Number.isInteger(itemTotal)||itemTotal<0||itemTotal>10000||!Array.isArray(sites)||sites.length>1000)thrownewJobError('bad_inventory_batch','인벤토리 배치 형식이 올바르지 않습니다');
if(!site||typeofsite!=='object'||Array.isArray(site)||!/^[a-f0-9]{32}$/.test(String(site.id||'')))thrownewJobError('bad_inventory_batch','사이트 항목 형식이 올바르지 않습니다');
if(meta.total!==itemTotal||meta.truncated)thrownewJobError('inventory_incomplete','완전한 Webmaker 인벤토리만 반영할 수 있습니다');
returnd.store.tx(()=>{
d.store.run('DELETE FROM agent_inventory_staging WHERE updated_at < ?',newDate(Date.now()-24*3600*1000).toISOString());
letrow=d.store.get('SELECT * FROM agent_inventory_staging WHERE agent_id=? AND kind=?',agent.id,'webmaker');
if(offset===0){d.store.run('INSERT INTO agent_inventory_staging(agent_id, kind, scan_id, total, next_offset, meta_json, items_json, updated_at) VALUES(?,?,?,?,0,?,?,?) ON CONFLICT(agent_id, kind) DO UPDATE SET scan_id=excluded.scan_id,total=excluded.total,next_offset=0,meta_json=excluded.meta_json,items_json=\'[]\',updated_at=excluded.updated_at',agent.id,'webmaker',scanId,itemTotal,toJson(meta),'[]',nowIso());row=d.store.get('SELECT * FROM agent_inventory_staging WHERE agent_id=? AND kind=?',agent.id,'webmaker');}
if(!row||row.scan_id!==scanId||row.total!==itemTotal||row.next_offset!==offset)thrownewJobError('inventory_batch_order','인벤토리 배치 순서가 맞지 않습니다 — 처음부터 다시 보내세요');
constitems=parseJson(row.items_json,[]);items.push(...cleanSites);if(items.length>itemTotal||offset+cleanSites.length!==items.length)thrownewJobError('bad_inventory_batch','인벤토리 항목 수가 선언값과 다릅니다');constitemsJson=toJson(items);if(Buffer.byteLength(itemsJson)>16*1024*1024)thrownewJobError('inventory_too_large','인벤토리 전체 크기가 안전 한도를 넘습니다');
constcomplete=items.length===itemTotal;d.store.run('UPDATE agent_inventory_staging SET next_offset=?, items_json=?, updated_at=? WHERE agent_id=? AND kind=?',items.length,itemsJson,nowIso(),agent.id,'webmaker');
constinventory={...parseJson(row.meta_json,{}),returned:items.length,sites:items};constaFacts=parseJson(d.store.get('SELECT facts_json FROM agents WHERE id=?',agent.id)?.facts_json,{})||{};aFacts.webmakerInventory=inventory;
constsFacts=parseJson(d.store.get('SELECT facts_json FROM servers WHERE id=?',agent.server_id)?.facts_json,{})||{};sFacts.webmakerInventory=inventory;
d.store.run('UPDATE agents SET facts_json=? WHERE id=?',toJson(aFacts),agent.id);d.store.run('UPDATE servers SET facts_json=?, updated_at=? WHERE id=?',toJson(sFacts),nowIso(),agent.server_id);d.store.run('DELETE FROM agent_inventory_staging WHERE agent_id=? AND kind=?',agent.id,'webmaker');
constagentFacts=mergeFacts(agent.facts_json);d.store.run('UPDATE agents SET last_seen_at=?, version=?, facts_json=COALESCE(?, facts_json) WHERE id=?',now,body.version||agent.version,agentFacts,agent.id);
constserverRow=d.store.get('SELECT facts_json FROM servers WHERE id=?',agent.server_id);constserverFacts=mergeFacts(serverRow?.facts_json);d.store.run('UPDATE servers SET last_seen_at=?, clock_skew_ms=?, agent_version=?, metrics_json=?, capacity_json=?, facts_json=COALESCE(?, facts_json), status=CASE WHEN status=\'offline\' THEN \'ready\' ELSE status END, updated_at=? WHERE id=?',now,skew,body.version||null,toJson(m),toJson(capacity),serverFacts,now,agent.server_id);
if(skew!=null&&Math.abs(skew)>60000)notify(d.store,{category:'ops',severity:'warning',incidentKey:`server:${agent.server_id}:clock`,title:`서버 시계 차이 ${Math.round(skew/1000)}초`,link:`/ops/servers/${agent.server_id}`});
if(m.disk&&m.disk.total&&m.disk.used/m.disk.total>=0.9)notify(d.store,{category:'ops',severity:m.disk.used/m.disk.total>=0.95?'critical':'warning',incidentKey:`server:${agent.server_id}:disk`,title:`서버 디스크 ${Math.round(m.disk.used/m.disk.total*100)}% 사용`,link:`/ops/servers/${agent.server_id}`});
return{ok:true,serverTime:now,pendingJobs:d.store.get('SELECT COUNT(*) AS n FROM agent_jobs WHERE agent_id=? AND status=\'queued\'',agent.id).n};
}
/** 오래 heartbeat 가 없는 서버를 offline 으로 표시(주기 작업). 변경된 서버 목록 반환 */
constrows=d.store.all('SELECT id, name FROM servers WHERE status IN (\'ready\',\'maintenance\') AND agent_id IS NOT NULL AND (last_seen_at IS NULL OR last_seen_at<?)',cut);
for(constrofrows){d.store.run('UPDATE servers SET status=\'offline\', updated_at=? WHERE id=?',nowIso(),r.id);notify(d.store,{category:'incident',severity:'critical',mandatory:false,incidentKey:`server:${r.id}:offline`,title:`서버 ${r.name} 응답 없음(Node Agent heartbeat 중단)`,link:`/ops/servers/${r.id}`});}
returnrows;
}
exportfunctionpurgeMetrics(d){constcut=newDate(Date.now()-d.config.metricsRetentionDays*86400000).toISOString();returnd.store.run('DELETE FROM metrics_samples WHERE at<?',cut).changes;}
exportfunctionmetrics(d,serverId,{since=null,until=null,limit=5000}={}){returnd.store.all('SELECT * FROM metrics_samples WHERE server_id=? AND at>=? AND at<=? ORDER BY at DESC LIMIT ?',serverId,since||newDate(Date.now()-90*86400000).toISOString(),until||nowIso(),Math.min(limit,50000));}
// ------------------------------------------------------------------ Agent 작업
/**
* 재진입 가능한 배정: 'dispatch' 단계에 저장된 Agent 작업이 실패·취소로 끝났으면(재시도) 새 Agent 작업을 만든다.
* 호출자는 ctx.step('dispatch', …) 대신 이 함수를 쓴다.
*/
exportasyncfunctionensureAgentJob(d,ctx,job,{stepId='dispatch',agentId,kind,payload,title='대상 Agent 에 작업 배정'}){
constagent=getAgent(d,agentId);if(!agent||agent.status!=='active')thrownewJobError('agent_unavailable','Agent 가 없거나 해지되었습니다');
if(agent.kind==='external'&&!EXTERNAL_KINDS.has(kind))thrownewJobError('kind_not_allowed_external',`외부 Agent 는 ${kind} 작업을 받을 수 없습니다`);
if(agent.kind==='external'&&payload?.siteId&&agent.site_id&&payload.siteId!==agent.site_id)thrownewJobError('site_mismatch','이 외부 Agent 는 다른 사이트에 귀속되어 있습니다');
constex=d.store.get('SELECT * FROM agent_jobs WHERE hal_job_id=? AND kind=? AND agent_id=? AND json_extract(payload_json, \'$._dedupe\')=?',halJobId,kind,agentId,key);
exportfunctiongetAgentJob(d,id){constr=d.store.get('SELECT * FROM agent_jobs WHERE id=?',id);returnr?{...r,payload:parseJson(r.payload_json,{}),steps:parseJson(r.steps_json,[]),result:parseJson(r.result_json,null),log:parseJson(r.log_json,[])}:null;}
/** Agent 가 다음 작업을 가져간다(long-poll 은 라우트가 반복 호출). queued 중 가장 오래된 것 하나 → running */
exportfunctionnextAgentJob(d,agentId){
constr=d.store.get('SELECT id FROM agent_jobs WHERE agent_id=? AND status=\'queued\' ORDER BY created_at LIMIT 1',agentId);if(!r)returnnull;
constgot=d.store.run('UPDATE agent_jobs SET status=\'running\', assigned_at=?, started_at=?, heartbeat_at=? WHERE id=? AND status=\'queued\'',nowIso(),nowIso(),nowIso(),r.id).changes;
exportfunctionunfinishedAgentJobs(d,agentId){returnd.store.all('SELECT id, kind, payload_json, steps_json FROM agent_jobs WHERE agent_id=? AND status=\'running\'',agentId).map((r)=>({id:r.id,kind:r.kind,payload:(()=>{constp=parseJson(r.payload_json,{});deletep._dedupe;returnp;})(),steps:parseJson(r.steps_json,[])}));}
exportfunctionagentJobProgress(d,id,agentId,{progress,steps,log}){constr=d.store.run('UPDATE agent_jobs SET progress=COALESCE(?, progress), steps_json=COALESCE(?, steps_json), log_json=COALESCE(?, log_json), heartbeat_at=? WHERE id=? AND agent_id=? AND status=\'running\'',progress??null,steps?toJson(redact(steps)):null,log?toJson(redact(log)):null,nowIso(),id,agentId).changes;return!!r;}
if(!['done','failed','needs_user'].includes(status))thrownewJobError('bad_status','Agent 결과 상태는 done/failed/needs_user 중 하나입니다');
constch=d.store.run('UPDATE agent_jobs SET status=?, result_json=?, error=?, error_code=?, user_message=?, steps_json=COALESCE(?, steps_json), log_json=COALESCE(?, log_json), progress=CASE WHEN ?=\'done\' THEN 100 ELSE progress END, ended_at=? WHERE id=? AND agent_id=? AND status=\'running\'',
if(silent>10*60*1000)ctx.needsUser(`대상 서버(${srv?.name||agent?.label||j.agent_id})의 Agent 가 10분 넘게 응답하지 않습니다. 서버 상태를 확인한 뒤 이어서 진행하세요`,[{id:'resume',label:'Agent 복구 후 이어서 진행'}]);
if(Date.now()-t0>timeoutMs)thrownewJobError('agent_timeout','Agent 작업이 제한 시간을 넘겼습니다');
exportfunctionapproveRelease(d,version,by){constr=getRelease(d,version);if(!r)thrownewJobError('release_not_found','릴리스가 없습니다');constchecks=r.checks||{};if(checks.testsPassed===false)thrownewJobError('release_checks_failed','검사에 실패한 릴리스는 승인할 수 없습니다');d.store.run('UPDATE releases SET status=\'approved\', approved_by=?, approved_at=? WHERE version=?',by,nowIso(),version);audit(d.store,{actorType:'operator',actorId:by,action:'release.approve',subjectType:'release',subjectId:version,immutable:true});returngetRelease(d,version);}
exportfunctiongetRelease(d,version){constr=d.store.get('SELECT * FROM releases WHERE version=?',version);returnr?{...r,artifacts:parseJson(r.artifacts_json,{}),checks:parseJson(r.checks_json,null)}:null;}
exportfunctionlistReleases(d){returnd.store.all('SELECT * FROM releases ORDER BY registered_at DESC').map((r)=>({...r,artifacts:parseJson(r.artifacts_json,{}),checks:parseJson(r.checks_json,null)}));}
exportfunctioncreateTransfer(d,{id=newId('tr_'),agentId,sha256:sum,size,purpose='',meta=null}){constpath=join(transferDir(d),id+'.bin');d.store.run('INSERT OR REPLACE INTO transfers(id, created_by_agent, path, size_bytes, sha256, created_at, expires_at, purpose, meta_json) VALUES(?,?,?,?,?,?,?,?,?)',id,agentId,path,size,sum,nowIso(),newDate(Date.now()+d.config.transferTtlMs).toISOString(),purpose,toJson(meta));return{id,path};}
exportfunctiongetTransfer(d,id){constt=d.store.get('SELECT * FROM transfers WHERE id=? AND deleted_at IS NULL',id);returnt&&existsSync(t.path)?t:null;}
// ------------------------------------------------------------------ 포털 이벤트 outbox(§6)
exportfunctionqueuePortalEvent(d,payload){d.store.run('INSERT INTO portal_events_outbox(id, payload_json, created_at, next_at) VALUES(?,?,?,?)',newId('ev_'),toJson(redact(payload)),nowIso(),nowIso());}
constrows=d.store.all('SELECT * FROM portal_events_outbox WHERE delivered_at IS NULL AND next_at<=? ORDER BY created_at LIMIT 50',nowIso());letn=0;
for(constrofrows){
try{constres=awaitserviceFetch(d.config.portalUrl,'hal',secret,'POST','/internal/hal-events',parseJson(r.payload_json,{}),{timeoutMs:8000,fetchImpl});if(!res.ok)thrownewError(`portal ${res.status}${res.json?.error||''}`);d.store.run('UPDATE portal_events_outbox SET delivered_at=? WHERE id=?',nowIso(),r.id);n++;}
catch(e){constattempts=r.attempts+1;d.store.run('UPDATE portal_events_outbox SET attempts=?, next_at=?, last_error=? WHERE id=?',attempts,newDate(Date.now()+Math.min(600000,5000*2**attempts)).toISOString(),safeError(e),r.id);}
}
d.store.run('DELETE FROM portal_events_outbox WHERE delivered_at IS NOT NULL AND delivered_at<?',newDate(Date.now()-7*86400000).toISOString());
returnn;
}
exportfunctionensureOpsAccountExists(d){returnd.store.get('SELECT COUNT(*) AS n FROM ops_accounts').n>0;}
