1import type * as Duration from "effect/Duration";
2
3export interface ProviderReplayGate {
4 readonly beforeEmit: (label: string | undefined, signal?: AbortSignal) => Promise<void>;
5 readonly waitForReached: (label: string) => Promise<boolean>;
6 readonly hasReached: (label: string) => boolean;
7 readonly release: (label: string) => boolean;
8 readonly releaseAll: () => void;
9 /**
10 * Receipts from adapters that hold a settled turn open and finalize it after
11 * a debounce: each arming records the debounce it waits. Replay runs on a
12 * test clock, so a scenario advances it by exactly that on the receipt.
13 */
14 readonly recordFinishArmed: (debounce: Duration.Input) => void;
15 readonly finishArmedCount: () => number;
16 /** Resolves with the latest armed debounce once more than `seen` were recorded. */
17 readonly waitForFinishArmed: (seen: number) => Promise<Duration.Input>;
18}
19
20interface GateState {
21 reached: boolean;
22 released: boolean;
23 readonly reachedPromise: Promise<void>;
24 readonly resolveReached: () => void;
25 readonly promise: Promise<void>;
26 readonly resolve: () => void;
27}
28
29export function makeProviderReplayGate(labels: ReadonlyArray<string>): ProviderReplayGate {
30 const states = new Map<string, GateState>();
31 for (const label of labels) {
32 if (states.has(label)) {
33 throw new Error(`Duplicate provider replay gate label ${label}.`);
34 }
35 let resolve = () => {};
36 const promise = new Promise<void>((resume) => {
37 resolve = resume;
38 });
39 let resolveReached = () => {};
40 const reachedPromise = new Promise<void>((resume) => {
41 resolveReached = resume;
42 });
43 states.set(label, {
44 reached: false,
45 released: false,
46 reachedPromise,
47 resolveReached,
48 promise,
49 resolve,
50 });
51 }
52
53 const finishArmed: Array<Duration.Input> = [];
54 const finishArmedWaiters: Array<() => void> = [];
55
56 return {
57 recordFinishArmed: (debounce) => {
58 finishArmed.push(debounce);
59 for (const wake of finishArmedWaiters.splice(0)) wake();
60 },
61 finishArmedCount: () => finishArmed.length,
62 waitForFinishArmed: (seen) =>
63 new Promise((resolve) => {
64 const check = () => {
65 if (finishArmed.length > seen) resolve(finishArmed[finishArmed.length - 1]!);
66 else finishArmedWaiters.push(check);
67 };
68 check();
69 }),
70 beforeEmit: (label, signal) => {
71 if (label === undefined) {
72 return Promise.resolve();
73 }
74 const state = states.get(label);
75 if (state === undefined) {
76 return Promise.resolve();
77 }
78 state.reached = true;
79 state.resolveReached();
80 if (signal === undefined) {
81 return state.promise;
82 }
83 if (signal.aborted) {
84 return Promise.resolve();
85 }
86 return new Promise<void>((resolve) => {
87 const stopWaiting = () => resolve();
88 signal.addEventListener("abort", stopWaiting, { once: true });
89 void state.promise.then(() => {
90 signal.removeEventListener("abort", stopWaiting);
91 resolve();
92 });
93 });
94 },
95 waitForReached: (label) => {
96 const state = states.get(label);
97 return state === undefined ? Promise.resolve(false) : state.reachedPromise.then(() => true);
98 },
99 hasReached: (label) => states.get(label)?.reached ?? false,
100 release: (label) => {
101 const state = states.get(label);
102 if (state === undefined || state.released) {
103 return false;
104 }
105 state.released = true;
106 state.resolve();
107 return true;
108 },
109 releaseAll: () => {
110 for (const state of states.values()) {
111 if (state.released) {
112 continue;
113 }
114 state.released = true;
115 state.resolve();
116 }
117 },
118 };
119}