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 // Monotonic per-message write counter. Every accepted write (enqueue publish
52 // or update) bumps it, so a writer that captured a revision before slow work
53 // (an attachment upload) is rejected before its stale payload reaches disk.
54 const revisions = new Map<MessageId, number>();
55 const bumpRevision = (messageId: MessageId): void => {
56 revisions.set(messageId, (revisions.get(messageId) ?? 0) + 1);
57 };
58
59 const serialize = <A>(mutation: () => Promise<A>): Promise<A> => {
60 const result = mutationQueue.then(mutation, mutation);
61 mutationQueue = result.then(
62 () => undefined,
63 () => undefined,
64 );
65 return result;
66 };
67
68 const currentMessages = (): ReadonlyArray<QueuedThreadMessage> =>
69 flattenQueuedThreadMessages(options.registry.get(queuedMessagesByThreadKeyAtom));
70
71 const setMessages = (messages: ReadonlyArray<QueuedThreadMessage>): void => {
72 options.registry.set(queuedMessagesByThreadKeyAtom, groupQueuedThreadMessages(messages));
73 };
74
75 // Readable messages can be used after a partial load. Only a complete load
76 // returns true, so cleanup cannot delete files owned by unreadable records.
77 // A later call retries failed reads without replacing live message objects.
78 const load = (): Promise<boolean> => {
79 if (loadPromise !== null) {
80 return loadPromise;
81 }
82 loadPromise = serialize(async () => {
83 const result = await options.storage.load();
84 const current = currentMessages();
85 const currentIds = new Set(current.map((message) => message.messageId));
86 const recovered = result.messages.filter(
87 (message) => !currentIds.has(message.messageId) && !revisions.has(message.messageId),
88 );
89 // Accepted edits and removals win over a later disk read. Retaining
90 // current objects also keeps retries from restarting the drain.
91 if (recovered.length > 0) setMessages([...recovered, ...current]);
92 if (result.errors.length > 0) {
93 throw new AggregateError(result.errors, "Some queued messages could not be read.");
94 }
95 return true;
96 }).catch((cause) => {
97 loadPromise = null;
98 warn(
99 "[thread-outbox] failed to load persisted messages",
100 new ThreadOutboxManagerError({
101 operation: "load",
102 environmentId: null,
103 threadId: null,
104 messageId: null,
105 cause,
106 }),
107 );
108 return false;
109 });
110 return loadPromise;
111 };
112
113 // The queued atom drives the composer's immediate "queued" feedback, so it
114 // is published synchronously; the durable write happens behind it and rolls
115 // the message back out if it fails (durability only matters for crash
116 // recovery, not for the in-session queue).
117 const enqueue = (message: QueuedThreadMessage): Promise<void> => {
118 bumpRevision(message.messageId);
119 setMessages([
120 ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
121 message,
122 ]);
123 return serialize(async () => {
124 try {
125 await options.storage.write(message);
126 } catch (cause) {
127 // Roll back by reference, not messageId: a retry enqueue with the same
128 // id may have optimistically replaced this attempt while the write was
129 // in flight, and its entry must survive this attempt's failure.
130 setMessages(currentMessages().filter((candidate) => candidate !== message));
131 // A concurrent update losing its post-write race compensates by
132 // persisting this message's payload before this write settles. When
133 // no same-id entry survives the rollback, drop that disk copy too, or
134 // a restart resurrects a message the queue no longer holds.
135 if (!currentMessages().some((candidate) => candidate.messageId === message.messageId)) {
136 try {
137 await options.storage.remove(message);
138 } catch {
139 // Best effort: bootstrap reconciles the queue against storage.
140 }
141 }
142 throw new ThreadOutboxManagerError({
143 operation: "enqueue",
144 environmentId: message.environmentId,
145 threadId: message.threadId,
146 messageId: message.messageId,
147 cause,
148 });
149 }
150 });
151 };
152
153 // Resolves once all pending mutations (including any in-flight enqueue
154 // write) have settled, reporting whether the message is still queued. The
155 // drain awaits this before dispatching so a message whose durable write
156 // later fails can never have been delivered first.
157 const confirmQueued = (message: QueuedThreadMessage): Promise<boolean> =>
158 serialize(async () => currentMessages().some((candidate) => candidate === message));
159
160 // Rewrites an already-queued message. A no-op when the message has been
161 // removed in the meantime (e.g. deleted or delivered), so a trailing editor
162 // flush can never resurrect it. Returns whether the message was updated.
163 //
164 // `expectedRevision` makes the update a compare-and-set: pass the revision
165 // read before starting slow work, and the update is rejected before the
166 // stale payload is persisted when any other write was accepted since. An
167 // enqueue can still publish synchronously while the durable write below is
168 // in flight, so the revision is re-checked after the write too; the stale
169 // payload it just persisted is then overwritten with the winning payload
170 // inside this mutation, so a crash before the winner's own serialized write
171 // cannot leave stale state on disk.
172 const update = (message: QueuedThreadMessage, expectedRevision?: number): Promise<boolean> =>
173 serialize(async () => {
174 const staleOrMissing = (): boolean =>
175 !currentMessages().some((candidate) => candidate.messageId === message.messageId) ||
176 (expectedRevision !== undefined &&
177 (revisions.get(message.messageId) ?? 0) !== expectedRevision);
178 if (staleOrMissing()) {
179 return false;
180 }
181 try {
182 await options.storage.write(message);
183 } catch (cause) {
184 throw new ThreadOutboxManagerError({
185 operation: "update",
186 environmentId: message.environmentId,
187 threadId: message.threadId,
188 messageId: message.messageId,
189 cause,
190 });
191 }
192 if (staleOrMissing()) {
193 const winner = currentMessages().find(
194 (candidate) => candidate.messageId === message.messageId,
195 );
196 if (winner !== undefined) {
197 try {
198 await options.storage.write(winner);
199 } catch {
200 // The winner's own serialized write follows this mutation and
201 // owns the failure handling for its payload.
202 }
203 }
204 return false;
205 }
206 bumpRevision(message.messageId);
207 setMessages([
208 ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
209 message,
210 ]);
211 return true;
212 });
213
214 // `expectedRevision` makes the removal a compare-and-set too: an edit
215 // accepted after the caller decided to remove (restore-to-composer reads
216 // the payload it is about to delete) keeps the newer message queued.
217 // `canRemove` adds a live ownership check for state such as an open editor,
218 // which can change without writing a new message revision.
219 const remove = (
220 message: QueuedThreadMessage,
221 expectedRevision?: number,
222 canRemove?: () => boolean,
223 ): Promise<QueuedThreadMessage | null> =>
224 serialize(async () => {
225 const removalCanceled = (): boolean =>
226 (expectedRevision !== undefined &&
227 (revisions.get(message.messageId) ?? 0) !== expectedRevision) ||
228 canRemove?.() === false;
229 if (removalCanceled()) {
230 return null;
231 }
232 // The live payload may carry attachments an accepted update added after
233 // the caller's snapshot; the caller releases files from what actually
234 // leaves the queue.
235 const removed =
236 currentMessages().find((candidate) => candidate.messageId === message.messageId) ?? message;
237 try {
238 await options.storage.remove(message);
239 } catch (cause) {
240 throw new ThreadOutboxManagerError({
241 operation: "remove",
242 environmentId: message.environmentId,
243 threadId: message.threadId,
244 messageId: message.messageId,
245 cause,
246 });
247 }
248 if (removalCanceled()) {
249 // An enqueue or editor lock can win while storage removal is in
250 // flight. Restore the live payload here, before any queued mutation
251 // gets its turn, so this canceled removal is durable on its own.
252 const winner = currentMessages().find(
253 (candidate) => candidate.messageId === message.messageId,
254 );
255 if (winner !== undefined) {
256 try {
257 await options.storage.write(winner);
258 } catch (cause) {
259 throw new ThreadOutboxManagerError({
260 operation: "remove",
261 environmentId: message.environmentId,
262 threadId: message.threadId,
263 messageId: message.messageId,
264 cause,
265 });
266 }
267 }
268 return null;
269 }
270 setMessages(
271 currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
272 );
273 // Tombstone, not delete: a same-id retry restarting at revision 1 would
274 // otherwise match a stale writer's expectedRevision from before the
275 // removal (ABA).
276 bumpRevision(message.messageId);
277 return removed;
278 });
279
280 const clearEnvironment = (
281 environmentId: EnvironmentId,
282 ): Promise<ReadonlyArray<QueuedThreadMessage>> => {
283 // Enqueues publish before their serialized writes. Capture revisions now,
284 // but wait for earlier mutations before reading messages: a message that
285 // changes after this request must not enter the clear set.
286 const revisionsAtRequest = new Map(revisions);
287 return serialize(async () => {
288 const persisted = await options.storage
289 .load()
290 .then((result) => {
291 if (result.errors.length > 0) {
292 throw new AggregateError(result.errors, "Some queued messages could not be read.");
293 }
294 return result.messages;
295 })
296 .catch((cause) => {
297 throw new ThreadOutboxManagerError({
298 operation: "clear-environment-load",
299 environmentId,
300 threadId: null,
301 messageId: null,
302 cause,
303 });
304 });
305 const allMessages = flattenQueuedThreadMessages(
306 groupQueuedThreadMessages([...persisted, ...currentMessages()]),
307 );
308 const candidates = allMessages.filter(
309 (message) =>
310 message.environmentId === environmentId &&
311 (revisions.get(message.messageId) ?? 0) ===
312 (revisionsAtRequest.get(message.messageId) ?? 0),
313 );
314 const candidateRevisions = new Map(
315 candidates.map(
316 (message) => [message.messageId, revisions.get(message.messageId) ?? 0] as const,
317 ),
318 );
319 const removedFromStorage = new Set<MessageId>();
320
321 await Promise.all(
322 candidates.map(async (message) => {
323 try {
324 await options.storage.remove(message);
325 removedFromStorage.add(message.messageId);
326 } catch (cause) {
327 warn(
328 "[thread-outbox] failed to clear persisted message",
329 new ThreadOutboxManagerError({
330 operation: "clear-environment-remove",
331 environmentId: message.environmentId,
332 threadId: message.threadId,
333 messageId: message.messageId,
334 cause,
335 }),
336 );
337 }
338 }),
339 );
340
341 // A same-id enqueue can publish while one of the removes above waits.
342 // Put its payload back before the later serialized enqueue write runs.
343 await Promise.all(
344 candidates.map(async (message) => {
345 if (
346 !removedFromStorage.has(message.messageId) ||
347 (revisions.get(message.messageId) ?? 0) === candidateRevisions.get(message.messageId)
348 ) {
349 return;
350 }
351 const retained = currentMessages().find(
352 (candidate) => candidate.messageId === message.messageId,
353 );
354 if (retained === undefined) {
355 return;
356 }
357 try {
358 await options.storage.write(retained);
359 } catch (cause) {
360 warn(
361 "[thread-outbox] failed to restore message retained during environment clear",
362 new ThreadOutboxManagerError({
363 operation: "clear-environment-remove",
364 environmentId: retained.environmentId,
365 threadId: retained.threadId,
366 messageId: retained.messageId,
367 cause,
368 }),
369 );
370 }
371 }),
372 );
373
374 const removed = candidates.filter(
375 (message) =>
376 removedFromStorage.has(message.messageId) &&
377 (revisions.get(message.messageId) ?? 0) === candidateRevisions.get(message.messageId),
378 );
379 const removedMessageIds = new Set(removed.map((message) => message.messageId));
380 const reconciledMessages = flattenQueuedThreadMessages(
381 groupQueuedThreadMessages([...allMessages, ...currentMessages()]),
382 ).filter((message) => !removedMessageIds.has(message.messageId));
383 for (const message of removed) {
384 bumpRevision(message.messageId);
385 }
386 setMessages(reconciledMessages);
387 // The caller releases these messages' attachment files; reporting what
388 // was actually removed keeps the release set honest even when this
389 // function's own load produced the messages.
390 return removed;
391 });
392 };
393
394 return {
395 queuedMessagesByThreadKeyAtom,
396 serialize,
397 load,
398 enqueue,
399 confirmQueued,
400 /** Current write revision for a queued message; input to update's CAS. */
401 revisionOf: (messageId: MessageId): number => revisions.get(messageId) ?? 0,
402 update,
403 remove,
404 clearEnvironment,
405 };
406}