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",      "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;  };   const enqueue = (message: QueuedThreadMessage): Promise<void> =>    serialize(async () => {      try {        await options.storage.write(message);      } catch (cause) {        throw new ThreadOutboxManagerError({          operation: "enqueue",          environmentId: message.environmentId,          threadId: message.threadId,          messageId: message.messageId,          cause,        });      }      setMessages([...currentMessages(), message]);    });   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,    remove,    clearEnvironment,  };}