1import { isMergebotRepository } from "@t3tools/contracts";
2import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
3import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl";
4import {
5 CommandId,
6 type PullRequestSummary,
7 type ThreadId,
8 type ThreadPullRequestKey,
9 type ThreadPullRequestLink,
10 type ThreadPullRequestSnapshot,
11 type ThreadPullRequestStack,
12} from "@t3tools/contracts";
13import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
14import {
15 threadPullRequestKeyOf,
16 normalizeThreadPullRequestKey,
17 threadPullRequestKeysEqual,
18 visibleThreadPullRequests,
19} from "@t3tools/shared/threadPullRequests";
20import * as Cause from "effect/Cause";
21import * as Clock from "effect/Clock";
22import * as Context from "effect/Context";
23import * as Crypto from "effect/Crypto";
24import * as DateTime from "effect/DateTime";
25import * as Effect from "effect/Effect";
26import * as Layer from "effect/Layer";
27import * as Schema from "effect/Schema";
28import * as Schedule from "effect/Schedule";
29import type * as Scope from "effect/Scope";
30import * as Semaphore from "effect/Semaphore";
31import * as Stream from "effect/Stream";
32
33import * as GitManager from "../git/GitManager.ts";
34import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
35import * as PullRequestService from "../pullRequest/PullRequestService.ts";
36import { forkParked } from "../serverActivation.ts";
37import * as Orchestrator from "./Orchestrator.ts";
38import * as ProjectionStore from "./ProjectionStore.ts";
39import { isTerminalRunStatus } from "./ThreadManagementService.ts";
40
41const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
42/** Shell commands that can merge or close a pull request without a merge notification. */
43const PULL_REQUEST_CLOSE_COMMAND = /\b(?:gh\s+pr|glab\s+mr)\s+(?:merge|close)\b/u;
44
45const isPullRequestProviderError = Schema.is(PullRequestProviderError);
46
47type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;
48
49interface LinkEntry {
50 readonly thread: ProjectionStore.ProjectionThreadPullRequests;
51 readonly link: ThreadPullRequestLink;
52}
53
54function snapshotFieldsOf(summary: PullRequestSummary): SnapshotFields {
55 return {
56 state: summary.state,
57 title: summary.title,
58 headBranch: summary.headBranch,
59 baseBranch: summary.baseBranch,
60 isDraft: summary.isDraft ?? false,
61 updatedAt: summary.updatedAt,
62 closedAt: summary.closedAt ?? null,
63 mergedAt: summary.mergedAt ?? null,
64 ...(summary.author === undefined ? {} : { author: summary.author }),
65 ...(summary.additions === undefined ? {} : { additions: summary.additions }),
66 ...(summary.deletions === undefined ? {} : { deletions: summary.deletions }),
67 ...(summary.changedFiles === undefined ? {} : { changedFiles: summary.changedFiles }),
68 ...(summary.reviewDecision === undefined ? {} : { reviewDecision: summary.reviewDecision }),
69 ...(summary.checksState === undefined ? {} : { checksState: summary.checksState }),
70 ...(summary.mergeability === undefined ? {} : { mergeability: summary.mergeability }),
71 };
72}
73
74function snapshotFieldsEqual(left: SnapshotFields, right: SnapshotFields): boolean {
75 return (
76 left.state === right.state &&
77 left.title === right.title &&
78 left.headBranch === right.headBranch &&
79 left.baseBranch === right.baseBranch &&
80 left.isDraft === right.isDraft &&
81 left.updatedAt === right.updatedAt &&
82 (left.closedAt ?? null) === (right.closedAt ?? null) &&
83 (left.mergedAt ?? null) === (right.mergedAt ?? null) &&
84 (left.author?.login ?? null) === (right.author?.login ?? null) &&
85 (left.author?.avatarUrl ?? null) === (right.author?.avatarUrl ?? null) &&
86 left.additions === right.additions &&
87 left.deletions === right.deletions &&
88 left.changedFiles === right.changedFiles &&
89 (left.reviewDecision ?? null) === (right.reviewDecision ?? null) &&
90 (left.checksState ?? null) === (right.checksState ?? null) &&
91 left.mergeability === right.mergeability
92 );
93}
94
95function stacksEqual(
96 left: ThreadPullRequestStack | null,
97 right: ThreadPullRequestStack | null,
98): boolean {
99 if (left === null || right === null) return left === right;
100 return (
101 left.kind === right.kind &&
102 left.id === right.id &&
103 left.number === right.number &&
104 left.url === right.url &&
105 left.base === right.base &&
106 left.layers.length === right.layers.length &&
107 left.layers.every((layer, index) => {
108 const other = right.layers[index]!;
109 return (
110 layer.number === other.number &&
111 layer.headBranch === other.headBranch &&
112 layer.state === other.state
113 );
114 })
115 );
116}
117
118function skipReason(cause: Cause.Cause<unknown>): string {
119 const error = Cause.squash(cause);
120 return error instanceof Error ? error.message : String(error);
121}
122
123/** When a host read failed because the host is rate limited, the time that pause ends. */
124function rateLimitRetryAt(cause: Cause.Cause<unknown>): number | undefined {
125 let error: unknown = Cause.squash(cause);
126 while (error instanceof Error) {
127 if (isPullRequestProviderError(error) && error.reason === "rate-limited") return error.retryAt;
128 error = error.cause;
129 }
130 return undefined;
131}
132
133function isUnsettled(thread: ProjectionStore.ProjectionThreadPullRequests): boolean {
134 return thread.settledOverride !== "settled" && thread.settledAt === null;
135}
136
137/**
138 * Keeps every thread ↔ pull request link's host snapshot current. One sweep a minute reads
139 * only the active threads that have links, groups visible links by pull request so the host
140 * is asked once per PR no matter how many threads share it, and writes back only what
141 * changed. Native stacks the host reports are auto-linked to the thread as `source: "stack"`.
142 */
143export class PullRequestSyncReactor extends Context.Service<
144 PullRequestSyncReactor,
145 {
146 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
147 readonly drain: Effect.Effect<void>;
148 /**
149 * Force the next sweep to re-read this pull request, even when its snapshot is terminal.
150 * While its host is rate limited, the read waits for the first sweep after the pause.
151 */
152 readonly requestSync: (key: ThreadPullRequestKey) => Effect.Effect<void>;
153 }
154>()("t3/orchestration-v2/PullRequestSyncReactor") {}
155
156/** @public Service construction is part of the canonical Effect module API. */
157export const make = Effect.gen(function* () {
158 const engine = yield* Orchestrator.OrchestratorV2;
159 const projections = yield* ProjectionStore.ProjectionStoreV2;
160 const pullRequests = yield* PullRequestService.PullRequestService;
161 const mergebot = yield* MergebotQueue.MergebotQueue;
162 const git = yield* GitManager.GitManager;
163 const crypto = yield* Crypto.Crypto;
164
165 const lastSyncedAt = new Map<string, number>();
166 const requested = new Map<string, number>();
167 let requestGeneration = 0;
168 // Requested keys wait in `requested` for one queued sweep, so a burst of links (an agent
169 // linking dozens of pull requests) is read together and shares the summary batches.
170 let requestedSweepQueued = false;
171 const retryStacks = new Set<string>();
172 // Rate limit pauses by project and host, since each project reads with its own credential.
173 // A paused host refuses every read without asking it, so the sweep leaves its pull requests
174 // due until the pause ends rather than failing each of them every minute.
175 const pausedUntil = new Map<string, number>();
176
177 const isDue = (key: string, entries: ReadonlyArray<LinkEntry>, nowMs: number): boolean => {
178 if (requested.has(key) || retryStacks.has(key)) return true;
179 if (entries.some((entry) => entry.link.snapshot === null)) return true;
180 // Settled threads stop watching their pull requests. Unsettling one makes its links due on
181 // the next sweep, since the cadence clock below kept running while it was settled.
182 const active = entries.filter((entry) => isUnsettled(entry.thread));
183 if (active.every((entry) => entry.link.snapshot?.state === "merged")) return false;
184 if (active.some((entry) => entry.link.snapshot?.state === "open")) return true;
185 // Closed requests can reopen on the host.
186 const last = lastSyncedAt.get(key);
187 return last === undefined || nowMs - last >= SLOW_SYNC_INTERVAL_MS;
188 };
189
190 const logSkipped =
191 (message: string, fields: Record<string, unknown>) =>
192 <E>(cause: Cause.Cause<E>): Effect.Effect<void, E> =>
193 Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);
194
195 /** `requested` reads only keys asked for through `requestSync`; `all` is the periodic pass. */
196 const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (scope: "all" | "requested") {
197 const threads = yield* projections.getThreadsWithPullRequests();
198 const now = yield* DateTime.now;
199 const nowMs = DateTime.toEpochMillis(now);
200 const nowIso = DateTime.formatIso(now);
201
202 const groups = new Map<string, Array<LinkEntry>>();
203 for (const thread of threads) {
204 for (const link of visibleThreadPullRequests(thread.pullRequests ?? [])) {
205 const key = threadPullRequestKeyOf(link);
206 const entries = groups.get(key) ?? [];
207 entries.push({ thread, link });
208 groups.set(key, entries);
209 }
210 }
211
212 for (const key of lastSyncedAt.keys()) if (!groups.has(key)) lastSyncedAt.delete(key);
213 for (const key of retryStacks) if (!groups.has(key)) retryStacks.delete(key);
214 for (const key of requested.keys()) if (!groups.has(key)) requested.delete(key);
215
216 // Layers auto-linked this sweep, so two links of one thread that share a
217 // stack do not both try to add the same sibling.
218 const linkedThisSweep = new Set<string>();
219 const persistence = yield* Semaphore.make(1);
220
221 const syncEntry = Effect.fn("PullRequestSyncReactor.syncEntry")(function* (
222 entry: LinkEntry,
223 fields: SnapshotFields,
224 fetchedStack: { readonly stack: ThreadPullRequestStack | null } | null,
225 ) {
226 const { thread, link } = entry;
227 const nextStack = fetchedStack === null ? link.stack : fetchedStack.stack;
228 const changed =
229 link.snapshot === null ||
230 !snapshotFieldsEqual(link.snapshot, fields) ||
231 !stacksEqual(link.stack, nextStack);
232 // Persist discovered siblings before a terminal snapshot can trigger settlement. A settled
233 // thread that shares this pull request with an active one takes the fresh snapshot, but
234 // gains no links.
235 for (const layer of isUnsettled(thread) ? (fetchedStack?.stack?.layers ?? []) : []) {
236 const layerKey = {
237 host: normalizeThreadPullRequestKey(link).host,
238 repository: link.repository,
239 number: layer.number,
240 };
241 const dedupeKey = `${thread.id}:${threadPullRequestKeyOf(layerKey)}`;
242 if (linkedThisSweep.has(dedupeKey)) continue;
243 // Tombstones count as present: a dismissed layer is never re-added.
244 if (
245 (thread.pullRequests ?? []).some((existing) =>
246 threadPullRequestKeysEqual(existing, layerKey),
247 )
248 ) {
249 continue;
250 }
251 const url = siblingPullRequestUrl(link.url, layer.number);
252 if (url === null) continue;
253 const uuid = yield* crypto.randomUUIDv4();
254 yield* engine.dispatch({
255 type: "thread.pull-request.link",
256 commandId: CommandId.make(`server:pr-stack-link:${thread.id}:${uuid}`),
257 threadId: thread.id,
258 ...layerKey,
259 url,
260 source: "stack",
261 });
262 linkedThisSweep.add(dedupeKey);
263 }
264 if (changed) {
265 const uuid = yield* crypto.randomUUIDv4();
266 yield* engine.dispatch({
267 type: "thread.pull-request-link.sync",
268 commandId: CommandId.make(`server:pr-sync:${thread.id}:${uuid}`),
269 threadId: thread.id,
270 host: normalizeThreadPullRequestKey(link).host,
271 repository: link.repository,
272 number: link.number,
273 snapshot: { ...fields, syncedAt: nowIso },
274 stack: nextStack,
275 });
276 }
277 });
278
279 const syncGroup = Effect.fn("PullRequestSyncReactor.syncGroup")(function* (
280 key: string,
281 entries: ReadonlyArray<LinkEntry>,
282 ) {
283 const first = entries[0]!;
284 const ref = {
285 projectId: first.thread.projectId,
286 host: normalizeThreadPullRequestKey(first.link).host,
287 repository: first.link.repository,
288 number: first.link.number,
289 };
290 const generation = requested.get(key);
291 if (generation !== undefined) yield* pullRequests.invalidate({ reference: ref });
292 const summary = yield* pullRequests.summary(ref, { recoverTransientFailure: false });
293 const fields = snapshotFieldsOf(summary);
294 const work =
295 ref.host === "github.com" &&
296 isMergebotRepository(mergebot.configuration ?? undefined, ref.repository);
297 const needsStack =
298 !work &&
299 (generation !== undefined ||
300 retryStacks.has(key) ||
301 entries.some(
302 (entry) =>
303 entry.link.snapshot === null ||
304 !snapshotFieldsEqual(entry.link.snapshot, fields) ||
305 (summary.stack !== undefined &&
306 (entry.link.stack?.number ?? null) !== (summary.stack?.number ?? null)),
307 ));
308 // A summary that says the pull request is in no stack, for links that hold none, already
309 // answers what the stack read would.
310 const knownUnstacked =
311 summary.stack === null && entries.every((entry) => entry.link.stack === null);
312 const fetchedStack = !needsStack
313 ? null
314 : knownUnstacked
315 ? { stack: null }
316 : yield* pullRequests.stack(ref, { includeDetails: false }).pipe(
317 Effect.map((stack) => ({
318 stack: stack === null ? null : ({ kind: "native", ...stack } as const),
319 })),
320 Effect.catchCauseIf(
321 (cause) => !Cause.hasInterruptsOnly(cause),
322 (cause) =>
323 rateLimitRetryAt(cause) !== undefined
324 ? // The sweep records the pause and holds the host's other reads until it ends.
325 Effect.sync(() => retryStacks.add(key)).pipe(
326 Effect.andThen(Effect.failCause(cause)),
327 )
328 : Effect.logWarning("pull request stack lookup failed", {
329 key,
330 }).pipe(Effect.as(null)),
331 ),
332 );
333 if (needsStack) {
334 if (fetchedStack === null) {
335 retryStacks.add(key);
336 return;
337 }
338 retryStacks.delete(key);
339 }
340 // The host answered, so the cadence clock ticks even if a dispatch below is rejected.
341 lastSyncedAt.set(key, nowMs);
342 // A refresh requested while the host read was in flight belongs to the next sweep.
343 if (requested.get(key) === generation) requested.delete(key);
344 yield* Effect.forEach(
345 entries,
346 (entry) =>
347 syncEntry(entry, fields, fetchedStack).pipe(
348 persistence.withPermits(1),
349 Effect.catchCause((cause) => {
350 if (!Cause.hasInterruptsOnly(cause)) retryStacks.add(key);
351 return logSkipped("pull request sync skipped", { threadId: entry.thread.id, key })(
352 cause,
353 );
354 }),
355 ),
356 { discard: true },
357 );
358 });
359
360 // Failed host reads by reason: how many, and the first key that failed that way.
361 const skips = new Map<string, { count: number; readonly key: string }>();
362 const readGroup = (key: string, entries: ReadonlyArray<LinkEntry>, pauseKey: string) =>
363 syncGroup(key, entries).pipe(
364 Effect.catchCause((cause) => {
365 if (Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause);
366 const retryAt = rateLimitRetryAt(cause);
367 if (retryAt !== undefined) {
368 pausedUntil.set(pauseKey, Math.max(retryAt, pausedUntil.get(pauseKey) ?? 0));
369 }
370 const reason = skipReason(cause);
371 const skip = skips.get(reason);
372 if (skip === undefined) skips.set(reason, { count: 1, key });
373 else skip.count += 1;
374 return Effect.void;
375 }),
376 );
377 yield* Effect.forEach(
378 groups,
379 ([key, entries]) => {
380 if (!((scope === "all" || requested.has(key)) && isDue(key, entries, nowMs))) {
381 return Effect.void;
382 }
383 const first = entries[0]!;
384 const pauseKey = `${first.thread.projectId}\0${normalizeThreadPullRequestKey(first.link).host}`;
385 // Checked against the clock as each read starts, so a pause found earlier in this sweep
386 // holds the rest, and one that ends during the sweep lets the rest through.
387 return Clock.currentTimeMillis.pipe(
388 Effect.flatMap((startedAtMs) =>
389 (pausedUntil.get(pauseKey) ?? 0) > startedAtMs
390 ? Effect.void
391 : readGroup(key, entries, pauseKey),
392 ),
393 );
394 },
395 // As wide as one batched summary read, so the sweep's reads on a host arrive together and
396 // GitHub answers them in one request rather than one `gh pr view` apiece.
397 { concurrency: 25, discard: true },
398 );
399 // A host failure such as a signed-out CLI fails every due pull request the same way, so a
400 // sweep reports one line per reason rather than one per pull request.
401 for (const [reason, { count, key }] of skips) {
402 yield* Effect.logWarning("pull request sync skipped", { count, key, reason });
403 }
404 });
405
406 const worker = yield* makeDrainableWorker((scope: "all" | "requested") =>
407 Effect.suspend(() => {
408 // Requests from here on queue another sweep; the ones already recorded are read by this.
409 if (scope === "requested") requestedSweepQueued = false;
410 return sweep(scope);
411 }).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))),
412 );
413
414 // Threads whose current run ran a merge or close command, until that run ends.
415 const closeCommandThreads = new Set<ThreadId>();
416 const refreshOpenLinks = (threadId: ThreadId) =>
417 projections.getThreadsWithPullRequests(threadId).pipe(
418 Effect.flatMap((threads) =>
419 Effect.forEach(
420 threads.flatMap((thread) =>
421 visibleThreadPullRequests(thread.pullRequests ?? []).filter(
422 (link) => link.snapshot?.state === "open",
423 ),
424 ),
425 requestSync,
426 { discard: true },
427 ),
428 ),
429 Effect.catchCause(logSkipped("pull request refresh after run skipped", { threadId })),
430 );
431
432 const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn(
433 "PullRequestSyncReactor.start",
434 )(function* () {
435 const events = engine.streamDomainEvents;
436 // A client reading a pull request, or its branch status, can see it merge or close before
437 // the next sweep does.
438 const stateChanges = Stream.merge(
439 yield* pullRequests.subscribeStateChanges,
440 yield* git.subscribePullRequestStateChanges,
441 );
442 yield* forkParked(
443 Stream.runForEach(stateChanges, requestSync).pipe(
444 Effect.catchCause(logSkipped("pull request state change stream failed", {})),
445 ),
446 );
447 yield* forkParked(
448 Stream.runForEach(events, (event) => {
449 switch (event.type) {
450 case "thread.pull-request-link-updated":
451 return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void;
452 case "thread.pull-request-synced":
453 return Effect.forEach(
454 visibleThreadPullRequests(event.payload.pullRequests ?? []).filter(
455 (link) => link.snapshot === null,
456 ),
457 requestSync,
458 { discard: true },
459 );
460 // An agent can merge or close its pull request from a shell (`gh pr merge`), which
461 // sends no merge notification. When a run that ran such a command ends, read the
462 // thread's open links fresh, so settlement does not wait for the next sweep and the
463 // cached summary. Other runs add no host reads.
464 case "turn-item.updated":
465 if (
466 event.payload.type === "command_execution" &&
467 PULL_REQUEST_CLOSE_COMMAND.test(event.payload.input)
468 ) {
469 closeCommandThreads.add(event.threadId);
470 }
471 return Effect.void;
472 case "run.updated":
473 return isTerminalRunStatus(event.payload.status) &&
474 closeCommandThreads.delete(event.threadId)
475 ? refreshOpenLinks(event.threadId)
476 : Effect.void;
477 default:
478 return Effect.void;
479 }
480 }).pipe(Effect.catchCause(logSkipped("pull request sync event stream failed", {}))),
481 );
482 yield* forkParked(
483 Effect.gen(function* () {
484 yield* worker.enqueue("all");
485 yield* worker.drain;
486 }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid),
487 );
488 });
489
490 const requestSync: PullRequestSyncReactor["Service"]["requestSync"] = (key) =>
491 Effect.suspend(() => {
492 requested.set(threadPullRequestKeyOf(key), ++requestGeneration);
493 if (requestedSweepQueued) return Effect.void;
494 requestedSweepQueued = true;
495 return worker.enqueue("requested");
496 });
497
498 return { start, drain: worker.drain, requestSync } satisfies PullRequestSyncReactor["Service"];
499});
500
501export const layer = Layer.effect(PullRequestSyncReactor, make);