6c7eed7580
Fresh history cut from cynarAI/evoo antigravity-handoff/project-timeline (Voice First + project timeline). Brand docs/titles as evOo; keep @rakazo/* packages as tech debt. Excludes node_modules, caches, secrets, voice blobs.
81 lines
2.9 KiB
TypeScript
81 lines
2.9 KiB
TypeScript
import type { ProductEvent } from "@rakazo/contracts";
|
|
import { abortableDelay } from "./async.js";
|
|
|
|
type ThreadHead = { threadId: string; cursor: number };
|
|
|
|
/** Recover a durable event stream while its caller owns snapshot commits and UI effects. */
|
|
export async function runThreadSubscription(options: {
|
|
signal: AbortSignal;
|
|
loadInitial: () => Promise<ThreadHead | null>;
|
|
loadHead: () => Promise<ThreadHead | null>;
|
|
refresh: () => Promise<unknown>;
|
|
currentSnapshot: () => ThreadHead | null;
|
|
subscribe: (cursor: number) => Promise<AsyncIterable<ProductEvent>>;
|
|
beforeEvent?: (event: ProductEvent) => void;
|
|
applyEvent: (event: ProductEvent) => void;
|
|
onEvent: (event: ProductEvent, initial: ThreadHead) => void;
|
|
}): Promise<void> {
|
|
const { signal } = options;
|
|
const wait = async (ms: number) => {
|
|
await abortableDelay(ms, signal);
|
|
return !signal.aborted;
|
|
};
|
|
let initial = await options.loadInitial().catch(() => null);
|
|
if (signal.aborted) return;
|
|
let headRetryMs = 250;
|
|
while (!initial && !signal.aborted) {
|
|
initial = await options.loadHead().catch(() => null);
|
|
if (initial) break;
|
|
if (!(await wait(headRetryMs))) return;
|
|
headRetryMs = Math.min(headRetryMs * 2, 5_000);
|
|
}
|
|
if (!initial || signal.aborted) return;
|
|
const head = initial;
|
|
let cursor = head.cursor;
|
|
let snapshotReady = options.currentSnapshot()?.threadId === head.threadId;
|
|
const pendingSnapshotEvents: ProductEvent[] = [];
|
|
if (!snapshotReady) {
|
|
void (async () => {
|
|
let snapshotRetryMs = 250;
|
|
while (!snapshotReady && !signal.aborted) {
|
|
if (!(await wait(snapshotRetryMs))) return;
|
|
await options.refresh().catch(() => null);
|
|
if (signal.aborted) return;
|
|
const committed = options.currentSnapshot();
|
|
if (committed?.threadId === head.threadId) {
|
|
snapshotReady = true;
|
|
for (const event of pendingSnapshotEvents.splice(0)) {
|
|
if (event.seq > committed.cursor) options.applyEvent(event);
|
|
}
|
|
return;
|
|
}
|
|
snapshotRetryMs = Math.min(snapshotRetryMs * 2, 5_000);
|
|
}
|
|
})();
|
|
}
|
|
let retryMs = 250;
|
|
while (!signal.aborted) {
|
|
try {
|
|
const events = await options.subscribe(cursor);
|
|
for await (const event of events) {
|
|
if (signal.aborted) break;
|
|
cursor = Math.max(cursor, event.seq);
|
|
retryMs = 250;
|
|
options.beforeEvent?.(event);
|
|
if (snapshotReady && options.currentSnapshot()?.threadId === event.threadId) {
|
|
options.applyEvent(event);
|
|
} else if (!snapshotReady) {
|
|
pendingSnapshotEvents.push(event);
|
|
}
|
|
options.onEvent(event, head);
|
|
}
|
|
} catch {
|
|
// Reconnect from the last durable event after a transient transport failure.
|
|
}
|
|
if (signal.aborted) return;
|
|
await options.refresh().catch(() => null);
|
|
if (!(await wait(retryMs))) return;
|
|
retryMs = Math.min(retryMs * 2, 5_000);
|
|
}
|
|
}
|