1import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
2import * as Schema from "effect/Schema";
3import { Atom, type AtomRegistry } from "effect/reactivity";
4
5import {
6 flattenQueuedThreadMessages,
7 groupQueuedThreadMessages,
8 type QueuedThreadMessage,
9} from "./thread-outbox-model";
10import type { ThreadOutboxStorage } from "./thread-outbox-storage";
11
12export class ThreadOutboxManagerError extends Schema.TaggedError<ThreadOutboxManagerError>()(
13 "ThreadOutboxManagerError",
14 {
15 operation: Schema.Literals([
16 "load",
17 "enqueue",
18 "update",
19 "remove",
20 "clear-environment-load",
21 "clear-environment-remove",
22 ]),
23 environmentId: Schema.NullOr(EnvironmentId),
24 threadId: Schema.NullOr(ThreadId),
25 messageId: Schema.NullOr(MessageId),
26 cause: Schema.Defect(),
27 },
28) {
29 override get message(): string {
30 return `Thread outbox operation ${this.operation} failed for environment ${this.environmentId ?? "unknown"}, thread ${this.threadId ?? "unknown"}, message ${this.messageId ?? "unknown"}.`;
31 }
32}
33
34export interface ThreadOutboxManagerOptions {
35 readonly registry: AtomRegistry.AtomRegistry;
36 readonly storage: ThreadOutboxStorage;
37 readonly warn?: (message: string, error: unknown) => void;
38}
39
40export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) {
41 const queuedMessagesByThreadKeyAtom = Atom.make<
42 Record<string, ReadonlyArray<QueuedThreadMessage>>
43 >({}).pipe(Atom.keepAlive, Atom.withLabel("mobile:thread-outbox:queued-messages"));
44 const warn =
45 options.warn ??
46 ((message: string, error: unknown) => {
47 console.warn(message, error);
48 });
49 let loadPromise: Promise<boolean> | null = null;
50 let mutationQueue: Promise<void> = Promise.resolve();
51 const revisions = new Map<MessageId, number>();
52 const bumpRevision = (messageId: MessageId): void => {
53 revisions.set(messageId, (revisions.get(messageId) ?? 0) + 1);
54 };
55
56 const serialize = <A>(mutation: () => Promise<A>): Promise<A> => {
57 const result = mutationQueue.then(mutation, mutation);
58 mutationQueue = result.then(
59 () => undefined,
60 () => undefined,
61 );
62 return result;
63 };
64
65 const currentMessages = (): ReadonlyArray<QueuedThreadMessage> =>
66 flattenQueuedThreadMessages(options.registry.get(queuedMessagesByThreadKeyAtom));
67
68 const setMessages = (messages: ReadonlyArray<QueuedThreadMessage>): void => {
69 options.registry.set(queuedMessagesByThreadKeyAtom, groupQueuedThreadMessages(messages));
70 };
71
72 // Partial loads expose readable messages; return true only on complete loads to protect unreadable records' files.
73 const load = (): Promise<boolean> => {
74 if (loadPromise !== null) {
75 return loadPromise;
76 }
77 loadPromise = serialize(async () => {
78 const result = await options.storage.load();
79 const current = currentMessages();
80 const currentIds = new Set(current.map((message) => message.messageId));
81 const recovered = result.messages.filter(
82 (message) => !currentIds.has(message.messageId) && !revisions.has(message.messageId),
83 );
84 if (recovered.length > 0) setMessages([...recovered, ...current]);
85 if (result.errors.length > 0) {
86 throw new AggregateError(result.errors, "Some queued messages could not be read.");
87 }
88 return true;
89 }).catch((cause) => {
90 loadPromise = null;
91 warn(
92 "[thread-outbox] failed to load persisted messages",
93 new ThreadOutboxManagerError({
94 operation: "load",
95 environmentId: null,
96 threadId: null,
97 messageId: null,
98 cause,
99 }),
100 );
101 return false;
102 });
103 return loadPromise;
104 };
105
106 const enqueue = (message: QueuedThreadMessage): Promise<void> => {
107 bumpRevision(message.messageId);
108 setMessages([
109 ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
110 message,
111 ]);
112 return serialize(async () => {
113 try {
114 await options.storage.write(message);
115 } catch (cause) {
116 // Rollback by reference: a retry with the same id may already own a newer optimistic entry.
117 setMessages(currentMessages().filter((candidate) => candidate !== message));
118 // Remove compensating persisted copies after rollback unless a same-id entry survives, or restart resurrects rejected messages.
119 if (!currentMessages().some((candidate) => candidate.messageId === message.messageId)) {
120 try {
121 await options.storage.remove(message);
122 } catch {}
123 }
124 throw new ThreadOutboxManagerError({
125 operation: "enqueue",
126 environmentId: message.environmentId,
127 threadId: message.threadId,
128 messageId: message.messageId,
129 cause,
130 });
131 }
132 });
133 };
134
135 // Confirm durability before dispatch so a failed write cannot follow successful delivery.
136 const confirmQueued = (message: QueuedThreadMessage): Promise<boolean> =>
137 serialize(async () => currentMessages().some((candidate) => candidate === message));
138
139 // Enqueue may publish during writes; persist the winning revision before settling to avoid stale crash recovery.
140 const update = (message: QueuedThreadMessage, expectedRevision?: number): Promise<boolean> =>
141 serialize(async () => {
142 const staleOrMissing = (): boolean =>
143 !currentMessages().some((candidate) => candidate.messageId === message.messageId) ||
144 (expectedRevision !== undefined &&
145 (revisions.get(message.messageId) ?? 0) !== expectedRevision);
146 if (staleOrMissing()) {
147 return false;
148 }
149 try {
150 await options.storage.write(message);
151 } catch (cause) {
152 throw new ThreadOutboxManagerError({
153 operation: "update",
154 environmentId: message.environmentId,
155 threadId: message.threadId,
156 messageId: message.messageId,
157 cause,
158 });
159 }
160 if (staleOrMissing()) {
161 const winner = currentMessages().find(
162 (candidate) => candidate.messageId === message.messageId,
163 );
164 if (winner !== undefined) {
165 try {
166 await options.storage.write(winner);
167 } catch {}
168 }
169 return false;
170 }
171 bumpRevision(message.messageId);
172 setMessages([
173 ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
174 message,
175 ]);
176 return true;
177 });
178
179 // Check live ownership too; editor locks can change without message revision updates.
180 const remove = (
181 message: QueuedThreadMessage,
182 expectedRevision?: number,
183 canRemove?: () => boolean,
184 ): Promise<QueuedThreadMessage | null> =>
185 serialize(async () => {
186 const removalCanceled = (): boolean =>
187 (expectedRevision !== undefined &&
188 (revisions.get(message.messageId) ?? 0) !== expectedRevision) ||
189 canRemove?.() === false;
190 if (removalCanceled()) {
191 return null;
192 }
193 const removed =
194 currentMessages().find((candidate) => candidate.messageId === message.messageId) ?? message;
195 try {
196 await options.storage.remove(message);
197 } catch (cause) {
198 throw new ThreadOutboxManagerError({
199 operation: "remove",
200 environmentId: message.environmentId,
201 threadId: message.threadId,
202 messageId: message.messageId,
203 cause,
204 });
205 }
206 if (removalCanceled()) {
207 // Enqueue or editor locks can win during removal; persist the live winner before the next queued mutation.
208 const winner = currentMessages().find(
209 (candidate) => candidate.messageId === message.messageId,
210 );
211 if (winner !== undefined) {
212 try {
213 await options.storage.write(winner);
214 } catch (cause) {
215 throw new ThreadOutboxManagerError({
216 operation: "remove",
217 environmentId: message.environmentId,
218 threadId: message.threadId,
219 messageId: message.messageId,
220 cause,
221 });
222 }
223 }
224 return null;
225 }
226 setMessages(
227 currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
228 );
229 // Tombstone revisions: restarting a same-id retry at one would match stale pre-removal writers (ABA).
230 bumpRevision(message.messageId);
231 return removed;
232 });
233
234 const clearEnvironment = (
235 environmentId: EnvironmentId,
236 ): Promise<ReadonlyArray<QueuedThreadMessage>> => {
237 // Capture revisions now but read messages after prior mutations; later changes must escape this clear request.
238 const revisionsAtRequest = new Map(revisions);
239 return serialize(async () => {
240 const persisted = await options.storage
241 .load()
242 .then((result) => {
243 if (result.errors.length > 0) {
244 throw new AggregateError(result.errors, "Some queued messages could not be read.");
245 }
246 return result.messages;
247 })
248 .catch((cause) => {
249 throw new ThreadOutboxManagerError({
250 operation: "clear-environment-load",
251 environmentId,
252 threadId: null,
253 messageId: null,
254 cause,
255 });
256 });
257 const allMessages = flattenQueuedThreadMessages(
258 groupQueuedThreadMessages([...persisted, ...currentMessages()]),
259 );
260 const candidates = allMessages.filter(
261 (message) =>
262 message.environmentId === environmentId &&
263 (revisions.get(message.messageId) ?? 0) ===
264 (revisionsAtRequest.get(message.messageId) ?? 0),
265 );
266 const candidateRevisions = new Map(
267 candidates.map(
268 (message) => [message.messageId, revisions.get(message.messageId) ?? 0] as const,
269 ),
270 );
271 const removedFromStorage = new Set<MessageId>();
272
273 await Promise.all(
274 candidates.map(async (message) => {
275 try {
276 await options.storage.remove(message);
277 removedFromStorage.add(message.messageId);
278 } catch (cause) {
279 warn(
280 "[thread-outbox] failed to clear persisted message",
281 new ThreadOutboxManagerError({
282 operation: "clear-environment-remove",
283 environmentId: message.environmentId,
284 threadId: message.threadId,
285 messageId: message.messageId,
286 cause,
287 }),
288 );
289 }
290 }),
291 );
292
293 await Promise.all(
294 candidates.map(async (message) => {
295 if (
296 !removedFromStorage.has(message.messageId) ||
297 (revisions.get(message.messageId) ?? 0) === candidateRevisions.get(message.messageId)
298 ) {
299 return;
300 }
301 const retained = currentMessages().find(
302 (candidate) => candidate.messageId === message.messageId,
303 );
304 if (retained === undefined) {
305 return;
306 }
307 try {
308 await options.storage.write(retained);
309 } catch (cause) {
310 warn(
311 "[thread-outbox] failed to restore message retained during environment clear",
312 new ThreadOutboxManagerError({
313 operation: "clear-environment-remove",
314 environmentId: retained.environmentId,
315 threadId: retained.threadId,
316 messageId: retained.messageId,
317 cause,
318 }),
319 );
320 }
321 }),
322 );
323
324 const removed = candidates.filter(
325 (message) =>
326 removedFromStorage.has(message.messageId) &&
327 (revisions.get(message.messageId) ?? 0) === candidateRevisions.get(message.messageId),
328 );
329 const removedMessageIds = new Set(removed.map((message) => message.messageId));
330 const reconciledMessages = flattenQueuedThreadMessages(
331 groupQueuedThreadMessages([...allMessages, ...currentMessages()]),
332 ).filter((message) => !removedMessageIds.has(message.messageId));
333 for (const message of removed) {
334 bumpRevision(message.messageId);
335 }
336 setMessages(reconciledMessages);
337 // Report the actual removed messages so caller-owned attachment cleanup includes records loaded here.
338 return removed;
339 });
340 };
341
342 return {
343 queuedMessagesByThreadKeyAtom,
344 serialize,
345 load,
346 enqueue,
347 confirmQueued,
348 revisionOf: (messageId: MessageId): number => revisions.get(messageId) ?? 0,
349 update,
350 remove,
351 clearEnvironment,
352 };
353}