146 lines
8.2 KiB
TypeScript
146 lines
8.2 KiB
TypeScript
// Durable event entries and reactor cursors share the handler's transaction.
|
|
// Transactional effects run exactly once. External effects can repeat after a crash.
|
|
import type { JsonValue } from '@earendil-works/chord';
|
|
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
|
|
import { defineDoc, ROOT_CONVERSATION_ID as ROOT } from '@earendil-works/pi-durable';
|
|
import type { Harness, Tx, EntryRecord, ConversationId, EntryId } from '@earendil-works/pi-durable';
|
|
export type EventRecord<D extends JsonValue = JsonValue> = {id:number;name:string;kind:string;v:number;at:string;data:D};
|
|
export interface EventToken<D extends JsonValue = JsonValue> {
|
|
readonly name:string; readonly kind:string; readonly version:number;
|
|
is(event:EventRecord|undefined):event is EventRecord<D>;
|
|
}
|
|
export type EventData<T> = T extends EventToken<infer D> ? D : never;
|
|
export function defineEvent<D extends JsonValue>(name:string, options:{version?:number}={}):EventToken<D> {
|
|
const kind=`event.${name}`;
|
|
return {name,kind,version:options.version??1,is:(event):event is EventRecord<D>=>event?.kind===kind};
|
|
}
|
|
export const NoticeEvent=defineEvent<{extension:string;message:string;level:string}>('sausage.notice');
|
|
export const EVENT_FEED='sausage.event-feed', EVENT_FEED_SIZE=100;
|
|
export const EventLog=defineDoc<{conversation?:number;cursors:Record<string,number>}>({
|
|
kind:'sausage.events',version:1,scope:'conversation',history:'latest',fork:'initial',
|
|
checkpointWhen:()=>false,initial:()=>({cursors:{}}),
|
|
});
|
|
export const EventFeed=defineDoc<{events:EventRecord[]}>({
|
|
kind:EVENT_FEED,version:1,scope:'conversation',history:'latest',fork:'initial',
|
|
checkpointWhen:()=>false,initial:()=>({events:[]}),
|
|
});
|
|
export type Reactor<D extends JsonValue>=(tx:Tx,event:EventRecord<D>)=>void|Promise<void>;
|
|
export type ReactOptions={from?:'start'|'now'};
|
|
export type ReactorStatus={name:string;events:string[];cursor:number;lag:number;running:boolean;failure?:{error:string;at:string;entryId?:number}};
|
|
export interface EventHost {
|
|
emit<D extends JsonValue>(tx:Tx,event:EventToken<D>,data:D):Promise<EventRecord<D>>;
|
|
react<T extends EventToken>(name:string,events:readonly T[],handler:Reactor<EventData<T>>,options?:ReactOptions):void;
|
|
eventLog:{recent<T extends EventToken=EventToken>(options?:{events?:readonly T[];since?:number;before?:number;order?:'ascending'|'descending';limit?:number}):Promise<EventRecord<EventData<T>>[]>;reactors():Promise<ReactorStatus[]>};
|
|
}
|
|
const record=(entry:EntryRecord):EventRecord=>structuredClone({id:entry.id,name:entry.kind.slice(6),kind:entry.kind,...entry.data as {v:number;at:string;data:JsonValue}});
|
|
export async function emit<D extends JsonValue>(tx:Tx,event:EventToken<D>,data:D):Promise<EventRecord<D>> {
|
|
const index=await tx.doc(EventLog,ROOT);
|
|
if(index.conversation===undefined)index.conversation=(await tx.createConversation({ownership:{kind:'ownerless'}})).id;
|
|
const entry=await tx.appendEntry(index.conversation as ConversationId,{kind:event.kind,data:{v:event.version,at:new Date().toISOString(),data}});
|
|
const emitted=record(entry) as EventRecord<D>;
|
|
const feed=(await tx.doc(EventFeed,ROOT)).events;
|
|
feed.push(emitted);
|
|
if(feed.length>EVENT_FEED_SIZE)feed.splice(0,feed.length-EVENT_FEED_SIZE);
|
|
return emitted;
|
|
}
|
|
async function scan(tx:Tx,log:ConversationId,after:number,visit:(entry:EntryRecord)=>boolean) {
|
|
let cursor:Parameters<Tx['scanEntries']>[2];
|
|
do {
|
|
const page=await tx.scanEntries({conversationId:log,minEntryId:(after+1) as EntryId,order:'ascending'},256,cursor);
|
|
for(const entry of page.items)if(visit(entry))return;
|
|
cursor=page.next;
|
|
} while(cursor);
|
|
}
|
|
/** One service per loader. Names retain their cursors when facets change. */
|
|
export function eventLog(harness:()=>Harness) {
|
|
type Active={kinds:Set<string>;handler:Reactor<JsonValue>;from:'start'|'now'};
|
|
const reactors=new Map<string,{active?:Active;kinds:Set<string>;chain:Promise<void>;failure?:ReactorStatus['failure']}>();
|
|
function drain(name:string) {
|
|
const state=reactors.get(name)!;
|
|
state.chain=state.chain.then(async()=>{
|
|
for(;;) {
|
|
const active=state.active;
|
|
if(!active)return;
|
|
let attempted:number|undefined;
|
|
try {
|
|
const handled=await harness().commit(async tx=>{
|
|
const index=await tx.doc(EventLog,ROOT);
|
|
if(index.cursors[name]===undefined && active.from==='now') {
|
|
const newest=index.conversation===undefined?undefined:(await tx.scanEntries({conversationId:index.conversation as ConversationId},1)).items[0];
|
|
index.cursors[name]=newest?.id??0;
|
|
return false;
|
|
}
|
|
if(index.conversation===undefined)return false;
|
|
const after=index.cursors[name]??0;
|
|
let pending:EntryRecord|undefined,last=after;
|
|
await scan(tx,index.conversation as ConversationId,after,entry=>{
|
|
last=entry.id;if(active.kinds.has(entry.kind))pending=entry;return pending!==undefined;
|
|
});
|
|
if(pending){attempted=pending.id;await active.handler(tx,record(pending));}
|
|
if(last!==after)index.cursors[name]=last;
|
|
return pending!==undefined;
|
|
},context);
|
|
state.failure=undefined;
|
|
if(!handled)return;
|
|
} catch(error) {
|
|
state.failure={error:String(error),at:new Date().toISOString(),entryId:attempted};
|
|
return;
|
|
}
|
|
}
|
|
});
|
|
}
|
|
return {
|
|
emit,
|
|
reactor<T extends EventToken>(name:string,events:readonly T[],handler:Reactor<EventData<T>>,options:ReactOptions={}) {
|
|
const active:Active={kinds:new Set(events.map(event=>event.kind)),handler:handler as Reactor<JsonValue>,from:options.from??'start'};
|
|
if(!reactors.has(name))reactors.set(name,{kinds:active.kinds,chain:Promise.resolve()});
|
|
const state=reactors.get(name)!;
|
|
let detach:(()=>void)|undefined;
|
|
const stop=()=>{detach?.();detach=undefined;if(state.active===active)state.active=undefined;return state.chain;};
|
|
return {
|
|
start:async()=>{
|
|
await stop();state.active=active;state.kinds=active.kinds;
|
|
const observation=harness().observeCommits({advance:publication=>{
|
|
if(publication.changes.some(change=>change.type==='entry' && active.kinds.has(change.value.kind)))drain(name);
|
|
},end:()=>{}});
|
|
detach=()=>observation.detach();drain(name);
|
|
},stop,
|
|
};
|
|
},
|
|
eventLog:{recent,reactors:statuses} satisfies EventHost['eventLog'],
|
|
};
|
|
async function recent<T extends EventToken=EventToken>(options:{events?:readonly T[];since?:number;before?:number;order?:'ascending'|'descending';limit?:number}={}) {
|
|
const log=(await harness().snapshot(EventLog,ROOT,context))?.conversation;
|
|
if(log===undefined)return [];
|
|
const kinds=options.events && new Set(options.events.map(event=>event.kind));
|
|
const limit=options.limit??20;
|
|
return harness().commit(async tx=>{
|
|
const found:EventRecord<EventData<T>>[]=[];
|
|
let cursor:Parameters<Tx['scanEntries']>[2];
|
|
do {
|
|
const page=await tx.scanEntries({conversationId:log as ConversationId,
|
|
...(options.since===undefined?{}:{minEntryId:(options.since+1) as EntryId}),
|
|
...(options.before===undefined?{}:{maxEntryId:(options.before-1) as EntryId}),order:options.order??'descending'},256,cursor);
|
|
for(const entry of page.items) {
|
|
if(!(kinds?kinds.has(entry.kind):entry.kind.startsWith('event.')))continue;
|
|
found.push(record(entry) as EventRecord<EventData<T>>);if(found.length>=limit)return found;
|
|
}
|
|
cursor=page.next;
|
|
} while(cursor);
|
|
return found;
|
|
},context);
|
|
}
|
|
async function statuses():Promise<ReactorStatus[]> {
|
|
const index=await harness().snapshot(EventLog,ROOT,context);
|
|
return harness().commit(async tx=>{
|
|
const list:ReactorStatus[]=[];
|
|
for(const [name,state] of reactors) {
|
|
const cursor=index?.cursors[name]??0;let lag=0;
|
|
if(index?.conversation!==undefined)await scan(tx,index.conversation as ConversationId,cursor,entry=>{if(state.kinds.has(entry.kind))lag++;return false;});
|
|
list.push({name,events:[...state.kinds].map(kind=>kind.slice(6)),cursor,lag,running:state.active!==undefined,...(state.failure?{failure:state.failure}:{})});
|
|
}
|
|
return list;
|
|
},context);
|
|
}
|
|
}
|