46 lines
1.6 KiB
TypeScript
46 lines
1.6 KiB
TypeScript
|
import type { CatchUpWindow } from "./chat-utils";
|
||
|
|
|
||
|
|
type Page<T> = { messages: T[]; next: string; through: string; hasMore: boolean };
|
||
|
|
|
||
|
|
export function createMutationRecovery<T>(
|
||
|
|
fetchPage: (cursor: CatchUpWindow) => Promise<Page<T>>,
|
||
|
|
apply: (messages: T[]) => void,
|
||
|
|
initialCursor: CatchUpWindow = { after: "0" },
|
||
|
|
onCursor?: (cursor: CatchUpWindow) => void,
|
||
|
|
) {
|
||
|
|
let cursor: CatchUpWindow = { ...initialCursor };
|
||
|
|
let stopped = false;
|
||
|
|
let active: Promise<void> | undefined;
|
||
|
|
let requested = false;
|
||
|
|
async function drain() {
|
||
|
|
do {
|
||
|
|
requested = false;
|
||
|
|
let more: boolean;
|
||
|
|
do {
|
||
|
|
const page = await fetchPage({ ...cursor });
|
||
|
|
if (stopped) return;
|
||
|
|
if (
|
||
|
|
BigInt(page.through) < BigInt(cursor.after) ||
|
||
|
|
BigInt(page.next) < BigInt(cursor.after) ||
|
||
|
|
BigInt(page.next) > BigInt(page.through) ||
|
||
|
|
(cursor.through !== undefined && cursor.through !== page.through) ||
|
||
|
|
(page.hasMore && BigInt(page.next) <= BigInt(cursor.after))
|
||
|
|
) throw new Error("Invalid mutation recovery window");
|
||
|
|
apply(page.messages);
|
||
|
|
cursor = page.hasMore ? { after: page.next, through: page.through } : { after: page.through };
|
||
|
|
onCursor?.({ ...cursor });
|
||
|
|
more = page.hasMore;
|
||
|
|
} while (more && !stopped);
|
||
|
|
} while (requested && !stopped);
|
||
|
|
}
|
||
|
|
return {
|
||
|
|
run(): Promise<void> {
|
||
|
|
if (stopped) return Promise.resolve();
|
||
|
|
if (active) { requested = true; return active; }
|
||
|
|
active = drain().finally(() => { active = undefined; });
|
||
|
|
return active;
|
||
|
|
},
|
||
|
|
stop() { stopped = true; },
|
||
|
|
};
|
||
|
|
}
|