Open History to see the last commit
1import * as Cache from "effect/Cache";
2import * as Clock from "effect/Clock";
3import * as Equal from "effect/Equal";
4import * as Hash from "effect/Hash";
5import { PullRequestOperationError, PullRequestUnavailableError } from "@t3tools/contracts";
6import * as Digest from "effect/crypto/Digest";
7import * as SecureRandom from "effect/crypto/SecureRandom";
8import * as Hex from "effect/encoding/Hex";
9import * as Context from "effect/Context";
10import * as Duration from "effect/Duration";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import * as FileSystem from "effect/FileSystem";
14import * as Layer from "effect/Layer";
15import * as Option from "effect/Option";
16import * as Path from "effect/Path";
17import * as Schedule from "effect/Schedule";
18import * as Schema from "effect/Schema";
19import * as Semaphore from "effect/Semaphore";
20import * as KeyValueStore from "effect/persistence/KeyValueStore";
21import * as Persistable from "effect/persistence/Persistable";
22import * as PersistedCache from "effect/persistence/PersistedCache";
23import * as Persistence from "effect/persistence/Persistence";
24import * as ServerConfig from "../config.ts";
25import { detachStackFrame } from "../git/detachStackFrame.ts";
26import { forkParked } from "../serverActivation.ts";
27
28const CONCURRENT_READS = 512;
29/** Invalidation excludes only reads whose scopes share its stripe, not every PR read. */
30const SCOPE_STRIPES = 64;
31// Entry filenames are the URI-encoded `<storeId>:<key>`; the scope revisions file has
32// no store id. Effect before 4.1 wrote `<storeId><key>`, still pruned here.
33const STORE_ID = "pr-v2";
34const ENTRY_FILE_NAME = new RegExp(`^${STORE_ID}(?:%3A)?[0-9a-f]{64}$`);
35/**
36 * A file untouched for a day is long expired; the day only leaves slack for
37 * clock changes.
38 */
39const ENTRY_FILE_MAX_AGE = Duration.days(1);
40type ReadError = PullRequestOperationError | PullRequestUnavailableError;
41const revisionCodec = Schema.fromJsonString(
42 Schema.Record(
43 Schema.String,
44 Schema.Struct({ revision: Schema.String, expiresAt: Schema.Finite }),
45 ),
46);
47class Read extends Persistable.Class<{
48 payload: { key: string; revision: string; lookup: Effect.Effect<string, ReadError> };
49}>()("PullRequestRead", {
50 primaryKey: ({ key }) => key,
51 success: Schema.Struct({
52 payload: Schema.String,
53 expiresAt: Schema.Finite,
54 revision: Schema.optionalKey(Schema.String),
55 }),
56 error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),
57}) {
58 matchesRevision(revision: string | undefined): boolean {
59 const stored = revision?.split(":") ?? [];
60 return this.revision
61 .split(":")
62 .every((value, index) => value === "" || value === stored[index]);
63 }
64
65 [Equal.symbol](that: unknown): boolean {
66 return that instanceof Read && that.key === this.key && that.revision === this.revision;
67 }
68 [Hash.symbol](): number {
69 return Hash.string(`${this.key}:${this.revision}`);
70 }
71}
72
73export class PullRequestReadCache extends Context.Service<
74 PullRequestReadCache,
75 {
76 readonly get: (
77 key: string,
78 lookup: Effect.Effect<string, ReadError>,
79 scopes?: ReadonlyArray<string>,
80 ) => Effect.Effect<string, ReadError>;
81 readonly invalidate: (scope: string) => Effect.Effect<void>;
82 }
83>()("t3/pullRequest/PullRequestReadCache") {}
84
85export const make = Effect.gen(function* () {
86 const backing = yield* KeyValueStore.KeyValueStore;
87 const random = yield* SecureRandom.SecureRandom;
88 const digester = yield* Digest.Digest;
89 const clock = yield* Clock.Clock;
90 let enabled = true;
91 const lock = yield* Semaphore.make(CONCURRENT_READS);
92 const stripes = yield* Effect.forEach(Array.from({ length: SCOPE_STRIPES }), () =>
93 Semaphore.make(CONCURRENT_READS),
94 );
95 const stripeOf = (scope: string) => (Hash.string(scope) >>> 0) % SCOPE_STRIPES;
96 const revisionWrites = yield* Semaphore.make(1);
97 const digest = (key: string) =>
98 digester.digest(Digest.sha256, new TextEncoder().encode(key)).pipe(Effect.map(Hex.encode));
99 const revisions = yield* Cache.makeWith(
100 () =>
101 backing.get("revisions").pipe(
102 Effect.flatMap((raw) => Schema.decodeEffect(revisionCodec)(raw ?? "{}")),
103 Effect.catchTags({
104 SchemaError: Effect.fn("PullRequestReadCache.recoverRevisions")(function* () {
105 // Without scope revisions, persisted PR entries cannot be trusted.
106 yield* backing.clear;
107 yield* Effect.logWarning("PR cache reset after invalid scope revisions");
108 return yield* Schema.decodeEffect(revisionCodec)("{}");
109 }),
110 }),
111 ),
112 {
113 capacity: 1,
114 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
115 },
116 );
117 const timeToLive: Persistable.TimeToLiveFn<Read> = (exit) =>
118 Exit.isSuccess(exit)
119 ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))
120 : Duration.zero;
121 const cache = yield* PersistedCache.make(
122 (request: Read) =>
123 request.lookup.pipe(
124 Effect.map((payload) => ({
125 payload,
126 expiresAt: clock.currentTimeMillisUnsafe() + 60_000,
127 revision: request.revision,
128 })),
129 ),
130 {
131 storeId: STORE_ID,
132 timeToLive,
133 inMemoryTTL: timeToLive,
134 inMemoryCapacity: CONCURRENT_READS,
135 },
136 ).pipe(Effect.provide(Persistence.layerKvs));
137 const refreshes = yield* Cache.makeWith(
138 Effect.fn("PullRequestReadCache.refresh")(function* (request: Read) {
139 const stored = yield* cache.get(request);
140 if (request.matchesRevision(stored.revision)) return stored;
141 yield* cache.invalidate(request);
142 return yield* cache.get(request);
143 }),
144 { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero },
145 );
146 // A cache entry keeps the fiber that filled it, whose lazy stack frames would keep the caller's
147 // data (e.g. a whole sync sweep) alive until eviction.
148 return PullRequestReadCache.of({
149 get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) {
150 if (!enabled) return yield* lookup;
151 const read = yield* Effect.cached(lookup);
152 return yield* Effect.gen(function* () {
153 const current = yield* Cache.get(revisions, undefined);
154 const now = clock.currentTimeMillisUnsafe();
155 const revision = scopes
156 .map((scope) => {
157 const value = current[scope];
158 return value !== undefined && value.expiresAt > now ? value.revision : "";
159 })
160 .join(":");
161 const request = new Read({ key: yield* digest(key), revision, lookup: read });
162 const stored = yield* cache.get(request);
163 return (
164 request.matchesRevision(stored.revision) ? stored : yield* Cache.get(refreshes, request)
165 ).payload;
166 }).pipe(
167 Effect.catchTags({
168 KeyValueStoreError: () => read,
169 PersistenceError: () => read,
170 SchemaError: () => read,
171 }),
172 // A read stores its result under the revisions it started with, so an invalidation of
173 // one of its scopes must wait for it; reads of other scopes run on.
174 (effect) =>
175 [...new Set(scopes.map(stripeOf))]
176 .toSorted((left, right) => left - right)
177 .reduceRight((inner, stripe) => stripes[stripe]!.withPermits(1)(inner), effect),
178 lock.withPermits(1),
179 );
180 }, detachStackFrame),
181 invalidate: Effect.fn("PullRequestReadCache.invalidate")(
182 function* (scope) {
183 const now = clock.currentTimeMillisUnsafe();
184 const current = yield* Cache.get(revisions, undefined);
185 const next = Object.fromEntries(
186 Object.entries(current).filter(([, value]) => value.expiresAt > now),
187 );
188 next[scope] = { revision: yield* random.uuidV4(), expiresAt: now + 60_000 };
189 const encoded = yield* Schema.encodeEffect(revisionCodec)(next);
190 yield* backing.set("revisions", encoded);
191 yield* Cache.set(revisions, undefined, next);
192 },
193 Effect.catch(() => {
194 enabled = false;
195 return Effect.logWarning("PR cache disabled after clearing failed");
196 }),
197 Effect.uninterruptible,
198 revisionWrites.withPermits(1),
199 (effect, scope) => stripes[stripeOf(scope)]!.withPermits(CONCURRENT_READS)(effect),
200 detachStackFrame,
201 ),
202 });
203});
204
205/**
206 * The persisted cache drops an expired entry only when it is read again, so
207 * files for PRs nobody reopens would otherwise stay forever.
208 */
209export const pruneExpiredEntryFiles = Effect.fn("PullRequestReadCache.pruneExpiredEntryFiles")(
210 function* (directory: string) {
211 const fileSystem = yield* FileSystem.FileSystem;
212 const path = yield* Path.Path;
213 const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(ENTRY_FILE_MAX_AGE);
214 const entries = (yield* fileSystem.readDirectory(directory)).filter((name) =>
215 ENTRY_FILE_NAME.test(name),
216 );
217 const [, failures] = yield* Effect.partition(entries, (name) => {
218 const entryPath = path.join(directory, name);
219 return fileSystem.stat(entryPath).pipe(
220 Effect.flatMap((info) =>
221 Option.exists(info.mtime, (mtime) => mtime.getTime() < cutoff)
222 ? fileSystem.remove(entryPath)
223 : Effect.void,
224 ),
225 Effect.catchReason("PlatformError", "NotFound", () => Effect.void),
226 );
227 });
228 if (failures.length > 0) {
229 yield* Effect.logWarning("Failed to prune some PR cache files", {
230 failed: failures.length,
231 cause: failures[0],
232 });
233 }
234 },
235);
236
237export const layer = Layer.unwrap(
238 Effect.gen(function* () {
239 const config = yield* ServerConfig.ServerConfig;
240 const path = yield* Path.Path;
241 const directory = path.join(config.providerStatusCacheDir, "pull-requests");
242 return Layer.effect(PullRequestReadCache, make).pipe(
243 Layer.provide(
244 // Memoize cleanup acquisition with the store; Layer.tap runs per consumer.
245 Layer.effectDiscard(
246 forkParked(
247 pruneExpiredEntryFiles(directory).pipe(
248 Effect.catch((cause) =>
249 Effect.logWarning("Failed to prune PR cache files", { cause }),
250 ),
251 Effect.repeat(Schedule.spaced(Duration.hours(1))),
252 ),
253 ),
254 ).pipe(
255 Layer.provideMerge(KeyValueStore.layerFileSystem(directory)),
256 Layer.catch(() =>
257 Layer.effectDiscard(
258 Effect.logWarning("PR cache directory unavailable; using memory cache"),
259 ).pipe(Layer.provideMerge(KeyValueStore.layerMemory)),
260 ),
261 ),
262 ),
263 );
264 }),
265);