1/**
2 * The marks a reader has ticked off, for a host that keeps none of its own, and the held record of
3 * what the head has of those files. The revisions cache is filed here because this is its only
4 * consumer: when a second one appears, export `makeFileRevisions` and split it into its own file.
5 */
6import * as Clock from "effect/Clock";
7import * as DateTime from "effect/DateTime";
8import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Semaphore from "effect/Semaphore";
11import {
12 PullRequestOperationError,
13 type PullRequestFilesViewedResult,
14 type PullRequestRef,
15 type PullRequestSetFilesViewedInput,
16} from "@t3tools/contracts";
17
18import type * as PullRequestFilesViewed from "../persistence/PullRequestFilesViewed.ts";
19import type {
20 ProviderFileRevisions,
21 PullRequestProviderError,
22} from "@t3tools/source-control-core/server/PullRequestProvider";
23import type { PullRequestError, SupportedProject } from "./PullRequestService.ts";
24
25/**
26 * How long the head's version of a file is believed, and how long a held answer stands while the
27 * next one is fetched. This is the host call behind the **Changed** badge alone, so a held answer
28 * costs a badge that is a minute behind rather than a stale tick.
29 */
30const FILE_REVISIONS_CACHE_TTL = Duration.seconds(60);
31const FILE_REVISIONS_STALE_WINDOW = Duration.minutes(10);
32export const FILE_REVISIONS_CACHE_CAPACITY = 64;
33
34/**
35 * How many paths one scope's entry carries. Bounds a single scope, not how many scopes are held:
36 * a reader ticking one file after another renews the same scope and grows it without limit
37 * otherwise.
38 */
39export const MAX_FILE_REVISION_PATHS = 1_000;
40
41interface FileRevisionsDependencies {
42 readonly runFork: (effect: Effect.Effect<void>) => unknown;
43 readonly refEpoch: (ref: PullRequestRef) => number;
44 readonly fileRevisionsEpoch: () => number;
45 readonly toPullRequestError: (
46 operation: string,
47 ) => (error: PullRequestProviderError) => PullRequestError;
48}
49
50const makeFileRevisions = (dependencies: FileRevisionsDependencies) => {
51 const { runFork, refEpoch, fileRevisionsEpoch, toPullRequestError } = dependencies;
52 /**
53 * What the head has of the files a reader has marked, held between reads. The entry tracks what
54 * has been asked as well as what was heard, since a path missing from an answer keeps whatever
55 * version was last given for it rather than being cleared.
56 */
57 interface HeldFileRevisions {
58 readonly at: number;
59 readonly asked: ReadonlySet<string>;
60 readonly revisions: ReadonlyMap<string, string>;
61 }
62 const heldFileRevisions = new Map<string, HeldFileRevisions>();
63 const refreshingFileRevisions = new Set<string>();
64
65 /**
66 * Carries the reference's epoch, so whatever moved the head strands what was held (or in
67 * flight) against the old one. Spelled from the project rather than the reference, since a
68 * reference arrives however the client spelled it while the epoch is bumped against the
69 * remote's own spelling.
70 */
71 const fileRevisionsKey = (project: SupportedProject, ref: PullRequestRef) =>
72 [
73 refEpoch({ ...ref, host: project.host, repository: project.repository }),
74 fileRevisionsEpoch(),
75 ref.projectId,
76 project.repository.trim().toLowerCase(),
77 ref.number,
78 ].join(" ");
79
80 /**
81 * `paths` are what was asked about, and are held as answered for whether the host had a version
82 * for them or not: that is what stops the same question being asked again. A `complete` answer
83 * adds every other path it carries, since a host that read the whole change to answer for one
84 * file has already paid for all of them, and the tick after this one names a file nothing has
85 * asked about yet.
86 */
87 const recordFileRevisions = (
88 key: string,
89 paths: ReadonlyArray<string>,
90 answer: ProviderFileRevisions,
91 ) =>
92 Effect.map(Clock.currentTimeMillis, (at) => {
93 const held = heldFileRevisions.get(key);
94 // Past the stale window the old entry is not worth merging into: it would carry paths
95 // nobody has asked about since, at revisions the head has long moved off.
96 const carried =
97 held !== undefined && at - held.at <= Duration.toMillis(FILE_REVISIONS_STALE_WINDOW)
98 ? held
99 : null;
100 const revisions = new Map(carried?.revisions ?? []);
101 const asked = new Set(carried?.asked ?? []);
102 // The paths asked for go last, so a whole-change answer wider than the cap is trimmed down
103 // to the reader's own files rather than over them.
104 const learned = answer.complete === true ? [...answer.revisions.keys(), ...paths] : paths;
105 for (const path of learned) {
106 // Reinserted rather than added, so what a full entry drops below is the path nobody has
107 // asked about in the longest rather than one just asked for.
108 asked.delete(path);
109 asked.add(path);
110 const revision = answer.revisions.get(path);
111 // A path left out of the answer keeps its last known version rather than being cleared.
112 if (revision !== undefined) {
113 revisions.delete(path);
114 revisions.set(path, revision);
115 }
116 }
117 for (const path of asked) {
118 if (asked.size <= MAX_FILE_REVISION_PATHS) break;
119 asked.delete(path);
120 revisions.delete(path);
121 }
122 heldFileRevisions.delete(key);
123 if (heldFileRevisions.size >= FILE_REVISIONS_CACHE_CAPACITY) {
124 const oldest = heldFileRevisions.keys().next().value;
125 if (oldest !== undefined) heldFileRevisions.delete(oldest);
126 }
127 // The entry is only as fresh as its oldest revision: stamping it with `now` on a partial
128 // answer would let an old revision ride past the point it should have been re-read.
129 const stamped = [...revisions.keys()].every((path) => answer.revisions.has(path))
130 ? at
131 : (carried?.at ?? at);
132 heldFileRevisions.set(key, { at: stamped, asked, revisions });
133 return revisions;
134 });
135
136 /** A held entry that covers every path asked for and is still worth answering from. */
137 const heldFileRevisionsFor = (key: string, paths: ReadonlyArray<string>, now: number) => {
138 const held = heldFileRevisions.get(key);
139 if (held === undefined) return null;
140 // Put back at the end on every read, so the scope a reader is working through is not the one
141 // evicted by a listing walking scopes nobody has open.
142 heldFileRevisions.delete(key);
143 heldFileRevisions.set(key, held);
144 if (now - held.at > Duration.toMillis(FILE_REVISIONS_STALE_WINDOW)) return null;
145 return paths.every((path) => held.asked.has(path)) ? held : null;
146 };
147
148 /**
149 * What the head has of these files, or null where the host cannot say (not an error: the marks
150 * just stop reporting staleness). `held` answers from a stale value and refetches off the
151 * critical path, since a badge a moment behind beats a page that won't paint until the host
152 * answers; `fresh` is for the press itself, which must not store a revision the head already
153 * moved off.
154 */
155 const fileRevisionsOf = (
156 project: SupportedProject,
157 ref: PullRequestRef,
158 paths: ReadonlyArray<string>,
159 operation: string,
160 freshness: "held" | "fresh" = "held",
161 ): Effect.Effect<ReadonlyMap<string, string> | null, PullRequestError> => {
162 const read = project.api.getFileRevisions;
163 if (read === undefined) return Effect.succeed(null);
164 // Suspended, so a held answer costs the host nothing: a provider is free to do its work as
165 // the request is built rather than as the effect is run.
166 const fetch = Effect.suspend(() => {
167 const key = fileRevisionsKey(project, ref);
168 return read({
169 cwd: project.project.workspaceRoot,
170 repository: project.repository,
171 host: project.host,
172 number: ref.number,
173 paths,
174 }).pipe(
175 Effect.mapError(toPullRequestError(operation)),
176 Effect.flatMap((answer) => recordFileRevisions(key, paths, answer)),
177 );
178 });
179 return Effect.flatMap(Clock.currentTimeMillis, (now) => {
180 const key = fileRevisionsKey(project, ref);
181 const held = heldFileRevisionsFor(key, paths, now);
182 if (held === null) return fetch;
183 if (now - held.at <= Duration.toMillis(FILE_REVISIONS_CACHE_TTL))
184 return Effect.succeed(held.revisions);
185 if (freshness === "fresh") return fetch;
186 if (refreshingFileRevisions.has(key)) return Effect.succeed(held.revisions);
187 // Its own fiber rather than a child: the caller has been answered and is gone before this
188 // lands. One at a time per change request, so a page of files costs one host read.
189 return Effect.sync(() => {
190 refreshingFileRevisions.add(key);
191 runFork(
192 Effect.ignore(fetch).pipe(
193 Effect.ensuring(Effect.sync(() => refreshingFileRevisions.delete(key))),
194 ),
195 );
196 }).pipe(Effect.as(held.revisions));
197 });
198 };
199
200 return { fileRevisionsOf } as const;
201};
202
203export interface Dependencies extends FileRevisionsDependencies {
204 readonly filesViewedStore: PullRequestFilesViewed.PullRequestFilesViewedRepository["Service"];
205 readonly requireProject: (
206 ref: PullRequestRef,
207 ) => Effect.Effect<SupportedProject, PullRequestError>;
208 readonly requiredViewerOf: (
209 project: SupportedProject,
210 operation: string,
211 ) => Effect.Effect<string | null, PullRequestError>;
212}
213
214// A plain factory rather than a `Context.Service` (against the preference in
215// `.repos/effect-smol/LLMS.md`): the held revisions, refresh set, and write gates are only correct
216// at one instance per service, and a layer provided at two points would give two of each behind
217// one epoch counter.
218export const make = (dependencies: Dependencies) => {
219 const { filesViewedStore, requireProject, requiredViewerOf, toPullRequestError } = dependencies;
220 const { fileRevisionsOf } = makeFileRevisions(dependencies);
221 /**
222 * Which change request's marks, and whose. Provider and host lead the key because the same
223 * repository can exist on more than one install; the reader is part of it because a host's own
224 * record is per-account. A host that names no reader is one reader, not none.
225 */
226 const filesViewedScope = (project: SupportedProject, number: number, viewer: string | null) => ({
227 provider: project.api.kind,
228 host: project.host,
229 repository: project.remote,
230 number,
231 viewer: viewer ?? "",
232 });
233
234 const toFilesViewedStoreError = (operation: string) => (cause: unknown) =>
235 new PullRequestOperationError({
236 operation,
237 detail: "This environment could not reach its record of which files you have seen.",
238 cause,
239 });
240
241 /**
242 * The marks this environment keeps for a host that keeps none of its own. A file the head
243 * still has at the revision it was cleared at is cleared; one the head has moved on from is
244 * reported as changed. Revisions are asked for the marked paths alone, so a reader who has
245 * marked nothing costs no host call.
246 */
247 const environmentFilesViewed = (
248 project: SupportedProject,
249 ref: PullRequestRef,
250 ): Effect.Effect<PullRequestFilesViewedResult, PullRequestError> =>
251 Effect.gen(function* () {
252 const viewer = yield* requiredViewerOf(project, "filesViewed");
253 const held = yield* filesViewedStore
254 .list(filesViewedScope(project, ref.number, viewer))
255 .pipe(Effect.mapError(toFilesViewedStoreError("filesViewed")));
256 const marks = held.files;
257 if (marks.length === 0) return { files: [], truncated: held.truncated };
258 // A host that won't say what its head has costs the marks their staleness (`fileRevisionsOf`
259 // returns null), rather than costing the reader every tick they've made.
260 const revisions = yield* fileRevisionsOf(
261 project,
262 ref,
263 marks.map((mark) => mark.path),
264 "filesViewed",
265 ).pipe(
266 Effect.catch((error) =>
267 Effect.logWarning("reporting viewed files without what the head has of them", {
268 operation: "filesViewed",
269 reason: error._tag,
270 }).pipe(Effect.as(null)),
271 ),
272 );
273 return {
274 files: marks.map((mark) => {
275 // A mark stamped with no baseline holds until the reader presses it again.
276 if (mark.revision === null) return { path: mark.path, state: "viewed" as const };
277 const revision = revisions?.get(mark.path);
278 // A deleted file is answered as the empty revision, matching its stamp, so it stays
279 // cleared; a path the host had no answer for (`undefined`) also holds as cleared.
280 return {
281 path: mark.path,
282 state:
283 revision === undefined || revision === mark.revision
284 ? ("viewed" as const)
285 : ("dismissed" as const),
286 };
287 }),
288 // The store caps marks per scope; a reader over that cap is told so, like a paginated read.
289 truncated: held.truncated,
290 };
291 });
292
293 /**
294 * One environment-backed write at a time per change request. A tick's host round trip is
295 * slower than an untick's, so unordered presses could finish out of order and leave a stale
296 * tick standing over a later untick.
297 */
298 const filesViewedGates = new Map<
299 string,
300 { readonly gate: Semaphore.Semaphore; pending: number }
301 >();
302
303 const inFilesViewedOrder = (
304 project: SupportedProject,
305 number: number,
306 write: Effect.Effect<void, PullRequestError>,
307 ) =>
308 // Suspended rather than generated, so finding the gate, putting it in and taking a place in
309 // its queue are one step: yielding for `Semaphore.make` between the lookup and the insert
310 // lets two presses each make a gate of their own and neither wait on the other.
311 Effect.suspend(() => {
312 const key = `${project.project.id} ${project.remote} ${number}`;
313 const held = filesViewedGates.get(key);
314 const entry = held ?? { gate: Semaphore.makeUnsafe(1), pending: 0 };
315 if (held === undefined) filesViewedGates.set(key, entry);
316 entry.pending += 1;
317 // Dropped once nobody is queued behind it, so a long-lived server does not keep a gate per
318 // change request anyone has ever ticked a file in.
319 return entry.gate
320 .withPermits(1)(write)
321 .pipe(
322 Effect.ensuring(
323 Effect.sync(() => {
324 entry.pending -= 1;
325 if (entry.pending === 0) filesViewedGates.delete(key);
326 }),
327 ),
328 );
329 });
330
331 const environmentSetFilesViewed = (
332 project: SupportedProject,
333 input: PullRequestSetFilesViewedInput,
334 ): Effect.Effect<void, PullRequestError> =>
335 Effect.gen(function* () {
336 const viewer = yield* requiredViewerOf(project, "setFilesViewed");
337 // Only the files being cleared need a revision. An unticked one is about to lose its row,
338 // and what the head has of it changes nothing about deleting it.
339 const cleared = input.files.filter((file) => file.viewed).map((file) => file.path);
340 const revisions =
341 cleared.length === 0
342 ? null
343 : yield* fileRevisionsOf(project, input, cleared, "setFilesViewed", "fresh").pipe(
344 // A host that won't say what its head has costs the press its baseline, not the
345 // press itself: the mark is stored with none and holds until pressed again.
346 Effect.catch((error) =>
347 Effect.logWarning("recording viewed files without what the head has of them", {
348 operation: "setFilesViewed",
349 reason: error._tag,
350 }).pipe(Effect.as(null)),
351 ),
352 );
353 const viewedAt = DateTime.formatIso(yield* DateTime.now);
354 yield* filesViewedStore
355 .set({
356 ...filesViewedScope(project, input.number, viewer),
357 // A path left out of the answer stores with no baseline, not the empty revision, since
358 // the empty revision is itself an answer and would misreport the file once it turns
359 // out to have a version after all.
360 files: input.files.map((file) => ({
361 path: file.path,
362 revision: revisions?.get(file.path) ?? null,
363 viewed: file.viewed,
364 })),
365 viewedAt,
366 })
367 .pipe(Effect.mapError(toFilesViewedStoreError("setFilesViewed")));
368 });
369
370 const filesViewed = (input: PullRequestRef) =>
371 requireProject(input).pipe(
372 Effect.flatMap((project): Effect.Effect<PullRequestFilesViewedResult, PullRequestError> => {
373 const read = project.api.getFilesViewed;
374 if (project.api.capabilities.viewedFiles === "host" && read) {
375 return read({
376 cwd: project.project.workspaceRoot,
377 repository: project.repository,
378 host: project.host,
379 number: input.number,
380 }).pipe(Effect.mapError(toPullRequestError("filesViewed")));
381 }
382 if (project.api.capabilities.viewedFiles === "environment") {
383 return environmentFilesViewed(project, input);
384 }
385 return Effect.fail(
386 new PullRequestOperationError({
387 operation: "filesViewed",
388 detail: "This host does not track which files a reader has seen.",
389 }),
390 );
391 }),
392 );
393
394 const setFilesViewed = (
395 input: PullRequestSetFilesViewedInput,
396 ): Effect.Effect<void, PullRequestError> =>
397 requireProject(input).pipe(
398 Effect.flatMap((project): Effect.Effect<void, PullRequestError> => {
399 const write = project.api.setFilesViewed;
400 if (project.api.capabilities.viewedFiles === "host" && write) {
401 return write({
402 cwd: project.project.workspaceRoot,
403 repository: project.repository,
404 host: project.host,
405 number: input.number,
406 files: input.files,
407 }).pipe(Effect.mapError(toPullRequestError("setFilesViewed")));
408 }
409 if (project.api.capabilities.viewedFiles === "environment") {
410 return inFilesViewedOrder(
411 project,
412 input.number,
413 environmentSetFilesViewed(project, input),
414 );
415 }
416 return Effect.fail(
417 new PullRequestOperationError({
418 operation: "setFilesViewed",
419 detail: "This host does not track which files a reader has seen.",
420 }),
421 );
422 }),
423 );
424
425 return { filesViewed, setFilesViewed };
426};