Open History to see the last commit
1import {
2 WS_METHODS,
3 PullRequestActivity,
4 PullRequestDetail as PullRequestDetailSchema,
5 PullRequestPreview,
6 PullRequestReviewGuideResult,
7 PullRequestListResult,
8 PullRequestChecks,
9 PullRequestDiffResult,
10 PullRequestChangeBreakdown,
11 PullRequestFilesViewedResult,
12 PullRequestLinkedThreadsResult,
13 type EnvironmentId,
14 type PullRequestActor,
15 type PullRequestActionInput,
16 type PullRequestListInput,
17 type PullRequestListStatsInput,
18 type PullRequestListStatsResult,
19 type PullRequestDetail,
20 type PullRequestDiffInput,
21 type PullRequestListEntry,
22 type PullRequestRef,
23 type PullRequestMergeMethod,
24 PullRequestOperationError,
25 PullRequestSummary,
26 type VcsStatusResult,
27} from "@t3tools/contracts";
28import * as Effect from "effect/Effect";
29import * as Schema from "effect/Schema";
30import { persistPullRequestStats } from "./pullRequestStatsPersistence.ts";
31import { persistQuery, type QueryPersistence } from "./queryPersistence.ts";
32import * as Option from "effect/Option";
33import * as SubscriptionRef from "effect/SubscriptionRef";
34import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
35
36import {
37 createAtomCommandScheduler,
38 createEnvironmentRpcCommand,
39 createEnvironmentRpcQueryAtomFamily,
40 createEnvironmentRpcSubscriptionAtomFamily,
41 createEnvironmentQueryAtomFamily,
42 queryPersistenceKey,
43 type QueryAnswerHandler,
44} from "./runtime.ts";
45import { createPullRequestRouter } from "./pullRequestRouting.ts";
46import * as PullRequestDiffLoader from "./pullRequestDiffHttp.ts";
47import type { EnvironmentRegistry } from "../connection/registry.ts";
48import * as EnvironmentSupervisor from "../connection/supervisor.ts";
49import { isStalePullRequestReply } from "../pullRequestReply.ts";
50import { parsePullRequestQuery } from "../pullRequestQuery.ts";
51import { reuseUnchanged } from "./serverConfigProjection.ts";
52
53export {
54 type PullRequestDiffLoadError,
55 PullRequestDiffCredentialRejectedError,
56} from "./pullRequestDiffHttp.ts";
57export * as PullRequestDiffLoader from "./pullRequestDiffHttp.ts";
58
59/** @public Required to name the error in consumers' inferred pull request results. */
60export class EnvironmentHttpConnectionNotReadyError extends Schema.TaggedError<EnvironmentHttpConnectionNotReadyError>()(
61 "EnvironmentHttpConnectionNotReadyError",
62 { message: Schema.String },
63) {}
64
65const LINKED_PULL_REQUEST_IDLE_TTL_MS = 5_000;
66
67function normalizePullRequestQuery<A>(
68 query: (target: { readonly environmentId: EnvironmentId; readonly input: PullRequestRef }) => A,
69) {
70 return ({
71 environmentId,
72 input: { projectId, host, repository, number, allowStale, expectedAccountId },
73 }: Parameters<typeof query>[0]) =>
74 query({
75 environmentId,
76 input: {
77 projectId,
78 ...(host === undefined ? {} : { host }),
79 repository,
80 number,
81 ...(allowStale === undefined ? {} : { allowStale }),
82 ...(expectedAccountId === undefined ? {} : { expectedAccountId }),
83 },
84 });
85}
86
87function writableQuery<A, E>(
88 source: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
89 idleTtlMs = 5 * 60_000,
90) {
91 return Atom.writable(
92 (get) => {
93 const result = get(source);
94 if (result._tag === "Success" && !result.waiting) return result;
95 const previous = get.self<AsyncResult.AsyncResult<A, E>>();
96 const value = Option.flatMap(previous, AsyncResult.value);
97 if (Option.isNone(value)) return result;
98 return result._tag === "Failure"
99 ? AsyncResult.failureWithPrevious(result.cause, { previous, waiting: result.waiting })
100 : AsyncResult.success<A, E>(value.value, result);
101 },
102 (context, value: AsyncResult.AsyncResult<A, E>) => context.setSelf(value),
103 (refresh) => refresh(source),
104 ).pipe(Atom.setIdleTTL(idleTtlMs));
105}
106
107function writableQueryFamily<A, E>(
108 family: (target: {
109 readonly environmentId: EnvironmentId;
110 readonly input: PullRequestRef;
111 }) => Atom.Atom<AsyncResult.AsyncResult<A, E>>,
112) {
113 const writable = Atom.family((source: Atom.Atom<AsyncResult.AsyncResult<A, E>>) =>
114 writableQuery(source),
115 );
116 return normalizePullRequestQuery((target) => writable(family(target)));
117}
118
119/** Restart pre-mutation reads before patching so they cannot restore stale values. */
120function updateCached<A, E>(
121 registry: AtomRegistry.AtomRegistry,
122 atom: Atom.Writable<AsyncResult.AsyncResult<A, E>>,
123 update: (value: A) => A,
124 refresh = false,
125) {
126 if (refresh || registry.get(atom).waiting) registry.refresh(atom);
127 registry.update(atom, AsyncResult.map(update));
128}
129
130interface ActionReplyPatch {
131 /** The read this reader is waiting on, if any. */
132 readonly pendingRead: (registry: AtomRegistry.AtomRegistry) => object | undefined;
133 readonly apply: (
134 registry: AtomRegistry.AtomRegistry,
135 environmentId: EnvironmentId,
136 reference: PullRequestRef,
137 reply: PullRequestSummary | undefined,
138 readsBeforeSend: ReadonlySet<object>,
139 ) => Effect.Effect<void>;
140}
141
142// Module-wide so readers built by separate factories, such as linked summaries, are patched too.
143const actionReplyPatches = new WeakMap<object, ActionReplyPatch>();
144
145function samePullRequest(reader: PullRequestRef, acted: PullRequestRef) {
146 return (
147 reader.projectId === acted.projectId &&
148 // A missing host is not a wildcard: one project can reach the same repository on two hosts.
149 reader.host?.toLowerCase() === acted.host?.toLowerCase() &&
150 reader.repository.toLowerCase() === acted.repository.toLowerCase() &&
151 reader.number === acted.number &&
152 reader.expectedAccountId === acted.expectedAccountId
153 );
154}
155
156function pickDefined<T extends object, K extends keyof T>(
157 source: T,
158 keys: ReadonlyArray<K>,
159): { [P in K]?: Exclude<T[P], undefined> } {
160 const picked: { [P in K]?: T[P] } = {};
161 for (const key of keys) if (source[key] !== undefined) picked[key] = source[key];
162 return picked as { [P in K]?: Exclude<T[P], undefined> };
163}
164
165function listAccepts(input: PullRequestListInput, state: string, isDraft: boolean | undefined) {
166 const draft = parsePullRequestQuery(input.query ?? "").filters.draft ?? input.filters?.draft;
167 return (
168 (input.state === "all" || input.state === state) &&
169 (draft !== "only" || isDraft !== false) &&
170 (draft !== "hide" || isDraft !== true)
171 );
172}
173
174/**
175 * Listing rows with an action's reply applied, the same array when it changes none, and `refresh`
176 * when the reply adds the pull request to a listing, which only a read can place.
177 */
178export function patchPullRequestListEntries<Entry extends PullRequestListEntry>(
179 input: PullRequestListInput,
180 entries: ReadonlyArray<Entry>,
181 reference: PullRequestRef,
182 reply: PullRequestSummary,
183): ReadonlyArray<Entry> | "refresh" {
184 let found = false;
185 let changed = false;
186 const patched = entries.flatMap((entry): Entry[] => {
187 if (!samePullRequest(entry, reference)) return [entry];
188 found = true;
189 if (isStalePullRequestReply(entry, reply)) return [entry];
190 changed = true;
191 const next = {
192 ...entry,
193 ...pickDefined(reply, [
194 "title",
195 "url",
196 "author",
197 "headBranch",
198 "baseBranch",
199 "state",
200 "isDraft",
201 "mergeability",
202 "updatedAt",
203 ]),
204 };
205 return listAccepts(input, next.state, next.isDraft) ? [next] : [];
206 });
207 if (!found) {
208 return input.cursors === undefined && listAccepts(input, reply.state, reply.isDraft)
209 ? "refresh"
210 : entries;
211 }
212 return changed
213 ? patched.sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt))
214 : entries;
215}
216
217function patchListWithReply(
218 input: PullRequestListInput,
219 value: PullRequestListResult,
220 reference: PullRequestRef,
221 reply: PullRequestSummary,
222): PullRequestListResult | "refresh" {
223 const entries = patchPullRequestListEntries(input, value.entries, reference, reply);
224 if (entries === "refresh") return entries;
225 return entries === value.entries ? value : { ...value, entries };
226}
227
228/**
229 * The query atom for one target, writable so an action's reply can replace what it shows. The
230 * patch is saved and announced the way an answer would be, so cold starts and snapshots see it.
231 */
232function actionPatchedQuery<Input, A, E>(
233 source: (target: {
234 readonly environmentId: EnvironmentId;
235 readonly input: Input;
236 }) => Atom.Atom<AsyncResult.AsyncResult<A, E>>,
237 options: {
238 readonly label: string;
239 readonly persistence?: QueryPersistence<A> | undefined;
240 readonly onAnswer?: QueryAnswerHandler<Input, A> | undefined;
241 readonly idleTtlMs?: number | undefined;
242 readonly concerns: (input: Input, reference: PullRequestRef) => boolean;
243 readonly patch: (
244 input: Input,
245 value: A,
246 reference: PullRequestRef,
247 reply: PullRequestSummary,
248 ) => A | "refresh";
249 /** Fields the reply lacks also changed, so the patched value shows while a read fetches them. */
250 readonly rereads?: (value: A, reply: PullRequestSummary) => boolean;
251 /** Re-read when the host's reply did not carry the pull request. */
252 readonly rereadWithoutReply?: boolean;
253 /** False for reads the refresh stream does not restart; defaults to true. */
254 readonly refreshedByServer?: (input: Input) => boolean;
255 },
256) {
257 const writable = Atom.family((atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>) =>
258 writableQuery(atom, options.idleTtlMs),
259 );
260 return (target: { readonly environmentId: EnvironmentId; readonly input: Input }) => {
261 const query = source(target);
262 const atom = writable(query);
263 if (actionReplyPatches.has(atom)) return atom;
264 const pendingRead = (registry: AtomRegistry.AtomRegistry) => {
265 const result = Option.getOrUndefined(
266 registry.getNodes().get(query)?.valueOption() ?? Option.none(),
267 ) as AsyncResult.AsyncResult<A, E> | undefined;
268 return result?.waiting === true ? result : undefined;
269 };
270 actionReplyPatches.set(atom, {
271 pendingRead,
272 apply: Effect.fnUntraced(
273 function* (registry, environmentId, reference, reply, readsBeforeSend) {
274 if (environmentId !== target.environmentId || !options.concerns(target.input, reference))
275 return;
276 const pending = pendingRead(registry);
277 // A read sent before the action may miss its result, and so may one sent before the host
278 // committed. The server's post-commit refresh restarts that one, but not a read outside
279 // its stream, whose re-read need not even show as waiting, so a mounted one the reply
280 // changes restarts here.
281 const unrefreshed = options.refreshedByServer?.(target.input) === false;
282 const freshRead = pending !== undefined && !unrefreshed && !readsBeforeSend.has(pending);
283 const reread = () => (freshRead ? undefined : registry.refresh(atom));
284 if (reply === undefined) {
285 return options.rereadWithoutReply === true ? reread() : undefined;
286 }
287 const shown = Option.flatMap(
288 registry.getNodes().get(atom)?.valueOption() ?? Option.none(),
289 (current: AsyncResult.AsyncResult<A, E>) => AsyncResult.value(current),
290 );
291 if (Option.isNone(shown)) return reread();
292 const patched = options.patch(target.input, shown.value, reference, reply);
293 if (patched === "refresh") return reread();
294 const lacking = options.rereads?.(shown.value, reply) === true;
295 const next = reuseUnchanged(shown.value, patched);
296 if (next === shown.value)
297 return lacking || (unrefreshed && pending !== undefined) ? reread() : undefined;
298 yield* options.onAnswer?.(target, next, registry) ?? Effect.void;
299 const mounted = (registry.getNodes().get(atom)?.listeners.size ?? 0) > 0;
300 if (!freshRead && (pending !== undefined || lacking || (unrefreshed && mounted)))
301 registry.refresh(atom);
302 registry.update(
303 atom,
304 AsyncResult.map(() => next),
305 );
306 if (options.persistence !== undefined) {
307 yield* options.persistence.save(
308 environmentId,
309 queryPersistenceKey(options.label, target.input),
310 next,
311 );
312 }
313 },
314 ),
315 });
316 return atom;
317 };
318}
319
320function makePullRequestRefreshAtomFamily<R, E>(
321 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
322) {
323 return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
324 label: "environment-data:pull-requests:turn-refreshes",
325 tag: WS_METHODS.pullRequestsSubscribeRefreshes,
326 });
327}
328
329const refreshFamiliesByRuntime = new WeakMap<
330 object,
331 ReturnType<typeof makePullRequestRefreshAtomFamily>
332>();
333
334function createPullRequestRefreshAtomFamily<R, E>(
335 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
336): ReturnType<typeof makePullRequestRefreshAtomFamily<R, E>> {
337 const existing = refreshFamiliesByRuntime.get(runtime);
338 if (existing) return existing as ReturnType<typeof makePullRequestRefreshAtomFamily<R, E>>;
339 const refreshes = makePullRequestRefreshAtomFamily(runtime);
340 refreshFamiliesByRuntime.set(runtime, refreshes);
341 return refreshes;
342}
343
344export function createLinkedPullRequestSummaryAtomFamily<R, E>(
345 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
346 refreshes = createPullRequestRefreshAtomFamily(runtime),
347 onAnswer?: QueryAnswerHandler<PullRequestRef, PullRequestSummary>,
348) {
349 const routedRequest = createPullRequestRouter();
350 const label = "environment-data:pull-requests:linked-summary";
351 const persistence = persistQuery(PullRequestSummary);
352 return actionPatchedQuery(
353 createEnvironmentRpcQueryAtomFamily(runtime, {
354 label,
355 persistence,
356 tag: WS_METHODS.pullRequestsSummary,
357 execute: (input) => routedRequest(WS_METHODS.pullRequestsSummary, input),
358 staleTimeMs: 60_000,
359 refreshIntervalMs: 60_000,
360 idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS,
361 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
362 isFinal: (summary) => summary.state === "merged",
363 ...(onAnswer === undefined ? {} : { onAnswer }),
364 }),
365 {
366 label,
367 persistence,
368 onAnswer,
369 idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS,
370 concerns: samePullRequest,
371 // A routed action replies with the destination's project and repository names.
372 patch: (_input, value, _reference, reply) =>
373 isStalePullRequestReply(value, reply)
374 ? value
375 : {
376 ...value,
377 ...reply,
378 projectId: value.projectId,
379 repository: value.repository,
380 number: value.number,
381 ...(value.observedAt === undefined ? {} : { observedAt: value.observedAt }),
382 },
383 },
384 );
385}
386
387/** Lets a client record what each answer says where it lands, rather than watching the data. */
388export interface PullRequestAnswerHandlers {
389 readonly list?: QueryAnswerHandler<PullRequestListInput, PullRequestListResult>;
390 readonly listStats?: QueryAnswerHandler<PullRequestListStatsInput, PullRequestListStatsResult>;
391 readonly detail?: QueryAnswerHandler<PullRequestRef, PullRequestDetail>;
392 readonly checks?: QueryAnswerHandler<PullRequestRef, PullRequestChecks | null>;
393 readonly reviewGuide?: QueryAnswerHandler<PullRequestRef, PullRequestReviewGuideResult>;
394 /** Once per action reply, after the loaded readers show it; unscoped references only. */
395 readonly actionReply?: QueryAnswerHandler<PullRequestRef, PullRequestSummary>;
396}
397
398export function createPullRequestStackAtomFamily<R, E>(
399 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
400 refreshes = createPullRequestRefreshAtomFamily(runtime),
401) {
402 const routedRequest = createPullRequestRouter();
403 return createEnvironmentRpcQueryAtomFamily(runtime, {
404 label: "environment-data:pull-requests:stack",
405 tag: WS_METHODS.pullRequestsStack,
406 execute: (input) => routedRequest(WS_METHODS.pullRequestsStack, input),
407 staleTimeMs: 60_000,
408 idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS,
409 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
410 });
411}
412
413export function pullRequestDetailToVcsStatus(
414 detail: PullRequestDetail | PullRequestSummary,
415): NonNullable<VcsStatusResult["pr"]> {
416 return {
417 number: detail.number,
418 title: detail.title,
419 url: detail.url,
420 baseRef: detail.baseBranch,
421 headRef: detail.headBranch,
422 state: detail.state,
423 ...(detail.isDraft === true ? { isDraft: true } : {}),
424 updatedAt: detail.updatedAt,
425 };
426}
427
428export function createPullRequestEnvironmentAtoms<R, E>(
429 runtime: Atom.AtomRuntime<
430 EnvironmentRegistry | PullRequestDiffLoader.PullRequestDiffLoader | R,
431 E
432 >,
433 onAnswer: PullRequestAnswerHandlers = {},
434) {
435 const refreshes = createPullRequestRefreshAtomFamily(runtime);
436 const commandScheduler = createAtomCommandScheduler();
437 const routedRequest = createPullRequestRouter();
438 const serialPerEnvironment = {
439 mode: "serial",
440 key: ({ environmentId }: { readonly environmentId: string }) => environmentId,
441 } as const;
442 const activity = writableQueryFamily(
443 createEnvironmentRpcQueryAtomFamily(runtime, {
444 label: "environment-data:pull-requests:activity",
445 persistence: persistQuery(PullRequestActivity),
446 tag: WS_METHODS.pullRequestsActivity,
447 execute: (input) => routedRequest(WS_METHODS.pullRequestsActivity, input),
448 staleTimeMs: 60_000,
449 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
450 }),
451 );
452 const detailAnswerAtoms = Atom.family((_key: string) =>
453 Atom.make<{ readonly value: PullRequestDetail; readonly version: number } | null>(null).pipe(
454 Atom.setIdleTTL(5 * 60_000),
455 ),
456 );
457 const detailAnswers = normalizePullRequestQuery((target) =>
458 detailAnswerAtoms(JSON.stringify(target)),
459 );
460 const detailLabel = "environment-data:pull-requests:detail";
461 const detailPersistence = persistQuery(PullRequestDetailSchema);
462 const detail = normalizePullRequestQuery(
463 actionPatchedQuery(
464 createEnvironmentRpcQueryAtomFamily(runtime, {
465 label: detailLabel,
466 persistence: detailPersistence,
467 tag: WS_METHODS.pullRequestsDetail,
468 execute: (input) => routedRequest(WS_METHODS.pullRequestsDetail, input),
469 staleTimeMs: 60_000,
470 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
471 onAnswer: (target, value, registry) =>
472 Effect.sync(() => {
473 // Cache reads bypass onAnswer, and server freshness stays separate from local readbacks.
474 registry.update(detailAnswers(target), (previous) => ({
475 value,
476 version: (previous?.version ?? 0) + 1,
477 }));
478 }).pipe(Effect.andThen(() => onAnswer.detail?.(target, value, registry) ?? Effect.void)),
479 }),
480 {
481 label: detailLabel,
482 persistence: detailPersistence,
483 onAnswer: onAnswer.detail,
484 concerns: samePullRequest,
485 rereadWithoutReply: true,
486 // Reopening or a new head changes branch data and permissions the reply lacks.
487 rereads: (value, reply) =>
488 !isStalePullRequestReply(value, reply) &&
489 ((value.state === "closed" && reply.state === "open") ||
490 (reply.headSha !== undefined && reply.headSha !== value.headSha)),
491 patch: (_input, value, _reference, reply) => {
492 if (isStalePullRequestReply(value, reply)) return value;
493 const patched = {
494 ...value,
495 ...pickDefined(reply, [
496 "title",
497 "url",
498 "author",
499 "state",
500 "isDraft",
501 "mergeability",
502 "additions",
503 "deletions",
504 "changedFiles",
505 "headBranch",
506 "baseBranch",
507 "updatedAt",
508 "mergedAt",
509 "closedAt",
510 "autoMergeEnabled",
511 "autoMergeMethod",
512 ]),
513 };
514 if (reply.autoMergeEnabled !== false) return patched;
515 const { autoMergeMethod: _disarmed, ...withoutAutoMerge } = patched;
516 return withoutAutoMerge;
517 },
518 },
519 ),
520 );
521 // Pages read the strict detail and other surfaces the default one; patch whichever is loaded
522 // without starting a read of the other.
523 const loadedDetails = (
524 registry: AtomRegistry.AtomRegistry,
525 target: { readonly environmentId: EnvironmentId; readonly input: PullRequestRef },
526 ) =>
527 [
528 detail({ ...target, input: { ...target.input, allowStale: undefined } }),
529 detail({ ...target, input: { ...target.input, allowStale: false } }),
530 ].filter((atom) => registry.getNodes().has(atom));
531 const detailReadVersions = Atom.family((source: ReturnType<typeof detail>) =>
532 Atom.family((answer: ReturnType<typeof detailAnswers>) =>
533 Atom.make((get) => {
534 const result = get(source);
535 const latest = get(answer);
536 return result._tag === "Success" && !result.waiting && result.value === latest?.value
537 ? latest.version
538 : Option.getOrElse(get.self<number>(), () => 0);
539 }).pipe(Atom.setIdleTTL(5 * 60_000)),
540 ),
541 );
542 const detailReadVersion = normalizePullRequestQuery((target) =>
543 detailReadVersions(detail(target))(detailAnswers(target)),
544 );
545 const previewPersistence = persistQuery(PullRequestPreview);
546 const previewQuery = (label: string) =>
547 normalizePullRequestQuery(
548 actionPatchedQuery(
549 createEnvironmentRpcQueryAtomFamily(runtime, {
550 label,
551 persistence: previewPersistence,
552 tag: WS_METHODS.pullRequestsPreview,
553 execute: (input) => routedRequest(WS_METHODS.pullRequestsPreview, input),
554 staleTimeMs: 60_000,
555 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
556 }),
557 {
558 label,
559 persistence: previewPersistence,
560 concerns: samePullRequest,
561 rereadWithoutReply: true,
562 // Servers before action replies send unstamped previews, which only a read can order.
563 patch: (_input, value, _reference, reply) =>
564 value.updatedAt === undefined && value.observedAt === undefined
565 ? "refresh"
566 : isStalePullRequestReply(value, reply)
567 ? value
568 : {
569 ...value,
570 ...pickDefined(reply, [
571 "title",
572 "url",
573 "author",
574 "state",
575 "isDraft",
576 "updatedAt",
577 "observedAt",
578 ]),
579 },
580 },
581 ),
582 );
583 const preview = previewQuery("environment-data:pull-requests:preview");
584 // Hover cards read their own atoms, so a pending or failed hover read never decides where a
585 // click on the same link goes.
586 const previewCard = previewQuery("environment-data:pull-requests:preview-card");
587 const labelCandidates = writableQueryFamily(
588 createEnvironmentRpcQueryAtomFamily(runtime, {
589 label: "environment-data:pull-requests:label-candidates",
590 tag: WS_METHODS.pullRequestsLabelCandidates,
591 execute: (input) => routedRequest(WS_METHODS.pullRequestsLabelCandidates, input),
592 staleTimeMs: 60_000,
593 }),
594 );
595 const reviewGuide = writableQueryFamily(
596 createEnvironmentRpcQueryAtomFamily(runtime, {
597 label: "environment-data:pull-requests:review-guide",
598 persistence: persistQuery(PullRequestReviewGuideResult),
599 tag: WS_METHODS.pullRequestsReviewGuide,
600 execute: (input) => routedRequest(WS_METHODS.pullRequestsReviewGuide, input),
601 staleTimeMs: 60_000,
602 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
603 ...(onAnswer.reviewGuide === undefined ? {} : { onAnswer: onAnswer.reviewGuide }),
604 }),
605 );
606 const reviewerCandidates = writableQueryFamily(
607 createEnvironmentRpcQueryAtomFamily(runtime, {
608 label: "environment-data:pull-requests:reviewer-candidates",
609 tag: WS_METHODS.pullRequestsReviewerCandidates,
610 execute: (input) => routedRequest(WS_METHODS.pullRequestsReviewerCandidates, input),
611 staleTimeMs: 60_000,
612 }),
613 );
614 // Asked only when the menu's first page is not everybody, so a query cannot be answered from it.
615 const reviewerSearch = createEnvironmentRpcQueryAtomFamily(runtime, {
616 label: "environment-data:pull-requests:reviewer-search",
617 tag: WS_METHODS.pullRequestsReviewerCandidates,
618 execute: (input) => routedRequest(WS_METHODS.pullRequestsReviewerCandidates, input),
619 staleTimeMs: 60_000,
620 });
621 const loadDiff = Effect.fn("PullRequests.loadDiff")(function* (input: PullRequestDiffInput) {
622 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
623 const loader = yield* PullRequestDiffLoader.PullRequestDiffLoader;
624 const prepared = yield* SubscriptionRef.get(supervisor.prepared);
625 if (Option.isNone(prepared)) {
626 return yield* new EnvironmentHttpConnectionNotReadyError({
627 message: "The environment HTTP connection is not ready.",
628 });
629 }
630 return yield* loader.load(prepared.value, input);
631 });
632
633 const runAction = Effect.fn("PullRequests.runAction")(function* (
634 input: PullRequestActionInput & {
635 readonly resolveMergeMethod?: (detail: PullRequestDetail) => PullRequestMergeMethod;
636 },
637 ) {
638 const { resolveMergeMethod, ...actionInput } = input;
639 let preparedInput = actionInput;
640 if (actionInput.action === "merge" && resolveMergeMethod !== undefined) {
641 const { projectId, host, repository, number } = actionInput;
642 const reference = { projectId, host, repository, number, allowStale: false };
643 const detail = yield* routedRequest(WS_METHODS.pullRequestsDetail, reference);
644 if (
645 detail.state !== "open" ||
646 detail.isDraft ||
647 !detail.capabilities.actions.includes("merge") ||
648 !detail.viewerPermissions.actions.includes("merge")
649 ) {
650 return yield* new PullRequestOperationError({
651 operation: "runAction",
652 detail: "This pull request cannot be merged.",
653 });
654 }
655 if (detail.capabilities.stackActions) {
656 const stack = yield* routedRequest(WS_METHODS.pullRequestsStack, reference);
657 if (stack !== null) {
658 return yield* new PullRequestOperationError({
659 operation: "runAction",
660 detail: "Open this pull request to merge its stack.",
661 });
662 }
663 }
664 const mergeMethod = yield* Effect.try({
665 try: () => resolveMergeMethod(detail),
666 catch: (cause) =>
667 new PullRequestOperationError({
668 operation: "runAction",
669 detail: cause instanceof Error ? cause.message : "Could not choose a merge method.",
670 }),
671 });
672 preparedInput = { ...actionInput, mergeMethod };
673 }
674 return yield* routedRequest(WS_METHODS.pullRequestsRunAction, {
675 ...preparedInput,
676 replyWithSummary: true,
677 });
678 });
679 const actionReplyReaders = (registry: AtomRegistry.AtomRegistry) =>
680 Array.from(registry.getNodes().keys()).flatMap((atom) => {
681 const patch = typeof atom === "string" ? undefined : actionReplyPatches.get(atom);
682 return patch === undefined ? [] : [patch];
683 });
684 const readsInFlight = (registry: AtomRegistry.AtomRegistry): ReadonlySet<object> =>
685 new Set(actionReplyReaders(registry).flatMap((reader) => reader.pendingRead(registry) ?? []));
686 const applyActionReply = Effect.fn("PullRequests.applyActionReply")(function* (
687 registry: AtomRegistry.AtomRegistry,
688 environmentId: EnvironmentId,
689 reference: PullRequestRef,
690 reply: PullRequestSummary | undefined,
691 readsBeforeSend: ReadonlySet<object>,
692 ) {
693 for (const reader of actionReplyReaders(registry)) {
694 yield* reader.apply(registry, environmentId, reference, reply, readsBeforeSend);
695 }
696 if (reply !== undefined && reference.expectedAccountId === undefined) {
697 yield* (
698 onAnswer.actionReply?.({ environmentId, input: reference }, reply, registry) ?? Effect.void
699 );
700 }
701 });
702 const listLabel = "environment-data:pull-requests:list";
703 const listPersistence = persistQuery(PullRequestListResult);
704
705 return {
706 refreshes,
707 linkedThreads: createEnvironmentRpcQueryAtomFamily(runtime, {
708 label: "environment-data:pull-requests:linked-threads",
709 persistence: persistQuery(PullRequestLinkedThreadsResult),
710 tag: WS_METHODS.pullRequestsLinkedThreads,
711 staleTimeMs: 0,
712 refreshIntervalMs: 10_000,
713 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
714 }),
715 // For callers that open a thread from the answer: a persisted one reads as final before
716 // the connection is known, and a kept one can predate the thread.
717 linkedThreadsUncached: createEnvironmentRpcQueryAtomFamily(runtime, {
718 label: "environment-data:pull-requests:linked-threads-uncached",
719 tag: WS_METHODS.pullRequestsLinkedThreads,
720 staleTimeMs: 0,
721 idleTtlMs: 0,
722 }),
723 list: actionPatchedQuery(
724 createEnvironmentRpcQueryAtomFamily(runtime, {
725 label: listLabel,
726 persistence: listPersistence,
727 tag: WS_METHODS.pullRequestsList,
728 staleTimeMs: 30_000,
729 refreshTrigger: ({ environmentId, input }) =>
730 input.cursors === undefined ? refreshes({ environmentId, input: {} }) : undefined,
731 ...(onAnswer.list === undefined ? {} : { onAnswer: onAnswer.list }),
732 }),
733 {
734 label: listLabel,
735 persistence: listPersistence,
736 onAnswer: onAnswer.list,
737 // Rows carry no account scope, so a reply read through another account leaves them alone.
738 concerns: (input, reference) =>
739 reference.expectedAccountId === undefined &&
740 (input.projectId === undefined || input.projectId === reference.projectId) &&
741 (input.projectIds === undefined || input.projectIds.includes(reference.projectId)) &&
742 (input.host === undefined ||
743 reference.host === undefined ||
744 input.host.toLowerCase() === reference.host.toLowerCase()),
745 refreshedByServer: (input) => input.cursors === undefined,
746 patch: patchListWithReply,
747 },
748 ),
749 /**
750 * Its own query because the listing is quicker without them — measured over twelve
751 * repositories, ~4.0s against ~7.1s — so the rows arrive first and their stats a moment
752 * later.
753 */
754 listStats: createEnvironmentRpcQueryAtomFamily(runtime, {
755 label: "environment-data:pull-requests:list-stats",
756 persistence: persistPullRequestStats(),
757 tag: WS_METHODS.pullRequestsListStats,
758 staleTimeMs: 60_000,
759 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
760 ...(onAnswer.listStats === undefined ? {} : { onAnswer: onAnswer.listStats }),
761 }),
762 filterOptions: createEnvironmentRpcQueryAtomFamily(runtime, {
763 label: "environment-data:pull-requests:filter-options",
764 tag: WS_METHODS.pullRequestsFilterOptions,
765 staleTimeMs: 5 * 60_000,
766 }),
767 detail,
768 detailReadVersion,
769 preview,
770 previewCard,
771 checks: normalizePullRequestQuery(
772 createEnvironmentRpcQueryAtomFamily(runtime, {
773 label: "environment-data:pull-requests:checks",
774 persistence: persistQuery(Schema.NullOr(PullRequestChecks)),
775 tag: WS_METHODS.pullRequestsChecks,
776 execute: (input) => routedRequest(WS_METHODS.pullRequestsChecks, input),
777 staleTimeMs: 45_000,
778 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
779 ...(onAnswer.checks === undefined ? {} : { onAnswer: onAnswer.checks }),
780 }),
781 ),
782 activity,
783 threadComments: createEnvironmentRpcCommand(runtime, {
784 label: "environment-data:pull-requests:thread-comments",
785 tag: WS_METHODS.pullRequestsThreadComments,
786 execute: (input) => routedRequest(WS_METHODS.pullRequestsThreadComments, input),
787 scheduler: commandScheduler,
788 concurrency: {
789 mode: "singleFlight",
790 key: ({ environmentId, input }) =>
791 JSON.stringify([environmentId, input.threadId, input.cursor]),
792 },
793 }),
794 diff: createEnvironmentQueryAtomFamily(runtime, {
795 label: "environment-data:pull-requests:diff",
796 persistence: persistQuery(PullRequestDiffResult),
797 staleTimeMs: 60_000,
798 execute: loadDiff,
799 }),
800 diffFileContents: createEnvironmentRpcCommand(runtime, {
801 label: "environment-data:pull-requests:diff-file-contents",
802 tag: WS_METHODS.pullRequestsDiffFileContents,
803 execute: (input) => routedRequest(WS_METHODS.pullRequestsDiffFileContents, input),
804 scheduler: commandScheduler,
805 concurrency: {
806 mode: "singleFlight",
807 key: ({ environmentId, input }) =>
808 JSON.stringify([
809 environmentId,
810 input.projectId,
811 input.host?.toLowerCase() ?? null,
812 input.repository,
813 input.number,
814 input.commit ?? null,
815 input.changeType,
816 input.oldPath,
817 input.newPath,
818 ]),
819 },
820 }),
821 filesViewed: createEnvironmentRpcQueryAtomFamily(runtime, {
822 label: "environment-data:pull-requests:files-viewed",
823 persistence: persistQuery(PullRequestFilesViewedResult),
824 tag: WS_METHODS.pullRequestsFilesViewed,
825 execute: (input) => routedRequest(WS_METHODS.pullRequestsFilesViewed, input),
826 staleTimeMs: 15_000,
827 }),
828 changeBreakdown: createEnvironmentRpcQueryAtomFamily(runtime, {
829 label: "environment-data:pull-requests:change-breakdown",
830 persistence: persistQuery(PullRequestChangeBreakdown),
831 tag: WS_METHODS.pullRequestsChangeBreakdown,
832 execute: (input) => routedRequest(WS_METHODS.pullRequestsChangeBreakdown, input),
833 staleTimeMs: 60_000,
834 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
835 }),
836 /**
837 * One write in flight per change request: the host applies these in order, and a reader
838 * ticking down a file list faster than the round trip would otherwise race their own presses.
839 */
840 setFilesViewed: createEnvironmentRpcCommand(runtime, {
841 label: "environment-data:pull-requests:set-files-viewed",
842 tag: WS_METHODS.pullRequestsSetFilesViewed,
843 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetFilesViewed, input),
844 scheduler: commandScheduler,
845 concurrency: {
846 mode: "serial",
847 key: ({ environmentId, input }) =>
848 JSON.stringify([environmentId, input.projectId, input.repository, input.number]),
849 },
850 }),
851 runAction: createEnvironmentRpcCommand(runtime, {
852 tag: WS_METHODS.pullRequestsRunAction,
853 label: "environment-data:pull-requests:run-action",
854 // Preparation belongs to the write's lane. Refreshable queries would restart it after
855 // every preceding action, and preparing outside the lane could reorder the clicks.
856 execute: runAction,
857 onStart: (_target, registry) => readsInFlight(registry),
858 // Loaded readers show the reply, or re-read when the host's reply lacked the pull request.
859 onSuccess: ({ environmentId, input }, registry, reply, readsBeforeSend) => {
860 const { projectId, host, repository, number, expectedAccountId } = input;
861 return applyActionReply(
862 registry,
863 environmentId,
864 {
865 projectId,
866 ...(host === undefined ? {} : { host }),
867 repository,
868 number,
869 ...(expectedAccountId === undefined ? {} : { expectedAccountId }),
870 },
871 reply,
872 readsBeforeSend ?? new Set(),
873 );
874 },
875 scheduler: commandScheduler,
876 concurrency: serialPerEnvironment,
877 }),
878 update: createEnvironmentRpcCommand(runtime, {
879 label: "environment-data:pull-requests:update",
880 tag: WS_METHODS.pullRequestsUpdate,
881 execute: (input) => routedRequest(WS_METHODS.pullRequestsUpdate, input),
882 onSuccess: (target, registry) => Effect.sync(() => registry.refresh(preview(target))),
883 scheduler: commandScheduler,
884 concurrency: serialPerEnvironment,
885 }),
886 comment: createEnvironmentRpcCommand(runtime, {
887 label: "environment-data:pull-requests:comment",
888 tag: WS_METHODS.pullRequestsComment,
889 execute: (input) => routedRequest(WS_METHODS.pullRequestsComment, input),
890 scheduler: commandScheduler,
891 concurrency: serialPerEnvironment,
892 }),
893 updateComment: createEnvironmentRpcCommand(runtime, {
894 label: "environment-data:pull-requests:update-comment",
895 tag: WS_METHODS.pullRequestsUpdateComment,
896 execute: (input) => routedRequest(WS_METHODS.pullRequestsUpdateComment, input),
897 scheduler: commandScheduler,
898 concurrency: serialPerEnvironment,
899 onSuccess: ({ environmentId, input: { projectId, host, repository, number } }, registry) =>
900 Effect.sync(() =>
901 registry.refresh(
902 activity({
903 environmentId,
904 input: { projectId, ...(host === undefined ? {} : { host }), repository, number },
905 }),
906 ),
907 ),
908 }),
909 submitReview: createEnvironmentRpcCommand(runtime, {
910 label: "environment-data:pull-requests:submit-review",
911 tag: WS_METHODS.pullRequestsSubmitReview,
912 execute: (input) => routedRequest(WS_METHODS.pullRequestsSubmitReview, input),
913 scheduler: commandScheduler,
914 concurrency: serialPerEnvironment,
915 }),
916 replyToThread: createEnvironmentRpcCommand(runtime, {
917 label: "environment-data:pull-requests:reply-to-thread",
918 tag: WS_METHODS.pullRequestsReplyToThread,
919 execute: (input) => routedRequest(WS_METHODS.pullRequestsReplyToThread, input),
920 scheduler: commandScheduler,
921 concurrency: serialPerEnvironment,
922 }),
923 reviewerCandidates,
924 /** The host's search for reviewers past the first page, keyed by what was typed. */
925 reviewerSearch,
926 requestReviewers: createEnvironmentRpcCommand(runtime, {
927 label: "environment-data:pull-requests:request-reviewers",
928 tag: WS_METHODS.pullRequestsRequestReviewers,
929 execute: (input) => routedRequest(WS_METHODS.pullRequestsRequestReviewers, input),
930 scheduler: commandScheduler,
931 concurrency: serialPerEnvironment,
932 onSuccess: (target, registry) =>
933 Effect.sync(() => {
934 const { reviewers, requested } = target.input;
935 const candidatesAtom = reviewerCandidates(target);
936 const candidates = AsyncResult.getOrElse(registry.get(candidatesAtom), () => null);
937 const selected =
938 candidates?.candidates.filter((candidate) =>
939 reviewers.some(
940 (reviewer) => reviewer.id === candidate.id && reviewer.kind === candidate.kind,
941 ),
942 ) ?? [];
943 const missingIdentities = selected.length < reviewers.length;
944 // Somebody found only by the host's search is not in the loaded list yet; the host
945 // puts requested reviewers first, so reading it again brings them in.
946 updateCached(
947 registry,
948 candidatesAtom,
949 (value) => ({
950 ...value,
951 candidates: value.candidates.map((candidate) =>
952 selected.includes(candidate) ? { ...candidate, isRequested: requested } : candidate,
953 ),
954 }),
955 missingIdentities,
956 );
957 const selectedLogins = new Set(
958 selected.map((candidate) => candidate.login.toLowerCase()),
959 );
960 const updateReviewers = (
961 actors: ReadonlyArray<PullRequestActor>,
962 keep = (_actor: PullRequestActor) => false,
963 ) =>
964 requested
965 ? [
966 ...actors,
967 ...selected
968 .filter(
969 (candidate) =>
970 !actors.some(
971 (actor) => actor.login.toLowerCase() === candidate.login.toLowerCase(),
972 ),
973 )
974 .map(({ login, name, avatarUrl }) => ({ login, name, avatarUrl })),
975 ]
976 : actors.filter(
977 (actor) => !selectedLogins.has(actor.login.toLowerCase()) || keep(actor),
978 );
979 for (const atom of loadedDetails(registry, target)) {
980 updateCached(
981 registry,
982 atom,
983 (value) => ({
984 ...value,
985 reviewers: updateReviewers(value.reviewers),
986 }),
987 missingIdentities,
988 );
989 }
990 updateCached(
991 registry,
992 activity(target),
993 (value) => ({
994 ...value,
995 reviewers:
996 value.reviewers === undefined
997 ? undefined
998 : updateReviewers(value.reviewers, (actor) =>
999 value.comments.some(
1000 (comment) =>
1001 (comment.kind === "review" || comment.kind === "review-comment") &&
1002 comment.author?.login.toLowerCase() === actor.login.toLowerCase(),
1003 ),
1004 ),
1005 }),
1006 missingIdentities,
1007 );
1008 }),
1009 }),
1010 reviewGuide,
1011 generateReviewGuide: createEnvironmentRpcCommand(runtime, {
1012 label: "environment-data:pull-requests:generate-review-guide",
1013 tag: WS_METHODS.pullRequestsGenerateReviewGuide,
1014 execute: (input) => routedRequest(WS_METHODS.pullRequestsGenerateReviewGuide, input),
1015 onSuccess: (target, registry, result) =>
1016 Effect.sync(() => {
1017 updateCached(registry, reviewGuide(target), () => result, true);
1018 }),
1019 }),
1020 labelCandidates,
1021 setLabels: createEnvironmentRpcCommand(runtime, {
1022 label: "environment-data:pull-requests:set-labels",
1023 tag: WS_METHODS.pullRequestsSetLabels,
1024 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetLabels, input),
1025 scheduler: commandScheduler,
1026 concurrency: serialPerEnvironment,
1027 onSuccess: (target, registry) =>
1028 Effect.sync(() => {
1029 const { labels, applied } = target.input;
1030 const candidatesAtom = labelCandidates(target);
1031 const candidates = AsyncResult.getOrElse(registry.get(candidatesAtom), () => null);
1032 const names = new Set(labels);
1033 updateCached(registry, candidatesAtom, (value) => ({
1034 ...value,
1035 candidates: value.candidates.map((candidate) =>
1036 names.has(candidate.name) ? { ...candidate, isApplied: applied } : candidate,
1037 ),
1038 }));
1039 for (const atom of loadedDetails(registry, target)) {
1040 updateCached(registry, atom, (value) => ({
1041 ...value,
1042 labels: applied
1043 ? [
1044 ...value.labels,
1045 ...labels
1046 .filter((name) => !value.labels.some((label) => label.name === name))
1047 .map((name) => ({
1048 name,
1049 color:
1050 candidates?.candidates.find((candidate) => candidate.name === name)
1051 ?.color ?? null,
1052 })),
1053 ]
1054 : value.labels.filter((label) => !names.has(label.name)),
1055 }));
1056 }
1057 }),
1058 }),
1059 setThreadResolution: createEnvironmentRpcCommand(runtime, {
1060 label: "environment-data:pull-requests:set-thread-resolution",
1061 tag: WS_METHODS.pullRequestsSetThreadResolution,
1062 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetThreadResolution, input),
1063 scheduler: commandScheduler,
1064 concurrency: serialPerEnvironment,
1065 }),
1066 setReaction: createEnvironmentRpcCommand(runtime, {
1067 label: "environment-data:pull-requests:set-reaction",
1068 tag: WS_METHODS.pullRequestsSetReaction,
1069 execute: (input) => routedRequest(WS_METHODS.pullRequestsSetReaction, input),
1070 scheduler: commandScheduler,
1071 concurrency: serialPerEnvironment,
1072 }),
1073 /**
1074 * A separate request rather than a flag on a read, so only a person's refresh spends
1075 * host requests while every silent re-read shares the cache.
1076 */
1077 invalidate: createEnvironmentRpcCommand(runtime, {
1078 label: "environment-data:pull-requests:invalidate",
1079 tag: WS_METHODS.pullRequestsInvalidate,
1080 execute: (input) => routedRequest(WS_METHODS.pullRequestsInvalidate, input),
1081 onSuccess: ({ environmentId, input }, registry) =>
1082 Effect.sync(() => {
1083 if (input.reference !== undefined) {
1084 registry.refresh(preview({ environmentId, input: input.reference }));
1085 registry.refresh(previewCard({ environmentId, input: input.reference }));
1086 }
1087 }),
1088 scheduler: commandScheduler,
1089 concurrency: serialPerEnvironment,
1090 }),
1091 };
1092}