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/** Waits for in-flight writes before app-update restart. */
22export async function flushThreadOutboxWrites(): Promise<void> {
23 while (inFlightWrites.size > 0) {
24 await Promise.allSettled(inFlightWrites);
25 }
26}
27
28export class ThreadOutboxStorageError extends Schema.TaggedError<ThreadOutboxStorageError>()(
29 "ThreadOutboxStorageError",
30 {
31 operation: Schema.Literals(["load", "read-message", "write", "remove"]),
32 environmentId: Schema.NullOr(EnvironmentId),
33 threadId: Schema.NullOr(ThreadId),
34 messageId: Schema.NullOr(MessageId),
35 fileName: Schema.NullOr(Schema.String),
36 cause: Schema.Defect(),
37 },
38) {
39 override get message(): string {
40 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"}.`;
41 }
42}
43
44export interface ThreadOutboxLoadResult {
45 readonly messages: ReadonlyArray<QueuedThreadMessage>;
46 readonly errors: ReadonlyArray<ThreadOutboxStorageError>;
47}
48
49export interface ThreadOutboxStorage {
50 readonly load: () => Promise<ThreadOutboxLoadResult>;
51 readonly write: (message: QueuedThreadMessage) => Promise<void>;
52 readonly remove: (message: QueuedThreadMessage) => Promise<void>;
53}
54
55function messageFileName(messageId: MessageId): string {
56 return `${encodeURIComponent(messageId)}.json`;
57}
58
59async function getOutboxDirectory() {
60 const { Directory, Paths } = await import("expo-file-system");
61 const directory = new Directory(Paths.document, THREAD_OUTBOX_DIRECTORY);
62 directory.create({ idempotent: true, intermediates: true });
63 return directory;
64}
65
66async function getMessageFile(messageId: MessageId) {
67 const { File } = await import("expo-file-system");
68 return new File(await getOutboxDirectory(), messageFileName(messageId));
69}
70
71export const expoThreadOutboxStorage: ThreadOutboxStorage = {
72 load: async () => {
73 const messages: QueuedThreadMessage[] = [];
74 const errors: ThreadOutboxStorageError[] = [];
75 try {
76 const { File } = await import("expo-file-system");
77 const directory = await getOutboxDirectory();
78
79 for (const entry of directory.list()) {
80 if (!(entry instanceof File) || !entry.name.endsWith(".json")) {
81 continue;
82 }
83 try {
84 messages.push(decodeQueuedThreadMessage(JSON.parse(await entry.text()) as unknown));
85 } catch (cause) {
86 errors.push(
87 new ThreadOutboxStorageError({
88 operation: "read-message",
89 environmentId: null,
90 threadId: null,
91 messageId: null,
92 fileName: entry.name,
93 cause,
94 }),
95 );
96 }
97 }
98 } catch (cause) {
99 throw new ThreadOutboxStorageError({
100 operation: "load",
101 environmentId: null,
102 threadId: null,
103 messageId: null,
104 fileName: null,
105 cause,
106 });
107 }
108 return { messages, errors };
109 },
110 write: async (message) => {
111 const fileName = messageFileName(message.messageId);
112 try {
113 await trackInFlightWrite(
114 (async () => {
115 const file = await getMessageFile(message.messageId);
116 await writeFileAtomically(file, JSON.stringify(encodeQueuedThreadMessage(message)));
117 })(),
118 );
119 } catch (cause) {
120 throw new ThreadOutboxStorageError({
121 operation: "write",
122 environmentId: message.environmentId,
123 threadId: message.threadId,
124 messageId: message.messageId,
125 fileName,
126 cause,
127 });
128 }
129 },
130 remove: async (message) => {
131 const fileName = messageFileName(message.messageId);
132 try {
133 const file = await getMessageFile(message.messageId);
134 if (file.exists) {
135 file.delete();
136 }
137 } catch (cause) {
138 throw new ThreadOutboxStorageError({
139 operation: "remove",
140 environmentId: message.environmentId,
141 threadId: message.threadId,
142 messageId: message.messageId,
143 fileName,
144 cause,
145 });
146 }
147 },
148};