1import * as NodeV8 from "node:v8";
2import { assert, it } from "@effect/vitest";
3import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import * as Layer from "effect/Layer";
6import { PullRequestOperationError } from "@t3tools/contracts";
7import * as Deferred from "effect/Deferred";
8import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as FileSystem from "effect/FileSystem";
12import * as Option from "effect/Option";
13import * as Path from "effect/Path";
14import * as KeyValueStore from "effect/persistence/KeyValueStore";
15import * as TestClock from "effect/testing/TestClock";
16import * as PullRequestReadCache from "./PullRequestReadCache.ts";
17
18const cacheLayer = (directory: string) =>
19 PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));
20
21it.layer(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))("PR filesystem cache", (it) => {
22 it.effect("clears in-flight reads before a new service can reuse them", () =>
23 Effect.gen(function* () {
24 const fs = yield* FileSystem.FileSystem;
25 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
26 const started = yield* Deferred.make<void>();
27 const release = yield* Deferred.make<void>();
28 const cache = yield* cacheLayer(directory);
29 const read = yield* cache
30 .get(
31 "summary",
32 Deferred.succeed(started, undefined).pipe(
33 Effect.andThen(Deferred.await(release)),
34 Effect.as("old"),
35 ),
36 ["pr"],
37 )
38 .pipe(Effect.forkChild);
39 yield* Deferred.await(started);
40 const invalidate = yield* cache
41 .invalidate("pr")
42 .pipe(Effect.forkChild({ startImmediately: true }));
43 // Another PR's invalidation does not wait on this read; this one's does.
44 yield* cache.invalidate("other-pr");
45 assert.isUndefined(invalidate.pollUnsafe());
46 yield* Deferred.succeed(release, undefined);
47 yield* Fiber.join(read);
48 yield* Fiber.join(invalidate);
49 const restarted = yield* cacheLayer(directory);
50 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("new"), ["pr"]), "new");
51 }),
52 );
53
54 it.effect("prunes expired entry files the store wrote", () =>
55 Effect.gen(function* () {
56 const fs = yield* FileSystem.FileSystem;
57 const path = yield* Path.Path;
58 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
59 const cache = yield* cacheLayer(directory);
60 yield* cache.get("summary", Effect.succeed("value"));
61 const [entry] = yield* fs.readDirectory(directory);
62 assert.isDefined(entry);
63 const info = yield* fs.stat(path.join(directory, entry));
64 yield* TestClock.setTime(
65 Option.getOrThrow(info.mtime).getTime() + Duration.toMillis(Duration.days(2)),
66 );
67 yield* PullRequestReadCache.pruneExpiredEntryFiles(directory);
68 assert.deepStrictEqual(yield* fs.readDirectory(directory), []);
69 }),
70 );
71
72 it.effect("invalidates only the changed scope across restarts and coalesces its next reads", () =>
73 Effect.gen(function* () {
74 const fs = yield* FileSystem.FileSystem;
75 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
76 let reads = 0;
77 const lookup = Effect.sync(() => String(++reads));
78 const cache = yield* cacheLayer(directory);
79 yield* cache.get("first", lookup, ["project", "pr-1"]);
80 yield* cache.get("second", lookup, ["project", "pr-2"]);
81 yield* cache.get("third", lookup, ["other-project", "pr-3"]);
82 yield* cache.invalidate("pr-1");
83 const restarted = yield* cacheLayer(directory);
84 const answers = yield* Effect.all(
85 Array.from({ length: 10 }, () => restarted.get("first", lookup, ["project", "pr-1"])),
86 { concurrency: 10 },
87 );
88 assert.deepStrictEqual(answers, Array(10).fill("4"));
89 assert.strictEqual(yield* restarted.get("second", lookup, ["project", "pr-2"]), "2");
90 yield* restarted.invalidate("project");
91 const again = yield* cacheLayer(directory);
92 assert.strictEqual(yield* again.get("second", lookup, ["project", "pr-2"]), "5");
93 assert.strictEqual(yield* again.get("third", lookup, ["other-project", "pr-3"]), "3");
94 assert.strictEqual(reads, 5);
95 const files = (yield* fs.readDirectory(directory)).length;
96 for (let index = 0; index < 3; index++) {
97 yield* again.invalidate("pr-1");
98 yield* again.get("first", lookup, ["project", "pr-1"]);
99 }
100 assert.strictEqual((yield* fs.readDirectory(directory)).length, files);
101 }),
102 );
103
104 it.effect("shares a pending refresh without blocking an unrelated cached PR", () =>
105 Effect.gen(function* () {
106 const cache = yield* PullRequestReadCache.make;
107 yield* cache.get("first", Effect.succeed("old"), ["pr-1"]);
108 yield* cache.get("second", Effect.succeed("warm"), ["pr-2"]);
109 yield* cache.invalidate("pr-1");
110 const started = yield* Deferred.make<void>();
111 const release = yield* Deferred.make<void>();
112 let reads = 0;
113 const refresh = cache.get(
114 "first",
115 Effect.gen(function* () {
116 reads++;
117 yield* Deferred.succeed(started, undefined);
118 yield* Deferred.await(release);
119 return "fresh";
120 }),
121 ["pr-1"],
122 );
123 const pending = yield* Effect.all(
124 Array.from({ length: 10 }, () => refresh),
125 {
126 concurrency: 10,
127 },
128 ).pipe(Effect.forkChild);
129 yield* Deferred.await(started);
130 assert.strictEqual(yield* cache.get("second", Effect.die("cache miss"), ["pr-2"]), "warm");
131 yield* Deferred.succeed(release, undefined);
132 assert.deepStrictEqual(yield* Fiber.join(pending), Array(10).fill("fresh"));
133 assert.strictEqual(reads, 1);
134 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
135 );
136
137 it.effect("does not persist failed GitHub reads", () =>
138 Effect.gen(function* () {
139 const fs = yield* FileSystem.FileSystem;
140 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
141 const cache = yield* cacheLayer(directory);
142 const error = new PullRequestOperationError({ operation: "summary", detail: "unavailable" });
143 yield* cache.get("summary", Effect.fail(error)).pipe(Effect.flip);
144 const restarted = yield* cacheLayer(directory);
145 assert.strictEqual(yield* restarted.get("summary", Effect.succeed("recovered")), "recovered");
146 }),
147 );
148
149 it.effect.each(["", "{", '{"pr":{"revision":42}}'])(
150 "recovers from malformed scope revisions without reusing stale entries: %j",
151 (corrupt) =>
152 Effect.gen(function* () {
153 const fs = yield* FileSystem.FileSystem;
154 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
155 const original = yield* cacheLayer(directory);
156 yield* original.get("summary", Effect.succeed("stale"), ["pr"]);
157 yield* fs.writeFileString(directory + "/revisions", corrupt);
158 const recovered = yield* cacheLayer(directory);
159 let reads = 0;
160 const lookup = Effect.sync(() => String(++reads));
161 assert.strictEqual(yield* recovered.get("summary", lookup, ["pr"]), "1");
162 assert.strictEqual(yield* recovered.get("summary", lookup, ["pr"]), "1");
163 yield* recovered.invalidate("pr");
164 assert.strictEqual(yield* recovered.get("summary", lookup, ["pr"]), "2");
165 const restarted = yield* cacheLayer(directory);
166 assert.strictEqual(yield* restarted.get("summary", lookup, ["pr"]), "2");
167 assert.strictEqual(reads, 2);
168 }),
169 );
170 it.effect("resumes caching after a failed scope read", () =>
171 Effect.gen(function* () {
172 const backing = yield* KeyValueStore.KeyValueStore;
173 let fail = true;
174 let reads = 0;
175 const cache = yield* PullRequestReadCache.make.pipe(
176 Effect.provideService(KeyValueStore.KeyValueStore, {
177 ...backing,
178 get: (key) =>
179 Effect.suspend(() => {
180 if (!fail) return backing.get(key);
181 fail = false;
182 return Effect.fail(
183 new KeyValueStore.KeyValueStoreError({ method: "get", message: "unavailable" }),
184 );
185 }),
186 }),
187 );
188 const read = cache.get(
189 "summary",
190 Effect.sync(() => String(++reads)),
191 ["pr"],
192 );
193 assert.strictEqual(yield* read, "1");
194 assert.strictEqual(yield* read, "2");
195 assert.strictEqual(yield* read, "2");
196 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
197 );
198
199 it.effect("does not keep a caller's data alive after its reads and invalidations", () =>
200 Effect.gen(function* () {
201 const cache = yield* PullRequestReadCache.make;
202 class SweepData extends Map<string, number> {}
203 // Like a sync sweep: an Effect.fn made per run that closes over that run's data.
204 const sweep = Effect.fn("sweep")(function* (index: number) {
205 const groups = new SweepData(Array.from({ length: 1_000 }, (_, id) => [`group-${id}`, id]));
206 const read = Effect.fn("read")(function* (pr: string) {
207 const size = groups.size;
208 yield* cache.invalidate(pr);
209 const answer = yield* cache.get(`${pr}-${index}`, Effect.succeed("answer"), [pr]);
210 return size + answer.length;
211 });
212 yield* Effect.forEach(["pr-1", "pr-2"], read, { discard: true });
213 });
214 for (let index = 0; index < 5; index++) yield* sweep(index);
215 // Counts after a full GC. A WeakRef would keep its target alive until the job ends.
216 assert.strictEqual(NodeV8.queryObjects(SweepData, { format: "count" }), 0);
217 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
218 );
219
220 it.effect("cancels abandoned reads without blocking invalidation", () =>
221 Effect.gen(function* () {
222 const cache = yield* PullRequestReadCache.make;
223 const started = yield* Deferred.make<void>();
224 const read = yield* cache
225 .get("summary", Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), [
226 "pr",
227 ])
228 .pipe(Effect.forkChild);
229 yield* Deferred.await(started);
230 yield* Fiber.interrupt(read);
231 yield* cache.invalidate("pr");
232 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
233 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
234 );
235
236 it.effect(
237 "finishes the in-memory revision update when invalidation is canceled after writing",
238 () =>
239 Effect.gen(function* () {
240 const backing = yield* KeyValueStore.KeyValueStore;
241 const written = yield* Deferred.make<void>();
242 const release = yield* Deferred.make<void>();
243 const cache = yield* PullRequestReadCache.make.pipe(
244 Effect.provideService(KeyValueStore.KeyValueStore, {
245 ...backing,
246 set: (key, value) =>
247 backing
248 .set(key, value)
249 .pipe(
250 Effect.andThen(
251 key === "revisions"
252 ? Deferred.succeed(written, undefined).pipe(
253 Effect.andThen(Deferred.await(release)),
254 )
255 : Effect.void,
256 ),
257 ),
258 }),
259 );
260 yield* cache.get("summary", Effect.succeed("old"), ["pr"]);
261 const invalidation = yield* cache.invalidate("pr").pipe(Effect.forkChild);
262 yield* Deferred.await(written);
263 const interrupt = yield* Fiber.interrupt(invalidation).pipe(
264 Effect.forkChild({ startImmediately: true }),
265 );
266 yield* Deferred.succeed(release, undefined);
267 yield* Fiber.join(interrupt);
268 assert.strictEqual(yield* cache.get("summary", Effect.succeed("fresh"), ["pr"]), "fresh");
269 }).pipe(Effect.provide(KeyValueStore.layerMemory)),
270 );
271});