1// @effect-diagnostics globalTimers:off - Vendored @clo/lib has no Effect dependency.
2/**
3 * helpers to deal with promises and asynchronous execution.
4 *
5 * @module
6 */
7
8/*** @deprecated */
9interface ARCEValue<T> {
10 value: T;
11 [Symbol.dispose]: () => void;
12}
13
14/*** @deprecated */
15export function RefCountedExpirable<T>(
16 init: () => Promise<T>,
17 deinit: (value: T) => void,
18 expire: number = 5 * 60 * 1000,
19): () => Promise<ARCEValue<T>> {
20 let refs = 0;
21 let item: ARCEValue<T> | null = null;
22 let loading: Promise<ARCEValue<T>> | null = null;
23 let timer: ReturnType<typeof setTimeout> | null = null;
24
25 function deref() {
26 ASSERT(item !== null);
27 if (--refs !== 0) return;
28 ASSERT(timer === null);
29 timer = setTimeout(() => {
30 ASSERT(refs === 0);
31 ASSERT(loading === null);
32 ASSERT(item !== null);
33 deinit(item.value);
34 item = null;
35 timer = null;
36 }, expire);
37 }
38
39 return async function (): Promise<ARCEValue<T>> {
40 if (timer !== null) {
41 clearTimeout(timer);
42 timer = null;
43 }
44 if (item !== null) {
45 refs++;
46 return item;
47 }
48 if (loading !== null) {
49 refs++;
50 return loading;
51 }
52 const p = Promise.withResolvers<ARCEValue<T>>();
53 loading = p.promise;
54 try {
55 const value = await init();
56 item = { value, [Symbol.dispose]: deref };
57 refs++;
58 p.resolve(item);
59 return item;
60 } catch (e) {
61 p.reject(e);
62 throw e;
63 } finally {
64 loading = null;
65 }
66 };
67}
68
69/** evaluates `fn` one time, caching the result forever. */
70export function once<T>(fn: () => Promise<T>): () => Promise<T> {
71 let result: T | Promise<T> | null = null;
72 return async () => {
73 if (result) return result;
74 try {
75 result = await (result = fn());
76 } catch (err) {
77 result = null;
78 throw err;
79 }
80 return result;
81 };
82}
83
84/** a variable that can be watched for changes */
85export class Watch<T> {
86 #value: T;
87 #observers = new Set<WatchCallback<T>>();
88 constructor(value: T) {
89 this.#value = value;
90 }
91
92 on(cb: WatchCallback<T>): () => void {
93 this.#observers.add(cb);
94 return () => void this.#observers.delete(cb);
95 }
96 once(cb: WatchCallback<T>): void {
97 const once = (n: T, p: T) => (cb(n, p), this.#observers.delete(once));
98 this.#observers.add(once);
99 }
100
101 next(): Promise<T> {
102 return new Promise((resolve) => this.once(resolve));
103 }
104
105 until(condition: (v: T) => boolean): Promise<T> {
106 if (condition(this.#value)) return Promise.resolve(this.#value);
107 return new Promise<T>((resolve) => {
108 function check(next: T) {
109 if (condition(next)) (resolve(next), release());
110 }
111 const release = this.on(check);
112 });
113 }
114
115 get value(): T {
116 return this.#value;
117 }
118 set value(next: T) {
119 const prev = this.#value;
120 if (prev !== next) {
121 this.#value = next;
122 this.#observers.forEach((cb) => cb(next, prev));
123 }
124 }
125}
126export type WatchCallback<T> = (next: T, prev: T) => void;
127
128/**
129 * when two requests with the same args come in at the same time, the response
130 * from them is shared. after returning, the data is no longer stored.
131 */
132export class DedupeConcurrent<Args extends unknown[], Ret> {
133 pending: Map<unknown, Promise<Ret>> = new Map();
134 fn: (...args: Args) => Promise<Ret>;
135 keyFn: (args: Args) => unknown = JSON.stringify;
136
137 constructor(
138 fn: (...args: Args) => Promise<Ret>,
139 keyFn: (args: Args) => unknown = JSON.stringify,
140 ) {
141 this.fn = fn;
142 this.keyFn = keyFn;
143 }
144
145 isPending(...args: Args): boolean {
146 return this.pending.has(this.keyFn(args));
147 }
148
149 /** Unbound to allow easily exporting this callback */
150 get = (...args: Args): Promise<Ret> => {
151 const k = this.keyFn(args);
152 let promise = this.pending.get(k);
153 if (promise) return promise;
154 promise = this.fn(...args);
155 this.pending.set(k, promise);
156 void promise.finally(() => void this.pending.delete(k)).catch(() => {});
157 return promise;
158 };
159}
160
161/**
162 * when two requests with the same args come, the result
163 * of the first is memoized.
164 */
165export class OnceMap<Args extends unknown[], Ret> {
166 cache: Map<unknown, Promise<Ret>> = new Map();
167 fn: (...args: Args) => Promise<Ret>;
168 keyFn: (args: Args) => unknown = JSON.stringify;
169
170 constructor(
171 fn: (...args: Args) => Promise<Ret>,
172 keyFn: (args: Args) => unknown = JSON.stringify,
173 ) {
174 this.fn = fn;
175 this.keyFn = keyFn;
176 }
177
178 has(...args: Args): boolean {
179 return this.cache.has(this.keyFn(args));
180 }
181
182 /** Unbound to allow easily exporting this callback */
183 getOrRun = (...args: Args): Promise<Ret> => {
184 const k = this.keyFn(args);
185 let promise = this.cache.get(k);
186 if (promise) return promise;
187 promise = this.fn(...args);
188 this.cache.set(k, promise);
189 return promise;
190 };
191}
192
193export class PromiseAggregator {
194 failures: unknown[] = [];
195 promises: Set<Promise<unknown>> = new Set();
196
197 push(p: Promise<unknown>) {
198 p = p.then(
199 () => {
200 this.promises.delete(p);
201 },
202 (err) => {
203 this.failures.push(err);
204 },
205 );
206 this.promises.add(p);
207 }
208
209 async all() {
210 while (this.promises.size > 0) {
211 const wait = [...this.promises];
212 this.promises.clear();
213 await Promise.all(wait);
214 }
215 if (this.failures.length > 0) {
216 const agg = new AggregateError(this.failures);
217 this.failures = [];
218 throw agg;
219 }
220 }
221
222 [Symbol.asyncDispose](): Promise<void> {
223 return this.all();
224 }
225}
226
227/** a promise that can be cancelled. */
228export type Cancelable<T> = Promise<T> &
229 Disposable & {
230 cancel: (reason?: unknown) => void;
231 };
232
233/** make an existing promise cancelable. */
234export function makeCancelable<T>(
235 promise: Promise<T>,
236 cancel: (reason?: unknown) => void,
237): Cancelable<T> {
238 const p = promise as Cancelable<T>;
239 p[Symbol.dispose] = cancel;
240 p.cancel = cancel;
241 return p;
242}
243
244/** wait `ms` milliseconds, then resolve. can be cancelled. */
245export function delay(ms: number): Cancelable<void> {
246 if (ms === Infinity) return makeCancelable(new Promise(() => {}), () => {});
247 let t: ts.Timer | null = null;
248 const maxTimer = 0x7fffffff;
249 return makeCancelable(
250 new Promise(function tick(resolve) {
251 if (ms <= maxTimer) {
252 t = setTimeout(() => ((t = null), resolve()), ms);
253 } else {
254 t = setTimeout(() => {
255 ms -= maxTimer;
256 tick(resolve);
257 }, maxTimer);
258 }
259 }),
260 () => {
261 if (t != null) clearTimeout(t);
262 },
263 );
264}
265
266export interface Deferred<T> extends AsyncDisposable, PromiseLike<T> {}
267
268/**
269 * Return something that can be awaited later without worry of an unhandled
270 * rejection. By disposing it, the promise exception will always be reported.
271 *
272 * ```ts
273 * // incorrect, possible unhandled rejection between a and b.
274 * const a = promise();
275 * const b = await promise();
276 * const c = await promise(b, await a);
277 * // correct
278 * await using a = async.deferred(promise());
279 * const b = await promise();
280 * const c = await promise(b, await a);
281 * ```
282 */
283export function deferred<T>(promise: Promise<T>): Deferred<T> {
284 let handled = false;
285 promise.catch(() => {}); // potential error is re-thrown on asyncDispose
286 return {
287 async [Symbol.asyncDispose]() {
288 if (!handled) await promise;
289 },
290 then(onFufilled, onRejected) {
291 return promise.finally(() => (handled = true)).then(onFufilled, onRejected);
292 },
293 };
294}
295
296import { ASSERT } from "./assert.ts";
297import * as ts from "./ts.ts";