1import { RegistryContext, useAtomMount, useAtomSubscribe, useAtomValue } from "@effect/atom-react";
2import { executeAtomQuery, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime";
3import type { EnvironmentId, PullRequestDiffStat, PullRequestListInput } from "@t3tools/contracts";
4import * as Cause from "effect/Cause";
5import * as Effect from "effect/Effect";
6import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity";
7import { use, useMemo, useState } from "react";
8import { environmentCatalog } from "../../connection/catalog";
9import {
10 composerPullRequests,
11 pullRequestListPagesFamily,
12 pullRequestListPagesKey,
13 pullRequestListStatsFamily,
14} from "../../state/pull-requests";
15import {
16 appendPullRequestListPage,
17 mergeEnvironmentPullRequestPages,
18 nextPullRequestListInput,
19 patchPullRequestListEntry,
20 pullRequestListHead,
21 shownPullRequestListPages,
22 pullRequestStatKey as statKey,
23 scopedPullRequestListKey,
24 withShownPullRequestStats,
25 type EnvironmentPullRequestListEntry,
26 type PullRequestListPages,
27} from "./pullRequestList";
28
29type ListTarget = { readonly environmentId: EnvironmentId; readonly input: PullRequestListInput };
30type Read = <A>(atom: Atom.Atom<A>) => A;
31
32/** The head the list query holds now, cached or answered. */
33function currentHead(read: Read, target: ListTarget) {
34 return AsyncResult.getOrElse(read(composerPullRequests.list(target)), () => null);
35}
36
37function currentPages(read: Read, target: ListTarget): PullRequestListPages | null {
38 return shownPullRequestListPages(
39 read(pullRequestListPagesFamily(pullRequestListPagesKey(target))),
40 currentHead(read, target),
41 );
42}
43
44const sourcesFamily = Atom.family((key: string) => {
45 const targets: readonly ListTarget[] = JSON.parse(key);
46 return Atom.make((get) =>
47 targets.map((target) => ({
48 target,
49 result: get(composerPullRequests.list(target)),
50 pages: currentPages(get, target),
51 stats: get(pullRequestListStatsFamily(target.environmentId)),
52 })),
53 );
54});
55
56function continuations(registry: AtomRegistry.AtomRegistry, targets: readonly ListTarget[]) {
57 return targets.flatMap((target) => {
58 const pages = currentPages((atom) => registry.get(atom), target);
59 const input = pages && nextPullRequestListInput(pages, target.input);
60 return pages && input ? [{ target, pages, input }] : [];
61 });
62}
63
64async function loadNextPages(
65 registry: AtomRegistry.AtomRegistry,
66 targets: readonly ListTarget[],
67 retry: boolean,
68 signal: AbortSignal,
69) {
70 const responses = await Promise.allSettled(
71 continuations(registry, targets).map(async ({ target, pages, input }) => {
72 const response = await executeAtomQuery(
73 registry,
74 composerPullRequests.list({ ...target, input }),
75 {
76 signal,
77 refresh: retry,
78 reportFailure: false,
79 },
80 );
81 if (signal.aborted) return;
82 if (response._tag === "Failure") throw squashAtomCommandFailure(response);
83 if (response.value.errors.length)
84 throw new Error(response.value.errors.map((error) => error.message).join("\n"));
85 const pagesAtom = pullRequestListPagesFamily(pullRequestListPagesKey(target));
86 const next = appendPullRequestListPage(
87 currentPages((atom) => registry.get(atom), target) ?? pages,
88 response.value,
89 input,
90 );
91 registry.set(pagesAtom, next);
92 if (
93 input.cursors &&
94 Object.keys(next.nextCursors).length &&
95 JSON.stringify(Object.entries(input.cursors).sort()) ===
96 JSON.stringify(Object.entries(next.nextCursors).sort())
97 )
98 throw new Error("Couldn't advance the list. Try again.");
99 }),
100 );
101 return requestErrors(responses);
102}
103
104async function refreshHeads(
105 registry: AtomRegistry.AtomRegistry,
106 targets: readonly ListTarget[],
107 signal: AbortSignal,
108) {
109 const responses = await Promise.allSettled(
110 targets.map(async (target) => {
111 await composerPullRequests.invalidate.run(registry, {
112 environmentId: target.environmentId,
113 input: {},
114 });
115 if (signal.aborted) return;
116 const response = await executeAtomQuery(registry, composerPullRequests.list(target), {
117 refresh: true,
118 signal,
119 reportFailure: false,
120 });
121 if (signal.aborted) return;
122 if (response._tag === "Failure") throw squashAtomCommandFailure(response);
123 registry.set(
124 pullRequestListPagesFamily(pullRequestListPagesKey(target)),
125 pullRequestListHead(null, response.value),
126 );
127 }),
128 );
129 return requestErrors(responses);
130}
131
132// One load of each kind per list scope. The atoms are not kept alive, so leaving a scope (a
133// filter change, unmount) interrupts its load and aborts the requests.
134const pageLoadFamily = Atom.family((scopeKey: string) => {
135 const targets: readonly ListTarget[] = JSON.parse(scopeKey);
136 return Atom.fn((retry: boolean, get) =>
137 Effect.promise((signal) => loadNextPages(get.registry, targets, retry, signal)),
138 );
139});
140const refreshLoadFamily = Atom.family((scopeKey: string) => {
141 const targets: readonly ListTarget[] = JSON.parse(scopeKey);
142 return Atom.fn((_: void, get) =>
143 Effect.promise((signal) => refreshHeads(get.registry, targets, signal)),
144 );
145});
146
147// Changes when any of the scope's environments brings up a new session.
148const sessionsFamily = Atom.family((scopeKey: string) =>
149 Atom.make((get) =>
150 (JSON.parse(scopeKey) as readonly ListTarget[])
151 .map(({ environmentId }) => get(environmentCatalog.sessionGenerationAtom(environmentId)))
152 .join(),
153 ),
154);
155
156const settledError = (result: AsyncResult.AsyncResult<string | null, never>) =>
157 AsyncResult.isSuccess(result) && !result.waiting ? result.value : null;
158
159export function usePullRequestList(targets: readonly ListTarget[]) {
160 const registry = use(RegistryContext);
161 const scopeKey = JSON.stringify(targets);
162 const sources = useAtomValue(sourcesFamily(scopeKey));
163 const pageAtom = pageLoadFamily(scopeKey);
164 const refreshAtom = refreshLoadFamily(scopeKey);
165 const page = useAtomValue(pageAtom);
166 const refresh = useAtomValue(refreshAtom);
167 const loadError = settledError(page);
168
169 // A page or refresh asked for while an environment was unreachable fails at once, and its
170 // error stops loading on scroll. Once it is back, ask for the page again; a refresh's error is
171 // stale by then because the list queries re-run on reconnect.
172 // A listener alone does not compute the atom, so nothing would follow the connections.
173 useAtomMount(sessionsFamily(scopeKey));
174 useAtomSubscribe(sessionsFamily(scopeKey), () => {
175 const pageResult = registry.get(pageAtom);
176 const refreshResult = registry.get(refreshAtom);
177 if (settledError(refreshResult) !== null) registry.set(refreshAtom, Atom.Reset);
178 else if (settledError(pageResult) !== null && !refreshResult.waiting)
179 registry.set(pageAtom, true);
180 });
181
182 // Read from the registry, not the render: a second onEndReached before React re-renders must
183 // see the load the first one started.
184 const loadMore = (retry = false) => {
185 const pageResult = registry.get(pageAtom);
186 const refreshResult = registry.get(refreshAtom);
187 if (pageResult.waiting || refreshResult.waiting) return;
188 if (!retry && settledError(pageResult)) return;
189 if (!continuations(registry, targets).length) return;
190 registry.set(refreshAtom, Atom.Reset);
191 registry.set(pageAtom, retry);
192 };
193
194 const onRefresh = () => {
195 if (!targets.length || registry.get(refreshAtom).waiting) return;
196 registry.set(pageAtom, Atom.Reset);
197 registry.set(refreshAtom, undefined);
198 };
199
200 const data = useMemo(
201 () =>
202 mergeEnvironmentPullRequestPages(
203 sources.flatMap(({ target, pages, stats }) =>
204 pages
205 ? [
206 {
207 environmentId: target.environmentId,
208 pages: {
209 ...pages,
210 entries: pages.entries.map((entry) => {
211 const stat = stats.stats.get(statKey(entry));
212 return stat
213 ? { ...entry, additions: stat.additions, deletions: stat.deletions }
214 : entry;
215 }),
216 },
217 },
218 ]
219 : [],
220 ),
221 ),
222 [sources],
223 );
224 const [visible, setVisible] = useState<{
225 key: string;
226 rows: readonly EnvironmentPullRequestListEntry[];
227 }>({ key: scopeKey, rows: [] });
228 const onViewableItemsChanged = ({
229 viewableItems,
230 }: {
231 viewableItems: readonly { item: EnvironmentPullRequestListEntry; isViewable: boolean }[];
232 }) => {
233 const rows = viewableItems.filter((token) => token.isViewable).map((token) => token.item);
234 setVisible((previous) =>
235 previous.key === scopeKey &&
236 previous.rows.map(scopedPullRequestListKey).join() ===
237 rows.map(scopedPullRequestListKey).join()
238 ? previous
239 : { key: scopeKey, rows },
240 );
241 };
242 const visibleRows =
243 visible.key === scopeKey && visible.rows.length ? visible.rows : data.entries.slice(0, 8);
244 // Visible rows carry the cached-answer overlay below. Decide which still need stats on the rows
245 // without it, so a cached answer shown offline does not unmount the query that holds it.
246 const withoutCachedOverlay = new Map(
247 data.entries.map((row) => [scopedPullRequestListKey(row), row]),
248 );
249 const statsKey = JSON.stringify(
250 targets.flatMap(({ environmentId }) => {
251 const refs = visibleRows
252 .filter((row) => {
253 const base = withoutCachedOverlay.get(scopedPullRequestListKey(row)) ?? row;
254 return row.environmentId === environmentId && base.additions + base.deletions === 0;
255 })
256 .map(({ projectId, host, repository, number }) => ({
257 projectId,
258 host,
259 repository,
260 number,
261 }));
262 return refs.length ? [{ environmentId, input: { refs } }] : [];
263 }),
264 );
265 const statsAtom = useMemo(() => {
266 const queries: readonly Parameters<typeof composerPullRequests.listStats>[0][] =
267 JSON.parse(statsKey);
268 return Atom.make((get) =>
269 queries.map((target) => ({
270 environmentId: target.environmentId,
271 result: get(composerPullRequests.listStats(target)),
272 })),
273 );
274 }, [statsKey]);
275 const counts = useAtomValue(statsAtom);
276 // Server answers land in pullRequestListStatsFamily; these cover a cached answer, which is
277 // all an offline environment has. Scrolling makes a new stats query, which has no answer until
278 // its cached rows load, so each environment keeps the answer it showed until the new one lands.
279 const [shown, setShown] = useState({ counts, answers: answeredStats(counts, []) });
280 if (shown.counts !== counts) setShown({ counts, answers: answeredStats(counts, shown.answers) });
281 const entries = useMemo(
282 () => withShownPullRequestStats(data.entries, shown.answers),
283 [data.entries, shown.answers],
284 );
285
286 const updateEntry = (
287 entry: EnvironmentPullRequestListEntry,
288 patch: Partial<EnvironmentPullRequestListEntry>,
289 ) => {
290 const target = targets.find((target) => target.environmentId === entry.environmentId);
291 if (!target) return;
292 const pagesAtom = pullRequestListPagesFamily(pullRequestListPagesKey(target));
293 const read = <A>(atom: Atom.Atom<A>) => registry.get(atom);
294 const patched = patchPullRequestListEntry(
295 read(pagesAtom),
296 currentHead(read, target),
297 entry,
298 patch,
299 Date.now(),
300 );
301 if (patched) registry.set(pagesAtom, patched);
302 registry.refresh(composerPullRequests.list(target));
303 };
304 const errors = useMemo(
305 () =>
306 sources.flatMap(({ target, result }) => {
307 if (result._tag !== "Failure") return [];
308 const cause = Cause.squash(result.cause);
309 return [
310 {
311 environmentId: target.environmentId,
312 projectTitle: "",
313 message: cause instanceof Error ? cause.message : "Couldn't load pull requests.",
314 },
315 ];
316 }),
317 [sources],
318 );
319 const listing = useMemo(
320 () => ({ ...data, entries, errors: [...data.errors, ...errors] }),
321 [data, entries, errors],
322 );
323 return {
324 data: listing,
325 entries,
326 error: errors.map((error) => error.message).join("\n") || null,
327 isPending: sources.some(({ result }) => result.waiting) && !data.entries.length,
328 continued: data.continued,
329 loadMore,
330 onRefresh,
331 onViewableItemsChanged,
332 updateEntry,
333 refreshing: refresh.waiting,
334 loadingMore: page.waiting,
335 loadMoreError: loadError,
336 hasMore: sources.some(
337 ({ pages, target }) => pages && nextPullRequestListInput(pages, target.input),
338 ),
339 };
340}
341
342function requestErrors(results: readonly PromiseSettledResult<unknown>[]) {
343 return (
344 results
345 .flatMap((result) =>
346 result.status === "rejected"
347 ? [
348 result.reason instanceof Error
349 ? result.reason.message
350 : "Couldn't load pull requests.",
351 ]
352 : [],
353 )
354 .join("\n") || null
355 );
356}
357
358type StatsAnswer = {
359 readonly environmentId: EnvironmentId;
360 readonly stats: readonly PullRequestDiffStat[];
361};
362
363/** Each environment's newest stats answer, or the one it had while its query has none. */
364function answeredStats(
365 counts: readonly {
366 readonly environmentId: EnvironmentId;
367 readonly result: AsyncResult.AsyncResult<
368 { readonly stats: readonly PullRequestDiffStat[] },
369 unknown
370 >;
371 }[],
372 previous: readonly StatsAnswer[],
373): readonly StatsAnswer[] {
374 const next = counts.flatMap(({ environmentId, result }) => {
375 const kept = previous.find((answer) => answer.environmentId === environmentId);
376 if (!AsyncResult.isSuccess(result)) return kept ? [kept] : [];
377 return kept?.stats === result.value.stats
378 ? [kept]
379 : [{ environmentId, stats: result.value.stats }];
380 });
381 return next.length === previous.length &&
382 next.every((answer, index) => answer === previous[index])
383 ? previous
384 : next;
385}