1import {
2 WS_METHODS,
3 PullRequestActivity,
4 PullRequestDetail as PullRequestDetailSchema,
5 PullRequestPreview,
6 PullRequestReviewGuideResult,
7 PullRequestListResult,
8 PullRequestChecks,
9 PullRequestDiffResult,
10 PullRequestFilesViewedResult,
11 PullRequestLinkedThreadsResult,
12 type EnvironmentId,
13 type PullRequestActor,
14 type PullRequestActionInput,
15 type PullRequestDetail,
16 type PullRequestDiffInput,
17 type PullRequestRef,
18 type PullRequestMergeMethod,
19 PullRequestOperationError,
20 type PullRequestSummary,
21 type VcsStatusResult,
22} from "@t3tools/contracts";
23import * as Data from "effect/Data";
24import * as Effect from "effect/Effect";
25import * as Schema from "effect/Schema";
26import { persistPullRequestStats } from "./pullRequestStatsPersistence.ts";
27import { persistQuery } from "./queryPersistence.ts";
28import * as Option from "effect/Option";
29import * as SubscriptionRef from "effect/SubscriptionRef";
30import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
31
32import {
33 createAtomCommandScheduler,
34 createEnvironmentRpcCommand,
35 createEnvironmentRpcQueryAtomFamily,
36 createEnvironmentRpcSubscriptionAtomFamily,
37 createEnvironmentQueryAtomFamily,
38} from "./runtime.ts";
39import { createPullRequestRouter } from "./pullRequestRouting.ts";
40import * as PullRequestDiffLoader from "./pullRequestDiffHttp.ts";
41import type { EnvironmentRegistry } from "../connection/registry.ts";
42import * as EnvironmentSupervisor from "../connection/supervisor.ts";
43
44export {
45 type PullRequestDiffLoadError,
46 PullRequestDiffCredentialRejectedError,
47} from "./pullRequestDiffHttp.ts";
48export * as PullRequestDiffLoader from "./pullRequestDiffHttp.ts";
49
50/** @public Required to name the error in consumers' inferred pull request results. */
51export class EnvironmentHttpConnectionNotReadyError extends Data.TaggedError(
52 "EnvironmentHttpConnectionNotReadyError",
53)<{ readonly message: string }> {}
54
55const LINKED_PULL_REQUEST_IDLE_TTL_MS = 5_000;
56
57/** Keep confirmed edits on the same cached reference regardless of input property order. */
58function writableQueryFamily<A, E>(
59 family: (target: {
60 readonly environmentId: EnvironmentId;
61 readonly input: PullRequestRef;
62 }) => Atom.Atom<AsyncResult.AsyncResult<A, E>>,
63) {
64 const writable = Atom.family((source: Atom.Atom<AsyncResult.AsyncResult<A, E>>) =>
65 Atom.writable(
66 (get) => {
67 const result = get(source);
68 if (result._tag === "Success" && !result.waiting) return result;
69 const previous = get.self<AsyncResult.AsyncResult<A, E>>();
70 const value = Option.flatMap(previous, AsyncResult.value);
71 if (Option.isNone(value)) return result;
72 return result._tag === "Failure"
73 ? AsyncResult.failureWithPrevious(result.cause, { previous, waiting: result.waiting })
74 : AsyncResult.success<A, E>(value.value, result);
75 },
76 (context, value: AsyncResult.AsyncResult<A, E>) => context.setSelf(value),
77 (refresh) => refresh(source),
78 ).pipe(Atom.setIdleTTL(5 * 60_000)),
79 );
80 return ({
81 environmentId,
82 input: { projectId, host, repository, number, allowStale },
83 }: Parameters<typeof family>[0]) =>
84 writable(
85 family({
86 environmentId,
87 input: {
88 projectId,
89 ...(host === undefined ? {} : { host }),
90 repository,
91 number,
92 ...(allowStale === undefined ? {} : { allowStale }),
93 },
94 }),
95 );
96}
97
98/** Restart pre-mutation reads before patching so they cannot restore stale values. */
99function updateCached<A, E>(
100 registry: AtomRegistry.AtomRegistry,
101 atom: Atom.Writable<AsyncResult.AsyncResult<A, E>>,
102 update: (value: A) => A,
103 refresh = false,
104) {
105 if (refresh || registry.get(atom).waiting) registry.refresh(atom);
106 registry.update(atom, AsyncResult.map(update));
107}
108
109function createPullRequestRefreshAtomFamily<R, E>(
110 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
111) {
112 return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
113 label: "environment-data:pull-requests:turn-refreshes",
114 tag: WS_METHODS.pullRequestsSubscribeRefreshes,
115 });
116}
117
118/** Refresh only the live fields a linked thread renders. A merged pull request is not re-read. */
119export function createLinkedPullRequestSummaryAtomFamily<R, E>(
120 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
121 refreshes = createPullRequestRefreshAtomFamily(runtime),
122) {
123 const routedRequest = createPullRequestRouter();
124 return createEnvironmentRpcQueryAtomFamily(runtime, {
125 label: "environment-data:pull-requests:linked-summary",
126 tag: WS_METHODS.pullRequestsSummary,
127 execute: (input) => routedRequest(WS_METHODS.pullRequestsSummary, input),
128 staleTimeMs: 60_000,
129 refreshIntervalMs: 60_000,
130 idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS,
131 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
132 isFinal: (summary) => summary.state === "merged",
133 });
134}
135
136/** The host-native stack a pull request belongs to; null where it is not stacked. */
137export function createPullRequestStackAtomFamily<R, E>(
138 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
139 refreshes = createPullRequestRefreshAtomFamily(runtime),
140) {
141 const routedRequest = createPullRequestRouter();
142 return createEnvironmentRpcQueryAtomFamily(runtime, {
143 label: "environment-data:pull-requests:stack",
144 tag: WS_METHODS.pullRequestsStack,
145 execute: (input) => routedRequest(WS_METHODS.pullRequestsStack, input),
146 staleTimeMs: 60_000,
147 idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS,
148 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
149 });
150}
151
152export function pullRequestDetailToVcsStatus(
153 detail: PullRequestDetail | PullRequestSummary,
154): NonNullable<VcsStatusResult["pr"]> {
155 return {
156 number: detail.number,
157 title: detail.title,
158 url: detail.url,
159 baseRef: detail.baseBranch,
160 headRef: detail.headBranch,
161 state: detail.state,
162 ...(detail.isDraft === true ? { isDraft: true } : {}),
163 updatedAt: detail.updatedAt,
164 };
165}
166
167/**
168 * Reopening a PR within a minute reuses detail and activity. Explicit refreshes and
169 * turn notifications still revalidate. Mutations run serially per environment: actions on the same
170 * pull request are order-sensitive. Confirmed label and reviewer edits update cached state.
171 */
172export function createPullRequestEnvironmentAtoms<R, E>(
173 runtime: Atom.AtomRuntime<
174 EnvironmentRegistry | PullRequestDiffLoader.PullRequestDiffLoader | R,
175 E
176 >,
177) {
178 const refreshes = createPullRequestRefreshAtomFamily(runtime);
179 const commandScheduler = createAtomCommandScheduler();
180 const routedRequest = createPullRequestRouter();
181 const serialPerEnvironment = {
182 mode: "serial",
183 key: ({ environmentId }: { readonly environmentId: string }) => environmentId,
184 } as const;
185 const activity = writableQueryFamily(
186 createEnvironmentRpcQueryAtomFamily(runtime, {
187 label: "environment-data:pull-requests:activity",
188 persistence: persistQuery(PullRequestActivity),
189 tag: WS_METHODS.pullRequestsActivity,
190 execute: (input) => routedRequest(WS_METHODS.pullRequestsActivity, input),
191 staleTimeMs: 60_000,
192 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
193 }),
194 );
195 const detail = writableQueryFamily(
196 createEnvironmentRpcQueryAtomFamily(runtime, {
197 label: "environment-data:pull-requests:detail",
198 persistence: persistQuery(PullRequestDetailSchema),
199 tag: WS_METHODS.pullRequestsDetail,
200 execute: (input) => routedRequest(WS_METHODS.pullRequestsDetail, input),
201 staleTimeMs: 60_000,
202 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
203 }),
204 );
205 const preview = writableQueryFamily(
206 createEnvironmentRpcQueryAtomFamily(runtime, {
207 label: "environment-data:pull-requests:preview",
208 persistence: persistQuery(PullRequestPreview),
209 tag: WS_METHODS.pullRequestsPreview,
210 execute: (input) => routedRequest(WS_METHODS.pullRequestsPreview, input),
211 staleTimeMs: 60_000,
212 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
213 }),
214 );
215 const labelCandidates = writableQueryFamily(
216 createEnvironmentRpcQueryAtomFamily(runtime, {
217 label: "environment-data:pull-requests:label-candidates",
218 tag: WS_METHODS.pullRequestsLabelCandidates,
219 execute: (input) => routedRequest(WS_METHODS.pullRequestsLabelCandidates, input),
220 staleTimeMs: 60_000,
221 }),
222 );
223 // The stored guide only changes when somebody generates one, which refreshes it below.
224 const reviewGuide = writableQueryFamily(
225 createEnvironmentRpcQueryAtomFamily(runtime, {
226 label: "environment-data:pull-requests:review-guide",
227 persistence: persistQuery(PullRequestReviewGuideResult),
228 tag: WS_METHODS.pullRequestsReviewGuide,
229 execute: (input) => routedRequest(WS_METHODS.pullRequestsReviewGuide, input),
230 staleTimeMs: 60_000,
231 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
232 }),
233 );
234 const reviewerCandidates = writableQueryFamily(
235 createEnvironmentRpcQueryAtomFamily(runtime, {
236 label: "environment-data:pull-requests:reviewer-candidates",
237 tag: WS_METHODS.pullRequestsReviewerCandidates,
238 execute: (input) => routedRequest(WS_METHODS.pullRequestsReviewerCandidates, input),
239 staleTimeMs: 60_000,
240 }),
241 );
242 return {
243 refreshes,
244 linkedThreads: createEnvironmentRpcQueryAtomFamily(runtime, {
245 label: "environment-data:pull-requests:linked-threads",
246 persistence: persistQuery(PullRequestLinkedThreadsResult),
247 tag: WS_METHODS.pullRequestsLinkedThreads,
248 staleTimeMs: 0,
249 refreshIntervalMs: 10_000,
250 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
251 }),
252 list: createEnvironmentRpcQueryAtomFamily(runtime, {
253 label: "environment-data:pull-requests:list",
254 persistence: persistQuery(PullRequestListResult),
255 tag: WS_METHODS.pullRequestsList,
256 staleTimeMs: 30_000,
257 refreshTrigger: ({ environmentId, input }) =>
258 input.cursors === undefined ? refreshes({ environmentId, input: {} }) : undefined,
259 }),
260 /**
261 * The line counts for rows the listing has already handed over. Its own query because the
262 * listing is quicker without them — measured over twelve repositories, ~4.0s against ~7.1s —
263 * so the rows arrive first and their stats a moment later. Kept longer than the listing:
264 * a change request's size only moves when somebody pushes to it.
265 */
266 listStats: createEnvironmentRpcQueryAtomFamily(runtime, {
267 label: "environment-data:pull-requests:list-stats",
268 persistence: persistPullRequestStats(),
269 tag: WS_METHODS.pullRequestsListStats,
270 staleTimeMs: 60_000,
271 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
272 }),
273 detail,
274 preview,
275 checks: createEnvironmentRpcQueryAtomFamily(runtime, {
276 label: "environment-data:pull-requests:checks",
277 persistence: persistQuery(Schema.NullOr(PullRequestChecks)),
278 tag: WS_METHODS.pullRequestsChecks,
279 execute: (input) => routedRequest(WS_METHODS.pullRequestsChecks, input),
280 staleTimeMs: 45_000,
281 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
282 }),
283 activity,
284 threadComments: createEnvironmentRpcCommand(runtime, {
285 label: "environment-data:pull-requests:thread-comments",
286 tag: WS_METHODS.pullRequestsThreadComments,
287 execute: (input) => routedRequest(WS_METHODS.pullRequestsThreadComments, input),
288 scheduler: commandScheduler,
289 concurrency: {
290 mode: "singleFlight",
291 key: ({ environmentId, input }) =>
292 JSON.stringify([environmentId, input.threadId, input.cursor]),
293 },
294 }),
295 diff: createEnvironmentQueryAtomFamily(runtime, {
296 label: "environment-data:pull-requests:diff",
297 persistence: persistQuery(PullRequestDiffResult),
298 staleTimeMs: 60_000,
299 execute: (input: PullRequestDiffInput) =>
300 Effect.gen(function* () {
301 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
302 const loader = yield* PullRequestDiffLoader.PullRequestDiffLoader;
303 const prepared = yield* SubscriptionRef.get(supervisor.prepared);
304 if (Option.isNone(prepared)) {
305 return yield* new EnvironmentHttpConnectionNotReadyError({
306 message: "The environment HTTP connection is not ready.",
307 });
308 }
309 return yield* loader.load(prepared.value, input);
310 }),
311 }),
312 diffFileContents: createEnvironmentRpcCommand(runtime, {
313 label: "environment-data:pull-requests:diff-file-contents",
314 tag: WS_METHODS.pullRequestsDiffFileContents,
315 execute: (input) => routedRequest(WS_METHODS.pullRequestsDiffFileContents, input),
316 scheduler: commandScheduler,
317 concurrency: {
318 mode: "singleFlight",
319 key: ({ environmentId, input }) =>
320 JSON.stringify([
321 environmentId,
322 input.projectId,
323 input.host?.toLowerCase() ?? null,
324 input.repository,
325 input.number,
326 input.commit ?? null,
327 input.changeType,
328 input.oldPath,
329 input.newPath,
330 ]),
331 },
332 }),
333 filesViewed: createEnvironmentRpcQueryAtomFamily(runtime, {
334 label: "environment-data:pull-requests:files-viewed",
335 persistence: persistQuery(PullRequestFilesViewedResult),
336 tag: WS_METHODS.pullRequestsFilesViewed,
337 execute: (input) => routedRequest(WS_METHODS.pullRequestsFilesViewed, input),
338 staleTimeMs: 15_000,
339 }),
340 /**
341 * One write in flight per change request: the host applies these in order, and a reader
342 * ticking down a file list faster than the round trip would otherwise race their own presses.
343 */
344 setFilesViewed: createEnvironmentRpcCommand(runtime, {
345 label: "environment-data:pull-requests:set-files-viewed",
346 tag: WS_METHODS.pullRequestsSetFilesViewed,
347 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetFilesViewed, input),
348 scheduler: commandScheduler,
349 concurrency: {
350 mode: "serial",
351 key: ({ environmentId, input }) =>
352 JSON.stringify([environmentId, input.projectId, input.repository, input.number]),
353 },
354 }),
355 runAction: createEnvironmentRpcCommand(runtime, {
356 tag: WS_METHODS.pullRequestsRunAction,
357 label: "environment-data:pull-requests:run-action",
358 // Preparation belongs to the write's lane. Refreshable queries would restart it after
359 // every preceding action, and preparing outside the lane could reorder the clicks.
360 execute: (
361 input: PullRequestActionInput & {
362 readonly resolveMergeMethod?: (detail: PullRequestDetail) => PullRequestMergeMethod;
363 },
364 ) =>
365 Effect.gen(function* () {
366 const { resolveMergeMethod, ...actionInput } = input;
367 let preparedInput = actionInput;
368 if (actionInput.action === "merge" && resolveMergeMethod !== undefined) {
369 const { projectId, host, repository, number } = actionInput;
370 const reference = { projectId, host, repository, number, allowStale: false };
371 const detail = yield* routedRequest(WS_METHODS.pullRequestsDetail, reference);
372 if (
373 detail.state !== "open" ||
374 detail.isDraft ||
375 !detail.capabilities.actions.includes("merge") ||
376 !detail.viewerPermissions.actions.includes("merge")
377 ) {
378 return yield* new PullRequestOperationError({
379 operation: "runAction",
380 detail: "This pull request cannot be merged.",
381 });
382 }
383 if (detail.capabilities.stackActions) {
384 const stack = yield* routedRequest(WS_METHODS.pullRequestsStack, reference);
385 if (stack !== null) {
386 return yield* new PullRequestOperationError({
387 operation: "runAction",
388 detail: "Open this pull request to merge its stack.",
389 });
390 }
391 }
392 const mergeMethod = yield* Effect.try({
393 try: () => resolveMergeMethod(detail),
394 catch: (cause) =>
395 new PullRequestOperationError({
396 operation: "runAction",
397 detail:
398 cause instanceof Error ? cause.message : "Could not choose a merge method.",
399 }),
400 });
401 preparedInput = { ...actionInput, mergeMethod };
402 }
403 return yield* routedRequest(WS_METHODS.pullRequestsRunAction, preparedInput);
404 }),
405 onSuccess: ({ environmentId, input }, registry) =>
406 Effect.sync(() => registry.refresh(preview({ environmentId, input }))),
407 scheduler: commandScheduler,
408 concurrency: serialPerEnvironment,
409 }),
410 update: createEnvironmentRpcCommand(runtime, {
411 label: "environment-data:pull-requests:update",
412 tag: WS_METHODS.pullRequestsUpdate,
413 execute: (input) => routedRequest(WS_METHODS.pullRequestsUpdate, input),
414 onSuccess: (target, registry) => Effect.sync(() => registry.refresh(preview(target))),
415 scheduler: commandScheduler,
416 concurrency: serialPerEnvironment,
417 }),
418 comment: createEnvironmentRpcCommand(runtime, {
419 label: "environment-data:pull-requests:comment",
420 tag: WS_METHODS.pullRequestsComment,
421 execute: (input) => routedRequest(WS_METHODS.pullRequestsComment, input),
422 scheduler: commandScheduler,
423 concurrency: serialPerEnvironment,
424 }),
425 updateComment: createEnvironmentRpcCommand(runtime, {
426 label: "environment-data:pull-requests:update-comment",
427 tag: WS_METHODS.pullRequestsUpdateComment,
428 execute: (input) => routedRequest(WS_METHODS.pullRequestsUpdateComment, input),
429 scheduler: commandScheduler,
430 concurrency: serialPerEnvironment,
431 onSuccess: ({ environmentId, input: { projectId, host, repository, number } }, registry) =>
432 Effect.sync(() =>
433 registry.refresh(
434 activity({
435 environmentId,
436 input: { projectId, ...(host === undefined ? {} : { host }), repository, number },
437 }),
438 ),
439 ),
440 }),
441 submitReview: createEnvironmentRpcCommand(runtime, {
442 label: "environment-data:pull-requests:submit-review",
443 tag: WS_METHODS.pullRequestsSubmitReview,
444 execute: (input) => routedRequest(WS_METHODS.pullRequestsSubmitReview, input),
445 scheduler: commandScheduler,
446 concurrency: serialPerEnvironment,
447 }),
448 replyToThread: createEnvironmentRpcCommand(runtime, {
449 label: "environment-data:pull-requests:reply-to-thread",
450 tag: WS_METHODS.pullRequestsReplyToThread,
451 execute: (input) => routedRequest(WS_METHODS.pullRequestsReplyToThread, input),
452 scheduler: commandScheduler,
453 concurrency: serialPerEnvironment,
454 }),
455 /**
456 * Its own query rather than part of the detail: the people who may be asked are only wanted
457 * once somebody opens the reviewer menu, so this atom is read then and not before. Kept fresh
458 * for a minute, because who has access to a repository changes far more slowly than the
459 * change request it is being read for.
460 */
461 reviewerCandidates,
462 requestReviewers: createEnvironmentRpcCommand(runtime, {
463 label: "environment-data:pull-requests:request-reviewers",
464 tag: WS_METHODS.pullRequestsRequestReviewers,
465 execute: (input) => routedRequest(WS_METHODS.pullRequestsRequestReviewers, input),
466 scheduler: commandScheduler,
467 concurrency: serialPerEnvironment,
468 onSuccess: (target, registry) =>
469 Effect.sync(() => {
470 const { reviewers, requested } = target.input;
471 const candidatesAtom = reviewerCandidates(target);
472 const candidates = Option.getOrNull(AsyncResult.value(registry.get(candidatesAtom)));
473 const selected =
474 candidates?.candidates.filter((candidate) =>
475 reviewers.some(
476 (reviewer) => reviewer.id === candidate.id && reviewer.kind === candidate.kind,
477 ),
478 ) ?? [];
479 const missingIdentities = selected.length < reviewers.length;
480 updateCached(registry, candidatesAtom, (value) => ({
481 ...value,
482 candidates: value.candidates.map((candidate) =>
483 selected.includes(candidate) ? { ...candidate, isRequested: requested } : candidate,
484 ),
485 }));
486 const selectedLogins = new Set(
487 selected.map((candidate) => candidate.login.toLowerCase()),
488 );
489 const updateReviewers = (
490 actors: ReadonlyArray<PullRequestActor>,
491 keep = (_actor: PullRequestActor) => false,
492 ) =>
493 requested
494 ? [
495 ...actors,
496 ...selected
497 .filter(
498 (candidate) =>
499 !actors.some(
500 (actor) => actor.login.toLowerCase() === candidate.login.toLowerCase(),
501 ),
502 )
503 .map(({ login, name, avatarUrl }) => ({ login, name, avatarUrl })),
504 ]
505 : actors.filter(
506 (actor) => !selectedLogins.has(actor.login.toLowerCase()) || keep(actor),
507 );
508 updateCached(
509 registry,
510 detail(target),
511 (value) => ({
512 ...value,
513 reviewers: updateReviewers(value.reviewers),
514 }),
515 missingIdentities,
516 );
517 updateCached(
518 registry,
519 activity(target),
520 (value) => ({
521 ...value,
522 reviewers:
523 value.reviewers === undefined
524 ? undefined
525 : updateReviewers(value.reviewers, (actor) =>
526 value.comments.some(
527 (comment) =>
528 (comment.kind === "review" || comment.kind === "review-comment") &&
529 comment.author?.login.toLowerCase() === actor.login.toLowerCase(),
530 ),
531 ),
532 }),
533 missingIdentities,
534 );
535 }),
536 }),
537 reviewGuide,
538 /** Runs a model on the environment; never fired without the reader asking for it. */
539 generateReviewGuide: createEnvironmentRpcCommand(runtime, {
540 label: "environment-data:pull-requests:generate-review-guide",
541 tag: WS_METHODS.pullRequestsGenerateReviewGuide,
542 execute: (input) => routedRequest(WS_METHODS.pullRequestsGenerateReviewGuide, input),
543 onSuccess: (target, registry, result) =>
544 Effect.sync(() => {
545 // The command returns the accepted run. Publish it before starting ends so readers
546 // never fall back to the empty/failed state while the follow-up query is in flight.
547 updateCached(registry, reviewGuide(target), () => result, true);
548 }),
549 }),
550 /** Read when the label menu opens, and kept for a minute, like the reviewer candidates. */
551 labelCandidates,
552 setLabels: createEnvironmentRpcCommand(runtime, {
553 label: "environment-data:pull-requests:set-labels",
554 tag: WS_METHODS.pullRequestsSetLabels,
555 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetLabels, input),
556 scheduler: commandScheduler,
557 concurrency: serialPerEnvironment,
558 onSuccess: (target, registry) =>
559 Effect.sync(() => {
560 const { labels, applied } = target.input;
561 const candidatesAtom = labelCandidates(target);
562 const candidates = Option.getOrNull(AsyncResult.value(registry.get(candidatesAtom)));
563 const names = new Set(labels);
564 updateCached(registry, candidatesAtom, (value) => ({
565 ...value,
566 candidates: value.candidates.map((candidate) =>
567 names.has(candidate.name) ? { ...candidate, isApplied: applied } : candidate,
568 ),
569 }));
570 updateCached(registry, detail(target), (value) => ({
571 ...value,
572 labels: applied
573 ? [
574 ...value.labels,
575 ...labels
576 .filter((name) => !value.labels.some((label) => label.name === name))
577 .map((name) => ({
578 name,
579 color:
580 candidates?.candidates.find((candidate) => candidate.name === name)
581 ?.color ?? null,
582 })),
583 ]
584 : value.labels.filter((label) => !names.has(label.name)),
585 }));
586 }),
587 }),
588 setThreadResolution: createEnvironmentRpcCommand(runtime, {
589 label: "environment-data:pull-requests:set-thread-resolution",
590 tag: WS_METHODS.pullRequestsSetThreadResolution,
591 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetThreadResolution, input),
592 scheduler: commandScheduler,
593 concurrency: serialPerEnvironment,
594 }),
595 setReaction: createEnvironmentRpcCommand(runtime, {
596 label: "environment-data:pull-requests:set-reaction",
597 tag: WS_METHODS.pullRequestsSetReaction,
598 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetReaction, input),
599 scheduler: commandScheduler,
600 concurrency: serialPerEnvironment,
601 }),
602 /**
603 * Explicit refresh: forget the server's cached answers, then re-run the reads. A separate
604 * request rather than a flag on a read, so only a person's refresh spends host requests
605 * while every silent re-read shares the cache.
606 */
607 invalidate: createEnvironmentRpcCommand(runtime, {
608 label: "environment-data:pull-requests:invalidate",
609 tag: WS_METHODS.pullRequestsInvalidate,
610 execute: (input) => routedRequest(WS_METHODS.pullRequestsInvalidate, input),
611 onSuccess: ({ environmentId, input }, registry) =>
612 Effect.sync(() => {
613 if (input.reference !== undefined) {
614 registry.refresh(preview({ environmentId, input: input.reference }));
615 }
616 }),
617 scheduler: commandScheduler,
618 concurrency: serialPerEnvironment,
619 }),
620 };
621}
622
623export function createMergebotQueueAtomFamily<R, E>(
624 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
625) {
626 return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
627 label: "mergebot:queue",
628 tag: WS_METHODS.mergebotSubscribe,
629 });
630}
631
632export function createMergebotPullRequestStatusAtomFamily<R, E>(
633 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
634) {
635 return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
636 label: "mergebot:pull-request-status",
637 tag: WS_METHODS.mergebotSubscribePullRequestStatus,
638 });
639}
640
641export function createMergebotDeploymentsAtomFamily<R, E>(
642 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
643) {
644 return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
645 label: "mergebot:deployments",
646 tag: WS_METHODS.mergebotSubscribeDeployments,
647 });
648}