AzerothAI/core/ale-store.ts

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<AleQueuedCommand|null>{
 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<?")
  .bind(expiredAt,expiredBefore).run();

 // If the host relay dies after leasing a command but before placing it in the
 // durable MySQL inbox, make that command eligible again. Normal commands are
 // completed within seconds; two minutes is intentionally conservative.
 const staleBefore=new Date(Date.now()-120_000).toISOString();
 await db.prepare("UPDATE ale_commands SET status='pending',dispatched_at=NULL WHERE status='dispatched' AND completed_at IS NULL AND dispatched_at IS NOT NULL AND dispatched_at<?")
  .bind(staleBefore).run();
 const row=await db.prepare('SELECT id,created_at,status,source FROM ale_commands WHERE status=? ORDER BY created_at ASC LIMIT 1').bind('pending').first<any>();
 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<any>();
 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<any>();
 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}));
}