1import { useAtomValue } from "@effect/atom-react";
2import { sourceControlClients } from "@t3tools/client-runtime/source-control-clients";
3import {
4 createLinkedPullRequestSummaryAtomFamily,
5 createPullRequestEnvironmentAtoms,
6 createPullRequestStackAtomFamily,
7} from "@t3tools/client-runtime/state/pull-requests";
8import type {
9 EnvironmentId,
10 ProjectId,
11 PullRequestListInput,
12 PullRequestListStatsInput,
13 PullRequestListEntry,
14 PullRequestRef,
15 PullRequestSummary,
16 ThreadPullRequestLink,
17} from "@t3tools/contracts";
18import {
19 threadPullRequestKeyOf,
20 visibleThreadPullRequests,
21} from "@t3tools/shared/threadPullRequests";
22import * as Option from "effect/Option";
23import { AsyncResult, Atom } from "effect/reactivity";
24import { useCallback, useLayoutEffect, useMemo } from "react";
25
26import { connectionAtomRuntime } from "../connection/runtime";
27import { appAtomRegistry } from "../rpc/atomRegistry";
28import {
29 mergePullRequestLists,
30 type EnvironmentPullRequestStat,
31 type MergedPullRequestList,
32} from "../components/pullRequest/pullRequestList.logic";
33import { formatEnvironmentQueryError } from "./query";
34import { environmentThreadShells } from "./threads";
35
36export const pullRequestEnvironment = createPullRequestEnvironmentAtoms(connectionAtomRuntime);
37export const linkedPullRequestDetailAtom = createLinkedPullRequestSummaryAtomFamily(
38 connectionAtomRuntime,
39 pullRequestEnvironment.refreshes,
40);
41
42export interface ObservedPullRequestSummary {
43 readonly summary: PullRequestSummary;
44 /** Client arrival time, the only ordering older servers leave us for same-dated snapshots. */
45 readonly observedAt: number;
46}
47
48const observedPullRequestSummaryAtom = Atom.family((key: string) =>
49 Atom.make<ObservedPullRequestSummary | null>(null).pipe(
50 Atom.setIdleTTL(5 * 60_000),
51 Atom.withLabel(`web-pull-requests:observed-summary:${key}`),
52 ),
53);
54
55/**
56 * Positive when `incoming` is the newer snapshot. Merged is final. Then the host's own update
57 * time, then the server's read-start time, which survives its caches; a snapshot without one
58 * never beats a stamped read. Zero when neither side carries a read time.
59 */
60function compareSummaries(current: PullRequestSummary, incoming: PullRequestSummary): number {
61 const merged = Number(incoming.state === "merged") - Number(current.state === "merged");
62 if (merged !== 0) return merged;
63 const updated = Date.parse(incoming.updatedAt) - Date.parse(current.updatedAt);
64 if (updated !== 0) return updated;
65 if (current.observedAt === undefined && incoming.observedAt === undefined) return 0;
66 return (incoming.observedAt ?? -Infinity) - (current.observedAt ?? -Infinity);
67}
68
69export function newestPullRequestSummary(
70 current: PullRequestSummary | null,
71 observed: PullRequestSummary | null,
72): PullRequestSummary | null {
73 if (current === null) return observed;
74 if (observed === null) return current;
75 return compareSummaries(current, observed) >= 0 ? observed : current;
76}
77
78/** Reuse list status without treating its deferred line-count placeholders as real stats. */
79export function pullRequestListEntryToSummary(entry: PullRequestListEntry): PullRequestSummary {
80 return {
81 provider: entry.provider,
82 projectId: entry.projectId,
83 repository: entry.repository,
84 number: entry.number,
85 title: entry.title,
86 url: entry.url,
87 state: entry.state,
88 isDraft: entry.isDraft,
89 headBranch: entry.headBranch,
90 baseBranch: entry.baseBranch,
91 updatedAt: entry.updatedAt,
92 ...(entry.observedAt === undefined ? {} : { observedAt: entry.observedAt }),
93 author: entry.author,
94 ...(entry.reviewDecision === undefined ? {} : { reviewDecision: entry.reviewDecision }),
95 ...(entry.checksState === undefined ? {} : { checksState: entry.checksState }),
96 mergeability: entry.mergeability,
97 };
98}
99
100// A project has one remote, so its id already pins the host. Leaving the host out lets a list
101// row, a hostless legacy reference and a URL-derived thread reference share one entry.
102function pullRequestSummaryKey(environmentId: EnvironmentId, reference: PullRequestRef): string {
103 return JSON.stringify([
104 environmentId,
105 reference.projectId,
106 reference.repository.toLowerCase(),
107 reference.number,
108 ]);
109}
110
111/** The observation to hold after `incoming` arrives. Returns `current` itself on a tie. */
112export function newestPullRequestObservation(
113 current: ObservedPullRequestSummary | null,
114 incoming: ObservedPullRequestSummary | null,
115): ObservedPullRequestSummary | null {
116 if (current === null) return incoming;
117 if (incoming === null) return current;
118 let order = compareSummaries(current.summary, incoming.summary);
119 // Client clocks only break ties between snapshots that both lack a server read time.
120 if (
121 order === 0 &&
122 current.summary.observedAt === undefined &&
123 incoming.summary.observedAt === undefined
124 ) {
125 order = incoming.observedAt - current.observedAt;
126 }
127 if (!(order > 0)) return current;
128 // A sparse summary must not erase known status, or carry old detail stats into a new list read.
129 return {
130 ...incoming,
131 summary: {
132 ...incoming.summary,
133 isDraft: incoming.summary.isDraft ?? current.summary.isDraft,
134 mergeability: incoming.summary.mergeability ?? current.summary.mergeability,
135 reviewDecision:
136 incoming.summary.reviewDecision === undefined
137 ? current.summary.reviewDecision
138 : incoming.summary.reviewDecision,
139 checksState:
140 incoming.summary.checksState === undefined
141 ? current.summary.checksState
142 : incoming.summary.checksState,
143 },
144 };
145}
146
147function observePullRequestSummary(
148 environmentId: EnvironmentId,
149 reference: PullRequestRef,
150 summary: PullRequestSummary,
151 observedAt: number,
152): void {
153 const atom = observedPullRequestSummaryAtom(pullRequestSummaryKey(environmentId, reference));
154 appAtomRegistry.modify(atom, (previous) => {
155 const next = newestPullRequestObservation(previous, { summary, observedAt });
156 return next === previous ? [false, previous] : [true, next];
157 });
158}
159
160export function threadPullRequestLinkToSummary(
161 link: ThreadPullRequestLink,
162 projectId: ProjectId,
163): PullRequestSummary | null {
164 // Without the host update time a snapshot cannot be ordered against a fresh read, so it sits out.
165 const updatedAt = link.snapshot?.updatedAt;
166 if (link.snapshot === null || updatedAt == null) return null;
167 return {
168 ...link.snapshot,
169 provider: (
170 sourceControlClients.findByChangeRequestUrl(link.url) ?? sourceControlClients.get(undefined)
171 ).kind,
172 projectId,
173 repository: link.repository,
174 number: link.number,
175 url: link.url,
176 updatedAt,
177 };
178}
179
180// Saved thread snapshots by environment and host-qualified link, indexed once per shell update
181// rather than per control. The host keeps a link to another forge from matching the project's PR.
182function savedPullRequestKey(
183 environmentId: EnvironmentId,
184 reference: Pick<PullRequestRef, "host" | "repository" | "number">,
185): string {
186 return reference.host === undefined
187 ? ""
188 : `${environmentId}:${threadPullRequestKeyOf({ ...reference, host: reference.host })}`;
189}
190
191const savedPullRequestSummariesAtom = Atom.make((get) => {
192 const summaries = new Map<string, PullRequestSummary>();
193 for (const thread of get(environmentThreadShells.threadShellsAtom)) {
194 for (const link of visibleThreadPullRequests(thread.pullRequests)) {
195 const summary = threadPullRequestLinkToSummary(link, thread.projectId);
196 if (summary === null) continue;
197 const key = savedPullRequestKey(thread.environmentId, link);
198 summaries.set(key, newestPullRequestSummary(summaries.get(key) ?? null, summary) ?? summary);
199 }
200 }
201 return summaries;
202});
203
204const savedPullRequestSummaryAtom = Atom.family((key: string) =>
205 Atom.make((get) => get(savedPullRequestSummariesAtom).get(key) ?? null).pipe(
206 // Shell updates also resend unchanged links; keep their subscribers quiet.
207 Atom.withEquality((previous, next) => JSON.stringify(previous) === JSON.stringify(next)),
208 Atom.setIdleTTL(5 * 60_000),
209 ),
210);
211
212export function useSharedPullRequestSummary(
213 environmentId: EnvironmentId | null,
214 reference: PullRequestRef | null,
215 current: PullRequestSummary | null,
216 observedAt: number | null = null,
217): PullRequestSummary | null {
218 const key =
219 environmentId === null || reference === null
220 ? "none"
221 : pullRequestSummaryKey(environmentId, reference);
222 const atom = observedPullRequestSummaryAtom(key);
223 const observed = useAtomValue(atom);
224 const saved = useAtomValue(
225 savedPullRequestSummaryAtom(
226 environmentId === null || reference === null
227 ? ""
228 : savedPullRequestKey(environmentId, reference),
229 ),
230 );
231 useLayoutEffect(() => {
232 if (environmentId === null || reference === null || current === null || observedAt === null)
233 return;
234 observePullRequestSummary(environmentId, reference, current, observedAt);
235 }, [current, environmentId, reference, observedAt]);
236 const summary =
237 newestPullRequestObservation(
238 observed,
239 current === null || observedAt === null ? null : { summary: current, observedAt },
240 )?.summary ?? current;
241 return newestPullRequestSummary(summary, saved);
242}
243export const pullRequestStackAtom = createPullRequestStackAtomFamily(
244 connectionAtomRuntime,
245 pullRequestEnvironment.refreshes,
246);
247
248export interface EnvironmentQueryTarget<Input> {
249 readonly environmentId: EnvironmentId;
250 readonly input: Input;
251}
252
253interface MergedEnvironmentQueryView<A> {
254 /** One entry per query target that has answered, in the order the targets were given. */
255 readonly values: ReadonlyArray<readonly [EnvironmentId, A]>;
256 /** The first environment that failed. Others may still have answered — this is not fatal. */
257 readonly error: string | null;
258 readonly isPending: boolean;
259 readonly observations: ReadonlyArray<readonly [EnvironmentId, A, number]>;
260}
261
262/**
263 * The same per-environment query read across several environments at once. React cannot subscribe
264 * to a list of atoms whose length changes, so the fan-out happens inside one derived atom keyed by
265 * the targets — the same shape the cross-environment thread search uses.
266 *
267 * An environment that fails contributes nothing rather than blanking the page: the pull request
268 * list is a union, and one unreachable machine should not hide the others' rows.
269 */
270function createMergedEnvironmentQuery<Input, A>(
271 label: string,
272 atomFor: (
273 target: EnvironmentQueryTarget<Input>,
274 ) => Atom.Atom<AsyncResult.AsyncResult<A, unknown>>,
275) {
276 const family = Atom.family((key: string) =>
277 Atom.make((get): MergedEnvironmentQueryView<A> => {
278 const targets = JSON.parse(key) as ReadonlyArray<EnvironmentQueryTarget<Input>>;
279 const values: Array<readonly [EnvironmentId, A]> = [];
280 const observations: Array<readonly [EnvironmentId, A, number]> = [];
281 let error: string | null = null;
282 let isPending = false;
283 for (const target of targets) {
284 const result = get(atomFor(target));
285 isPending ||= result.waiting;
286 if (result._tag === "Failure" && error === null) {
287 error = formatEnvironmentQueryError(result.cause);
288 }
289 const value = Option.getOrNull(AsyncResult.value(result));
290 if (value !== null) values.push([target.environmentId, value]);
291 if (result._tag === "Success") {
292 observations.push([target.environmentId, result.value, result.timestamp]);
293 }
294 }
295 return { values, error, isPending, observations };
296 }).pipe(Atom.withLabel(`${label}:${key}`)),
297 );
298 const empty = Atom.make<MergedEnvironmentQueryView<A>>({
299 values: [],
300 observations: [],
301 error: null,
302 isPending: false,
303 }).pipe(Atom.withLabel(`${label}:empty`));
304 return function useMergedQuery(targets: ReadonlyArray<EnvironmentQueryTarget<Input>>) {
305 const key = JSON.stringify(targets);
306 const view = useAtomValue(targets.length === 0 ? empty : family(key));
307 const refresh = useCallback(
308 (override?: ReadonlyArray<EnvironmentQueryTarget<Input>>) => {
309 const refreshTargets =
310 override ?? (JSON.parse(key) as ReadonlyArray<EnvironmentQueryTarget<Input>>);
311 for (const atom of new Set(refreshTargets.map(atomFor))) {
312 appAtomRegistry.refresh(atom);
313 }
314 },
315 [key],
316 );
317 return { ...view, refresh };
318 };
319}
320
321const usePullRequestListsQuery = createMergedEnvironmentQuery(
322 "web-pull-requests:list",
323 pullRequestEnvironment.list,
324);
325
326const usePullRequestStatsQuery = createMergedEnvironmentQuery(
327 "web-pull-requests:list-stats",
328 pullRequestEnvironment.listStats,
329);
330
331const usePullRequestTurnRefreshQuery = createMergedEnvironmentQuery(
332 "web-pull-requests:turn-refreshes",
333 ({ environmentId }: EnvironmentQueryTarget<Readonly<Record<string, never>>>) =>
334 pullRequestEnvironment.refreshes({ environmentId, input: {} }),
335);
336
337export function usePullRequestTurnRefreshes(
338 environmentIds: ReadonlyArray<EnvironmentId>,
339): ReadonlyArray<readonly [EnvironmentId, number]> {
340 return usePullRequestTurnRefreshQuery(
341 environmentIds.map((environmentId) => ({ environmentId, input: {} })),
342 ).values;
343}
344
345export function usePullRequestTurnRefresh(environmentId: EnvironmentId): number | null {
346 const result = useAtomValue(pullRequestEnvironment.refreshes({ environmentId, input: {} }));
347 return Option.getOrNull(AsyncResult.value(result));
348}
349
350export interface MergedPullRequestListView {
351 readonly data: MergedPullRequestList | null;
352 readonly error: string | null;
353 readonly isPending: boolean;
354 readonly refresh: (targets?: ReadonlyArray<EnvironmentQueryTarget<PullRequestListInput>>) => void;
355}
356
357/** One listing per environment, merged into the single list the page renders. */
358export function usePullRequestList(
359 targets: ReadonlyArray<EnvironmentQueryTarget<PullRequestListInput>>,
360): MergedPullRequestListView {
361 const query = usePullRequestListsQuery(targets);
362 useLayoutEffect(() => {
363 for (const [environmentId, answer, observedAt] of query.observations) {
364 for (const entry of answer.entries) {
365 observePullRequestSummary(
366 environmentId,
367 {
368 projectId: entry.projectId,
369 host: entry.host,
370 repository: entry.repository,
371 number: entry.number,
372 },
373 pullRequestListEntryToSummary(entry),
374 observedAt,
375 );
376 }
377 }
378 }, [query.observations]);
379 const data = useMemo(() => mergePullRequestLists(query.values), [query.values]);
380 return { data, error: query.error, isPending: query.isPending, refresh: query.refresh };
381}
382
383/** The line counts for the rows on screen, asked of each environment for its own rows. */
384export function usePullRequestListStats(
385 targets: ReadonlyArray<EnvironmentQueryTarget<PullRequestListStatsInput>>,
386): {
387 readonly stats: ReadonlyArray<EnvironmentPullRequestStat> | null;
388 readonly isPending: boolean;
389 readonly refresh: (
390 targets?: ReadonlyArray<EnvironmentQueryTarget<PullRequestListStatsInput>>,
391 ) => void;
392} {
393 const query = usePullRequestStatsQuery(targets);
394 const stats = useMemo(
395 () =>
396 query.values.length === 0
397 ? null
398 : query.values.flatMap(([environmentId, result]) =>
399 result.stats.map((stat) => ({ ...stat, environmentId })),
400 ),
401 [query.values],
402 );
403 return { stats, isPending: query.isPending, refresh: query.refresh };
404}