Add nightly realm reports and observed bot activity explanations

app/api/realm-insights/route.ts

@@ -0,0 +1,51 @@
+import {env} from 'cloudflare:workers';
+import {z} from 'zod';
+import {D1Store} from '@/core/store';
+import type {RuntimeConfig} from '@/core/types';
+import {observeInsights,activityReason,reportWindow,summarize,DAY,type Cursor,type Evidence} from '@/core/realm-insights';
+const recovery=z.object({id:z.number().int().positive(),guid:z.number().int().positive(),name:z.string().max(64),at:z.number().finite(),detail:z.string().max(255)});
+async function ensure(){await env.DB!.batch([
+ env.DB!.prepare('CREATE TABLE IF NOT EXISTS insights_cursor (id INTEGER PRIMARY KEY,payload TEXT)'),
+ env.DB!.prepare('CREATE TABLE IF NOT EXISTS insights_evidence (id TEXT PRIMARY KEY,at INTEGER,payload TEXT)'),
+ env.DB!.prepare('CREATE INDEX IF NOT EXISTS insights_evidence_at ON insights_evidence(at)'),
+ env.DB!.prepare('CREATE TABLE IF NOT EXISTS insights_samples (at INTEGER PRIMARY KEY)'),
+ env.DB!.prepare('CREATE TABLE IF NOT EXISTS insights_reports (start INTEGER PRIMARY KEY,payload TEXT)'),
+ env.DB!.prepare('CREATE TABLE IF NOT EXISTS insights_recoveries (id INTEGER PRIMARY KEY,payload TEXT)'),
+ env.DB!.prepare('CREATE TABLE IF NOT EXISTS insights_clock (id INTEGER PRIMARY KEY,at INTEGER)'),
+ env.DB!.prepare('INSERT OR IGNORE INTO insights_clock VALUES (1,0)')]);}
+async function report(start:number,end:number,startedAt:number){
+ const rows=await env.DB!.prepare('SELECT payload FROM insights_evidence WHERE at>=? AND at<? ORDER BY at').bind(start,end).all<{payload:string}>();
+ const samples=await env.DB!.prepare('SELECT COUNT(*) AS n FROM insights_samples WHERE at>=? AND at<?').bind(start,end).first<{n:number}>();
+ return summarize(rows.results.map(r=>JSON.parse(r.payload)),start,end,startedAt,samples?.n??0);
+}
+export async function GET(){
+ await ensure();const now=Math.floor(Date.now()/1000),s=await new D1Store().load();
+ const row=await env.DB!.prepare('SELECT payload FROM insights_cursor WHERE id=1').first<{payload:string}>();const cursor:Cursor|undefined=row?JSON.parse(row.payload):undefined;
+ const history=await env.DB!.prepare('SELECT payload FROM insights_reports ORDER BY start DESC LIMIT 30').all<{payload:string}>();
+ const recoveries=await env.DB!.prepare('SELECT payload FROM insights_recoveries ORDER BY id DESC LIMIT 100').all<{payload:string}>();const attempts=recoveries.results.map(r=>JSON.parse(r.payload));
+ return Response.json({collectedAt:cursor?.at??null,startedAt:cursor?.startedAt??null,nextReportAt:reportWindow(now)+DAY,current:await report(reportWindow(now),now+1,cursor?.startedAt??now),reports:history.results.map(r=>JSON.parse(r.payload)),activity:s.bots.filter(b=>b.autonomy?.managed&&b.status==='online').map(b=>({name:b.name,zone:b.zone,action:b.lastDecision?.action??'unknown',observedAt:b.world?.at??null,stationarySince:b.world?.stationarySince??null,lastProgressAt:cursor?.bots[b.id]?.lastProgressAt??null,...activityReason(b,now),recoveries:attempts.filter(r=>r.guid===b.autonomy?.characterGuid).slice(0,3).map(({at,detail})=>({at,detail}))}))},{headers:{'Cache-Control':'no-store'}});
+}
+export async function POST(request:Request){
+ const token=(env as unknown as RuntimeConfig).AUTONOMY_TOKEN;if(!token||request.headers.get('authorization')!==`Bearer ${token}`)return Response.json({error:'Unauthorized'},{status:401});
+ try{
+  const raw=await request.text();if(raw.length>50000)throw new Error('Payload too large');const body=z.object({at:z.number().finite(),recoveries:z.array(recovery).max(100)}).strict().parse(JSON.parse(raw));
+  const now=Math.floor(Date.now()/1000);if(Math.abs(now-body.at)>180)throw new Error('Stale sample');await ensure();
+  const gate=await env.DB!.prepare('UPDATE insights_clock SET at=? WHERE id=1 AND at<?').bind(now,now-40).run();if(!gate.meta.changes)return Response.json({ok:true,skipped:true});
+  const previous=await env.DB!.prepare('SELECT payload FROM insights_cursor WHERE id=1').first<{payload:string}>();const old:Cursor|undefined=previous?JSON.parse(previous.payload):undefined;
+  const {cursor,events}=observeInsights(await new D1Store().load(),old,now);
+  const statements=events.map(e=>env.DB!.prepare('INSERT OR IGNORE INTO insights_evidence VALUES (?,?,?)').bind(e.id,e.at,JSON.stringify(e)));
+  for(const r of body.recoveries){
+   if(r.at>now+60)continue;
+   statements.push(env.DB!.prepare('INSERT OR IGNORE INTO insights_recoveries VALUES (?,?)').bind(r.id,JSON.stringify(r)));
+   if(r.at>=cursor.startedAt){const e:Evidence={id:'recovery:'+r.id,at:r.at,kind:'recoveries',name:r.name,amount:1,detail:'Native recovery attempted'};statements.push(env.DB!.prepare('INSERT OR IGNORE INTO insights_evidence VALUES (?,?,?)').bind(e.id,e.at,JSON.stringify(e)));}
+  }
+  statements.push(env.DB!.prepare('INSERT INTO insights_cursor VALUES (1,?) ON CONFLICT(id) DO UPDATE SET payload=excluded.payload').bind(JSON.stringify(cursor)),env.DB!.prepare('INSERT OR IGNORE INTO insights_samples VALUES (?)').bind(now));
+  await env.DB!.batch(statements);
+  for(let start=Math.max(reportWindow(cursor.startedAt),reportWindow(now)-30*DAY);start<reportWindow(now);start+=DAY){
+   if(await env.DB!.prepare('SELECT start FROM insights_reports WHERE start=?').bind(start).first())continue;
+   await env.DB!.prepare('INSERT OR IGNORE INTO insights_reports VALUES (?,?)').bind(start,JSON.stringify(await report(start,start+DAY,cursor.startedAt))).run();
+  }
+  await env.DB!.batch([env.DB!.prepare('DELETE FROM insights_evidence WHERE at<?').bind(now-31*DAY),env.DB!.prepare('DELETE FROM insights_samples WHERE at<?').bind(now-31*DAY),env.DB!.prepare('DELETE FROM insights_reports WHERE start<?').bind(reportWindow(now)-30*DAY),env.DB!.prepare('DELETE FROM insights_recoveries WHERE id NOT IN (SELECT id FROM insights_recoveries ORDER BY id DESC LIMIT 100)')]);
+  return Response.json({ok:true});
+ }catch(e){return Response.json({error:e instanceof Error?e.message:'Insight collection failed'},{status:400});}
+}