1import {
2 canonicalRepositoryKey,
3 sourceControlRepositorySelector,
4} from "@t3tools/shared/sourceControl";
5import {
6 CommandId,
7 type OrchestrationProjectShell,
8 type OrchestrationV2DomainEvent,
9 type OrchestrationV2ThreadShell,
10 type ThreadId,
11 type ThreadLinkedPullRequest,
12} from "@t3tools/contracts";
13import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
14import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
15import * as Cause from "effect/Cause";
16import * as Context from "effect/Context";
17import * as Crypto from "effect/Crypto";
18import * as Effect from "effect/Effect";
19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as Schedule from "effect/Schedule";
23import type * as Scope from "effect/Scope";
24import * as Stream from "effect/Stream";
25
26import * as GitManager from "../git/GitManager.ts";
27import * as ProjectStore from "./ProjectStore.ts";
28import * as PullRequestService from "../pullRequest/PullRequestService.ts";
29import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
30import { forkParked } from "../serverActivation.ts";
31import * as Orchestrator from "./Orchestrator.ts";
32
33class ThreadPullRequestServiceV2 extends Context.Service<
34 ThreadPullRequestServiceV2,
35 {
36 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
37 readonly drain: Effect.Effect<void>;
38 }
39>()("t3/orchestration-v2/ThreadPullRequestService/ThreadPullRequestServiceV2") {}
40
41function samePullRequest(
42 left: ThreadLinkedPullRequest | null | undefined,
43 right: ThreadLinkedPullRequest | null,
44): boolean {
45 if (left == null || right === null) return left == null && right === null;
46 return (
47 left.projectId === right.projectId &&
48 left.repository.toLowerCase() === right.repository.toLowerCase() &&
49 left.number === right.number &&
50 left.url === right.url
51 );
52}
53
54function pullRequestMatchesProject(
55 pullRequest: GitManager.GitBranchPullRequest,
56 project: OrchestrationProjectShell,
57): boolean {
58 return (
59 pullRequest.repositoryKey !== null &&
60 project.repositoryIdentity != null &&
61 canonicalRepositoryKey(pullRequest.repositoryKey) ===
62 canonicalRepositoryKey(project.repositoryIdentity.canonicalKey)
63 );
64}
65
66export const resolveProjectForPullRequestDiscovery = Effect.fn(
67 "ThreadPullRequestServiceV2.resolveProject",
68)(function* (
69 project: OrchestrationProjectShell,
70 repositoryIdentities: RepositoryIdentityResolver.RepositoryIdentityResolver["Service"],
71 options?: { readonly refresh?: boolean },
72) {
73 // Identities stay cached for 15 minutes. A finished turn may have added the
74 // remote its pull request lives on, so post-turn discovery refreshes.
75 const repositoryIdentity = yield* repositoryIdentities.resolve(project.workspaceRoot, {
76 refresh: options?.refresh ?? false,
77 });
78 return {
79 project: { ...project, repositoryIdentity },
80 repository: sourceControlRepositorySelector(repositoryIdentity),
81 };
82});
83
84export function projectWorkspaceMatchesSnapshot(
85 currentProject: Option.Option<Pick<OrchestrationProjectShell, "workspaceRoot">>,
86 expectedWorkspaceRoot: string,
87): boolean {
88 return (
89 Option.isSome(currentProject) && currentProject.value.workspaceRoot === expectedWorkspaceRoot
90 );
91}
92
93const BACKFILL_ATTEMPTS = 5;
94
95interface RefreshRequest {
96 readonly threadId: ThreadId | null;
97 readonly refresh: boolean;
98 readonly backfill?: boolean;
99}
100
101export const make = Effect.gen(function* () {
102 const orchestrator = yield* Orchestrator.OrchestratorV2;
103 const projectStore = yield* ProjectStore.ProjectStoreV2;
104 const git = yield* GitManager.GitManager;
105 const pullRequests = yield* PullRequestService.PullRequestService;
106 const repositoryIdentities = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
107 const crypto = yield* Crypto.Crypto;
108 const fileSystem = yield* FileSystem.FileSystem;
109 const pendingBackfill = new Map<ThreadId, number>();
110
111 const finishBackfill = (threads: ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id">>) => {
112 for (const thread of threads) pendingBackfill.delete(thread.id);
113 };
114 const failBackfill = (threads: ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id">>) => {
115 for (const thread of threads) {
116 const remaining = pendingBackfill.get(thread.id);
117 if (remaining === undefined) continue;
118 if (remaining <= 1) pendingBackfill.delete(thread.id);
119 else pendingBackfill.set(thread.id, remaining - 1);
120 }
121 };
122
123 /**
124 * A sweep for one thread reads only that thread's shell, not every thread's.
125 * Finished runs and checkpoints queue one of these each. A sweep over all
126 * threads reads only active, unsettled ones, since discovery skips the rest;
127 * backfill looks up settled threads, so its passes read every active thread.
128 */
129 const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) =>
130 threadId === null
131 ? orchestrator
132 .getShellSnapshot({
133 location: "active",
134 unsettledOnly: !(backfill || pendingBackfill.size > 0),
135 })
136 // Reads every active thread once a minute: run it off the event loop. Writes keep
137 // landing during the read, and the snapshot's sequence guards the sync against them.
138 .pipe(NodeSqliteClient.readOnly)
139 : Effect.gen(function* () {
140 // Read the sequence first. The thread is then at least this new, so a
141 // sync guarded by the sequence is rejected rather than missing a change.
142 const snapshotSequence = yield* orchestrator.getThreadEventSequence(threadId);
143 const thread = yield* orchestrator.getThreadShell(threadId);
144 return { snapshotSequence, threads: thread === null ? [] : [thread] };
145 });
146
147 const synchronize = Effect.fn("ThreadPullRequestServiceV2.synchronize")(function* (
148 request: RefreshRequest,
149 ) {
150 const [threadSnapshot, projectShells] = yield* Effect.all([
151 readThreadSnapshot(request),
152 projectStore.listShells(),
153 ]);
154 const projects = new Map(projectShells.map((project) => [project.id, project]));
155 if (request.backfill) {
156 for (const thread of threadSnapshot.threads) {
157 if (
158 (thread.settledOverride === "settled" || thread.settledAt !== null) &&
159 thread.branchPullRequest == null
160 ) {
161 pendingBackfill.set(thread.id, BACKFILL_ATTEMPTS);
162 }
163 }
164 }
165 // A single-thread read only shows whether its own thread is gone. A thread
166 // with no branch has nothing to look up, and its entry would keep every
167 // periodic pass on the full read.
168 const visibleThreadIds = new Set(
169 threadSnapshot.threads.filter((thread) => thread.branch !== null).map((thread) => thread.id),
170 );
171 const checkedIds = request.threadId === null ? [...pendingBackfill.keys()] : [request.threadId];
172 for (const threadId of checkedIds) {
173 if (!visibleThreadIds.has(threadId)) pendingBackfill.delete(threadId);
174 }
175 const threads = threadSnapshot.threads.filter(
176 (thread) =>
177 thread.archivedAt === null &&
178 ((thread.settledOverride !== "settled" && thread.settledAt === null) ||
179 request.threadId !== null ||
180 pendingBackfill.has(thread.id)) &&
181 (thread.branch !== null || thread.branchPullRequest != null),
182 );
183 const groups = Map.groupBy(threads, (thread) =>
184 JSON.stringify([thread.projectId, thread.worktreePath, thread.branch]),
185 );
186
187 yield* Effect.forEach(
188 groups.values(),
189 (group) =>
190 Effect.gen(function* () {
191 const first = group[0]!;
192 const project = projects.get(first.projectId);
193 if (project === undefined) return finishBackfill(group);
194 const { project: resolvedProject, repository } =
195 yield* resolveProjectForPullRequestDiscovery(project, repositoryIdentities, {
196 refresh: request.refresh,
197 });
198 if (first.branch !== null && repository === null) return finishBackfill(group);
199 const worktreeExists =
200 first.worktreePath !== null && (yield* fileSystem.exists(first.worktreePath));
201 const cwd =
202 worktreeExists && first.worktreePath !== null
203 ? first.worktreePath
204 : project.workspaceRoot;
205 const detected =
206 first.branch === null
207 ? null
208 : yield* git.branchPullRequest(
209 { cwd, branch: first.branch },
210 { refresh: request.refresh },
211 );
212 if (detected !== null && !pullRequestMatchesProject(detected, resolvedProject)) {
213 return finishBackfill(group);
214 }
215 const detectedReference =
216 detected !== null && repository !== null
217 ? {
218 projectId: project.id,
219 repository,
220 number: detected.number,
221 url: detected.url,
222 }
223 : null;
224
225 const plans = yield* Effect.forEach(group, (thread) =>
226 Effect.gen(function* () {
227 let branchPullRequest = detectedReference;
228 if (
229 branchPullRequest === null &&
230 thread.branch !== null &&
231 thread.worktreePath === null &&
232 thread.branchPullRequest != null
233 ) {
234 const previous = yield* pullRequests.summary(thread.branchPullRequest, {
235 recoverTransientFailure: false,
236 });
237 if (previous.state === "merged" || previous.state === "closed") {
238 branchPullRequest = thread.branchPullRequest;
239 }
240 }
241
242 let replacement: ThreadLinkedPullRequest | undefined;
243 if (
244 (thread.pullRequests ?? []).length === 0 &&
245 thread.linkedPullRequest != null &&
246 detected?.state === "open" &&
247 detectedReference !== null &&
248 !samePullRequest(thread.linkedPullRequest, detectedReference)
249 ) {
250 const linked = yield* pullRequests.summary(thread.linkedPullRequest, {
251 recoverTransientFailure: false,
252 });
253 if (linked.state === "merged" || linked.state === "closed") {
254 replacement = detectedReference;
255 }
256 }
257 if (
258 samePullRequest(thread.branchPullRequest, branchPullRequest) &&
259 replacement === undefined
260 ) {
261 pendingBackfill.delete(thread.id);
262 return null;
263 }
264 return { thread, branchPullRequest, replacement };
265 }).pipe(
266 Effect.catchCauseIf(
267 (cause) => !Cause.hasInterruptsOnly(cause),
268 (cause) =>
269 Effect.logWarning("thread pull request discovery failed", {
270 threadId: thread.id,
271 cause: Cause.pretty(cause),
272 }).pipe(
273 Effect.tap(() => Effect.sync(() => failBackfill([thread]))),
274 Effect.as(null),
275 ),
276 ),
277 ),
278 );
279 const updates = plans.filter((plan) => plan !== null);
280 if (updates.length === 0) return;
281
282 if (detected !== null && first.branch !== null) {
283 const current = yield* git.branchPullRequest(
284 { cwd, branch: first.branch },
285 { freshGitState: true },
286 );
287 const currentIdentity = yield* repositoryIdentities.resolve(project.workspaceRoot, {
288 refresh: true,
289 });
290 if (
291 current === null ||
292 current.number !== detected.number ||
293 current.url !== detected.url ||
294 current.state !== detected.state ||
295 current.repositoryKey !== detected.repositoryKey ||
296 !pullRequestMatchesProject(current, {
297 ...project,
298 repositoryIdentity: currentIdentity,
299 })
300 ) {
301 return failBackfill(updates.map((update) => update.thread));
302 }
303 }
304
305 yield* Effect.forEach(
306 updates,
307 ({ thread, branchPullRequest, replacement }) =>
308 Effect.gen(function* () {
309 // Discovery can perform network I/O. Re-read the project at
310 // the transaction boundary so a deleted project or changed
311 // workspace root cannot apply a result from the old checkout.
312 const currentProject = yield* projectStore.getShell(project.id);
313 if (!projectWorkspaceMatchesSnapshot(currentProject, project.workspaceRoot)) {
314 return failBackfill([thread]);
315 }
316 const uuid = yield* crypto.randomUUIDv4();
317 yield* orchestrator.dispatch({
318 type: "thread.pull-request.sync",
319 commandId: CommandId.make(`server:thread-pull-request:${thread.id}:${uuid}`),
320 threadId: thread.id,
321 projectId: project.id,
322 snapshotSequence: threadSnapshot.snapshotSequence,
323 expected: {
324 workspaceRoot: project.workspaceRoot,
325 branch: thread.branch,
326 worktreePath: thread.worktreePath,
327 linkedPullRequest: thread.linkedPullRequest ?? null,
328 branchPullRequest: thread.branchPullRequest ?? null,
329 },
330 branchPullRequest,
331 ...(replacement === undefined ? {} : { linkedPullRequest: replacement }),
332 });
333 pendingBackfill.delete(thread.id);
334 }).pipe(
335 Effect.catchCause((cause) =>
336 Cause.hasInterruptsOnly(cause)
337 ? Effect.failCause(cause)
338 : Effect.logWarning("thread pull request update failed", {
339 threadId: thread.id,
340 cause: Cause.pretty(cause),
341 }).pipe(Effect.tap(() => Effect.sync(() => failBackfill([thread])))),
342 ),
343 ),
344 { discard: true },
345 );
346 }).pipe(
347 Effect.catchCauseIf(
348 (cause) => !Cause.hasInterruptsOnly(cause),
349 (cause) =>
350 Effect.logWarning("thread branch pull request lookup failed", {
351 threadIds: group.map((thread) => thread.id),
352 cause: Cause.pretty(cause),
353 }).pipe(Effect.tap(() => Effect.sync(() => failBackfill(group)))),
354 ),
355 ),
356 // Wide enough that a sweep's GitHub branch lookups reach the GitHub provider together and share one
357 // GraphQL document, instead of one `gh pr list` per branch.
358 { concurrency: 32, discard: true },
359 );
360 });
361
362 const worker = yield* makeDrainableWorker((request: RefreshRequest) =>
363 synchronize(request).pipe(
364 Effect.catchCauseIf(
365 (cause) => !Cause.hasInterruptsOnly(cause),
366 (cause) =>
367 Effect.logWarning("thread pull request refresh failed", {
368 cause: Cause.pretty(cause),
369 }),
370 ),
371 ),
372 );
373
374 const processEvent = (event: OrchestrationV2DomainEvent) => {
375 switch (event.type) {
376 case "thread.created":
377 case "thread.unarchived":
378 case "thread.metadata-updated":
379 return worker.enqueue({ threadId: event.threadId, refresh: false });
380 case "thread.unsettled":
381 case "checkpoint.captured":
382 return worker.enqueue({ threadId: event.threadId, refresh: true });
383 case "run.updated":
384 if (
385 event.payload.status === "completed" ||
386 event.payload.status === "failed" ||
387 event.payload.status === "cancelled" ||
388 event.payload.status === "interrupted"
389 ) {
390 return worker.enqueue({ threadId: event.threadId, refresh: true });
391 }
392 break;
393 }
394 return Effect.void;
395 };
396
397 const start: ThreadPullRequestServiceV2["Service"]["start"] = Effect.fn(
398 "ThreadPullRequestServiceV2.start",
399 )(function* () {
400 yield* forkParked(Stream.runForEach(orchestrator.streamDomainEvents, processEvent));
401 yield* forkParked(
402 Effect.gen(function* () {
403 yield* worker.enqueue({ threadId: null, refresh: false, backfill: true });
404 yield* worker.drain;
405 yield* Effect.gen(function* () {
406 yield* worker.enqueue({ threadId: null, refresh: false });
407 yield* worker.drain;
408 }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.delay("1 minute"));
409 }).pipe(Effect.asVoid),
410 );
411 });
412
413 return { start, drain: worker.drain } satisfies ThreadPullRequestServiceV2["Service"];
414});
415
416const layer = Layer.effect(ThreadPullRequestServiceV2, make);