1import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets";
2import {
3 clampFileAttachmentUploadBytes,
4 fileAttachmentTooLargeMessage,
5 isAssetAttachmentNotFoundFailure,
6 runAttachmentUploadCycle,
7 verifyPersistedAttachmentUpload,
8} from "@t3tools/client-runtime/state/attachments";
9import { runAtomCommand, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime";
10import type {
11 ChatFileAttachment,
12 ChatImageAttachment,
13 EnvironmentId,
14 UploadChatImageAttachment,
15} from "@t3tools/contracts";
16import {
17 AuthOrchestrationOperateScope,
18 PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES,
19} from "@t3tools/contracts";
20import * as Option from "effect/Option";
21
22import { appAtomRegistry } from "../state/atom-registry";
23import { assetEnvironment } from "../state/assets";
24import { attachmentEnvironment } from "../state/attachments";
25import { environmentSession, readEnvironmentScope } from "../state/session";
26import { resolveOwnedComposerAttachmentFileUri } from "./composerAttachmentFiles";
27import { retainComposerAttachmentFileForPreview } from "./composerAttachmentPreviewRetention";
28import {
29 isComposerImageAttachment,
30 isFileBackedComposerAttachment,
31 type DraftComposerAttachment,
32 type DraftComposerImageAttachment,
33} from "./composerImages";
34import { imageMimeType } from "@t3tools/shared/image";
35import { uuidv4 } from "./uuid";
36
37/**
38 * This module owns the server side of a composer attachment's lifecycle.
39 * `prepareTurnAttachments` acquires pending uploads (verifying and reusing
40 * persisted ones), hands the uploaded ids back to the attachment's durable
41 * owner (queued outbox message or composer draft), and leaves their cleanup
42 * to that owner after it checks shared references. Nothing outside this module
43 * mints or deletes pending uploads. The local-file side of the lifecycle is
44 * owned by `removeThreadOutboxMessage` / the composer draft mutators, which
45 * release files through `releaseUnusedComposerAttachmentFiles`.
46 */
47export type UploadedMobileAttachment =
48 | UploadChatImageAttachment
49 | ChatImageAttachment
50 | ChatFileAttachment;
51
52export function validateDraftFileAttachments(input: {
53 readonly attachments: ReadonlyArray<DraftComposerAttachment>;
54 readonly serverConfig: {
55 readonly environment: {
56 readonly capabilities: {
57 readonly attachmentUploads?: boolean;
58 readonly fileAttachments?: { readonly maxUploadBytes: number };
59 };
60 };
61 } | null;
62}): string | null {
63 const files = input.attachments.filter((attachment) => attachment.type === "file");
64 if (files.length === 0) return null;
65 if (input.serverConfig === null) return "Server attachment support is still loading.";
66 const capabilities = input.serverConfig.environment.capabilities;
67 if (capabilities.attachmentUploads !== true || capabilities.fileAttachments === undefined) {
68 return "This server does not support file attachments.";
69 }
70 const maxBytes = clampFileAttachmentUploadBytes(capabilities.fileAttachments.maxUploadBytes);
71 const oversized = files.find((attachment) => attachment.sizeBytes > maxBytes);
72 return oversized ? fileAttachmentTooLargeMessage(oversized.name, maxBytes) : null;
73}
74
75/** Keep uploaded ids alongside the local bytes so a later send can reuse them. */
76export function withUploadedMobileAttachmentReferences(input: {
77 readonly environmentId: EnvironmentId;
78 readonly attachments: ReadonlyArray<DraftComposerAttachment>;
79 readonly uploadedAttachments: ReadonlyArray<UploadedMobileAttachment>;
80}): ReadonlyArray<DraftComposerAttachment> {
81 return input.attachments.map((attachment, index) => {
82 const uploaded = input.uploadedAttachments[index];
83 // A picture picked through Files stays `type: "file"` in the draft while it uploads as an
84 // image, so compare against the type it was actually sent under: comparing draft types
85 // drops the id, and the next send re-uploads bytes the server already holds.
86 const uploadedAs = isComposerImageAttachment(attachment) ? "image" : attachment.type;
87 if (
88 !uploaded ||
89 !("id" in uploaded) ||
90 uploadedAs !== uploaded.type ||
91 (attachment.uploadedAttachmentId === uploaded.id &&
92 attachment.uploadEnvironmentId === input.environmentId)
93 ) {
94 return attachment;
95 }
96 return {
97 ...attachment,
98 uploadedAttachmentId: uploaded.id,
99 uploadEnvironmentId: input.environmentId,
100 };
101 });
102}
103
104/**
105 * Deletes pending uploads the client no longer references. Every delete result
106 * is inspected; failed deletes are retried once and a persistent failure
107 * throws, so a caller can never silently leak the outcome. (The server also
108 * expires pending uploads, so a leaked id self-heals eventually.)
109 */
110export async function releasePendingAttachmentUploads(
111 environmentId: EnvironmentId,
112 attachmentIds: ReadonlyArray<string>,
113): Promise<void> {
114 const deleteOnce = async (attachmentId: string): Promise<boolean> => {
115 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) {
116 throw new Error("This connection cannot delete pending attachments.");
117 }
118 const result = await runAtomCommand(
119 appAtomRegistry,
120 attachmentEnvironment.remove,
121 { environmentId, input: { attachmentId } },
122 { reportFailure: false, reportDefect: false },
123 );
124 return (
125 result._tag === "Success" ||
126 isAssetAttachmentNotFoundFailure(squashAtomCommandFailure(result))
127 );
128 };
129
130 const failedAttachmentIds: string[] = [];
131 for (const attachmentId of attachmentIds) {
132 if (!(await deleteOnce(attachmentId)) && !(await deleteOnce(attachmentId))) {
133 failedAttachmentIds.push(attachmentId);
134 }
135 }
136 if (failedAttachmentIds.length > 0) {
137 throw new Error(
138 `Could not delete ${failedAttachmentIds.length} pending attachment upload(s): ${failedAttachmentIds.join(", ")}.`,
139 );
140 }
141}
142
143async function releaseCreatedUploadsQuietly(
144 environmentId: EnvironmentId,
145 attachmentIds: ReadonlyArray<string>,
146): Promise<void> {
147 try {
148 await releasePendingAttachmentUploads(environmentId, attachmentIds);
149 } catch (error) {
150 // The original failure must propagate; the leaked pending uploads expire
151 // on the server.
152 console.warn("[attachments] could not delete abandoned pending uploads", error);
153 }
154}
155
156export interface PreparedTurnAttachments {
157 readonly status: "ready";
158 /** Wire attachments for `startTurn`, in the original composer order. */
159 readonly attachments: ReadonlyArray<UploadedMobileAttachment>;
160 /** Composer attachments annotated with the uploaded pending ids. */
161 readonly draftAttachments: ReadonlyArray<DraftComposerAttachment>;
162 /** Every pending upload backing this turn (reused and newly minted). */
163 readonly pendingAttachmentIds: ReadonlyArray<string>;
164}
165
166export type PrepareTurnAttachmentsResult =
167 | PreparedTurnAttachments
168 | { readonly status: "abandoned" };
169
170/**
171 * The mime an attachment travels under. A picture picked through Files arrives typed as a plain
172 * file, often with no usable mime, so it is promoted to the type the provider accepts. Every
173 * place that names the attachment on the wire — the upload header, the upload input, and the
174 * message reference — has to agree on this one value, or the turn describes bytes that are not
175 * what was actually sent and `ChatImageAttachment` rejects it.
176 */
177export function composerAttachmentWireMimeType(attachment: DraftComposerAttachment): string {
178 if (!isComposerImageAttachment(attachment)) return attachment.mimeType;
179 return supportedImageWireMimeType(attachment);
180}
181
182function supportedImageWireMimeType(
183 attachment: DraftComposerAttachment,
184): (typeof PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES)[number] {
185 const inferred = imageMimeType(attachment);
186 const mimeType = PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES.find(
187 (type) => type === attachment.mimeType.toLowerCase() || type === inferred,
188 );
189 if (!mimeType) throw new Error(`Unsupported image type for '${attachment.name}'.`);
190 return mimeType;
191}
192
193function uploadedReference(
194 attachment: DraftComposerAttachment,
195 id: string,
196): ChatImageAttachment | ChatFileAttachment {
197 const fields = {
198 id,
199 name: attachment.name,
200 mimeType: composerAttachmentWireMimeType(attachment),
201 sizeBytes: attachment.sizeBytes,
202 };
203 // A picture picked through Files is typed as a plain file; uploading it as one leaves the
204 // chat view with nothing to show a thumbnail from, on every client.
205 return isComposerImageAttachment(attachment)
206 ? { type: "image", ...fields }
207 : {
208 type: "file",
209 ...fields,
210 ...(attachment.source ? { source: attachment.source } : {}),
211 };
212}
213
214function attachmentUploadInput(attachment: DraftComposerAttachment) {
215 const fields = { name: attachment.name, sizeBytes: attachment.sizeBytes };
216 return isComposerImageAttachment(attachment)
217 ? { ...fields, mimeType: supportedImageWireMimeType(attachment) }
218 : { type: "file" as const, ...fields, mimeType: attachment.mimeType };
219}
220
221/**
222 * Wire shape for startTurn on servers without attachment uploads: pure inline
223 * uploads without client draft id / previewUri. File-backed images read their
224 * base64 from disk lazily, only when this legacy path is actually taken.
225 */
226async function toUploadChatImageAttachments(
227 attachments: ReadonlyArray<DraftComposerImageAttachment>,
228): Promise<ReadonlyArray<UploadChatImageAttachment>> {
229 return Promise.all(
230 attachments.map(async (attachment) => ({
231 type: attachment.type,
232 name: attachment.name,
233 mimeType: attachment.mimeType,
234 sizeBytes: attachment.sizeBytes,
235 dataUrl: await composerImageAttachmentDataUrl(attachment),
236 })),
237 );
238}
239
240/** Inline bytes for one image: legacy drafts carry them, file-backed ones read them from disk. */
241async function composerImageAttachmentDataUrl(
242 attachment: DraftComposerImageAttachment,
243): Promise<string> {
244 if (attachment.dataUrl !== undefined) {
245 return attachment.dataUrl;
246 }
247 if (!isFileBackedComposerAttachment(attachment)) {
248 throw new Error(`'${attachment.name}' is no longer available. Attach the image again.`);
249 }
250 const release = retainComposerAttachmentFileForPreview(attachment);
251 try {
252 const { File, Paths } = await import("expo-file-system");
253 const uri =
254 resolveOwnedComposerAttachmentFileUri(attachment.fileUri, Paths.document.uri) ??
255 attachment.fileUri;
256 const base64 = await new File(uri).base64();
257 return `data:${attachment.mimeType};base64,${base64}`;
258 } finally {
259 release();
260 }
261}
262
263async function uploadFileBytes(
264 environmentId: EnvironmentId,
265 attachment: DraftComposerAttachment,
266 url: string,
267 signal: AbortSignal,
268 onProgress?: (progress: number) => void,
269): Promise<void> {
270 const { File, Paths, UploadType } = await import("expo-file-system");
271 if (signal.aborted) throw new Error("Upload cancelled.");
272 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) {
273 throw new Error("This connection cannot upload attachments.");
274 }
275 // Legacy image drafts persisted inline bytes and stage them in a temp cache
276 // file for the native uploader. Everything else uploads its owned copy.
277 const fileUri = attachment.fileUri;
278 const inlineDataUrl = attachment.type === "image" ? attachment.dataUrl : undefined;
279 if (fileUri === undefined && inlineDataUrl === undefined) {
280 throw new Error(`'${attachment.name}' is no longer available. Attach the image again.`);
281 }
282 const file =
283 fileUri === undefined
284 ? new File(Paths.cache, `t3-upload-${uuidv4()}`)
285 : new File(resolveOwnedComposerAttachmentFileUri(fileUri, Paths.document.uri) ?? fileUri);
286 try {
287 if (fileUri === undefined && inlineDataUrl !== undefined) {
288 file.create();
289 await file.write(inlineDataUrl.slice(inlineDataUrl.indexOf(",") + 1), {
290 encoding: "base64",
291 });
292 }
293 const result = await file.upload(url, {
294 httpMethod: "POST",
295 uploadType: UploadType.BINARY_CONTENT,
296 headers: { "Content-Type": composerAttachmentWireMimeType(attachment) },
297 signal,
298 ...(onProgress
299 ? {
300 onProgress: ({ bytesSent, totalBytes }) => {
301 if (totalBytes > 0) onProgress(bytesSent / totalBytes);
302 },
303 }
304 : {}),
305 });
306 if (result.status < 200 || result.status >= 300) {
307 throw new Error(`Upload failed for '${attachment.name}' (${result.status}).`);
308 }
309 } finally {
310 if (fileUri === undefined && file.exists) file.delete();
311 }
312}
313
314/**
315 * Acquires server-side uploads for one turn's attachments and persists the
316 * uploaded ids into the attachments' durable owner.
317 *
318 * `persistUploadedReferences` runs once the bytes are on the server and only
319 * when new ids appeared. It must write the annotated attachments into the
320 * owner (queued message or draft) so a retry after a crash reuses the bytes.
321 * Returning `"abandon"` (owner no longer wants the send) or throwing deletes
322 * the pending uploads this call minted, so the owner cannot leak them.
323 */
324export async function prepareTurnAttachments(input: {
325 readonly environmentId: EnvironmentId;
326 readonly attachments: ReadonlyArray<DraftComposerAttachment>;
327 /** Older environments continue to receive inline images. */
328 readonly supportsImageUploads?: boolean;
329 readonly signal?: AbortSignal;
330 readonly onUploadProgress?: (attachmentId: string, progress: number) => void;
331 readonly persistUploadedReferences?: (
332 draftAttachments: ReadonlyArray<DraftComposerAttachment>,
333 ) => Promise<"persisted" | "abandon">;
334}): Promise<PrepareTurnAttachmentsResult> {
335 const { environmentId } = input;
336 if (input.signal?.aborted) return { status: "abandoned" };
337 const files = input.attachments.filter((attachment) => attachment.type === "file");
338 const ready = (
339 attachments: ReadonlyArray<UploadedMobileAttachment>,
340 pendingAttachmentIds: ReadonlyArray<string>,
341 draftAttachments: ReadonlyArray<DraftComposerAttachment>,
342 ): PreparedTurnAttachments => ({
343 status: "ready",
344 attachments,
345 draftAttachments,
346 pendingAttachmentIds,
347 });
348
349 if (input.attachments.length === 0 || (files.length === 0 && !input.supportsImageUploads)) {
350 try {
351 const imageAttachments = await toUploadChatImageAttachments(
352 input.attachments.filter((attachment) => attachment.type === "image"),
353 );
354 if (input.signal?.aborted) return { status: "abandoned" };
355 return ready(imageAttachments, [], input.attachments);
356 } catch (error) {
357 if (input.signal?.aborted) return { status: "abandoned" };
358 throw error;
359 }
360 }
361
362 const requireUploadAccess = () => {
363 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) {
364 throw new Error("This connection cannot upload attachments.");
365 }
366 };
367 requireUploadAccess();
368
369 const connection = appAtomRegistry.get(
370 environmentSession.preparedConnectionValueAtom(environmentId),
371 );
372 if (Option.isNone(connection)) {
373 throw new Error("The environment is not connected.");
374 }
375
376 const uploadedAttachments: UploadedMobileAttachment[] = [];
377 const pendingAttachmentIds: string[] = [];
378 const createdAttachmentIds: string[] = [];
379 const controller = new AbortController();
380 const abort = () => controller.abort();
381 input.signal?.addEventListener("abort", abort, { once: true });
382 try {
383 for (const attachment of input.attachments) {
384 if (controller.signal.aborted) throw new Error("Upload cancelled.");
385 requireUploadAccess();
386 if (attachment.type === "image" && !input.supportsImageUploads) {
387 uploadedAttachments.push(...(await toUploadChatImageAttachments([attachment])));
388 continue;
389 }
390
391 // Reuse the bytes from a previous attempt when their pending upload is
392 // still alive on this environment.
393 if (
394 attachment.uploadEnvironmentId === environmentId &&
395 attachment.uploadedAttachmentId !== undefined
396 ) {
397 const verification = await verifyPersistedAttachmentUpload({
398 registry: appAtomRegistry,
399 createAssetUrl: assetEnvironment.createUrl,
400 environmentId,
401 attachmentId: attachment.uploadedAttachmentId,
402 });
403 if (verification.status === "failed") {
404 throw verification.error;
405 }
406 if (verification.status === "verified") {
407 pendingAttachmentIds.push(attachment.uploadedAttachmentId);
408 uploadedAttachments.push(uploadedReference(attachment, attachment.uploadedAttachmentId));
409 continue;
410 }
411 // "missing": the pending upload expired, upload the bytes again.
412 }
413
414 requireUploadAccess();
415 const result = await runAttachmentUploadCycle({
416 registry: appAtomRegistry,
417 createUploadUrl: attachmentEnvironment.createUploadUrl,
418 remove: attachmentEnvironment.remove,
419 environmentId,
420 upload: attachmentUploadInput(attachment),
421 // Read the connection at transfer time: the environment may have
422 // reconnected on a new base URL since this cycle started.
423 resolveUploadUrl: (relativeUrl) => {
424 requireUploadAccess();
425 const currentConnection = appAtomRegistry.get(
426 environmentSession.preparedConnectionValueAtom(environmentId),
427 );
428 return Option.isNone(currentConnection)
429 ? null
430 : resolveAssetUrl(currentConnection.value.httpBaseUrl, relativeUrl);
431 },
432 transport: (url) => ({
433 done: uploadFileBytes(
434 environmentId,
435 attachment,
436 url,
437 controller.signal,
438 input.onUploadProgress
439 ? (progress) => input.onUploadProgress?.(attachment.id, progress)
440 : undefined,
441 ),
442 abort,
443 }),
444 onMinted: (attachmentId) => {
445 if (controller.signal.aborted) return "cancel";
446 pendingAttachmentIds.push(attachmentId);
447 createdAttachmentIds.push(attachmentId);
448 return "continue";
449 },
450 });
451 if (result.status !== "uploaded") {
452 throw result.status === "failed" && result.error !== undefined
453 ? result.error
454 : new Error(`Upload failed for '${attachment.name}'.`);
455 }
456 uploadedAttachments.push(uploadedReference(attachment, result.attachmentId));
457 }
458
459 if (controller.signal.aborted) throw new Error("Upload cancelled.");
460
461 const draftAttachments = withUploadedMobileAttachmentReferences({
462 environmentId,
463 attachments: input.attachments,
464 uploadedAttachments,
465 });
466 const referencesChanged = draftAttachments.some(
467 (attachment, index) => attachment !== input.attachments[index],
468 );
469 if (referencesChanged && input.persistUploadedReferences) {
470 if ((await input.persistUploadedReferences(draftAttachments)) === "abandon") {
471 await releaseCreatedUploadsQuietly(environmentId, createdAttachmentIds);
472 return { status: "abandoned" };
473 }
474 }
475 return ready(uploadedAttachments, pendingAttachmentIds, draftAttachments);
476 } catch (error) {
477 await releaseCreatedUploadsQuietly(environmentId, createdAttachmentIds);
478 if (controller.signal.aborted) return { status: "abandoned" };
479 throw error;
480 } finally {
481 input.signal?.removeEventListener("abort", abort);
482 }
483}