1import {
2 OrchestrationProjectShell,
3 OrchestrationV2ThreadShell,
4 RepositoryIdentity,
5 ThreadPullRequestLink,
6 type OrchestrationV2ShellSnapshot,
7 type OrchestrationV2ShellStreamItem,
8} from "@t3tools/contracts";
9import * as Schema from "effect/Schema";
10
11const sameThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell);
12const sameProjectShell = Schema.toEquivalence(OrchestrationProjectShell);
13const sameLinks = Schema.toEquivalence(Schema.UndefinedOr(Schema.Array(ThreadPullRequestLink)));
14const sameRepositoryIdentity = Schema.toEquivalence(
15 Schema.UndefinedOr(Schema.NullOr(RepositoryIdentity)),
16);
17
18export function sameThreadPullRequests(
19 left: OrchestrationV2ThreadShell["pullRequests"],
20 right: OrchestrationV2ThreadShell["pullRequests"],
21): boolean {
22 return left === right || sameLinks(left, right);
23}
24
25/**
26 * Links are compared apart from the rest of the row because a deferred snapshot arrives without
27 * them; such a row keeps its links until they fill in.
28 */
29export function reuseUnchangedThreadShells(
30 previous: OrchestrationV2ShellSnapshot | null,
31 next: OrchestrationV2ShellSnapshot,
32): OrchestrationV2ShellSnapshot {
33 if (previous === null || previous.threads.length === 0) return next;
34 const previousById = new Map(previous.threads.map((thread) => [thread.id, thread] as const));
35 let changed = false;
36 const threads = next.threads.map((thread) => {
37 const prior = previousById.get(thread.id);
38 if (prior === undefined) return thread;
39 const candidate =
40 thread.pullRequests === undefined && prior.pullRequests !== undefined
41 ? { ...thread, pullRequests: prior.pullRequests }
42 : thread;
43 const retained = sameThreadShell(prior, candidate) ? prior : candidate;
44 if (retained !== thread) changed = true;
45 return retained;
46 });
47 return changed ? { ...next, threads } : next;
48}
49
50function upsertById<T extends { readonly id: unknown }>(
51 items: ReadonlyArray<T>,
52 item: T,
53): ReadonlyArray<T> {
54 const index = items.findIndex((candidate) => candidate.id === item.id);
55 if (index === -1) return [...items, item];
56 return items.map((candidate, candidateIndex) => (candidateIndex === index ? item : candidate));
57}
58
59function retainRepositoryIdentity(
60 previous: OrchestrationProjectShell | undefined,
61 next: OrchestrationProjectShell,
62): OrchestrationProjectShell {
63 if (
64 next.repositoryIdentity == null &&
65 previous?.repositoryIdentity != null &&
66 previous.workspaceRoot === next.workspaceRoot
67 ) {
68 return { ...next, repositoryIdentity: previous.repositoryIdentity };
69 }
70 return next;
71}
72
73export interface MergeShellSnapshotOptions {
74 /**
75 * Metadata-only enrichment refresh: structure and sequence never change;
76 * listed roots accept identity exactly (including null).
77 * Omit this options object for authoritative HTTP/initial WebSocket snapshots.
78 */
79 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
80}
81
82export function mergeShellSnapshotProjects(
83 previous: OrchestrationV2ShellSnapshot | null | undefined,
84 next: OrchestrationV2ShellSnapshot,
85 options?: MergeShellSnapshotOptions,
86): OrchestrationV2ShellSnapshot {
87 if (previous === null || previous === undefined) {
88 return next;
89 }
90
91 const isEnrichment = options !== undefined;
92 const resolvedRootSet = isEnrichment ? new Set(options.resolvedRepositoryIdentityRoots) : null;
93
94 if (isEnrichment) {
95 const nextById = new Map(next.projects.map((project) => [project.id, project] as const));
96 const projects = previous.projects.map((project) => {
97 const candidate = nextById.get(project.id);
98 if (candidate === undefined || candidate.workspaceRoot !== project.workspaceRoot) {
99 return project;
100 }
101 if (resolvedRootSet?.has(project.workspaceRoot) === true) {
102 // Refreshes re-resolve every root; an unchanged identity keeps the project's identity
103 // so every subscriber (each pull request row, the sidebar) skips re-rendering.
104 if (sameRepositoryIdentity(project.repositoryIdentity, candidate.repositoryIdentity)) {
105 return project;
106 }
107 return { ...project, repositoryIdentity: candidate.repositoryIdentity };
108 }
109 if (project.repositoryIdentity == null && candidate.repositoryIdentity != null) {
110 return { ...project, repositoryIdentity: candidate.repositoryIdentity };
111 }
112 return project;
113 });
114 return projects.every((project, index) => project === previous.projects[index])
115 ? previous
116 : { ...previous, projects };
117 }
118
119 const previousById = new Map(previous.projects.map((project) => [project.id, project] as const));
120 // Project atoms and shell persistence compare by reference, so unchanged projects keep their objects.
121 const projects = next.projects.map((project) => {
122 const prior = previousById.get(project.id);
123 const merged = retainRepositoryIdentity(prior, project);
124 return prior !== undefined && sameProjectShell(prior, merged) ? prior : merged;
125 });
126 const unchanged =
127 projects.length === previous.projects.length &&
128 projects.every((project, index) => project === previous.projects[index]);
129 return { ...next, projects: unchanged ? previous.projects : projects };
130}
131
132export function applyShellStreamEvent(
133 snapshot: OrchestrationV2ShellSnapshot,
134 event: ShellStreamEvent,
135): OrchestrationV2ShellSnapshot {
136 if (event.sequence <= snapshot.snapshotSequence) return snapshot;
137
138 switch (event.kind) {
139 case "project.updated": {
140 // Enrichment is async. A project mutation can land with null
141 // repositoryIdentity while an earlier snapshot already resolved it.
142 // Keep the prior identity for the same workspace root so multi-env
143 // grouping does not split until a full snapshot refresh arrives.
144 const previous = snapshot.projects.find((project) => project.id === event.project.id);
145 const project = retainRepositoryIdentity(previous, event.project);
146 return {
147 ...snapshot,
148 projects: upsertById(snapshot.projects, project),
149 snapshotSequence: event.sequence,
150 };
151 }
152 case "project.removed":
153 return {
154 ...snapshot,
155 projects: snapshot.projects.filter((project) => project.id !== event.projectId),
156 snapshotSequence: event.sequence,
157 };
158 case "thread.updated": {
159 const existing =
160 event.location === "active"
161 ? snapshot.threads.find((thread) => thread.id === event.thread.id)
162 : undefined;
163 if (existing !== undefined && sameThreadShell(existing, event.thread)) {
164 return { ...snapshot, snapshotSequence: event.sequence };
165 }
166 const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) =>
167 threads.filter((thread) => thread.id !== event.thread.id);
168 return {
169 ...snapshot,
170 threads:
171 event.location === "active"
172 ? upsertById(snapshot.threads, event.thread)
173 : withoutThread(snapshot.threads),
174 // The archive has its own bounded query/subscription. Older servers may
175 // still send archive-located deltas here; remove them from the normal
176 // shell instead of growing its persisted cache again.
177 archivedThreads: withoutThread(snapshot.archivedThreads),
178 snapshotSequence: event.sequence,
179 };
180 }
181 case "thread.removed":
182 return {
183 ...snapshot,
184 threads: snapshot.threads.filter((thread) => thread.id !== event.threadId),
185 archivedThreads: snapshot.archivedThreads.filter((thread) => thread.id !== event.threadId),
186 snapshotSequence: event.sequence,
187 };
188 default:
189 return snapshot;
190 }
191}
192
193/** Index a collection lazily and materialize it only if a delta changes its rows. */
194function batchRows<T extends { readonly id: unknown }>(rows: ReadonlyArray<T>) {
195 let byId: Map<T["id"], T> | undefined;
196 let changed = false;
197 const index = () => {
198 if (byId === undefined) {
199 byId = new Map();
200 for (const row of rows) byId.set(row.id, row);
201 }
202 return byId;
203 };
204 return {
205 get: (id: T["id"]) => index().get(id),
206 set: (row: T) => {
207 index().set(row.id, row);
208 changed = true;
209 },
210 remove: (id: T["id"]) => {
211 if (index().delete(id)) changed = true;
212 },
213 finish: () => (changed ? Array.from(index().values()) : rows),
214 };
215}
216
217export type ShellStreamEvent = Extract<
218 OrchestrationV2ShellStreamItem,
219 { readonly sequence: number }
220>;
221
222/** Apply ordered deltas with one index and at most one new array per collection. */
223export function applyShellStreamEvents(
224 snapshot: OrchestrationV2ShellSnapshot,
225 events: ReadonlyArray<ShellStreamEvent>,
226): OrchestrationV2ShellSnapshot {
227 // Avoid indexing the whole shell for the common single-event delivery.
228 if (events.length === 1) return applyShellStreamEvent(snapshot, events[0]!);
229 const projects = batchRows(snapshot.projects);
230 const threads = batchRows(snapshot.threads);
231 const archivedThreads = batchRows(snapshot.archivedThreads);
232 let sequence = snapshot.snapshotSequence;
233 for (const event of events) {
234 if (event.sequence <= sequence) continue;
235 switch (event.kind) {
236 case "project.updated":
237 projects.set(retainRepositoryIdentity(projects.get(event.project.id), event.project));
238 break;
239 case "project.removed":
240 projects.remove(event.projectId);
241 break;
242 case "thread.updated": {
243 const id = event.thread.id;
244 if (event.location === "active") {
245 const previous = threads.get(id);
246 if (previous !== undefined && sameThreadShell(previous, event.thread)) break;
247 threads.set(event.thread);
248 } else {
249 threads.remove(id);
250 }
251 // Legacy archive deltas remove rows from the shell; the archive is queried separately.
252 archivedThreads.remove(id);
253 break;
254 }
255 case "thread.removed":
256 threads.remove(event.threadId);
257 archivedThreads.remove(event.threadId);
258 break;
259 default:
260 continue;
261 }
262 sequence = event.sequence;
263 }
264 if (sequence === snapshot.snapshotSequence) return snapshot;
265 return {
266 ...snapshot,
267 snapshotSequence: sequence,
268 projects: projects.finish(),
269 threads: threads.finish(),
270 archivedThreads: archivedThreads.finish(),
271 };
272}