1import * as Clock from "effect/Clock";
2import * as Deferred from "effect/Deferred";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Ref from "effect/Ref";
7
8const REFRESH_TIMEOUT = Duration.seconds(30);
9
10type Entry<A> =
11 | { readonly _tag: "Empty" }
12 | { readonly _tag: "Scanning"; readonly scan: Deferred.Deferred<A> }
13 | {
14 readonly _tag: "Ready";
15 readonly value: A;
16 readonly expiresAtNanos: bigint;
17 readonly refreshing: boolean;
18 };
19
20// Shared scans survive caller cancellation; monotonic TTL bounds cache age.
21// Warm reads return stale data while a scope-owned, bounded refresh runs.
22export const makeStaleWhileRevalidate = Effect.fn("StaleWhileRevalidate.make")(function* <A>(
23 discover: Effect.Effect<A>,
24 ttl: Duration.Input,
25) {
26 const ttlNanos = Duration.toNanosUnsafe(Duration.fromInputUnsafe(ttl));
27 const scope = yield* Effect.scope;
28 const state = yield* Ref.make<Entry<A>>({ _tag: "Empty" });
29
30 const settle = Effect.fn("StaleWhileRevalidate.settle")(function* (exit: Exit.Exit<A>) {
31 const now = yield* Clock.monotonicTimeNanos;
32 yield* Ref.update(state, (entry): Entry<A> =>
33 Exit.isSuccess(exit)
34 ? {
35 _tag: "Ready",
36 value: exit.value,
37 expiresAtNanos: now + ttlNanos,
38 refreshing: false,
39 }
40 : entry._tag === "Ready"
41 ? { ...entry, refreshing: false }
42 : { _tag: "Empty" },
43 );
44 });
45 const fork = (scan: Effect.Effect<unknown>) =>
46 scan.pipe(Effect.ignoreCause({ log: true }), Effect.interruptible, Effect.forkIn(scope));
47 const firstScan = (scan: Deferred.Deferred<A>) =>
48 fork(
49 discover.pipe(
50 Effect.onExit((exit) => settle(exit).pipe(Effect.andThen(Deferred.done(scan, exit)))),
51 ),
52 );
53 const refresh = fork(discover.pipe(Effect.onExit(settle), Effect.timeoutOption(REFRESH_TIMEOUT)));
54
55 return Effect.gen(function* () {
56 const now = yield* Clock.monotonicTimeNanos;
57 // Claiming a scan and starting it are one uninterruptible step: a caller
58 // interrupted between them would leave a claimed scan that never runs.
59 const result = yield* Ref.modify(
60 state,
61 (entry): readonly [readonly [Effect.Effect<unknown>, Effect.Effect<A>], Entry<A>] => {
62 switch (entry._tag) {
63 case "Empty": {
64 const scan = Deferred.makeUnsafe<A>();
65 return [[firstScan(scan), Deferred.await(scan)], { _tag: "Scanning", scan }];
66 }
67 case "Scanning":
68 return [[Effect.void, Deferred.await(entry.scan)], entry];
69 case "Ready":
70 return entry.refreshing || entry.expiresAtNanos > now
71 ? [[Effect.void, Effect.succeed(entry.value)], entry]
72 : [[refresh, Effect.succeed(entry.value)], { ...entry, refreshing: true }];
73 }
74 },
75 ).pipe(
76 Effect.flatMap(([start, result]) => Effect.as(start, result)),
77 Effect.uninterruptible,
78 );
79 return yield* result;
80 });
81});