Open History to see the last commit
1import * as Clock from "effect/Clock";
2import * as DateTime from "effect/DateTime";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Semaphore from "effect/Semaphore";
6import {
7 PullRequestOperationError,
8 type PullRequestFilesViewedResult,
9 type PullRequestRef,
10 type PullRequestSetFilesViewedInput,
11} from "@t3tools/contracts";
12
13import type * as PullRequestFilesViewed from "../persistence/PullRequestFilesViewed.ts";
14import type {
15 ProviderFileRevisions,
16 PullRequestProviderError,
17} from "@t3tools/source-control-core/server/PullRequestProvider";
18import type { PullRequestError, SupportedProject } from "./PullRequestService.ts";
19
20const FILE_REVISIONS_CACHE_TTL = Duration.seconds(60);
21const FILE_REVISIONS_STALE_WINDOW = Duration.minutes(10);
22const FILE_REVISIONS_CACHE_CAPACITY = 64;
23
24/**
25 * Bounds a single scope, not how many scopes are held: a reader ticking one file after
26 * another renews the same scope and grows it without limit otherwise.
27 */
28const MAX_FILE_REVISION_PATHS = 1_000;
29
30interface FileRevisionsDependencies {
31 readonly runFork: (effect: Effect.Effect<void>) => unknown;
32 readonly refEpoch: (ref: PullRequestRef) => number;
33 readonly fileRevisionsEpoch: () => number;
34 readonly toPullRequestError: (
35 operation: string,
36 ) => (error: PullRequestProviderError) => PullRequestError;
37}
38
39const makeFileRevisions = (dependencies: FileRevisionsDependencies) => {
40 const { runFork, refEpoch, fileRevisionsEpoch, toPullRequestError } = dependencies;
41 interface HeldFileRevisions {
42 readonly at: number;
43 readonly asked: ReadonlySet<string>;
44 readonly revisions: ReadonlyMap<string, string>;
45 }
46 const heldFileRevisions = new Map<string, HeldFileRevisions>();
47 const refreshingFileRevisions = new Set<string>();
48
49 // Epochs use remote spelling; canonicalize client refs before keying.
50 const fileRevisionsKey = (project: SupportedProject, ref: PullRequestRef) =>
51 [
52 refEpoch({ ...ref, host: project.host, repository: project.repository }),
53 fileRevisionsEpoch(),
54 ref.projectId,
55 project.repository.trim().toLowerCase(),
56 ref.number,
57 ].join(" ");
58
59 const recordFileRevisions = (
60 key: string,
61 paths: ReadonlyArray<string>,
62 answer: ProviderFileRevisions,
63 ) =>
64 Effect.map(Clock.currentTimeMillis, (at) => {
65 const held = heldFileRevisions.get(key);
66 const carried =
67 held !== undefined && at - held.at <= Duration.toMillis(FILE_REVISIONS_STALE_WINDOW)
68 ? held
69 : null;
70 const revisions = new Map(carried?.revisions ?? []);
71 const asked = new Set(carried?.asked ?? []);
72 const learned = answer.complete === true ? [...answer.revisions.keys(), ...paths] : paths;
73 for (const path of learned) {
74 asked.delete(path);
75 asked.add(path);
76 const revision = answer.revisions.get(path);
77 if (revision !== undefined) {
78 revisions.delete(path);
79 revisions.set(path, revision);
80 }
81 }
82 for (const path of asked) {
83 if (asked.size <= MAX_FILE_REVISION_PATHS) break;
84 asked.delete(path);
85 revisions.delete(path);
86 }
87 heldFileRevisions.delete(key);
88 if (heldFileRevisions.size >= FILE_REVISIONS_CACHE_CAPACITY) {
89 const oldest = heldFileRevisions.keys().next().value;
90 if (oldest !== undefined) heldFileRevisions.delete(oldest);
91 }
92 // Partial answers retain the oldest timestamp so carried revisions cannot outlive their freshness.
93 const stamped = [...revisions.keys()].every((path) => answer.revisions.has(path))
94 ? at
95 : (carried?.at ?? at);
96 heldFileRevisions.set(key, { at: stamped, asked, revisions });
97 return revisions;
98 });
99
100 const heldFileRevisionsFor = (key: string, paths: ReadonlyArray<string>, now: number) => {
101 const held = heldFileRevisions.get(key);
102 if (held === undefined) return null;
103 heldFileRevisions.delete(key);
104 heldFileRevisions.set(key, held);
105 if (now - held.at > Duration.toMillis(FILE_REVISIONS_STALE_WINDOW)) return null;
106 return paths.every((path) => held.asked.has(path)) ? held : null;
107 };
108
109 // Held reads refresh off the critical path; presses need fresh revisions to avoid saving obsolete heads.
110 const fileRevisionsOf = (
111 project: SupportedProject,
112 ref: PullRequestRef,
113 paths: ReadonlyArray<string>,
114 operation: string,
115 freshness: "held" | "fresh" = "held",
116 ): Effect.Effect<ReadonlyMap<string, string> | null, PullRequestError> => {
117 const read = project.api.getFileRevisions;
118 if (read === undefined) return Effect.succeed(null);
119 // Suspend request construction too; providers may start work while building requests.
120 const fetch = Effect.suspend(() => {
121 const key = fileRevisionsKey(project, ref);
122 return read({
123 cwd: project.project.workspaceRoot,
124 repository: project.repository,
125 host: project.host,
126 number: ref.number,
127 paths,
128 }).pipe(
129 Effect.mapError(toPullRequestError(operation)),
130 Effect.flatMap((answer) => recordFileRevisions(key, paths, answer)),
131 );
132 });
133 return Effect.flatMap(Clock.currentTimeMillis, (now) => {
134 const key = fileRevisionsKey(project, ref);
135 const held = heldFileRevisionsFor(key, paths, now);
136 if (held === null) return fetch;
137 if (now - held.at <= Duration.toMillis(FILE_REVISIONS_CACHE_TTL))
138 return Effect.succeed(held.revisions);
139 if (freshness === "fresh") return fetch;
140 if (refreshingFileRevisions.has(key)) return Effect.succeed(held.revisions);
141 // Refresh independently; it outlives the answered caller.
142 return Effect.sync(() => {
143 refreshingFileRevisions.add(key);
144 runFork(
145 Effect.ignore(fetch).pipe(
146 Effect.ensuring(Effect.sync(() => refreshingFileRevisions.delete(key))),
147 ),
148 );
149 }).pipe(Effect.as(held.revisions));
150 });
151 };
152
153 return { fileRevisionsOf } as const;
154};
155
156export interface Dependencies extends FileRevisionsDependencies {
157 readonly filesViewedStore: PullRequestFilesViewed.PullRequestFilesViewedRepository["Service"];
158 readonly requireProject: (
159 ref: PullRequestRef,
160 ) => Effect.Effect<SupportedProject, PullRequestError>;
161 readonly requiredViewerOf: (
162 project: SupportedProject,
163 operation: string,
164 ) => Effect.Effect<string | null, PullRequestError>;
165}
166
167// Use one factory instance per service; multiple layers would duplicate revision state and gates behind one epoch.
168export const make = (dependencies: Dependencies) => {
169 const { filesViewedStore, requireProject, requiredViewerOf, toPullRequestError } = dependencies;
170 const { fileRevisionsOf } = makeFileRevisions(dependencies);
171 const filesViewedScope = (project: SupportedProject, number: number, viewer: string | null) => ({
172 provider: project.api.kind,
173 host: project.host,
174 repository: project.remote,
175 number,
176 viewer: viewer ?? "",
177 });
178
179 const toFilesViewedStoreError = (operation: string) => (cause: unknown) =>
180 new PullRequestOperationError({
181 operation,
182 detail: "This environment could not reach its record of which files you have seen.",
183 cause,
184 });
185
186 const environmentFilesViewed = Effect.fn("pullRequestViewedFiles.environmentFilesViewed")(
187 function* (
188 project: SupportedProject,
189 ref: PullRequestRef,
190 ): Effect.fn.Return<PullRequestFilesViewedResult, PullRequestError> {
191 const viewer = yield* requiredViewerOf(project, "filesViewed");
192 const held = yield* filesViewedStore
193 .list(filesViewedScope(project, ref.number, viewer))
194 .pipe(Effect.mapError(toFilesViewedStoreError("filesViewed")));
195 const marks = held.files;
196 if (marks.length === 0) return { files: [], truncated: held.truncated };
197 const revisions = yield* fileRevisionsOf(
198 project,
199 ref,
200 marks.map((mark) => mark.path),
201 "filesViewed",
202 ).pipe(
203 Effect.catch((error) =>
204 Effect.logWarning("reporting viewed files without what the head has of them", {
205 operation: "filesViewed",
206 reason: error._tag,
207 }).pipe(Effect.as(null)),
208 ),
209 );
210 return {
211 files: marks.map((mark) => {
212 if (mark.revision === null) return { path: mark.path, state: "viewed" as const };
213 const revision = revisions?.get(mark.path);
214 return {
215 path: mark.path,
216 state:
217 revision === undefined || revision === mark.revision
218 ? ("viewed" as const)
219 : ("dismissed" as const),
220 };
221 }),
222 truncated: held.truncated,
223 };
224 },
225 );
226
227 // Serialize presses so slower tick lookups cannot overwrite later unticks.
228 const filesViewedGates = new Map<
229 string,
230 { readonly gate: Semaphore.Semaphore; pending: number }
231 >();
232
233 const inFilesViewedOrder = (
234 project: SupportedProject,
235 number: number,
236 write: Effect.Effect<void, PullRequestError>,
237 ) =>
238 // Atomically find, register and queue on gates; yielding during creation lets concurrent presses use separate gates.
239 Effect.suspend(() => {
240 const key = `${project.project.id} ${project.remote} ${number}`;
241 const held = filesViewedGates.get(key);
242 const entry = held ?? { gate: Semaphore.makeUnsafe(1), pending: 0 };
243 if (held === undefined) filesViewedGates.set(key, entry);
244 entry.pending += 1;
245 return entry.gate
246 .withPermits(1)(write)
247 .pipe(
248 Effect.ensuring(
249 Effect.sync(() => {
250 entry.pending -= 1;
251 if (entry.pending === 0) filesViewedGates.delete(key);
252 }),
253 ),
254 );
255 });
256
257 const environmentSetFilesViewed = Effect.fn("pullRequestViewedFiles.environmentSetFilesViewed")(
258 function* (
259 project: SupportedProject,
260 input: PullRequestSetFilesViewedInput,
261 ): Effect.fn.Return<void, PullRequestError> {
262 const viewer = yield* requiredViewerOf(project, "setFilesViewed");
263 const cleared = input.files
264 .filter((file) => file.viewed && file.revision === undefined)
265 .map((file) => file.path);
266 const revisions =
267 cleared.length === 0
268 ? null
269 : yield* fileRevisionsOf(project, input, cleared, "setFilesViewed", "fresh").pipe(
270 Effect.catch((error) =>
271 Effect.logWarning("recording viewed files without what the head has of them", {
272 operation: "setFilesViewed",
273 reason: error._tag,
274 }).pipe(Effect.as(null)),
275 ),
276 );
277 const viewedAt = DateTime.formatIso(yield* DateTime.now);
278 yield* filesViewedStore
279 .set({
280 ...filesViewedScope(project, input.number, viewer),
281 // Shown revisions win over the head; missing paths have null baselines; empty means deleted.
282 files: input.files.map((file) => ({
283 path: file.path,
284 revision: file.revision ?? revisions?.get(file.path) ?? null,
285 viewed: file.viewed,
286 })),
287 viewedAt,
288 })
289 .pipe(Effect.mapError(toFilesViewedStoreError("setFilesViewed")));
290 },
291 );
292
293 const filesViewed = (input: PullRequestRef) =>
294 requireProject(input).pipe(
295 Effect.flatMap((project): Effect.Effect<PullRequestFilesViewedResult, PullRequestError> => {
296 const read = project.api.getFilesViewed;
297 if (project.api.capabilities.viewedFiles === "host" && read) {
298 return read({
299 cwd: project.project.workspaceRoot,
300 repository: project.repository,
301 host: project.host,
302 number: input.number,
303 }).pipe(Effect.mapError(toPullRequestError("filesViewed")));
304 }
305 if (project.api.capabilities.viewedFiles === "environment") {
306 return environmentFilesViewed(project, input);
307 }
308 return Effect.fail(
309 new PullRequestOperationError({
310 operation: "filesViewed",
311 detail: "This host does not track which files a reader has seen.",
312 }),
313 );
314 }),
315 );
316
317 const setFilesViewed = (
318 input: PullRequestSetFilesViewedInput,
319 ): Effect.Effect<void, PullRequestError> =>
320 requireProject(input).pipe(
321 Effect.flatMap((project): Effect.Effect<void, PullRequestError> => {
322 const write = project.api.setFilesViewed;
323 if (project.api.capabilities.viewedFiles === "host" && write) {
324 return write({
325 cwd: project.project.workspaceRoot,
326 repository: project.repository,
327 host: project.host,
328 number: input.number,
329 files: input.files,
330 }).pipe(Effect.mapError(toPullRequestError("setFilesViewed")));
331 }
332 if (project.api.capabilities.viewedFiles === "environment") {
333 return inFilesViewedOrder(
334 project,
335 input.number,
336 environmentSetFilesViewed(project, input),
337 );
338 }
339 return Effect.fail(
340 new PullRequestOperationError({
341 operation: "setFilesViewed",
342 detail: "This host does not track which files a reader has seen.",
343 }),
344 );
345 }),
346 );
347
348 return { filesViewed, setFilesViewed };
349};