1import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
2import * as Schema from "effect/Schema";
3
4import { writeFileAtomically } from "../lib/atomic-file";
5import {
6 decodeQueuedThreadMessage,
7 encodeQueuedThreadMessage,
8 type QueuedThreadMessage,
9} from "./thread-outbox-model";
10
11const THREAD_OUTBOX_DIRECTORY = "thread-outbox";
12
13const inFlightWrites = new Set<Promise<void>>();
14
15function trackInFlightWrite(operation: Promise<void>): Promise<void> {
16 inFlightWrites.add(operation);
17 void operation.catch(() => undefined).finally(() => inFlightWrites.delete(operation));
18 return operation;
19}
20
21/**
22 * Awaits queued-message writes so an app update restart cannot tear down the
23 * runtime while one is mid-file.
24 */
25export async function flushThreadOutboxWrites(): Promise<void> {
26 while (inFlightWrites.size > 0) {
27 await Promise.allSettled(inFlightWrites);
28 }
29}
30
31export class ThreadOutboxStorageError extends Schema.TaggedError<ThreadOutboxStorageError>()(
32 "ThreadOutboxStorageError",
33 {
34 operation: Schema.Literals(["load", "read-message", "write", "remove"]),
35 environmentId: Schema.NullOr(EnvironmentId),
36 threadId: Schema.NullOr(ThreadId),
37 messageId: Schema.NullOr(MessageId),
38 fileName: Schema.NullOr(Schema.String),
39 cause: Schema.Defect(),
40 },
41) {
42 override get message(): string {
43 return `Thread outbox storage operation ${this.operation} failed for environment ${this.environmentId ?? "unknown"}, thread ${this.threadId ?? "unknown"}, message ${this.messageId ?? "unknown"}, file ${this.fileName ?? "unknown"}.`;
44 }
45}
46
47export interface ThreadOutboxLoadResult {
48 readonly messages: ReadonlyArray<QueuedThreadMessage>;
49 readonly errors: ReadonlyArray<ThreadOutboxStorageError>;
50}
51
52export interface ThreadOutboxStorage {
53 readonly load: () => Promise<ThreadOutboxLoadResult>;
54 readonly write: (message: QueuedThreadMessage) => Promise<void>;
55 readonly remove: (message: QueuedThreadMessage) => Promise<void>;
56}
57
58function messageFileName(messageId: MessageId): string {
59 return `${encodeURIComponent(messageId)}.json`;
60}
61
62async function getOutboxDirectory() {
63 const { Directory, Paths } = await import("expo-file-system");
64 const directory = new Directory(Paths.document, THREAD_OUTBOX_DIRECTORY);
65 directory.create({ idempotent: true, intermediates: true });
66 return directory;
67}
68
69async function getMessageFile(messageId: MessageId) {
70 const { File } = await import("expo-file-system");
71 return new File(await getOutboxDirectory(), messageFileName(messageId));
72}
73
74export const expoThreadOutboxStorage: ThreadOutboxStorage = {
75 load: async () => {
76 const messages: QueuedThreadMessage[] = [];
77 const errors: ThreadOutboxStorageError[] = [];
78 try {
79 const { File } = await import("expo-file-system");
80 const directory = await getOutboxDirectory();
81
82 for (const entry of directory.list()) {
83 if (!(entry instanceof File) || !entry.name.endsWith(".json")) {
84 continue;
85 }
86 try {
87 messages.push(decodeQueuedThreadMessage(JSON.parse(await entry.text()) as unknown));
88 } catch (cause) {
89 // Recover readable messages without treating their attachment
90 // owners as the complete inventory needed for cleanup.
91 errors.push(
92 new ThreadOutboxStorageError({
93 operation: "read-message",
94 environmentId: null,
95 threadId: null,
96 messageId: null,
97 fileName: entry.name,
98 cause,
99 }),
100 );
101 }
102 }
103 } catch (cause) {
104 throw new ThreadOutboxStorageError({
105 operation: "load",
106 environmentId: null,
107 threadId: null,
108 messageId: null,
109 fileName: null,
110 cause,
111 });
112 }
113 return { messages, errors };
114 },
115 write: async (message) => {
116 const fileName = messageFileName(message.messageId);
117 try {
118 await trackInFlightWrite(
119 (async () => {
120 const file = await getMessageFile(message.messageId);
121 await writeFileAtomically(file, JSON.stringify(encodeQueuedThreadMessage(message)));
122 })(),
123 );
124 } catch (cause) {
125 throw new ThreadOutboxStorageError({
126 operation: "write",
127 environmentId: message.environmentId,
128 threadId: message.threadId,
129 messageId: message.messageId,
130 fileName,
131 cause,
132 });
133 }
134 },
135 remove: async (message) => {
136 const fileName = messageFileName(message.messageId);
137 try {
138 const file = await getMessageFile(message.messageId);
139 if (file.exists) {
140 file.delete();
141 }
142 } catch (cause) {
143 throw new ThreadOutboxStorageError({
144 operation: "remove",
145 environmentId: message.environmentId,
146 threadId: message.threadId,
147 messageId: message.messageId,
148 fileName,
149 cause,
150 });
151 }
152 },
153};