1import { assert, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { PullRequestOperationError } from "@t3tools/contracts";
4import * as DateTime from "effect/DateTime";
5import * as Deferred from "effect/Deferred";
6import * as Duration from "effect/Duration";
7import * as Effect from "effect/Effect";
8import * as Fiber from "effect/Fiber";
9import * as FileSystem from "effect/FileSystem";
10import * as Layer from "effect/Layer";
11import * as Path from "effect/Path";
12import * as Queue from "effect/Queue";
13import * as TestClock from "effect/testing/TestClock";
14import * as KeyValueStore from "effect/persistence/KeyValueStore";
15import * as ServerConfig from "../config.ts";
16import * as PullRequestReadCache from "./PullRequestReadCache.ts";
17
18const cacheLayer = (directory: string) =>
19 PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));
20
21/** Sets every file's mtime to `age` before the current test clock time. */
22const ageFiles = (directory: string, names: ReadonlyArray<string>, age: Duration.Duration) =>
23 Effect.gen(function* () {
24 const fs = yield* FileSystem.FileSystem;
25 const path = yield* Path.Path;
26 const mtime = DateTime.toDateUtc(DateTime.subtractDuration(yield* DateTime.now, age));
27 for (const name of names) yield* fs.utimes(path.join(directory, name), mtime, mtime);
28 });
29
30it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
31 it.effect("prunes entry files written before the max age and keeps fresh ones", () =>
32 Effect.gen(function* () {
33 const fs = yield* FileSystem.FileSystem;
34 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
35 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z")));
36 const cache = yield* cacheLayer(directory);
37 yield* cache.get("old", Effect.succeed("old"), ["pr"]);
38 yield* cache.invalidate("pr");
39 const [stale, revisions] = (yield* fs.readDirectory(directory)).toSorted();
40 assert.strictEqual(revisions, "revisions");
41 yield* cache.get("fresh", Effect.succeed("fresh"));
42 const fresh = (yield* fs.readDirectory(directory)).find(
43 (name) => name !== stale && name !== revisions,
44 );
45 const unrelated = "unrelated.json";
46 yield* fs.writeFileString(`${directory}/${unrelated}`, "{}");
47 // Starts with the store id but is not an entry file.
48 const prefixed = "pr-v2-backup.json";
49 yield* fs.writeFileString(`${directory}/${prefixed}`, "{}");
50 const justExpired = Duration.sum(
51 PullRequestReadCache.ENTRY_FILE_MAX_AGE,
52 Duration.seconds(1),
53 );
54 yield* ageFiles(directory, [stale!, revisions!, unrelated, prefixed], justExpired);
55 yield* ageFiles(directory, [fresh!], PullRequestReadCache.ENTRY_FILE_MAX_AGE);
56
57 yield* PullRequestReadCache.pruneExpiredEntryFiles(directory);
58
59 assert.deepStrictEqual(
60 (yield* fs.readDirectory(directory)).toSorted(),
61 [fresh!, revisions!, unrelated, prefixed].toSorted(),
62 );
63 }),
64 );
65
66 it.effect("sweeps the cache directory every hour once the layer is built", () =>
67 Effect.gen(function* () {
68 const fs = yield* FileSystem.FileSystem;
69 const { providerStatusCacheDir } = yield* ServerConfig.ServerConfig;
70 const directory = `${providerStatusCacheDir}/pull-requests`;
71 yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z")));
72 yield* (yield* cacheLayer(directory)).get("summary", Effect.succeed("cached"));
73 const [entry] = yield* fs.readDirectory(directory);
74 // Within the max age at the first two sweeps, past it at the third.
75 yield* ageFiles(
76 directory,
77 [entry!],
78 Duration.subtract(PullRequestReadCache.ENTRY_FILE_MAX_AGE, Duration.minutes(90)),
79 );
80 // A sweep's last file operation is the entry's stat when it keeps the
81 // entry, and its removal otherwise. Waiting for it means the sweep has
82 // finished before the clock moves.
83 const operations = yield* Queue.unbounded<string>();
84 const observedFs = FileSystem.FileSystem.of({
85 ...fs,
86 stat: (path) => fs.stat(path).pipe(Effect.tap(() => Queue.offer(operations, "stat"))),
87 remove: (path, options) =>
88 fs.remove(path, options).pipe(Effect.tap(() => Queue.offer(operations, "remove"))),
89 });
90 yield* Layer.build(
91 PullRequestReadCache.layer.pipe(
92 Layer.provide(Layer.succeed(FileSystem.FileSystem, observedFs)),
93 ),
94 );
95 assert.strictEqual(yield* Queue.take(operations), "stat");
96 yield* TestClock.adjust("1 hour");
97 assert.strictEqual(yield* Queue.take(operations), "stat");
98 assert.deepStrictEqual(yield* fs.readDirectory(directory), [entry]);
99 yield* TestClock.adjust("1 hour");
100 assert.deepStrictEqual(yield* Queue.takeN(operations, 2), ["stat", "remove"]);
101 assert.deepStrictEqual(yield* fs.readDirectory(directory), []);
102 }).pipe(
103 Effect.scoped,
104 Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-pr-cache-layer-" })),
105 ),
106 );
107
108 it.effect("reuses files after restart and respects the original expiry", () =>
109 Effect.gen(function* () {
110 const fs = yield* FileSystem.FileSystem;
111 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
112 let reads = 0;
113 const lookup = Effect.sync(() => String(++reads));
114 const first = yield* cacheLayer(directory);
115 const key = "long/repository/key".repeat(100);
116 assert.strictEqual(yield* first.get(key, lookup), "1");
117 yield* TestClock.adjust("59 seconds");
118 const restarted = yield* cacheLayer(directory);
119 assert.strictEqual(yield* restarted.get(key, lookup), "1");
120 yield* TestClock.adjust("1 second");
121 assert.strictEqual(yield* restarted.get(key, lookup), "2");
122 assert.strictEqual(reads, 2);
123 assert.strictEqual((yield* fs.readDirectory(directory)).length, 1);
124 }),
125 );
126
127 it.effect("clears in-flight reads before a new service can reuse them", () =>
128 Effect.gen(function* () {
129 const fs = yield* FileSystem.FileSystem;
130 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
131 const started = yield* Deferred.make<void>();
132 const release = yield* Deferred.make<void>();
133 const cache = yield* cacheLayer(directory);
134 const read = yield* cache
135 .get(
136 "summary",
137 Deferred.succeed(started, undefined).pipe(
138 Effect.andThen(Deferred.await(release)),
139 Effect.as("old"),
140 ),
141 ["pr"],
142 )
143 .pipe(Effect.forkChild);
144 yield* Deferred.await(started);
145 const invalidate = yield* cache
146 .invalidate("pr")
147 .pipe(Effect.forkChild({ startImmediately: true }));
148 yield* Deferred.succeed(release, undefined);
149 yield* Fiber.join(read);
150 yield* Fiber.join(invalidate);
151 const restarted = yield* cacheLayer(directory);
152 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new"), ["pr"]), "new");
153 }),
154 );
155
156 it.effect("invalidates only the changed scope across restarts and coalesces its next reads", () =>
157 Effect.gen(function* () {
158 const fs = yield* FileSystem.FileSystem;
159 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
160 let reads = 0;
161 const lookup = Effect.sync(() => String(++reads));
162 const cache = yield* cacheLayer(directory);
163 yield* cache.get("first", lookup, ["project", "pr-1"]);
164 yield* cache.get("second", lookup, ["project", "pr-2"]);
165 yield* cache.get("third", lookup, ["other-project", "pr-3"]);
166 yield* cache.invalidate("pr-1");
167 const restarted = yield* cacheLayer(directory);
168 const answers = yield* Effect.all(
169 Array.from({ length: 10 }, () => restarted.get("first", lookup, ["project", "pr-1"])),
170 { concurrency: 10 },
171 );
172 assert.deepStrictEqual(answers, Array(10).fill("4"));
173 assert.strictEqual(yield* restarted.get("second", lookup, ["project", "pr-2"]), "2");
174 yield* restarted.invalidate("project");
175 const again = yield* cacheLayer(directory);
176 assert.strictEqual(yield* again.get("second", lookup, ["project", "pr-2"]), "5");
177 assert.strictEqual(yield* again.get("third", lookup, ["other-project", "pr-3"]), "3");
178 assert.strictEqual(reads, 5);
179 const files = (yield* fs.readDirectory(directory)).length;
180 for (let index = 0; index < 3; index++) {
181 yield* again.invalidate("pr-1");
182 yield* again.get("first", lookup, ["project", "pr-1"]);
183 }
184 assert.strictEqual((yield* fs.readDirectory(directory)).length, files);
185 }),
186 );
187
188 it.effect("shares a pending refresh without blocking an unrelated cached PR", () =>
189 Effect.gen(function* () {
190 const cache = yield* PullRequestReadCache.make;
191 yield* cache.get("first", Effect.succeed("old"), ["pr-1"]);
192 yield* cache.get("second", Effect.succeed("warm"), ["pr-2"]);
193 yield* cache.invalidate("pr-1");
194 const started = yield* Deferred.make<void>();
195 const release = yield* Deferred.make<void>();
196 let reads = 0;
197 const refresh = cache.get(
198 "first",
199 Effect.gen(function* () {
200 reads++;
201 yield* Deferred.succeed(started, undefined);
202 yield* Deferred.await(release);
203 return "fresh";
204 }),
205 ["pr-1"],
206 );
207 const pending = yield* Effect.all(
208 Array.from({ length: 10 }, () => refresh),
209 {
210 concurrency: 10,
211 },
212 ).pipe(Effect.forkChild);
213 yield* Deferred.await(started);
214 assert.strictEqual(yield* cache.get("second", Effect.die("cache miss"), ["pr-2"]), "warm");
215 yield* Deferred.succeed(release, undefined);
216 assert.deepStrictEqual(yield* Fiber.join(pending), Array(10).fill("fresh"));
217 assert.strictEqual(reads, 1);
218 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
219 );
220
221 it.effect("compacts expired scope records without discarding fresh PR data", () =>
222 Effect.gen(function* () {
223 const fs = yield* FileSystem.FileSystem;
224 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
225 const cache = yield* cacheLayer(directory);
226 yield* cache.get("summary", Effect.succeed("old"), ["pr"]);
227 yield* cache.invalidate("pr");
228 for (let index = 0; index < 100; index++) yield* cache.invalidate(`pr-${index}`);
229 assert.strictEqual((yield* fs.readDirectory(directory)).length, 2);
230 const before = (yield* fs.stat(`${directory}/revisions`)).size;
231 yield* TestClock.adjust("59 seconds");
232 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
233 yield* TestClock.adjust("1 second");
234 yield* cache.invalidate("other-pr");
235 assert.isTrue((yield* fs.stat(`${directory}/revisions`)).size < before);
236 const restarted = yield* cacheLayer(directory);
237 assert.strictEqual(
238 yield* restarted.get("summary", Effect.die("cache miss"), ["pr"]),
239 "fresh",
240 );
241 }),
242 );
243
244 it.effect("does not persist failed GitHub reads", () =>
245 Effect.gen(function* () {
246 const fs = yield* FileSystem.FileSystem;
247 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
248 const cache = yield* cacheLayer(directory);
249 const error = new PullRequestOperationError({ operation: "summary", detail: "unavailable" });
250 yield* cache.get("summary", Effect.fail(error)).pipe(Effect.flip);
251 const restarted = yield* cacheLayer(directory);
252 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered");
253 }),
254 );
255
256 it.effect("resumes caching after a failed scope read", () =>
257 Effect.gen(function* () {
258 const backing = yield* KeyValueStore.KeyValueStore;
259 let fail = true;
260 let reads = 0;
261 const cache = yield* PullRequestReadCache.make.pipe(
262 Effect.provideService(KeyValueStore.KeyValueStore, {
263 ...backing,
264 get: (key) =>
265 Effect.suspend(() => {
266 if (!fail) return backing.get(key);
267 fail = false;
268 return Effect.fail(
269 new KeyValueStore.KeyValueStoreError({ method: "get", message: "unavailable" }),
270 );
271 }),
272 }),
273 );
274 const read = cache.get(
275 "summary",
276 Effect.sync(() => String(++reads)),
277 ["pr"],
278 );
279 assert.strictEqual(yield* read, "1");
280 assert.strictEqual(yield* read, "2");
281 assert.strictEqual(yield* read, "2");
282 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
283 );
284
285 it.effect("cancels abandoned reads without blocking invalidation", () =>
286 Effect.gen(function* () {
287 const cache = yield* PullRequestReadCache.make;
288 const started = yield* Deferred.make<void>();
289 const read = yield* cache
290 .get("summary", Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), [
291 "pr",
292 ])
293 .pipe(Effect.forkChild);
294 yield* Deferred.await(started);
295 yield* Fiber.interrupt(read);
296 yield* cache.invalidate("pr");
297 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
298 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
299 );
300
301 it.effect(
302 "finishes the in-memory revision update when invalidation is canceled after writing",
303 () =>
304 Effect.gen(function* () {
305 const backing = yield* KeyValueStore.KeyValueStore;
306 const written = yield* Deferred.make<void>();
307 const release = yield* Deferred.make<void>();
308 const cache = yield* PullRequestReadCache.make.pipe(
309 Effect.provideService(KeyValueStore.KeyValueStore, {
310 ...backing,
311 set: (key, value) =>
312 backing
313 .set(key, value)
314 .pipe(
315 Effect.andThen(
316 key === "revisions"
317 ? Deferred.succeed(written, undefined).pipe(
318 Effect.andThen(Deferred.await(release)),
319 )
320 : Effect.void,
321 ),
322 ),
323 }),
324 );
325 yield* cache.get("summary", Effect.succeed("old"), ["pr"]);
326 const invalidation = yield* cache.invalidate("pr").pipe(Effect.forkChild);
327 yield* Deferred.await(written);
328 const interrupt = yield* Fiber.interrupt(invalidation).pipe(
329 Effect.forkChild({ startImmediately: true }),
330 );
331 yield* Deferred.succeed(release, undefined);
332 yield* Fiber.join(interrupt);
333 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
334 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
335 );
336});