1import {
2 CommandId,
3 isMergebotRepository,
4 MessageId,
5 type OrchestrationV2Notification,
6 type PullRequestActivity,
7 type PullRequestDetail,
8 type PullRequestComment,
9 type PullRequestRef,
10 type PullRequestThreadCommentsResult,
11 type ThreadPullRequestLink,
12 type ThreadPullRequestWatch,
13} from "@t3tools/contracts";
14import {
15 normalizeThreadPullRequestKey,
16 threadPullRequestKeyOf,
17 visibleThreadPullRequests,
18} from "@t3tools/shared/threadPullRequests";
19import * as Cause from "effect/Cause";
20import * as Clock from "effect/Clock";
21import * as Context from "effect/Context";
22import * as Crypto from "effect/Crypto";
23import * as Effect from "effect/Effect";
24import * as Exit from "effect/Exit";
25import * as Layer from "effect/Layer";
26import * as Option from "effect/Option";
27import * as Schedule from "effect/Schedule";
28import * as Semaphore from "effect/Semaphore";
29import * as Stream from "effect/Stream";
30import * as Schema from "effect/Schema";
31import type * as Scope from "effect/Scope";
32
33import {
34 type ProviderChangeRequestWatchFingerprint,
35 PullRequestProviderError,
36} from "@t3tools/source-control-core/server/PullRequestProvider";
37import * as MergebotQueue from "../mergebot/MergebotQueue.ts";
38import {
39 mergebotChecks,
40 mergebotMergeability,
41 mergebotWatchFingerprint,
42} from "../mergebot/pullRequestRead.ts";
43import * as PullRequestService from "../pullRequest/PullRequestService.ts";
44import { forkParked } from "../serverActivation.ts";
45import * as Orchestrator from "./Orchestrator.ts";
46import * as ProjectionStore from "./ProjectionStore.ts";
47import { evaluatePullRequestWatch, pullRequestWatchMessage } from "./pullRequestWatch.ts";
48
49/**
50 * Minutes between passes. Checks take minutes, so a faster pass mostly spends the host's rate
51 * limit, which every machine on the same account shares.
52 */
53const SWEEP_MINUTES = 2;
54/** Reads in a row that failed for a reason other than a rate limit before the watch ends. */
55const READ_FAILURE_LIMIT = 8;
56/**
57 * Without a host fingerprint, a pull request with nothing in flight is read again only when its
58 * sync snapshot moves, or after this long, for news the snapshot cannot show, such as a bot
59 * editing its review.
60 */
61const QUIET_REREAD_MS = 10 * 60_000;
62/**
63 * With a host fingerprint, how long until the activity is read again anyway: the fingerprint
64 * does not see edits to comments inside review threads.
65 */
66const FINGERPRINT_REREAD_MS = 30 * 60_000;
67
68const isProviderError = Schema.is(PullRequestProviderError);
69
70/** A rate limit is the host asking us to wait, not a sign the pull request cannot be read. */
71const isRateLimited = (cause: Cause.Cause<PullRequestService.PullRequestError>) =>
72 Option.match(Cause.findErrorOption(cause), {
73 onNone: () => false,
74 onSome: (error) =>
75 error._tag === "PullRequestOperationError" &&
76 isProviderError(error.cause) &&
77 error.cause.reason === "rate-limited",
78 });
79
80const logFailure =
81 (message: string, fields: Record<string, unknown>) =>
82 <E>(cause: Cause.Cause<E>): Effect.Effect<void> =>
83 Cause.hasInterruptsOnly(cause)
84 ? Effect.interrupt
85 : Effect.logWarning(message, { ...fields, cause });
86
87interface WatchTarget {
88 readonly thread: ProjectionStore.ProjectionThreadPullRequests;
89 readonly link: ThreadPullRequestLink;
90 readonly watch: ThreadPullRequestWatch;
91}
92
93/** Every thread of one project that watches one pull request, read once per pass. */
94interface WatchGroup {
95 readonly key: string;
96 readonly targets: ReadonlyArray<WatchTarget>;
97 /** What sync last saw of the pull request, to skip a read when nothing moved. */
98 readonly fingerprint: string;
99}
100
101/** The last successful read of a pull request, kept in memory: a restart reads each once. */
102interface LastRead {
103 /** When the activity was last read. */
104 readonly at: number;
105 /** The sync snapshot this read saw. */
106 readonly fingerprint: string;
107 /** A check was still running or mergeability unknown, so the detail can move unannounced. */
108 readonly inFlight: boolean;
109 /** The last activity read listed every remark; an incomplete one is read again. */
110 readonly remarksComplete: boolean;
111 /** The watches this read evaluated; a watch started since takes its first look next pass. */
112 readonly watches: ReadonlySet<string>;
113 /** The host fingerprint parts these reads answer, when the host gives one. */
114 readonly status: string | null;
115 readonly remarks: string | null;
116}
117
118/** Which reads a pass makes for one pull request. */
119interface ReadPlan {
120 readonly detail: boolean;
121 readonly activity: boolean;
122}
123
124const watchKey = ({ thread, watch }: WatchTarget) => `${thread.id} ${watch.startedAt}`;
125
126/**
127 * Why a watch ended. `stopped` is anything outside this reactor: the agent unwatched, the user
128 * pressed Stop, or the thread settled, archived, or was deleted between passes.
129 */
130type WatchEndReason =
131 | "merged"
132 | "closed"
133 | "unreadable"
134 | "comment-limit"
135 | "settled"
136 | "subagent"
137 | "stopped";
138
139/**
140 * What one watch did while this server ran, logged once when it ends so we can see how long
141 * watches stay quiet. Kept in memory: a watch older than the server process only has partial
142 * numbers, and one that ends while the server is down, or starts and ends between two passes,
143 * is not logged.
144 */
145interface WatchLife {
146 readonly threadId: string;
147 readonly pullRequest: string;
148 readonly startedAt: number;
149 /** The head commit the last successful read saw. */
150 readonly headSha: string | null;
151 /** When a pass last saw the head commit move, or the start. */
152 readonly pushedAt: number;
153 /** Longest time between pushes, not counting the time since the last one. */
154 readonly longestQuietMs: number;
155 readonly wakes: number;
156 readonly reads: number;
157}
158
159const lifeKey = ({ thread, link, watch }: WatchTarget) =>
160 `${thread.id} ${threadPullRequestKeyOf(link)} ${watch.startedAt}`;
161
162const minutes = (ms: number) => Math.max(0, Math.round(ms / 60_000));
163
164const snapshotFingerprint = ({ link }: WatchTarget) => {
165 const snapshot = link.snapshot;
166 return snapshot === null
167 ? ""
168 : [
169 snapshot.state,
170 snapshot.updatedAt,
171 snapshot.checksState,
172 snapshot.mergeability,
173 snapshot.reviewDecision,
174 snapshot.isDraft,
175 ].join(" ");
176};
177
178/**
179 * With a host fingerprint, the detail (1 point on GitHub) is read when anything moved or a check
180 * is still running, since check counts by state cannot tell which check finished, and the
181 * activity (15 points) only when the remarks moved. Without one, both are read whenever the sync
182 * snapshot moved or something is in flight.
183 */
184function planRead(
185 group: WatchGroup,
186 last: LastRead | undefined,
187 fingerprint: ProviderChangeRequestWatchFingerprint | null,
188 now: number,
189): ReadPlan {
190 // A watch takes its first look with everything.
191 if (last === undefined || group.targets.some((target) => !last.watches.has(watchKey(target)))) {
192 return { detail: true, activity: true };
193 }
194 if (fingerprint === null) {
195 const read =
196 last.inFlight ||
197 !last.remarksComplete ||
198 last.fingerprint !== group.fingerprint ||
199 now - last.at >= QUIET_REREAD_MS;
200 return { detail: read, activity: read };
201 }
202 const activity =
203 last.remarks !== fingerprint.remarks ||
204 !last.remarksComplete ||
205 now - last.at >= FINGERPRINT_REREAD_MS;
206 return { detail: activity || last.status !== fingerprint.status || last.inFlight, activity };
207}
208
209function watchesEqual(left: ThreadPullRequestWatch, right: ThreadPullRequestWatch): boolean {
210 return (
211 left.startedAt === right.startedAt &&
212 left.headSha === right.headSha &&
213 left.failedChecks.join("\n") === right.failedChecks.join("\n") &&
214 left.passed === right.passed &&
215 left.passedChecks.join("\n") === right.passedChecks.join("\n") &&
216 left.remarksThrough === right.remarksThrough &&
217 left.remarkIds.join("\n") === right.remarkIds.join("\n") &&
218 left.conflicting === right.conflicting &&
219 left.wakes === right.wakes
220 );
221}
222
223/**
224 * Wakes a thread's agent when a pull request it watches (`watch_pull_request`) needs a look:
225 * checks finished on the head commit, someone else commented, or the branch started to
226 * conflict. A pass every two minutes looks at each watched pull request once for all the threads
227 * of a project that watch it. Where the host has a fingerprint, one batched read of those tells
228 * the pass which pull requests moved, and only those are read; elsewhere a pull request is read
229 * unless its sync snapshot has not moved while nothing is in flight.
230 * Settling or archiving a thread ends its watches. A merged or closed pull request wakes its
231 * active thread once and ends its watch. Each ended watch is logged once with why it ended
232 * and how long it went without a push, for debugging watches that live too long.
233 */
234export class PullRequestWatchReactor extends Context.Service<
235 PullRequestWatchReactor,
236 {
237 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
238 /** One pass over every watched pull request. */
239 readonly sweep: Effect.Effect<void>;
240 }
241>()("t3/orchestration-v2/PullRequestWatchReactor") {}
242
243/** @public Service construction is part of the canonical Effect module API. */
244export const make = Effect.gen(function* () {
245 const engine = yield* Orchestrator.OrchestratorV2;
246 const projections = yield* ProjectionStore.ProjectionStoreV2;
247 const pullRequests = yield* PullRequestService.PullRequestService;
248 const mergebot = yield* MergebotQueue.MergebotQueue;
249 const sweepLock = yield* Semaphore.make(1);
250 const crypto = yield* Crypto.Crypto;
251 const bootedAt = yield* Clock.currentTimeMillis;
252 const lives = new Map<string, WatchLife>();
253
254 const lifeOf = (target: WatchTarget): WatchLife => {
255 const existing = lives.get(lifeKey(target));
256 if (existing !== undefined) return existing;
257 const startedAt = Date.parse(target.watch.startedAt);
258 const life: WatchLife = {
259 threadId: target.thread.id,
260 pullRequest: threadPullRequestKeyOf(target.link),
261 startedAt,
262 headSha: target.watch.headSha,
263 pushedAt: startedAt,
264 longestQuietMs: 0,
265 wakes: 0,
266 reads: 0,
267 };
268 lives.set(lifeKey(target), life);
269 return life;
270 };
271 const woke = (target: WatchTarget) =>
272 Effect.sync(() => {
273 const life = lifeOf(target);
274 lives.set(lifeKey(target), { ...life, wakes: life.wakes + 1 });
275 });
276
277 const reportEnd = (key: string, life: WatchLife, reason: WatchEndReason) =>
278 Effect.gen(function* () {
279 lives.delete(key);
280 const now = yield* Clock.currentTimeMillis;
281 const quietMs = now - life.pushedAt;
282 yield* Effect.logInfo("pull request watch ended", {
283 threadId: life.threadId,
284 pullRequest: life.pullRequest,
285 reason,
286 minutes: minutes(now - life.startedAt),
287 quietMinutes: minutes(quietMs),
288 longestQuietMinutes: minutes(Math.max(life.longestQuietMs, quietMs)),
289 wakes: life.wakes,
290 reads: life.reads,
291 partial: life.startedAt < bootedAt,
292 });
293 });
294 const ended = (target: WatchTarget, reason: WatchEndReason) =>
295 Effect.suspend(() => reportEnd(lifeKey(target), lifeOf(target), reason));
296
297 // Reads in a row that failed, per pull request. Kept in memory: a restart only delays the stop.
298 const readFailures = new Map<string, number>();
299 const lastReads = new Map<string, LastRead>();
300 // Replies past each long thread's first page, per pull request, so a pass pages a thread
301 // again only when the host's count of it moves. A restart pages each thread once more.
302 const threadTails = new Map<
303 string,
304 Map<string, { readonly count: number; readonly comments: ReadonlyArray<PullRequestComment> }>
305 >();
306
307 // Host-level identity, with the repository as linked, the way pull request sync reads it.
308 const identityOf = (link: ThreadPullRequestLink) => ({
309 host: normalizeThreadPullRequestKey(link).host,
310 repository: link.repository,
311 number: link.number,
312 });
313
314 /**
315 * Records what a pass saw, and wakes the agent with it. The orchestrator applies this only
316 * while the same watch is on, so a stop or restart that lands during the host read wins.
317 */
318 const record = (
319 target: WatchTarget,
320 next: ThreadPullRequestWatch | null,
321 wake?: { readonly text: string; readonly notification: OrchestrationV2Notification },
322 ) =>
323 Effect.gen(function* () {
324 const uuid = yield* crypto.randomUUIDv4();
325 yield* engine.dispatch({
326 type: "thread.pull-request-watch.sync",
327 commandId: CommandId.make(`server:pr-watch:${target.thread.id}:${uuid}`),
328 threadId: target.thread.id,
329 ...identityOf(target.link),
330 startedAt: target.watch.startedAt,
331 watch: next,
332 ...(wake === undefined
333 ? {}
334 : { wake: { ...wake, messageId: MessageId.make(`message:pr-watch:${uuid}`) } }),
335 });
336 });
337
338 // A watch that cannot read its pull request ends with a wake saying so, rather than showing
339 // "Watching" while it learns nothing.
340 const giveUp = (target: WatchTarget) =>
341 record(target, null, {
342 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 to watch it again.`,
343 notification: {
344 source: { kind: "monitor" },
345 outcome: "failed",
346 summary: `#${target.link.number}: stopped watching, could not read it`,
347 },
348 }).pipe(
349 Effect.tap(() => woke(target)),
350 Effect.catch(() => record(target, null)),
351 Effect.tap(() => ended(target, "unreadable")),
352 );
353
354 const readRemarks = Effect.fn("PullRequestWatchReactor.readRemarks")(
355 function* (key: string, reference: PullRequestRef, activity: PullRequestActivity) {
356 // Comment cursors cannot account for missing threads. Only finish a truncated read
357 // when the host confirms that every thread was listed.
358 if (activity.commentsTruncated && activity.reviewThreadsTruncated !== false) return null;
359
360 let tails = threadTails.get(key);
361 if (tails === undefined) {
362 tails = new Map();
363 threadTails.set(key, tails);
364 }
365 const remarks = [...activity.comments];
366 for (const thread of activity.reviewThreads) {
367 let cursor = thread.nextCommentsCursor ?? null;
368 if (cursor === null) continue;
369 const count = thread.commentCount ?? 0;
370 let tail = tails.get(thread.id);
371 if (tail?.count !== count) {
372 const comments = new Map<string, PullRequestComment>();
373 const cursors = new Set<string>();
374 while (cursor !== null) {
375 if (cursors.has(cursor)) return null;
376 cursors.add(cursor);
377 const page: PullRequestThreadCommentsResult = yield* pullRequests.threadComments({
378 ...reference,
379 threadId: thread.id,
380 cursor,
381 });
382 for (const comment of page.comments) {
383 comments.set(comment.id, {
384 ...comment,
385 kind: "review-comment",
386 path: thread.path,
387 reviewState: null,
388 });
389 }
390 cursor = page.nextCursor;
391 }
392 if (thread.comments.length + comments.size < count) return null;
393 tail = { count, comments: [...comments.values()] };
394 tails.set(thread.id, tail);
395 }
396 remarks.push(...tail.comments);
397 }
398 return remarks.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
399 },
400 Effect.catch((error) =>
401 Effect.logWarning("pull request watch comment pagination failed", { error }).pipe(
402 Effect.as(null),
403 ),
404 ),
405 );
406
407 // A closed pull request can reopen, but the watch has nothing to report until then.
408 const closed = (target: WatchTarget) =>
409 record(target, null, {
410 text: `Pull request #${target.link.number} (${target.link.url}) was closed, so T3 Code stopped watching it. Call watch_pull_request if it reopens.`,
411 notification: {
412 source: { kind: "monitor" },
413 outcome: "updated",
414 summary: `#${target.link.number}: closed, stopped watching`,
415 },
416 }).pipe(
417 Effect.tap(() => woke(target)),
418 Effect.tap(() => ended(target, "closed")),
419 );
420
421 const merged = (target: WatchTarget) =>
422 record(target, null, {
423 text: `Pull request #${target.link.number} (${target.link.url}) merged. T3 Code stopped watching it. Continue any work you were waiting to do after this merge.`,
424 notification: {
425 source: { kind: "monitor" },
426 outcome: "completed",
427 summary: `Pull request #${target.link.number} merged`,
428 },
429 }).pipe(
430 Effect.tap(() => woke(target)),
431 Effect.tap(() => ended(target, "merged")),
432 );
433
434 /** Why a watch ends without a host read; the rest are read once per pull request. */
435 const isWorkLink = (link: ThreadPullRequestLink) =>
436 normalizeThreadPullRequestKey(link).host === "github.com" &&
437 isMergebotRepository(mergebot.configuration ?? undefined, link.repository);
438
439 const endsWithoutRead = ({ thread, link }: WatchTarget): WatchEndReason | undefined =>
440 // Inactive threads discard even a merged watch without waking.
441 thread.settledOverride === "settled" || thread.settledAt !== null
442 ? "settled"
443 : thread.lineage.relationshipToParent === "subagent"
444 ? "subagent"
445 : link.snapshot?.state === "merged" && !isWorkLink(link)
446 ? "merged"
447 : undefined;
448
449 /**
450 * Runs one thread's step for each thread in a group, so one refusal does not skip the rest.
451 * Succeeds with whether every step landed.
452 */
453 const eachTarget = <E>(
454 group: WatchGroup,
455 step: (target: WatchTarget) => Effect.Effect<void, E>,
456 ) =>
457 Effect.forEach(group.targets, (target) =>
458 step(target).pipe(
459 Effect.as(true),
460 Effect.catchCause((cause) => {
461 // A thread that did not get its update must not wait for the quiet reread.
462 lastReads.delete(group.key);
463 return logFailure("pull request watch update failed", {
464 threadId: target.thread.id,
465 pullRequest: group.key,
466 })(cause).pipe(Effect.as(false));
467 }),
468 ),
469 ).pipe(Effect.map((landed) => landed.every(Boolean)));
470
471 const referenceOf = (group: WatchGroup) => {
472 const first = group.targets[0]!;
473 return { projectId: first.thread.projectId, ...identityOf(first.link) };
474 };
475
476 /**
477 * The host fingerprint of a watched pull request, read for every group at once so they share
478 * one batched request. Null, for a host without one or one that gave no answer, keeps the
479 * sync-snapshot gating; "paused" skips the pass while the host is rate limited, since every
480 * read would be refused.
481 */
482 const fingerprintOf = Effect.fnUntraced(function* (
483 group: WatchGroup,
484 ): Effect.fn.Return<ProviderChangeRequestWatchFingerprint | null | "paused"> {
485 if (isWorkLink(group.targets[0]!.link)) {
486 const value = yield* mergebot.readPullRequest(referenceOf(group));
487 return value === null ? "paused" : mergebotWatchFingerprint(value);
488 }
489 return yield* pullRequests.watchFingerprint(referenceOf(group)).pipe(
490 Effect.catchCause(
491 (cause): Effect.Effect<ProviderChangeRequestWatchFingerprint | null | "paused"> =>
492 Cause.hasInterruptsOnly(cause)
493 ? Effect.interrupt
494 : isRateLimited(cause)
495 ? Effect.succeed("paused")
496 : // The snapshot-gated reads report a host that cannot be read, and count toward
497 // giving up.
498 Effect.logDebug("pull request watch fingerprint failed", {
499 pullRequest: group.key,
500 cause,
501 }).pipe(Effect.as(null)),
502 ),
503 );
504 });
505
506 const readGroup = Effect.fn("PullRequestWatchReactor.readGroup")(function* (
507 group: WatchGroup,
508 fingerprint: ProviderChangeRequestWatchFingerprint | null,
509 ) {
510 const now = yield* Clock.currentTimeMillis;
511 const last = lastReads.get(group.key);
512 const plan = planRead(group, last, fingerprint, now);
513 if (!plan.detail) return;
514 const reference = referenceOf(group);
515 // The reads are recorded against this fingerprint, so a cached answer older than it must not
516 // stand in for them. Clients are not told: this only drops this pull request's cached reads.
517 const work = isWorkLink(group.targets[0]!.link);
518 const native = work ? yield* mergebot.readPullRequest(reference) : null;
519 if (work && native === null) return;
520 if (!work && fingerprint !== null) yield* pullRequests.invalidate({ reference });
521 const reading: Effect.Effect<
522 readonly [
523 Pick<
524 PullRequestDetail,
525 "state" | "headSha" | "baseBranch" | "mergeability" | "checks" | "author" | "viewer"
526 >,
527 Option.Option<PullRequestActivity>,
528 ],
529 PullRequestService.PullRequestError
530 > =
531 native !== null
532 ? Effect.succeed([
533 {
534 state: native.lifecycle,
535 headSha: native.headSha,
536 baseBranch: native.baseBranch,
537 mergeability: mergebotMergeability(native),
538 checks: mergebotChecks(native),
539 author: native.author,
540 },
541 Option.none<PullRequestActivity>(),
542 ] as const)
543 : Effect.all(
544 [
545 pullRequests.detail({ ...reference, allowStale: false }),
546 plan.activity
547 ? pullRequests.activity(reference).pipe(Effect.map(Option.some))
548 : Effect.succeed(Option.none<PullRequestActivity>()),
549 ],
550 { concurrency: 2 },
551 );
552 const read = yield* Effect.exit(reading);
553 if (Exit.isFailure(read)) {
554 if (Cause.hasInterruptsOnly(read.cause)) return yield* Effect.failCause(read.cause);
555 lastReads.delete(group.key);
556 // The host's pause refuses later reads without a request, so waiting it out is free.
557 if (isRateLimited(read.cause)) return;
558 const failures = (readFailures.get(group.key) ?? 0) + 1;
559 readFailures.set(group.key, failures);
560 // The count stays until every stop lands, so a failed stop is tried again on the next
561 // failed read, and a watch started after the stops begins from zero.
562 if (failures >= READ_FAILURE_LIMIT && (yield* eachTarget(group, giveUp))) {
563 readFailures.delete(group.key);
564 }
565 return yield* Effect.failCause(read.cause);
566 }
567 readFailures.delete(group.key);
568 const [detail, activity] = read.value;
569 const headSha = detail.headSha ?? null;
570 for (const target of group.targets) {
571 const life = lifeOf(target);
572 // The first read only learns the head, so it is not a push.
573 const pushed = life.headSha !== null && headSha !== life.headSha;
574 lives.set(lifeKey(target), {
575 ...life,
576 headSha,
577 reads: life.reads + 1,
578 ...(pushed
579 ? { pushedAt: now, longestQuietMs: Math.max(life.longestQuietMs, now - life.pushedAt) }
580 : {}),
581 });
582 }
583 if (detail.state !== "open") {
584 lastReads.delete(group.key);
585 return yield* eachTarget(group, (target) =>
586 detail.state === "closed" ? closed(target) : merged(target),
587 );
588 }
589
590 // Never advance the remark watermark past comments an incomplete read could have missed.
591 // A pass that did not read the activity has no remarks to report.
592 const remarks =
593 native !== null
594 ? native.comments
595 : Option.isSome(activity)
596 ? yield* readRemarks(group.key, reference, activity.value)
597 : null;
598 lastReads.set(group.key, {
599 at: plan.activity || last === undefined ? now : last.at,
600 fingerprint: group.fingerprint,
601 inFlight:
602 detail.mergeability === "unknown" ||
603 detail.checks.some((check) => check.status === "pending"),
604 // Comments a partial read could not see are read again next pass.
605 remarksComplete: plan.activity ? remarks !== null : (last?.remarksComplete ?? false),
606 watches: new Set(group.targets.map(watchKey)),
607 status: fingerprint?.status ?? null,
608 remarks: fingerprint?.remarks ?? null,
609 });
610 yield* eachTarget(group, (target) => {
611 const report = evaluatePullRequestWatch(target.watch, detail, remarks);
612 if (report.changes.length > 0) {
613 return record(
614 target,
615 report.exhausted ? null : report.next,
616 pullRequestWatchMessage({
617 number: target.link.number,
618 url: target.link.url,
619 baseBranch: detail.baseBranch,
620 headSha: report.next.headSha,
621 report,
622 }),
623 ).pipe(
624 Effect.tap(() => woke(target)),
625 Effect.tap(() => (report.exhausted ? ended(target, "comment-limit") : Effect.void)),
626 );
627 }
628 return watchesEqual(report.next, target.watch) ? Effect.void : record(target, report.next);
629 });
630 });
631
632 const sweep = Effect.gen(function* () {
633 const threads = yield* projections.getThreadsWithPullRequests();
634 const targets = threads.flatMap((thread) =>
635 visibleThreadPullRequests(thread.pullRequests ?? []).flatMap((link) =>
636 link.watch === undefined ? [] : [{ thread, link, watch: link.watch }],
637 ),
638 );
639 // A watch seen last pass and gone now was ended outside this reactor.
640 const present = new Set(targets.map(lifeKey));
641 yield* Effect.forEach(
642 [...lives].filter(([key]) => !present.has(key)),
643 ([key, life]) => reportEnd(key, life, "stopped"),
644 { discard: true },
645 );
646 for (const target of targets) lifeOf(target);
647 const byPullRequest = new Map<string, Array<WatchTarget>>();
648 const ending: Array<readonly [WatchTarget, WatchEndReason]> = [];
649 for (const target of targets) {
650 const reason = endsWithoutRead(target);
651 if (reason !== undefined) {
652 ending.push([target, reason]);
653 continue;
654 }
655 // Grouped per project too: each project reads through its own checkout, so one that cannot
656 // read the pull request must not end another project's watches.
657 const key = `${target.thread.projectId} ${threadPullRequestKeyOf(target.link)}`;
658 byPullRequest.set(key, [...(byPullRequest.get(key) ?? []), target]);
659 }
660 const groups = [...byPullRequest].map(([key, members]): WatchGroup => ({
661 key,
662 targets: members,
663 fingerprint: [...new Set(members.map(snapshotFingerprint))].toSorted().join("\n"),
664 }));
665 for (const cache of [readFailures, lastReads, threadTails]) {
666 for (const key of cache.keys()) if (!byPullRequest.has(key)) cache.delete(key);
667 }
668 yield* Effect.forEach(
669 ending,
670 ([target, reason]) =>
671 (reason === "merged"
672 ? merged(target)
673 : record(target, null).pipe(Effect.tap(() => ended(target, reason)))
674 ).pipe(
675 Effect.catchCause(
676 logFailure("pull request watch stop failed", {
677 threadId: target.thread.id,
678 pullRequest: threadPullRequestKeyOf(target.link),
679 }),
680 ),
681 ),
682 { discard: true },
683 );
684 const fingerprints = yield* Effect.forEach(groups, fingerprintOf, {
685 concurrency: "unbounded",
686 });
687 yield* Effect.forEach(
688 groups,
689 (group, index) => {
690 const fingerprint = fingerprints[index]!;
691 return fingerprint === "paused"
692 ? Effect.void
693 : readGroup(group, fingerprint).pipe(
694 Effect.catchCause(
695 logFailure("pull request watch check failed", { pullRequest: group.key }),
696 ),
697 );
698 },
699 { concurrency: 4, discard: true },
700 );
701 }).pipe(
702 Effect.catchCause(logFailure("pull request watch sweep failed", {})),
703 Effect.withSpan("PullRequestWatchReactor.sweep"),
704 sweepLock.withPermit,
705 );
706
707 const start: PullRequestWatchReactor["Service"]["start"] = () =>
708 forkParked(
709 Effect.all(
710 [
711 sweep.pipe(Effect.repeat(Schedule.spaced(`${SWEEP_MINUTES} minutes`)), Effect.asVoid),
712 mergebot.streamChanges.pipe(Stream.runForEach(() => sweep)),
713 ],
714 { concurrency: "unbounded", discard: true },
715 ),
716 );
717
718 return { start, sweep } satisfies PullRequestWatchReactor["Service"];
719});
720
721export const layer = Layer.effect(PullRequestWatchReactor, make);