import type { EnvironmentId, MessageId } from "@t3tools/contracts";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 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((error) => {      loadPromise = null;      warn("[thread-outbox] failed to load persisted messages", error);    });    return loadPromise;  };   const enqueue = (message: QueuedThreadMessage): Promise<void> =>    serialize(async () => {      await options.storage.write(message);      setMessages([...currentMessages(), message]);    });   const remove = (message: QueuedThreadMessage): Promise<void> =>    serialize(async () => {      await options.storage.remove(message);      setMessages(        currentMessages().filter((candidate) => candidate.messageId !== message.messageId),      );    });   const clearEnvironment = (environmentId: EnvironmentId): Promise<void> =>    serialize(async () => {      const persisted = await options.storage.load().catch((error) => {        warn("[thread-outbox] failed to load messages while clearing environment", error);        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 (error) {              warn("[thread-outbox] failed to clear persisted message", error);            }          }),      );       setMessages(allMessages.filter((message) => !removedMessageIds.has(message.messageId)));    });   return {    queuedMessagesByThreadKeyAtom,    serialize,    load,    enqueue,    remove,    clearEnvironment,  };}