// Atomic events, durable cursors, feed copies, world facets, and crash recovery.
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
import { Harness,createRegistry,defineDoc,ROOT_CONVERSATION_ID as ROOT } from '@earendil-works/pi-durable';
import { createModels } from '@earendil-works/pi-ai/models';
import { openStorage } from './sqlite';
import { defineEvent,eventLog,EventFeed,EVENT_FEED_SIZE } from './events';
import { loadExtensions,fs,git } from './extensions';
const assert=(value:any,message:string)=>{if(!value)throw new Error(message);};
const equal=(a:any,b:any)=>assert(JSON.stringify(a)===JSON.stringify(b),`Expected ${JSON.stringify(b)}, got ${JSON.stringify(a)}`);
const until=async(test:()=>any)=>{const deadline=Date.now()+15000;while(!await test()){if(Date.now()>deadline)throw new Error('Event deadline');await new Promise(r=>setTimeout(r,0));}};
const doc=(kind:string)=>defineDoc({kind,scope:'conversation',version:1,history:'latest',fork:'initial',checkpointWhen:()=>false,initial:()=>({shas:[] as string[]})});
const Shipped=doc('test.shipped'),Seen=doc('test.seen');
const Landed=defineEvent<{sha:string}>('test.landed'),Branched=defineEvent<{id:number}>('test.branched');
const models=createModels();
const open=async()=>{const storage=await openStorage(globalThis.storagePath);const registry=createRegistry();const harness=await Harness.open(storage,{models,registry},context);await harness.root(context);return {harness,storage,registry};};
let {harness,storage,registry}=await open();
let service=eventLog(()=>harness);
const read=async(token:any)=>(await harness.snapshot(token,ROOT,context))?.shas??[];
const land=async(sha:string,fail=false)=>harness.commit(async tx=>{(await tx.doc(Shipped,ROOT)).shas.push(sha);await service.emit(tx,Landed,{sha});if(fail)throw new Error('rolled back');},context);
if(globalThis.eventsMode==='emit') {
  await land('crash');await fs.write('/emitted','ready');setInterval(()=>{},1000);await new Promise(()=>{});
} else if(globalThis.eventsMode==='consume') {
  const reactor=service.reactor('probe/counter',[Landed],async(tx,event)=>{(await tx.doc(Seen,ROOT)).shas.push(event.data.sha);});
  await reactor.start();await until(async()=>(await service.eventLog.reactors())[0].lag===0);await reactor.stop();equal(await read(Seen),['crash']);await harness.close(context);
} else if(globalThis.eventsMode==='world') {
  globalThis.eventHosts={};
  const repo='/src/sausage';
  await fs.mkdir(`${repo}/extensions`);
  for(const name of ['emitter','receiver'])await fs.mkdir(`${repo}/extensions/${name}`);
  await fs.write(`${repo}/extensions/emitter/server.ts`,"export default host=>{globalThis.eventHosts.emitter=host;return {name:'emitter'};}");
  const receiver=(version:number)=>`import {defineDoc,ROOT_CONVERSATION_ID as ROOT} from '@earendil-works/pi-durable';
import {defineEvent} from '@earendil-works/sausage-events';
const Seen=defineDoc({kind:'test.seen',scope:'conversation',version:1,history:'latest',fork:'initial',checkpointWhen:()=>false,initial:()=>({shas:[]})});
export default host=>{globalThis.eventHosts.receiver=host;globalThis.receiverVersion=${version};host.react('counter',[defineEvent('test.landed')],async(tx,event)=>{(await tx.doc(Seen,ROOT)).shas.push(event.data.sha);});return {name:'receiver'};};`;
  await fs.write(`${repo}/extensions/receiver/server.ts`,receiver(1));await git(repo,'add','.');await git(repo,'commit','--no-gpg-sign','-m','events');
  const loader=await loadExtensions(harness,registry,storage,models,context);
  try {
    await harness.commit(tx=>globalThis.eventHosts.emitter.emit(tx,Landed,{sha:'world-a'}),context);
    await until(async()=>(await read(Seen)).length===1);
    await fs.write(`${repo}/extensions/receiver/server.ts`,receiver(2));await git(repo,'add','.');await git(repo,'commit','--no-gpg-sign','-m','reload');
    const sha=await git(repo,'rev-parse','HEAD');await until(()=>loader.state().good===sha);
    await harness.commit(tx=>globalThis.eventHosts.emitter.emit(tx,Landed,{sha:'world-b'}),context);
    await until(async()=>(await read(Seen)).length===2);equal(await read(Seen),['world-a','world-b']);assert(globalThis.receiverVersion===2,'Replaced receiver');
    globalThis.eventHosts.emitter.notice('Notice');globalThis.eventHosts.emitter.notice('Notice');
    await until(async()=>(await service.eventLog.recent()).some(e=>e.name==='sausage.notice'));
    equal((await service.eventLog.recent()).filter(e=>e.name==='sausage.notice').length,1);
  } finally {await loader.close();await harness.close(context);}
} else {
  let throwOn:string|undefined;
  const handler=async(tx:any,event:any)=>{(await tx.doc(Seen,ROOT)).shas.push(event.data.sha);if(event.data.sha===throwOn)throw new Error('handler failed');};
  await land('a1');try{await land('bad',true);throw new Error('Missing rollback');}catch(error){assert(String(error).includes('rolled back'),'Rollback error');}
  await harness.commit(tx=>service.emit(tx,Branched,{id:7}),context);equal(await read(Shipped),['a1']);
  let reactor=service.reactor('probe/counter',[Landed],handler);await reactor.start();await until(async()=>(await read(Seen)).length===1);
  await land('b2');await until(async()=>(await read(Seen)).length===2);
  throwOn='c3';await land('c3');await until(async()=>!!(await service.eventLog.reactors())[0].failure);equal(await read(Seen),['a1','b2']);
  equal((await service.eventLog.reactors())[0].lag,1);throwOn=undefined;await land('d4');await until(async()=>(await read(Seen)).length===4);
  assert(!(await service.eventLog.reactors())[0].failure,'Failure cleared');
  await reactor.stop();await land('e5');equal((await service.eventLog.reactors())[0].running,false);
  reactor=service.reactor('probe/counter',[Landed],handler);await reactor.start();await until(async()=>(await read(Seen)).length===5);
  await reactor.stop();await harness.close(context);({harness,storage,registry}=await open());service=eventLog(()=>harness);
  await land('f6');reactor=service.reactor('probe/counter',[Landed],handler);await reactor.start();await until(async()=>(await read(Seen)).length===6);
  equal(await read(Seen),['a1','b2','c3','d4','e5','f6']);equal(await read(Shipped),await read(Seen));await reactor.stop();
  const all=await service.eventLog.recent({limit:20});equal(all.map(e=>e.data),[{sha:'f6'},{sha:'e5'},{sha:'d4'},{sha:'c3'},{sha:'b2'},{id:7},{sha:'a1'}]);
  const branch=all.find(e=>Branched.is(e))!;
  equal([branch.name,branch.kind,branch.v],['test.branched','event.test.branched',1]);assert(typeof branch.at==='string','Event timestamp');
  equal((await service.eventLog.recent({events:[Landed],since:branch.id,limit:2})).map(e=>e.data.sha),['f6','e5']);
  equal((await service.eventLog.recent({before:branch.id,order:'ascending'})).map(e=>e.data),[{sha:'a1'}]);
  (all[0].data as any).sha='changed';equal((await service.eventLog.recent({limit:1}))[0].data,{sha:'f6'});
  const feed=await harness.snapshot(EventFeed,ROOT,context);equal(feed?.events.map(e=>e.id),all.map(e=>e.id).toReversed());
  let nowCount=0;const now=service.reactor('probe/now',[Landed],()=>{nowCount++;},{from:'now'});await now.start();
  await until(async()=>(await service.eventLog.reactors()).find(r=>r.name==='probe/now')!.cursor===all[0].id);
  await land('g7');await until(()=>nowCount===1);await now.stop();
  await harness.commit(async tx=>{for(let i=0;i<EVENT_FEED_SIZE;i++)await service.emit(tx,Branched,{id:i});},context);
  equal((await harness.snapshot(EventFeed,ROOT,context))?.events.map(e=>e.data),Array.from({length:EVENT_FEED_SIZE},(_,id)=>({id})));
  await harness.close(context);
}