import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";import * as Schema from "effect/Schema";import { Atom, type AtomRegistry } from "effect/unstable/reactivity"; import { flattenQueuedThreadMessages, groupQueuedThreadMessages, type QueuedThreadMessage,} from "./thread-outbox-model";import type { ThreadOutboxStorage } from "./thread-outbox-storage"; export class ThreadOutboxManagerError extends Schema.TaggedErrorClass<ThreadOutboxManagerError>()( "ThreadOutboxManagerError", { operation: Schema.Literals([ "load", "enqueue", "update", "remove", "clear-environment-load", "clear-environment-remove", ]), environmentId: Schema.NullOr(EnvironmentId), threadId: Schema.NullOr(ThreadId), messageId: Schema.NullOr(MessageId), cause: Schema.Defect(), },) { override get message(): string { return `Thread outbox operation ${this.operation} failed for environment ${this.environmentId ?? "unknown"}, thread ${this.threadId ?? "unknown"}, message ${this.messageId ?? "unknown"}.`; }} export interface ThreadOutboxManagerOptions { readonly registry: AtomRegistry.AtomRegistry; readonly storage: ThreadOutboxStorage; readonly warn?: (message: string, error: unknown) => void;} export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { const queuedMessagesByThreadKeyAtom = Atom.make< Record<string, ReadonlyArray<QueuedThreadMessage>> >({}).pipe(Atom.keepAlive, Atom.withLabel("mobile:thread-outbox:queued-messages")); const warn = options.warn ?? ((message: string, error: unknown) => { console.warn(message, error); }); let loadPromise: Promise<void> | null = null; let mutationQueue: Promise<void> = Promise.resolve(); const serialize = <A>(mutation: () => Promise<A>): Promise<A> => { const result = mutationQueue.then(mutation, mutation); mutationQueue = result.then( () => undefined, () => undefined, ); return result; }; const currentMessages = (): ReadonlyArray<QueuedThreadMessage> => flattenQueuedThreadMessages(options.registry.get(queuedMessagesByThreadKeyAtom)); const setMessages = (messages: ReadonlyArray<QueuedThreadMessage>): void => { options.registry.set(queuedMessagesByThreadKeyAtom, groupQueuedThreadMessages(messages)); }; const load = (): Promise<void> => { if (loadPromise !== null) { return loadPromise; } loadPromise = serialize(async () => { const persistedMessages = await options.storage.load(); setMessages([...persistedMessages, ...currentMessages()]); }).catch((cause) => { loadPromise = null; warn( "[thread-outbox] failed to load persisted messages", new ThreadOutboxManagerError({ operation: "load", environmentId: null, threadId: null, messageId: null, cause, }), ); }); return loadPromise; }; // The queued atom drives the composer's immediate "queued" feedback, so it // is published synchronously; the durable write happens behind it and rolls // the message back out if it fails (durability only matters for crash // recovery, not for the in-session queue). const enqueue = (message: QueuedThreadMessage): Promise<void> => { setMessages([ ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), message, ]); return serialize(async () => { try { await options.storage.write(message); } catch (cause) { // Roll back by reference, not messageId: a retry enqueue with the same // id may have optimistically replaced this attempt while the write was // in flight, and its entry must survive this attempt's failure. setMessages(currentMessages().filter((candidate) => candidate !== message)); throw new ThreadOutboxManagerError({ operation: "enqueue", environmentId: message.environmentId, threadId: message.threadId, messageId: message.messageId, cause, }); } }); }; // Resolves once all pending mutations (including any in-flight enqueue // write) have settled, reporting whether the message is still queued. The // drain awaits this before dispatching so a message whose durable write // later fails can never have been delivered first. const confirmQueued = (message: QueuedThreadMessage): Promise<boolean> => serialize(async () => currentMessages().some((candidate) => candidate === message)); // Rewrites an already-queued message. A no-op when the message has been // removed in the meantime (e.g. deleted or delivered), so a trailing editor // flush can never resurrect it. Returns whether the message was updated. const update = (message: QueuedThreadMessage): Promise<boolean> => serialize(async () => { const exists = currentMessages().some( (candidate) => candidate.messageId === message.messageId, ); if (!exists) { return false; } try { await options.storage.write(message); } catch (cause) { throw new ThreadOutboxManagerError({ operation: "update", environmentId: message.environmentId, threadId: message.threadId, messageId: message.messageId, cause, }); } setMessages([ ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), message, ]); return true; }); const remove = (message: QueuedThreadMessage): Promise<void> => serialize(async () => { try { await options.storage.remove(message); } catch (cause) { throw new ThreadOutboxManagerError({ operation: "remove", environmentId: message.environmentId, threadId: message.threadId, messageId: message.messageId, cause, }); } setMessages( currentMessages().filter((candidate) => candidate.messageId !== message.messageId), ); }); const clearEnvironment = (environmentId: EnvironmentId): Promise<void> => serialize(async () => { const persisted = await options.storage.load().catch((cause) => { warn( "[thread-outbox] failed to load messages while clearing environment", new ThreadOutboxManagerError({ operation: "clear-environment-load", environmentId, threadId: null, messageId: null, cause, }), ); return []; }); const allMessages = flattenQueuedThreadMessages( groupQueuedThreadMessages([...persisted, ...currentMessages()]), ); const removedMessageIds = new Set<MessageId>(); await Promise.all( allMessages .filter((message) => message.environmentId === environmentId) .map(async (message) => { try { await options.storage.remove(message); removedMessageIds.add(message.messageId); } catch (cause) { warn( "[thread-outbox] failed to clear persisted message", new ThreadOutboxManagerError({ operation: "clear-environment-remove", environmentId: message.environmentId, threadId: message.threadId, messageId: message.messageId, cause, }), ); } }), ); setMessages(allMessages.filter((message) => !removedMessageIds.has(message.messageId))); }); return { queuedMessagesByThreadKeyAtom, serialize, load, enqueue, confirmQueued, update, remove, clearEnvironment, };}
import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
import * as Schema from "effect/Schema";
import { Atom, type AtomRegistry } from "effect/unstable/reactivity";
import {
flattenQueuedThreadMessages,
groupQueuedThreadMessages,
type QueuedThreadMessage,
} from "./thread-outbox-model";
import type { ThreadOutboxStorage } from "./thread-outbox-storage";
export class ThreadOutboxManagerError extends Schema.TaggedErrorClass<ThreadOutboxManagerError>()(
"ThreadOutboxManagerError",
{
operation: Schema.Literals([
"load",
"enqueue",
"update",
"remove",
"clear-environment-load",
"clear-environment-remove",
]),
environmentId: Schema.NullOr(EnvironmentId),
threadId: Schema.NullOr(ThreadId),
messageId: Schema.NullOr(MessageId),
cause: Schema.Defect(),
},
) {
override get message(): string {
return `Thread outbox operation ${this.operation} failed for environment ${this.environmentId ?? "unknown"}, thread ${this.threadId ?? "unknown"}, message ${this.messageId ?? "unknown"}.`;
}
}
export interface ThreadOutboxManagerOptions {
readonly registry: AtomRegistry.AtomRegistry;
readonly storage: ThreadOutboxStorage;
readonly warn?: (message: string, error: unknown) => void;
}
export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) {
const queuedMessagesByThreadKeyAtom = Atom.make<
Record<string, ReadonlyArray<QueuedThreadMessage>>
>({}).pipe(Atom.keepAlive, Atom.withLabel("mobile:thread-outbox:queued-messages"));
const warn =
options.warn ??
((message: string, error: unknown) => {
console.warn(message, error);
});
let loadPromise: Promise<void> | null = null;
let mutationQueue: Promise<void> = Promise.resolve();
const serialize = <A>(mutation: () => Promise<A>): Promise<A> => {
const result = mutationQueue.then(mutation, mutation);
mutationQueue = result.then(
() => undefined,
() => undefined,
);
return result;
};
const currentMessages = (): ReadonlyArray<QueuedThreadMessage> =>
flattenQueuedThreadMessages(options.registry.get(queuedMessagesByThreadKeyAtom));
const setMessages = (messages: ReadonlyArray<QueuedThreadMessage>): void => {
options.registry.set(queuedMessagesByThreadKeyAtom, groupQueuedThreadMessages(messages));
};
const load = (): Promise<void> => {
if (loadPromise !== null) {
return loadPromise;
}
loadPromise = serialize(async () => {
const persistedMessages = await options.storage.load();
setMessages([...persistedMessages, ...currentMessages()]);
}).catch((cause) => {
loadPromise = null;
warn(
"[thread-outbox] failed to load persisted messages",
new ThreadOutboxManagerError({
operation: "load",
environmentId: null,
threadId: null,
messageId: null,
cause,
}),
);
});
return loadPromise;
};
// The queued atom drives the composer's immediate "queued" feedback, so it
// is published synchronously; the durable write happens behind it and rolls
// the message back out if it fails (durability only matters for crash
// recovery, not for the in-session queue).
const enqueue = (message: QueuedThreadMessage): Promise<void> => {
setMessages([
...currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
message,
]);
return serialize(async () => {
try {
await options.storage.write(message);
} catch (cause) {
// Roll back by reference, not messageId: a retry enqueue with the same
// id may have optimistically replaced this attempt while the write was
// in flight, and its entry must survive this attempt's failure.
setMessages(currentMessages().filter((candidate) => candidate !== message));
throw new ThreadOutboxManagerError({
operation: "enqueue",
environmentId: message.environmentId,
threadId: message.threadId,
messageId: message.messageId,
cause,
});
}
});
};
// Resolves once all pending mutations (including any in-flight enqueue
// write) have settled, reporting whether the message is still queued. The
// drain awaits this before dispatching so a message whose durable write
// later fails can never have been delivered first.
const confirmQueued = (message: QueuedThreadMessage): Promise<boolean> =>
serialize(async () => currentMessages().some((candidate) => candidate === message));
// Rewrites an already-queued message. A no-op when the message has been
// removed in the meantime (e.g. deleted or delivered), so a trailing editor
// flush can never resurrect it. Returns whether the message was updated.
const update = (message: QueuedThreadMessage): Promise<boolean> =>
serialize(async () => {
const exists = currentMessages().some(
(candidate) => candidate.messageId === message.messageId,
);
if (!exists) {
return false;
}
try {
await options.storage.write(message);
} catch (cause) {
throw new ThreadOutboxManagerError({
operation: "update",
environmentId: message.environmentId,
threadId: message.threadId,
messageId: message.messageId,
cause,
});
}
setMessages([
...currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
message,
]);
return true;
});
const remove = (message: QueuedThreadMessage): Promise<void> =>
serialize(async () => {
try {
await options.storage.remove(message);
} catch (cause) {
throw new ThreadOutboxManagerError({
operation: "remove",
environmentId: message.environmentId,
threadId: message.threadId,
messageId: message.messageId,
cause,
});
}
setMessages(
currentMessages().filter((candidate) => candidate.messageId !== message.messageId),
);
});
const clearEnvironment = (environmentId: EnvironmentId): Promise<void> =>
serialize(async () => {
const persisted = await options.storage.load().catch((cause) => {
warn(
"[thread-outbox] failed to load messages while clearing environment",
new ThreadOutboxManagerError({
operation: "clear-environment-load",
environmentId,
threadId: null,
messageId: null,
cause,
}),
);
return [];
});
const allMessages = flattenQueuedThreadMessages(
groupQueuedThreadMessages([...persisted, ...currentMessages()]),
);
const removedMessageIds = new Set<MessageId>();
await Promise.all(
allMessages
.filter((message) => message.environmentId === environmentId)
.map(async (message) => {
try {
await options.storage.remove(message);
removedMessageIds.add(message.messageId);
} catch (cause) {
warn(
"[thread-outbox] failed to clear persisted message",
new ThreadOutboxManagerError({
operation: "clear-environment-remove",
environmentId: message.environmentId,
threadId: message.threadId,
messageId: message.messageId,
cause,
}),
);
}
}),
);
setMessages(allMessages.filter((message) => !removedMessageIds.has(message.messageId)));
});
return {
queuedMessagesByThreadKeyAtom,
serialize,
load,
enqueue,
confirmQueued,
update,
remove,
clearEnvironment,
};
}