export type AleCommandStatus='pending'|'dispatched'|'completed'|'failed'; export type AleQueuedCommand={ id:string; createdAt:string; status:AleCommandStatus; source:string; result?:string; error?:string; dispatchedAt?:string; completedAt?:string; }; async function ensure(db:D1Database){ await db.prepare(`CREATE TABLE IF NOT EXISTS ale_commands ( id TEXT PRIMARY KEY NOT NULL, created_at TEXT NOT NULL, status TEXT NOT NULL, source TEXT NOT NULL, result TEXT, error TEXT, dispatched_at TEXT, completed_at TEXT )`).run(); await db.prepare(`CREATE INDEX IF NOT EXISTS idx_ale_commands_status_created ON ale_commands(status,created_at)`).run(); await db.prepare(`CREATE TABLE IF NOT EXISTS ale_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, at TEXT NOT NULL, kind TEXT NOT NULL, payload TEXT NOT NULL )`).run(); await db.prepare(`CREATE INDEX IF NOT EXISTS idx_ale_events_at ON ale_events(at)`).run(); } export async function queueAleCommand(db:D1Database,source:string){ await ensure(db); const id=crypto.randomUUID(),createdAt=new Date().toISOString(); await db.prepare('INSERT INTO ale_commands (id,created_at,status,source) VALUES (?,?,?,?)').bind(id,createdAt,'pending',source).run(); return {id,createdAt,status:'pending' as const}; } export async function pollAleCommand(db:D1Database):Promise{ await ensure(db); // Never replay ancient control actions after a transport outage. Commands // that sat pending for more than 10 minutes are stale world intentions. const expiredBefore=new Date(Date.now()-10*60_000).toISOString(); const expiredAt=new Date().toISOString(); await db.prepare("UPDATE ale_commands SET status='failed',error='Expired before delivery',completed_at=? WHERE status='pending' AND created_at(); if(!row)return null; const dispatchedAt=new Date().toISOString(); const updated=await db.prepare('UPDATE ale_commands SET status=?,dispatched_at=? WHERE id=? AND status=?').bind('dispatched',dispatchedAt,row.id,'pending').run(); if(updated.meta.changes!==1)return null; return {id:row.id,createdAt:row.created_at,status:'dispatched',source:row.source,dispatchedAt}; } export async function completeAleCommand(db:D1Database,id:string,ok:boolean,result:string){ await ensure(db); const completedAt=new Date().toISOString(); const clipped=result.slice(0,16000); const status=ok?'completed':'failed'; const write=ok ? db.prepare('UPDATE ale_commands SET status=?,result=?,error=NULL,completed_at=? WHERE id=?').bind(status,clipped,completedAt,id) : db.prepare('UPDATE ale_commands SET status=?,error=?,completed_at=? WHERE id=?').bind(status,clipped,completedAt,id); await write.run(); } export async function recordAleEvent(db:D1Database,kind:string,payload:unknown){ await ensure(db); const at=new Date().toISOString(); const body=JSON.stringify(payload).slice(0,32000); await db.prepare('INSERT INTO ale_events (at,kind,payload) VALUES (?,?,?)').bind(at,kind.slice(0,100),body).run(); await db.prepare('DELETE FROM ale_events WHERE id NOT IN (SELECT id FROM ale_events ORDER BY id DESC LIMIT 2000)').run(); return {at}; } export async function recentAleEvents(db:D1Database,limit=100){ await ensure(db); const rows=await db.prepare('SELECT id,at,kind,payload FROM ale_events ORDER BY id DESC LIMIT ?').bind(Math.max(1,Math.min(limit,500))).all(); return rows.results.map(r=>({id:r.id,at:r.at,kind:r.kind,payload:JSON.parse(r.payload)})); } export async function recentAleCommands(db:D1Database,limit=50){ await ensure(db); const rows=await db.prepare('SELECT id,created_at,status,result,error,dispatched_at,completed_at FROM ale_commands ORDER BY created_at DESC LIMIT ?').bind(Math.max(1,Math.min(limit,200))).all(); return rows.results.map(r=>({id:r.id,createdAt:r.created_at,status:r.status,result:r.result,error:r.error,dispatchedAt:r.dispatched_at,completedAt:r.completed_at})); }