1import {
2 OrchestrationV2ThreadShell,
3 ThreadPullRequestLink,
4 type OrchestrationProjectShell,
5 type OrchestrationV2ShellSnapshot,
6 type OrchestrationV2ShellStreamItem,
7} from "@t3tools/contracts";
8import * as Schema from "effect/Schema";
9
10const sameThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell);
11const sameLinks = Schema.toEquivalence(Schema.UndefinedOr(Schema.Array(ThreadPullRequestLink)));
12
13/** Whether two rows' pull request links are equal, including both absent. */
14export function sameThreadPullRequests(
15 left: OrchestrationV2ThreadShell["pullRequests"],
16 right: OrchestrationV2ThreadShell["pullRequests"],
17): boolean {
18 return left === right || sameLinks(left, right);
19}
20
21/**
22 * Keep the previous object for each row a full snapshot left unchanged, so the list only
23 * re-renders rows that changed. Links are compared apart from the rest of the row because a
24 * deferred snapshot arrives without them; such a row keeps its links until they fill in.
25 */
26export function reuseUnchangedThreadShells(
27 previous: OrchestrationV2ShellSnapshot | null,
28 next: OrchestrationV2ShellSnapshot,
29): OrchestrationV2ShellSnapshot {
30 if (previous === null || previous.threads.length === 0) return next;
31 const previousById = new Map(previous.threads.map((thread) => [thread.id, thread] as const));
32 let reused = 0;
33 const threads = next.threads.map((thread) => {
34 const prior = previousById.get(thread.id);
35 if (prior === undefined) return thread;
36 const candidate =
37 thread.pullRequests === undefined && prior.pullRequests !== undefined
38 ? { ...thread, pullRequests: prior.pullRequests }
39 : thread;
40 if (!sameThreadShell(prior, candidate)) return thread;
41 reused += 1;
42 return prior;
43 });
44 return reused === 0 ? next : { ...next, threads };
45}
46
47function upsertById<T extends { readonly id: unknown }>(
48 items: ReadonlyArray<T>,
49 item: T,
50): ReadonlyArray<T> {
51 const index = items.findIndex((candidate) => candidate.id === item.id);
52 if (index === -1) return [...items, item];
53 return items.map((candidate, candidateIndex) => (candidateIndex === index ? item : candidate));
54}
55
56function retainRepositoryIdentity(
57 previous: OrchestrationProjectShell | undefined,
58 next: OrchestrationProjectShell,
59): OrchestrationProjectShell {
60 if (
61 next.repositoryIdentity == null &&
62 previous?.repositoryIdentity != null &&
63 previous.workspaceRoot === next.workspaceRoot
64 ) {
65 return { ...next, repositoryIdentity: previous.repositoryIdentity };
66 }
67 return next;
68}
69
70export interface MergeShellSnapshotOptions {
71 /**
72 * Metadata-only enrichment refresh: structure and sequence never change;
73 * listed roots accept identity exactly (including null).
74 * Omit this options object for authoritative HTTP/initial WebSocket snapshots.
75 */
76 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
77}
78
79/**
80 * Merge an incoming full shell snapshot into prior client state.
81 *
82 * Authoritative snapshots (no options) replace structure and sequence even when
83 * lower than cache, while retaining a prior non-null identity when the candidate
84 * is still unresolved/null for the same root.
85 *
86 * Enrichment snapshots (options present) only patch repository identity for
87 * matching current projects. They never replace projects, threads, archives,
88 * or sequence, regardless of the incoming snapshot sequence.
89 */
90export function mergeShellSnapshotProjects(
91 previous: OrchestrationV2ShellSnapshot | null | undefined,
92 next: OrchestrationV2ShellSnapshot,
93 options?: MergeShellSnapshotOptions,
94): OrchestrationV2ShellSnapshot {
95 if (previous === null || previous === undefined) {
96 return next;
97 }
98
99 const isEnrichment = options !== undefined;
100 const resolvedRootSet = isEnrichment ? new Set(options.resolvedRepositoryIdentityRoots) : null;
101
102 if (isEnrichment) {
103 const nextById = new Map(next.projects.map((project) => [project.id, project] as const));
104 return {
105 ...previous,
106 projects: previous.projects.map((project) => {
107 const candidate = nextById.get(project.id);
108 if (candidate === undefined || candidate.workspaceRoot !== project.workspaceRoot) {
109 return project;
110 }
111 if (resolvedRootSet?.has(project.workspaceRoot) === true) {
112 return { ...project, repositoryIdentity: candidate.repositoryIdentity };
113 }
114 if (project.repositoryIdentity == null && candidate.repositoryIdentity != null) {
115 return { ...project, repositoryIdentity: candidate.repositoryIdentity };
116 }
117 return project;
118 }),
119 };
120 }
121
122 const previousById = new Map(previous.projects.map((project) => [project.id, project] as const));
123 return {
124 ...next,
125 projects: next.projects.map((project) => {
126 const prior = previousById.get(project.id);
127 if (resolvedRootSet?.has(project.workspaceRoot) === true) {
128 return project;
129 }
130 return retainRepositoryIdentity(prior, project);
131 }),
132 };
133}
134
135/** Applies one committed V2 shell delta while preserving active/archive exclusivity. */
136export function applyShellStreamEvent(
137 snapshot: OrchestrationV2ShellSnapshot,
138 event: Exclude<
139 OrchestrationV2ShellStreamItem,
140 { readonly kind: "snapshot" } | { readonly kind: "synchronized" }
141 >,
142): OrchestrationV2ShellSnapshot {
143 if (event.sequence <= snapshot.snapshotSequence) return snapshot;
144
145 switch (event.kind) {
146 case "project.updated": {
147 // Enrichment is async. A project mutation can land with null
148 // repositoryIdentity while an earlier snapshot already resolved it.
149 // Keep the prior identity for the same workspace root so multi-env
150 // grouping does not split until a full snapshot refresh arrives.
151 const previous = snapshot.projects.find((project) => project.id === event.project.id);
152 const project = retainRepositoryIdentity(previous, event.project);
153 return {
154 ...snapshot,
155 projects: upsertById(snapshot.projects, project),
156 snapshotSequence: event.sequence,
157 };
158 }
159 case "project.removed":
160 return {
161 ...snapshot,
162 projects: snapshot.projects.filter((project) => project.id !== event.projectId),
163 snapshotSequence: event.sequence,
164 };
165 case "thread.updated": {
166 // An unchanged shell keeps its object and the list, so subscribers that
167 // compare by reference skip the update. Only the cursor moves.
168 const existing =
169 event.location === "active"
170 ? snapshot.threads.find((thread) => thread.id === event.thread.id)
171 : undefined;
172 if (existing !== undefined && sameThreadShell(existing, event.thread)) {
173 return { ...snapshot, snapshotSequence: event.sequence };
174 }
175 const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) =>
176 threads.filter((thread) => thread.id !== event.thread.id);
177 return {
178 ...snapshot,
179 threads:
180 event.location === "active"
181 ? upsertById(snapshot.threads, event.thread)
182 : withoutThread(snapshot.threads),
183 // The archive has its own bounded query/subscription. Older servers may
184 // still send archive-located deltas here; remove them from the normal
185 // shell instead of growing its persisted cache again.
186 archivedThreads: withoutThread(snapshot.archivedThreads),
187 snapshotSequence: event.sequence,
188 };
189 }
190 case "thread.removed":
191 return {
192 ...snapshot,
193 threads: snapshot.threads.filter((thread) => thread.id !== event.threadId),
194 archivedThreads: snapshot.archivedThreads.filter((thread) => thread.id !== event.threadId),
195 snapshotSequence: event.sequence,
196 };
197 default:
198 return snapshot;
199 }
200}