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 Crypto from "effect/Crypto";
7import * as Hex from "effect/encoding/Hex";
8import * as Context from "effect/Context";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Exit from "effect/Exit";
12import * as FileSystem from "effect/FileSystem";
13import * as Layer from "effect/Layer";
14import * as Option from "effect/Option";
15import * as Path from "effect/Path";
16import * as Schedule from "effect/Schedule";
17import * as Schema from "effect/Schema";
18import * as Semaphore from "effect/Semaphore";
19import * as KeyValueStore from "effect/persistence/KeyValueStore";
20import * as Persistable from "effect/persistence/Persistable";
21import * as PersistedCache from "effect/persistence/PersistedCache";
22import * as Persistence from "effect/persistence/Persistence";
23import * as ServerConfig from "../config.ts";
24import { forkParked } from "../serverActivation.ts";
25
26const CONCURRENT_READS = 512;
27// Persistence prefixes every entry key with the store id, so each entry file
28// name starts with it. The scope `revisions` file does not.
29const STORE_ID = "pr-v2";
30// An entry file is the store id followed by the SHA-256 hex digest of its key.
31const ENTRY_FILE_NAME = new RegExp(`^${STORE_ID}[0-9a-f]{64}$`);
32/**
33 * Entries expire a minute after they are written, and a write sets the file's
34 * mtime. A file untouched for a day is long expired; the day only leaves slack
35 * for clock changes. Pruning a live entry would cost one refetch.
36 */
37export const ENTRY_FILE_MAX_AGE = Duration.days(1);
38type ReadError = PullRequestOperationError | PullRequestUnavailableError;
39const revisionCodec = Schema.fromJsonString(
40 Schema.Record(
41 Schema.String,
42 Schema.Struct({ revision: Schema.String, expiresAt: Schema.Finite }),
43 ),
44);
45class Read extends Persistable.Class<{
46 payload: { key: string; revision: string; lookup: Effect.Effect<string, ReadError> };
47}>()("PullRequestRead", {
48 primaryKey: ({ key }) => key,
49 success: Schema.Struct({
50 payload: Schema.String,
51 expiresAt: Schema.Finite,
52 revision: Schema.optionalKey(Schema.String),
53 }),
54 error: Schema.Union([PullRequestOperationError, PullRequestUnavailableError]),
55}) {
56 matchesRevision(revision: string | undefined): boolean {
57 const stored = revision?.split(":") ?? [];
58 return this.revision
59 .split(":")
60 .every((value, index) => value === "" || value === stored[index]);
61 }
62
63 [Equal.symbol](that: unknown): boolean {
64 return that instanceof Read && that.key === this.key && that.revision === this.revision;
65 }
66 [Hash.symbol](): number {
67 return Hash.string(`${this.key}:${this.revision}`);
68 }
69}
70
71export class PullRequestReadCache extends Context.Service<
72 PullRequestReadCache,
73 {
74 readonly get: (
75 key: string,
76 lookup: Effect.Effect<string, ReadError>,
77 scopes?: ReadonlyArray<string>,
78 ) => Effect.Effect<string, ReadError>;
79 readonly invalidate: (scope: string) => Effect.Effect<void>;
80 }
81>()("t3/pullRequest/PullRequestReadCache") {}
82
83export const make = Effect.gen(function* () {
84 const backing = yield* KeyValueStore.KeyValueStore;
85 const crypto = yield* Crypto.Crypto;
86 const clock = yield* Clock.Clock;
87 let enabled = true;
88 const lock = yield* Semaphore.make(CONCURRENT_READS);
89 const digest = (key: string) =>
90 crypto.digest("SHA-256", new TextEncoder().encode(key)).pipe(Effect.map(Hex.encode));
91 const revisions = yield* Cache.makeWith(
92 () =>
93 backing
94 .get("revisions")
95 .pipe(Effect.flatMap((raw) => Schema.decodeEffect(revisionCodec)(raw ?? "{}"))),
96 {
97 capacity: 1,
98 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
99 },
100 );
101 const timeToLive: Persistable.TimeToLiveFn<Read> = (exit) =>
102 Exit.isSuccess(exit)
103 ? Duration.millis(Math.max(0, exit.value.expiresAt - clock.currentTimeMillisUnsafe()))
104 : Duration.zero;
105 const cache = yield* PersistedCache.make(
106 (request: Read) =>
107 request.lookup.pipe(
108 Effect.map((payload) => ({
109 payload,
110 expiresAt: clock.currentTimeMillisUnsafe() + 60_000,
111 revision: request.revision,
112 })),
113 ),
114 {
115 storeId: STORE_ID,
116 timeToLive,
117 inMemoryTTL: timeToLive,
118 inMemoryCapacity: CONCURRENT_READS,
119 },
120 ).pipe(Effect.provide(Persistence.layerKvs));
121 const refreshes = yield* Cache.makeWith(
122 Effect.fn("PullRequestReadCache.refresh")(function* (request: Read) {
123 const stored = yield* cache.get(request);
124 if (request.matchesRevision(stored.revision)) return stored;
125 yield* cache.invalidate(request);
126 return yield* cache.get(request);
127 }),
128 { capacity: CONCURRENT_READS, timeToLive: () => Duration.zero },
129 );
130 return PullRequestReadCache.of({
131 get: Effect.fn("PullRequestReadCache.get")(function* (key, lookup, scopes = []) {
132 if (!enabled) return yield* lookup;
133 const read = yield* Effect.cached(lookup);
134 return yield* Effect.gen(function* () {
135 const current = yield* Cache.get(revisions, undefined);
136 const now = clock.currentTimeMillisUnsafe();
137 const revision = scopes
138 .map((scope) => {
139 const value = current[scope];
140 return value !== undefined && value.expiresAt > now ? value.revision : "";
141 })
142 .join(":");
143 const request = new Read({ key: yield* digest(key), revision, lookup: read });
144 const stored = yield* cache.get(request);
145 return (
146 request.matchesRevision(stored.revision) ? stored : yield* Cache.get(refreshes, request)
147 ).payload;
148 }).pipe(
149 Effect.catchTags({
150 PlatformError: () => read,
151 KeyValueStoreError: () => read,
152 PersistenceError: () => read,
153 SchemaError: () => read,
154 }),
155 lock.withPermits(1),
156 );
157 }),
158 invalidate: (scope) =>
159 Effect.gen(function* () {
160 const now = clock.currentTimeMillisUnsafe();
161 const current = yield* Cache.get(revisions, undefined);
162 const next = Object.fromEntries(
163 Object.entries(current).filter(([, value]) => value.expiresAt > now),
164 );
165 next[scope] = { revision: yield* crypto.randomUUIDv4(), expiresAt: now + 60_000 };
166 const encoded = yield* Schema.encodeEffect(revisionCodec)(next);
167 yield* backing.set("revisions", encoded);
168 yield* Cache.set(revisions, undefined, next);
169 }).pipe(
170 Effect.catch(() => {
171 enabled = false;
172 return Effect.logWarning("PR cache disabled after clearing failed");
173 }),
174 Effect.uninterruptible,
175 lock.withPermits(CONCURRENT_READS),
176 ),
177 });
178});
179
180/**
181 * Deletes entry files in `directory` not written within `ENTRY_FILE_MAX_AGE`.
182 * The persisted cache drops an expired entry only when it is read again, so
183 * files for PRs nobody reopens would otherwise stay forever.
184 */
185export const pruneExpiredEntryFiles = Effect.fn("PullRequestReadCache.pruneExpiredEntryFiles")(
186 function* (directory: string) {
187 const fileSystem = yield* FileSystem.FileSystem;
188 const path = yield* Path.Path;
189 const cutoff = (yield* Clock.currentTimeMillis) - Duration.toMillis(ENTRY_FILE_MAX_AGE);
190 const entries = (yield* fileSystem.readDirectory(directory)).filter((name) =>
191 ENTRY_FILE_NAME.test(name),
192 );
193 // One file at a time, and `partition` visits every file, so one locked file
194 // does not stop the sweep.
195 const [, failures] = yield* Effect.partition(entries, (name) => {
196 const entryPath = path.join(directory, name);
197 return fileSystem.stat(entryPath).pipe(
198 Effect.flatMap((info) =>
199 Option.exists(info.mtime, (mtime) => mtime.getTime() < cutoff)
200 ? fileSystem.remove(entryPath)
201 : Effect.void,
202 ),
203 Effect.catchReason("PlatformError", "NotFound", () => Effect.void),
204 );
205 });
206 if (failures.length > 0) {
207 yield* Effect.logWarning("Failed to prune some PR cache files", {
208 failed: failures.length,
209 cause: failures[0],
210 });
211 }
212 },
213);
214
215export const layer = Layer.unwrap(
216 Effect.gen(function* () {
217 const config = yield* ServerConfig.ServerConfig;
218 const path = yield* Path.Path;
219 const directory = path.join(config.providerStatusCacheDir, "pull-requests");
220 return Layer.effect(PullRequestReadCache, make).pipe(
221 Layer.provide(
222 KeyValueStore.layerFileSystem(directory).pipe(
223 // Prunes once the server is active, then every hour.
224 Layer.tap(() =>
225 forkParked(
226 pruneExpiredEntryFiles(directory).pipe(
227 Effect.catch((cause) =>
228 Effect.logWarning("Failed to prune PR cache files", { cause }),
229 ),
230 Effect.repeat(Schedule.spaced(Duration.hours(1))),
231 ),
232 ),
233 ),
234 Layer.catch(() =>
235 Layer.effectDiscard(
236 Effect.logWarning("PR cache directory unavailable; using memory cache"),
237 ).pipe(Layer.provideMerge(KeyValueStore.layerMemory)),
238 ),
239 ),
240 ),
241 );
242 }),
243);