1import {
2 CommandId,
3 MessageId,
4 type OrchestrationV2Notification,
5 type PullRequestActivity,
6 type PullRequestComment,
7 type PullRequestRef,
8 type PullRequestThreadCommentsResult,
9 type ThreadPullRequestLink,
10 type ThreadPullRequestWatch,
11} from "@t3tools/contracts";
12import {
13 normalizeThreadPullRequestKey,
14 threadPullRequestKeyOf,
15 visibleThreadPullRequests,
16} from "../sourceControl/threadPullRequests.ts";
17import * as Cause from "effect/Cause";
18import * as Clock from "effect/Clock";
19import * as Context from "effect/Context";
20import * as SecureRandom from "effect/crypto/SecureRandom";
21import * as Effect from "effect/Effect";
22import * as Exit from "effect/Exit";
23import * as FiberMap from "effect/FiberMap";
24import * as Layer from "effect/Layer";
25import * as Option from "effect/Option";
26import * as Schedule from "effect/Schedule";
27import * as Schema from "effect/Schema";
28import * as Semaphore from "effect/Semaphore";
29import type * as Scope from "effect/Scope";
30import * as Stream from "effect/Stream";
31
32import {
33 type ProviderChangeRequestWatchFingerprint,
34 PullRequestProviderError,
35} from "@t3tools/source-control-core/server/PullRequestProvider";
36import * as PullRequestService from "../pullRequest/PullRequestService.ts";
37import { forkParked } from "../serverActivation.ts";
38import * as Orchestrator from "./Orchestrator.ts";
39import * as ProjectionStore from "./ProjectionStore.ts";
40import {
41 combinePullRequestWatchWakes,
42 evaluatePullRequestWatch,
43 PULL_REQUEST_WATCH_NO_CHECKS_MS,
44 pullRequestWatchMessage,
45} from "./pullRequestWatch.ts";
46
47/** Checks take minutes; faster sweeps waste the account's shared host quota. */
48const SWEEP_MINUTES = 2;
49const READ_FAILURE_LIMIT = 8;
50/** Group reads in flight at once, inside the host's own request limit. */
51const GROUP_READS = 4;
52/** Without fingerprints, reread quiet PRs on snapshot changes or this interval to catch edits. */
53const QUIET_REREAD_MS = 10 * 60_000;
54/** Reread despite unchanged fingerprints; they miss review-thread comment edits. */
55const FINGERPRINT_REREAD_MS = 30 * 60_000;
56
57const isProviderError = Schema.is(PullRequestProviderError);
58
59const providerFailureReason = (cause: Cause.Cause<PullRequestService.PullRequestError>) =>
60 Option.match(Cause.findErrorOption(cause), {
61 onNone: () => null,
62 onSome: (error) =>
63 error._tag === "PullRequestOperationError" && isProviderError(error.cause)
64 ? error.cause.reason
65 : null,
66 });
67
68const logFailure =
69 (message: string, fields: Record<string, unknown>) =>
70 <E>(cause: Cause.Cause<E>): Effect.Effect<void> =>
71 Cause.hasInterruptsOnly(cause)
72 ? Effect.interrupt
73 : Effect.logWarning(message, { ...fields, cause });
74
75interface WatchTarget {
76 readonly thread: ProjectionStore.ProjectionThreadPullRequests;
77 readonly link: ThreadPullRequestLink;
78 readonly watch: ThreadPullRequestWatch;
79}
80
81interface WatchGroup {
82 readonly key: string;
83 readonly targets: ReadonlyArray<WatchTarget>;
84 readonly fingerprint: string;
85}
86
87interface LastRead {
88 readonly at: number;
89 readonly fingerprint: string;
90 readonly inFlight: boolean;
91 readonly remarksComplete: boolean;
92 readonly watches: ReadonlySet<string>;
93 readonly status: string | null;
94 readonly remarks: string | null;
95}
96
97interface ReadPlan {
98 readonly detail: boolean;
99 readonly activity: boolean;
100}
101
102interface Wake {
103 readonly text: string;
104 readonly notification: OrchestrationV2Notification;
105 readonly includesRemarks?: boolean;
106}
107
108/** A wake a pass holds back so each thread gets one for everything the pass found. */
109interface HeldWake {
110 readonly groupKey: string;
111 readonly target: WatchTarget;
112 readonly next: ThreadPullRequestWatch | null;
113 readonly endedWatch?: ThreadPullRequestWatch;
114 readonly wake: Wake;
115 readonly reason?: WatchEndReason;
116}
117
118const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`;
119
120type WatchEndReason =
121 | "merged"
122 | "closed"
123 | "unreadable"
124 | "comment-limit"
125 | "settled"
126 | "subagent"
127 | "stopped";
128
129interface WatchLife {
130 readonly threadId: string;
131 readonly pullRequest: string;
132 readonly startedAt: number;
133 readonly headSha: string | null;
134 readonly pushedAt: number;
135 readonly longestQuietMs: number;
136 readonly wakes: number;
137 readonly reads: number;
138}
139
140const lifeKey = ({ thread, link, watch }: WatchTarget) =>
141 `${thread.id} ${threadPullRequestKeyOf(link)} ${watch.startedAt}`;
142
143const minutes = (ms: number) => Math.max(0, Math.round(ms / 60_000));
144
145const snapshotFingerprint = ({ link }: WatchTarget) => {
146 const snapshot = link.snapshot;
147 return snapshot === null
148 ? ""
149 : [
150 snapshot.state,
151 snapshot.updatedAt,
152 snapshot.checksState,
153 snapshot.mergeability,
154 snapshot.reviewDecision,
155 snapshot.isDraft,
156 ].join(" ");
157};
158
159function planRead(
160 group: WatchGroup,
161 last: LastRead | undefined,
162 fingerprint: ProviderChangeRequestWatchFingerprint | null,
163 now: number,
164): ReadPlan {
165 if (last === undefined || group.targets.some((target) => !last.watches.has(watchKey(target)))) {
166 return { detail: true, activity: true };
167 }
168 // Nothing on the host changes when the no-checks deadline passes, or when a push to another
169 // branch clears a watch's answer, so either alone prompts a read.
170 const checksDue = group.targets.some(
171 ({ watch }) =>
172 watch.checksPending === undefined ||
173 (watch.checksAwaitedSince != null &&
174 now - Date.parse(watch.checksAwaitedSince) >= PULL_REQUEST_WATCH_NO_CHECKS_MS),
175 );
176 if (fingerprint === null) {
177 const read =
178 last.inFlight ||
179 !last.remarksComplete ||
180 last.fingerprint !== group.fingerprint ||
181 now - last.at >= QUIET_REREAD_MS;
182 return { detail: read || checksDue, activity: read };
183 }
184 const activity =
185 last.remarks !== fingerprint.remarks ||
186 !last.remarksComplete ||
187 now - last.at >= FINGERPRINT_REREAD_MS;
188 return {
189 detail: activity || checksDue || last.status !== fingerprint.status || last.inFlight,
190 activity,
191 };
192}
193
194// Lets the orchestrator tell that a push cleared the answer while this read was in flight.
195const checksPendingBeforeOf = ({ watch }: WatchTarget) =>
196 watch.checksPending === undefined ? {} : { checksPendingBefore: watch.checksPending };
197
198function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean {
199 return (
200 left.startedAt === right.startedAt &&
201 left.headSha === right.headSha &&
202 left.failedChecks.join("\n") === right.failedChecks.join("\n") &&
203 left.passed === right.passed &&
204 left.passedChecks.join("\n") === right.passedChecks.join("\n") &&
205 (left.owedChecks ?? []).join("\n") === (right.owedChecks ?? []).join("\n") &&
206 (left.awaitingApproval ?? []).join("\n") === (right.awaitingApproval ?? []).join("\n") &&
207 (left.approvalAnnounced ?? false) === (right.approvalAnnounced ?? false) &&
208 left.remarksThrough === right.remarksThrough &&
209 left.remarkIds.join("\n") === right.remarkIds.join("\n") &&
210 left.conflicting === right.conflicting &&
211 left.mergeQueueRemovalId === right.mergeQueueRemovalId &&
212 left.wakes === right.wakes &&
213 left.awaitingBaseline === right.awaitingBaseline &&
214 left.checksAwaitedSince === right.checksAwaitedSince &&
215 left.checksPending === right.checksPending
216 );
217}
218
219export class PullRequestWatchReactor extends Context.Service<
220 PullRequestWatchReactor,
221 {
222 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
223 readonly sweep: Effect.Effect<void>;
224 }
225>()("t3/orchestration-v2/PullRequestWatchReactor") {}
226
227/** @public */
228export const make = Effect.gen(function* () {
229 const engine = yield* Orchestrator.OrchestratorV2;
230 const projections = yield* ProjectionStore.ProjectionStoreV2;
231 const pullRequests = yield* PullRequestService.PullRequestService;
232 const random = yield* SecureRandom.SecureRandom;
233 const bootedAt = yield* Clock.currentTimeMillis;
234 const lives = new Map<string, WatchLife>();
235
236 const lifeOf = (target: WatchTarget): WatchLife => {
237 const existing = lives.get(lifeKey(target));
238 if (existing !== undefined) return existing;
239 const startedAt = Date.parse(target.watch.startedAt);
240 const life: WatchLife = {
241 threadId: target.thread.id,
242 pullRequest: threadPullRequestKeyOf(target.link),
243 startedAt,
244 headSha: target.watch.headSha,
245 pushedAt: startedAt,
246 longestQuietMs: 0,
247 wakes: 0,
248 reads: 0,
249 };
250 lives.set(lifeKey(target), life);
251 return life;
252 };
253 const woke = (target: WatchTarget) =>
254 Effect.sync(() => {
255 const life = lifeOf(target);
256 lives.set(lifeKey(target), { ...life, wakes: life.wakes + 1 });
257 });
258
259 const reportEnd = Effect.fn("PullRequestWatchReactor.reportEnd")(function* (
260 key: string,
261 life: WatchLife,
262 reason: WatchEndReason,
263 ) {
264 lives.delete(key);
265 const now = yield* Clock.currentTimeMillis;
266 const quietMs = now - life.pushedAt;
267 yield* Effect.logInfo("pull request watch ended", {
268 threadId: life.threadId,
269 pullRequest: life.pullRequest,
270 reason,
271 minutes: minutes(now - life.startedAt),
272 quietMinutes: minutes(quietMs),
273 longestQuietMinutes: minutes(Math.max(life.longestQuietMs, quietMs)),
274 wakes: life.wakes,
275 reads: life.reads,
276 partial: life.startedAt < bootedAt,
277 });
278 });
279 const ended = (target: WatchTarget, reason: WatchEndReason) =>
280 Effect.suspend(() => reportEnd(lifeKey(target), lifeOf(target), reason));
281
282 const readFailures = new Map<string, number>();
283 /** The last sweep's groups, whose pull requests get host change feeds. */
284 let currentGroups: ReadonlyArray<WatchGroup> = [];
285 // A pass reads watches and records new ones, so passes must not overlap.
286 const passes = Semaphore.makeUnsafe(1);
287 const lastReads = new Map<string, LastRead>();
288 const threadTails = new Map<
289 string,
290 {
291 readonly fetchedAt: number;
292 readonly visibleThrough: number;
293 readonly tails: ReadonlyMap<
294 string,
295 { readonly count: number; readonly comments: ReadonlyArray<PullRequestComment> }
296 >;
297 }
298 >();
299
300 const identityOf = (link: ThreadPullRequestLink) => ({
301 host: normalizeThreadPullRequestKey(link).host,
302 repository: link.repository,
303 number: link.number,
304 });
305
306 /**
307 * The orchestrator applies this only while the same watch is on, so a stop or restart
308 * that lands during the host read wins.
309 */
310 const record = Effect.fn("PullRequestWatchReactor.record")(function* (
311 target: WatchTarget,
312 next: ThreadPullRequestWatch | null,
313 wake?: Wake,
314 endedWatch?: ThreadPullRequestWatch,
315 ) {
316 const uuid = yield* random.uuidV4();
317 yield* engine.dispatch({
318 type: "thread.pull-request-watch.sync",
319 commandId: CommandId.make(`server:pr-watch:${target.thread.id}:${uuid}`),
320 threadId: target.thread.id,
321 ...identityOf(target.link),
322 startedAt: target.watch.startedAt,
323 watch: next,
324 ...checksPendingBeforeOf(target),
325 ...(endedWatch === undefined ? {} : { endedWatch }),
326 ...(wake === undefined
327 ? {}
328 : { wake: { ...wake, messageId: MessageId.make(`message:pr-watch:${uuid}`) } }),
329 });
330 });
331
332 const giveUp = (target: WatchTarget) =>
333 record(target, null, {
334 text: `T3 Code stopped watching pull request #${target.link.number} (${target.link.url}) because it failed to read it from the host ${READ_FAILURE_LIMIT} times in a row. Check it yourself, and call watch_pull_request only if it is still open.`,
335 notification: {
336 source: { kind: "monitor" },
337 outcome: "failed",
338 summary: `#${target.link.number}: stopped watching, could not read it`,
339 },
340 }).pipe(
341 Effect.tap(() => woke(target)),
342 Effect.catch(() => record(target, null)),
343 Effect.tap(() => ended(target, "unreadable")),
344 );
345
346 const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")(
347 function* (
348 key: string,
349 reference: PullRequestRef,
350 activity: PullRequestActivity,
351 now: number,
352 rereadMs: number,
353 ) {
354 // Comment cursors cannot account for missing threads. Only finish a truncated read
355 // when the host confirms that every thread was listed.
356 if (activity.commentsTruncated && activity.reviewThreadsTruncated !== false) return null;
357
358 const visibleThrough = activity.comments.reduce(
359 (latest, comment) => Math.max(latest, Date.parse(comment.editedAt ?? comment.createdAt)),
360 0,
361 );
362 const paginated = activity.reviewThreads.filter(
363 (thread) => thread.nextCommentsCursor != null,
364 );
365 const cached = threadTails.get(key);
366 // All tails share one watermark, so newer visible remarks or a changed thread must
367 // refresh the entire group before any remarks can advance it.
368 const reuse =
369 cached !== undefined &&
370 now - cached.fetchedAt < rereadMs &&
371 cached.visibleThrough === visibleThrough &&
372 cached.tails.size === paginated.length &&
373 paginated.every(
374 (thread) => cached.tails.get(thread.id)?.count === (thread.commentCount ?? 0),
375 );
376 const tails = new Map<
377 string,
378 { count: number; comments: ReadonlyArray<PullRequestComment> }
379 >();
380 const remarks = [...activity.comments];
381 for (const thread of paginated) {
382 let cursor = thread.nextCommentsCursor ?? null;
383 const count = thread.commentCount ?? 0;
384 let tail = reuse ? cached.tails.get(thread.id) : undefined;
385 if (tail === undefined) {
386 const comments = new Map<string, PullRequestComment>();
387 const cursors = new Set<string>();
388 let hidden = 0;
389 while (cursor !== null) {
390 if (cursors.has(cursor)) return null;
391 cursors.add(cursor);
392 const page: PullRequestThreadCommentsResult = yield* pullRequests.threadComments({
393 ...reference,
394 threadId: thread.id,
395 cursor,
396 });
397 for (const comment of page.comments) {
398 comments.set(comment.id, {
399 ...comment,
400 kind: "review-comment",
401 path: thread.path,
402 reviewState: null,
403 });
404 }
405 hidden += page.hiddenCount ?? 0;
406 cursor = page.nextCursor;
407 }
408 if (thread.comments.length + comments.size + hidden < count) return null;
409 tail = { count, comments: [...comments.values()] };
410 }
411 tails.set(thread.id, tail);
412 remarks.push(...tail.comments);
413 }
414 threadTails.set(key, {
415 fetchedAt: reuse ? cached.fetchedAt : now,
416 visibleThrough,
417 tails,
418 });
419 return remarks.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
420 },
421 Effect.catch((error) =>
422 Effect.logWarning("pull request watch comment pagination failed", { error }).pipe(
423 Effect.as(null),
424 ),
425 ),
426 (effect, key) =>
427 effect.pipe(
428 Effect.onExit((exit) =>
429 Effect.sync(() => {
430 // A retry must refresh every tail before advancing the shared remark watermark.
431 if (Exit.isFailure(exit) || exit.value === null) threadTails.delete(key);
432 }),
433 ),
434 ),
435 );
436
437 /** Wakes the current pass found, sent by `sendHeld` once every read of the pass is done. */
438 let held: Array<HeldWake> = [];
439 const hold = (wake: HeldWake) => Effect.sync(() => void held.push(wake));
440
441 const sendOne = (item: HeldWake) =>
442 record(item.target, item.next, item.wake, item.endedWatch).pipe(
443 Effect.tap(() => woke(item.target)),
444 Effect.tap(() => (item.reason === undefined ? Effect.void : ended(item.target, item.reason))),
445 Effect.catchCause((cause) => {
446 lastReads.delete(item.groupKey);
447 return logFailure("pull request watch update failed", {
448 threadId: item.target.thread.id,
449 pullRequest: item.groupKey,
450 })(cause);
451 }),
452 );
453
454 const sendTogether = Effect.fn("PullRequestWatchReactor.sendTogether")(function* (
455 items: readonly [HeldWake, ...ReadonlyArray<HeldWake>],
456 ) {
457 const [first, ...rest] = items;
458 const uuid = yield* random.uuidV4();
459 yield* engine.dispatch({
460 type: "thread.pull-request-watch.sync",
461 commandId: CommandId.make(`server:pr-watch:${first.target.thread.id}:${uuid}`),
462 threadId: first.target.thread.id,
463 ...identityOf(first.target.link),
464 startedAt: first.target.watch.startedAt,
465 watch: first.next,
466 ...checksPendingBeforeOf(first.target),
467 ...(first.endedWatch === undefined ? {} : { endedWatch: first.endedWatch }),
468 others: rest.map((item) => ({
469 ...identityOf(item.target.link),
470 startedAt: item.target.watch.startedAt,
471 watch: item.next,
472 ...checksPendingBeforeOf(item.target),
473 ...(item.endedWatch === undefined ? {} : { endedWatch: item.endedWatch }),
474 ...(item.wake.includesRemarks === true ? { includesRemarks: true } : {}),
475 })),
476 wake: {
477 ...combinePullRequestWatchWakes([first.wake, ...rest.map((item) => item.wake)]),
478 ...(first.wake.includesRemarks === true ? { includesRemarks: true } : {}),
479 sections: items.map(({ wake }) => ({
480 text: wake.text,
481 outcome: wake.notification.outcome,
482 summary: wake.notification.summary,
483 })),
484 messageId: MessageId.make(`message:pr-watch:${uuid}`),
485 },
486 });
487 yield* Effect.forEach(
488 items,
489 (item) =>
490 woke(item.target).pipe(
491 Effect.andThen(item.reason === undefined ? Effect.void : ended(item.target, item.reason)),
492 ),
493 { discard: true },
494 );
495 });
496
497 const sendHeld = Effect.suspend(() => {
498 const byThread = new Map<string, Array<HeldWake>>();
499 for (const item of held) {
500 byThread.set(item.target.thread.id, [...(byThread.get(item.target.thread.id) ?? []), item]);
501 }
502 held = [];
503 return Effect.forEach(
504 byThread.values(),
505 ([first, ...rest]) =>
506 rest.length === 0
507 ? sendOne(first!)
508 : // A watch that ended or moved during the pass fails the whole command; each then
509 // goes on its own, as the orchestrator decides for it alone.
510 sendTogether([first!, ...rest]).pipe(
511 Effect.catchCauseIf(
512 (cause) => !Cause.hasInterruptsOnly(cause),
513 () => Effect.forEach([first!, ...rest], sendOne, { discard: true }),
514 ),
515 ),
516 { discard: true },
517 );
518 });
519
520 const closed = (groupKey: string, target: WatchTarget) =>
521 hold({
522 groupKey,
523 target,
524 next: null,
525 wake: {
526 text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`,
527 notification: {
528 source: { kind: "monitor" },
529 outcome: "updated",
530 summary: `#${target.link.number}: closed, stopped watching`,
531 },
532 },
533 reason: "closed",
534 });
535
536 const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined =>
537 link.snapshot?.state === "merged"
538 ? "merged"
539 : thread.settledOverride === "settled" || thread.settledAt !== null
540 ? "settled"
541 : thread.lineage.relationshipToParent === "subagent"
542 ? "subagent"
543 : undefined;
544
545 const eachTarget = <E>(
546 group: WatchGroup,
547 step: (target: WatchTarget) => Effect.Effect<void, E>,
548 ) =>
549 Effect.forEach(group.targets, (target) =>
550 step(target).pipe(
551 Effect.as(true),
552 Effect.catchCause((cause) => {
553 lastReads.delete(group.key);
554 return logFailure("pull request watch update failed", {
555 threadId: target.thread.id,
556 pullRequest: group.key,
557 })(cause).pipe(Effect.as(false));
558 }),
559 ),
560 ).pipe(Effect.map((landed) => landed.every(Boolean)));
561
562 const referenceOf = (group: WatchGroup) => {
563 const first = group.targets[0]!;
564 return { projectId: first.thread.projectId, ...identityOf(first.link) };
565 };
566
567 const fingerprintOf = (group: WatchGroup) =>
568 pullRequests.watchFingerprint(referenceOf(group)).pipe(
569 Effect.catchCause(
570 (cause): Effect.Effect<ProviderChangeRequestWatchFingerprint | null | "paused"> =>
571 Cause.hasInterruptsOnly(cause)
572 ? Effect.interrupt
573 : providerFailureReason(cause) === "rate-limited"
574 ? Effect.succeed("paused")
575 : Effect.logDebug("pull request watch fingerprint failed", {
576 pullRequest: group.key,
577 cause,
578 }).pipe(Effect.as(null)),
579 ),
580 );
581
582 const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* (
583 group: WatchGroup,
584 fingerprint: ProviderChangeRequestWatchFingerprint | null,
585 ) {
586 const now = yield* Clock.currentTimeMillis;
587 const last = lastReads.get(group.key);
588 const plan = planRead(group, last, fingerprint, now);
589 if (!plan.detail) return;
590 if (
591 last === undefined ||
592 group.targets.some((target) => !last.watches.has(watchKey(target))) ||
593 (fingerprint !== null && last.remarks !== fingerprint.remarks)
594 ) {
595 // A new watch or changed host remarks needs a fresh view of every tail.
596 threadTails.delete(group.key);
597 }
598 const reference = referenceOf(group);
599 // Invalidate cached reads before using the newer fingerprint; clients receive no notification.
600 if (fingerprint !== null) yield* pullRequests.invalidate({ reference });
601 const [detailRead, activityRead] = yield* Effect.all(
602 [
603 Effect.exit(pullRequests.detail({ ...reference, allowStale: false })),
604 Effect.exit(
605 plan.activity
606 ? pullRequests.activity(reference).pipe(Effect.asSome)
607 : Effect.succeed(Option.none<PullRequestActivity>()),
608 ),
609 ],
610 { concurrency: 2 },
611 );
612 // Activity failures must not cancel or hide a permanent detail failure.
613 // Activity is irrelevant once the pull request is closed or merged, so it cannot block ending.
614 const read = Exit.isFailure(detailRead)
615 ? Exit.failCause(detailRead.cause)
616 : detailRead.value.state !== "open"
617 ? Exit.succeed([detailRead.value, Option.none<PullRequestActivity>()] as const)
618 : Exit.isFailure(activityRead)
619 ? Exit.failCause(activityRead.cause)
620 : Exit.succeed([detailRead.value, activityRead.value] as const);
621 if (Exit.isFailure(read)) {
622 if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause);
623 lastReads.delete(group.key);
624 const reason = providerFailureReason(read.cause);
625 if (reason === "rate-limited") return;
626 // Transport and host failures can outlast the failure limit; keep the watch for recovery.
627 if (reason === "failed") {
628 readFailures.delete(group.key);
629 return yield* Effect.failCause(read.cause);
630 }
631 const failures = (readFailures.get(group.key) ?? 0) + 1;
632 readFailures.set(group.key, failures);
633 if (failures >= READ_FAILURE_LIMIT && (yield* eachTarget(group, giveUp))) {
634 readFailures.delete(group.key);
635 }
636 return yield* Effect.failCause(read.cause);
637 }
638 readFailures.delete(group.key);
639 const [detail, activity] = read.value;
640 for (const target of group.targets) {
641 const life = lifeOf(target);
642 const headSha = detail.headSha ?? life.headSha;
643 const pushed = life.headSha !== null && headSha !== life.headSha;
644 lives.set(lifeKey(target), {
645 ...life,
646 headSha,
647 reads: life.reads + 1,
648 ...(pushed
649 ? { pushedAt: now, longestQuietMs: Math.max(life.longestQuietMs, now - life.pushedAt) }
650 : {}),
651 });
652 }
653 if (detail.state !== "open") {
654 lastReads.delete(group.key);
655 return yield* eachTarget(group, (target) =>
656 detail.state === "closed"
657 ? closed(group.key, target)
658 : record(target, null).pipe(Effect.tap(() => ended(target, "merged"))),
659 );
660 }
661
662 // Never advance the remark watermark past comments an incomplete read could have missed.
663 const remarks = Option.isSome(activity)
664 ? yield* readRemarks(
665 group.key,
666 reference,
667 activity.value,
668 now,
669 fingerprint === null ? QUIET_REREAD_MS : FINGERPRINT_REREAD_MS,
670 )
671 : null;
672 lastReads.set(group.key, {
673 at:
674 plan.activity || last === undefined
675 ? (threadTails.get(group.key)?.fetchedAt ?? now)
676 : last.at,
677 fingerprint: group.fingerprint,
678 inFlight:
679 detail.mergeability === "unknown" ||
680 detail.checks.some((check) => check.status === "pending"),
681 remarksComplete: plan.activity ? remarks !== null : (last?.remarksComplete ?? false),
682 watches: new Set(group.targets.map(watchKey)),
683 status: fingerprint?.status ?? null,
684 remarks: fingerprint?.remarks ?? null,
685 });
686 yield* eachTarget(group, (target) => {
687 const report = evaluatePullRequestWatch(target.watch, detail, remarks, now);
688 if (report.changes.length > 0) {
689 const wake = {
690 ...pullRequestWatchMessage({
691 number: target.link.number,
692 url: target.link.url,
693 baseBranch: detail.baseBranch,
694 headSha: report.next.headSha,
695 report,
696 }),
697 includesRemarks: report.changes.some((change) => change.kind === "remarks"),
698 };
699 // An ending watch keeps what this pass saw, so watching again starts from it.
700 return hold(
701 report.exhausted
702 ? {
703 groupKey: group.key,
704 target,
705 next: null,
706 endedWatch: report.next,
707 wake,
708 reason: "comment-limit",
709 }
710 : { groupKey: group.key, target, next: report.next, wake },
711 );
712 }
713 return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next);
714 });
715 });
716
717 /** Every watch, or with `changed`, only that pull request's after its host reported a change. */
718 // The sweep and change reads each name their own span.
719 const pass = Effect.fnUntraced(function* (changed: string | undefined) {
720 held = [];
721 const threads = yield* projections.getThreadsWithPullRequests();
722 const targets = threads.flatMap((thread) =>
723 visibleThreadPullRequests(thread.pullRequests ?? []).flatMap((link) =>
724 link.watch === undefined ||
725 (changed !== undefined && threadPullRequestKeyOf(link) !== changed)
726 ? []
727 : [{ thread, link, watch: link.watch }],
728 ),
729 );
730 if (changed === undefined) {
731 const present = new Set(targets.map(lifeKey));
732 yield* Effect.forEach(
733 [...lives].filter(([key]) => !present.has(key)),
734 ([key, life]) => reportEnd(key, life, "stopped"),
735 { discard: true },
736 );
737 }
738 for (const target of targets) lifeOf(target);
739 const byPullRequest = new Map<string, Array<WatchTarget>>();
740 const ending: Array<readonly [WatchTarget, WatchEndReason]> = [];
741 for (const target of targets) {
742 const reason = endsWithoutRead(target);
743 if (reason !== undefined) {
744 ending.push([target, reason]);
745 continue;
746 }
747 // Group by project because checkout access differs; one failure must not end another project's watches.
748 const key = `${target.thread.projectId} ${threadPullRequestKeyOf(target.link)}`;
749 byPullRequest.set(key, [...(byPullRequest.get(key) ?? []), target]);
750 }
751 const groups = [...byPullRequest].map(([key, members]): WatchGroup => ({
752 key,
753 targets: members,
754 fingerprint: [...new Set(members.map(snapshotFingerprint))].toSorted().join("\n"),
755 }));
756 if (changed === undefined) {
757 currentGroups = groups;
758 for (const cache of [readFailures, lastReads, threadTails]) {
759 for (const key of cache.keys()) if (!byPullRequest.has(key)) cache.delete(key);
760 }
761 }
762 yield* Effect.forEach(
763 ending,
764 ([target, reason]) =>
765 record(target, null).pipe(
766 Effect.tap(() => ended(target, reason)),
767 Effect.catchCause(
768 logFailure("pull request watch stop failed", {
769 threadId: target.thread.id,
770 pullRequest: threadPullRequestKeyOf(target.link),
771 }),
772 ),
773 ),
774 { discard: true },
775 );
776 // Every fingerprint is asked for at once so the host can batch them, and each group reads
777 // as soon as its own answer lands rather than waiting on the slowest host.
778 const reads = yield* Semaphore.make(GROUP_READS);
779 yield* Effect.forEach(
780 groups,
781 (group) =>
782 // The host just reported a change, so a cached fingerprint may predate it.
783 (changed === undefined
784 ? Effect.void
785 : pullRequests.invalidate({ reference: referenceOf(group) })
786 ).pipe(
787 Effect.andThen(fingerprintOf(group)),
788 Effect.tap((fingerprint) => {
789 const last = lastReads.get(group.key);
790 // Open pull request views refresh only when the host state moved.
791 return changed !== undefined &&
792 fingerprint !== null &&
793 fingerprint !== "paused" &&
794 (last === undefined ||
795 last.status !== fingerprint.status ||
796 last.remarks !== fingerprint.remarks)
797 ? pullRequests.invalidate({ reference: referenceOf(group) }, { notifyReaders: true })
798 : Effect.void;
799 }),
800 Effect.flatMap((fingerprint) =>
801 fingerprint === "paused"
802 ? Effect.void
803 : reads.withPermit(
804 readGroup(group, fingerprint).pipe(
805 Effect.catchCause(
806 logFailure("pull request watch check failed", { pullRequest: group.key }),
807 ),
808 ),
809 ),
810 ),
811 ),
812 { concurrency: "unbounded", discard: true },
813 );
814 yield* sendHeld;
815 }, passes.withPermit);
816
817 const sweep = pass(undefined).pipe(
818 Effect.catchCause(logFailure("pull request watch sweep failed", {})),
819 Effect.withSpan("PullRequestWatchReactor.sweep"),
820 );
821
822 const pullRequestOf = (group: WatchGroup) => threadPullRequestKeyOf(group.targets[0]!.link);
823
824 const readOnChange = (pullRequest: string) =>
825 pass(pullRequest).pipe(
826 Effect.catchCause(logFailure("pull request watch change read failed", { pullRequest })),
827 Effect.withSpan("PullRequestWatchReactor.readOnChange"),
828 );
829
830 const start: PullRequestWatchReactor["Service"]["start"] = Effect.fn(
831 "PullRequestWatchReactor.start",
832 )(function* () {
833 // One feed per pull request: the host keeps one subscription per pull request and account.
834 const feeds = yield* FiberMap.make<string>();
835 const follow = Effect.gen(function* () {
836 const wanted = new Map(currentGroups.map((group) => [pullRequestOf(group), group]));
837 const stale = Array.from(feeds, ([pullRequest]) => pullRequest).filter(
838 (pullRequest) => !wanted.has(pullRequest),
839 );
840 yield* Effect.forEach(stale, (pullRequest) => FiberMap.remove(feeds, pullRequest), {
841 discard: true,
842 });
843 for (const [pullRequest, group] of wanted) {
844 // An ended feed restarts here, so a host that refused it is asked again each sweep.
845 yield* FiberMap.run(feeds, pullRequest, { onlyIfMissing: true })(
846 pullRequests.changes(referenceOf(group)).pipe(
847 // A review sends an event per comment; those landing during a read make one more.
848 Stream.buffer({ capacity: 1, strategy: "sliding" }),
849 Stream.runForEach(() => readOnChange(pullRequest)),
850 ),
851 );
852 }
853 });
854 yield* forkParked(
855 sweep.pipe(
856 Effect.andThen(follow),
857 Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)),
858 Effect.asVoid,
859 ),
860 );
861 });
862
863 return { start, sweep } satisfies PullRequestWatchReactor["Service"];
864});
865
866export const layer = Layer.effect(PullRequestWatchReactor, make);