1import {
2 AuthOrchestrationOperateScope,
3 PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES,
4 type ChatAttachment,
5 type EnvironmentId,
6} from "@t3tools/contracts";
7import { parseScopedThreadKey } from "@t3tools/client-runtime/environment";
8import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets";
9import {
10 deletePendingAttachmentUpload,
11 runAttachmentUploadCycle,
12 verifyPersistedAttachmentUpload,
13 type PersistedAttachmentVerification,
14} from "@t3tools/client-runtime/state/attachments";
15import { create } from "zustand";
16import * as Option from "effect/Option";
17import { AsyncResult } from "effect/reactivity";
18
19import {
20 DraftId,
21 useComposerDraftStore,
22 type ComposerFileAttachment,
23 type ComposerImageAttachment,
24 type ComposerThreadTarget,
25} from "../composerDraftStore";
26import { appAtomRegistry } from "../rpc/atomRegistry";
27import { environmentCatalog } from "../connection/catalog";
28import { assetEnvironment } from "../state/assets";
29import { attachmentEnvironment } from "../state/attachments";
30import { readEnvironmentScope, readPreparedConnection } from "../state/session";
31import type { AttachmentUploadState, ReadyAttachmentUpload } from "./attachmentUploadState";
32
33const MAX_UPLOADS_PER_ENVIRONMENT = 3;
34const UPLOAD_TIMEOUT_MS = 5 * 60_000;
35
36interface AttachmentUploadStore {
37 readonly uploadsByImageId: Readonly<Record<string, AttachmentUploadState>>;
38}
39
40export const useAttachmentUploadStore = create<AttachmentUploadStore>(() => ({
41 uploadsByImageId: {},
42}));
43
44interface UploadJob {
45 readonly image: ComposerImageAttachment | ComposerFileAttachment;
46 readonly environmentId: EnvironmentId;
47 /**
48 * The draft that owned this file when the job started. Completion resolves
49 * the current owner because the file can move while the upload is pending.
50 */
51 readonly draftTarget?: ComposerThreadTarget;
52 readonly previous?: ReadyAttachmentUpload;
53 /**
54 * The draft's persisted server-side upload, to verify instead of re-upload.
55 * The draft owns this id; the queue never deletes it on cancel or retry.
56 * Deleting it goes through `releasePersistedAttachmentUpload` only.
57 */
58 readonly persistedAttachmentId?: string;
59 readonly settled: Promise<void>;
60 resolveSettled: () => void;
61 /** Only ids this queue minted itself. Cancel and retry may delete these. */
62 attachmentId: string | null;
63 cancelled: boolean;
64 abort: (() => void) | null;
65 stopWatchingConnection: () => void;
66}
67
68// Failed jobs retain their source and connection subscription until retry or release.
69const jobsByImageId = new Map<string, UploadJob>();
70const queue: UploadJob[] = [];
71const activeUploadsByEnvironment = new Map<EnvironmentId, number>();
72
73function setUploadState(imageId: string, upload: AttachmentUploadState): void {
74 useAttachmentUploadStore.setState((state) => ({
75 uploadsByImageId: { ...state.uploadsByImageId, [imageId]: upload },
76 }));
77}
78
79function clearUploadState(imageId: string): void {
80 useAttachmentUploadStore.setState((state) => {
81 if (!(imageId in state.uploadsByImageId)) {
82 return state;
83 }
84 const uploadsByImageId = { ...state.uploadsByImageId };
85 delete uploadsByImageId[imageId];
86 return { uploadsByImageId };
87 });
88}
89
90export function readAttachmentUpload(imageId: string): AttachmentUploadState | undefined {
91 return useAttachmentUploadStore.getState().uploadsByImageId[imageId];
92}
93
94/** Finds the file's current same-environment draft after any in-flight move. */
95function resolveCurrentFileDraftTarget(job: UploadJob): ComposerThreadTarget | undefined {
96 if (job.draftTarget === undefined || job.image.type !== "file") {
97 return undefined;
98 }
99 const store = useComposerDraftStore.getState();
100 for (const [key, draft] of Object.entries(store.draftsByThreadKey)) {
101 if (!draft.files.some((file) => file.id === job.image.id)) {
102 continue;
103 }
104 const draftSession = store.draftThreadsByThreadKey[key];
105 if (draftSession !== undefined) {
106 if (draftSession.environmentId === job.environmentId) {
107 return DraftId.make(key);
108 }
109 continue;
110 }
111 // Tests and legacy callers can use a DraftId without session metadata.
112 // Only its original job supplies enough environment identity to trust it.
113 if (typeof job.draftTarget === "string" && job.draftTarget === key) {
114 return DraftId.make(key);
115 }
116 const threadRef = parseScopedThreadKey(key);
117 if (threadRef?.environmentId === job.environmentId) {
118 return threadRef;
119 }
120 }
121 return undefined;
122}
123
124/**
125 * Persists a finished upload's ids onto the draft that owns the file. The
126 * mounted composer effect performs the same write for live UI updates, but a
127 * background completion (user navigated away, upload finished, reload) must
128 * not depend on a mounted composer to survive. `setFileUpload` no-ops when
129 * the draft row is gone or already carries these ids.
130 */
131function stampDraftFileUpload(job: UploadJob, attachmentId: string): void {
132 const draftTarget = resolveCurrentFileDraftTarget(job);
133 if (draftTarget === undefined) {
134 return;
135 }
136 useComposerDraftStore
137 .getState()
138 .setFileUpload(draftTarget, job.image.id, job.environmentId, attachmentId);
139}
140
141function deletePendingUpload(environmentId: EnvironmentId, attachmentId: string): void {
142 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return;
143 deletePendingAttachmentUpload({
144 registry: appAtomRegistry,
145 remove: attachmentEnvironment.remove,
146 environmentId,
147 attachmentId,
148 });
149}
150
151function uploadBytes(input: {
152 readonly url: string;
153 readonly file: File;
154 readonly mimeType: string;
155 readonly onProgress: (progress: number) => void;
156}): { readonly done: Promise<void>; readonly abort: () => void } {
157 const xhr = new XMLHttpRequest();
158 const done = new Promise<void>((resolve, reject) => {
159 xhr.open("POST", input.url, true);
160 xhr.timeout = UPLOAD_TIMEOUT_MS;
161 xhr.setRequestHeader("Content-Type", input.mimeType);
162 xhr.upload.addEventListener("progress", (event) => {
163 if (event.lengthComputable && event.total > 0) {
164 input.onProgress(event.loaded / event.total);
165 }
166 });
167 xhr.addEventListener("load", () => {
168 if (xhr.status >= 200 && xhr.status < 300) {
169 resolve();
170 } else {
171 reject(new Error(`Upload rejected (${xhr.status})`));
172 }
173 });
174 xhr.addEventListener("error", () => reject(new Error("Upload failed")));
175 xhr.addEventListener("timeout", () => reject(new Error("Upload timed out")));
176 xhr.addEventListener("abort", () => reject(new Error("Upload cancelled")));
177 xhr.send(input.file);
178 });
179
180 return { done, abort: () => xhr.abort() };
181}
182
183async function runUpload(job: UploadJob): Promise<void> {
184 if (!readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)) {
185 setUploadState(job.image.id, {
186 status: "failed",
187 environmentId: job.environmentId,
188 reason: "This connection cannot upload attachments.",
189 ...(job.previous ? { previous: job.previous } : {}),
190 });
191 return;
192 }
193 if (job.persistedAttachmentId) {
194 const verification = await verifyPersistedAttachmentUpload({
195 registry: appAtomRegistry,
196 createAssetUrl: assetEnvironment.createUrl,
197 environmentId: job.environmentId,
198 attachmentId: job.persistedAttachmentId,
199 });
200 if (job.cancelled) {
201 return;
202 }
203 if (verification.status === "verified") {
204 setUploadState(job.image.id, {
205 status: "ready",
206 environmentId: job.environmentId,
207 attachmentId: job.persistedAttachmentId,
208 });
209 stampDraftFileUpload(job, job.persistedAttachmentId);
210 return;
211 }
212 if (verification.status === "missing" && !job.image.file && job.image.type === "file") {
213 const draftTarget = resolveCurrentFileDraftTarget(job);
214 if (
215 draftTarget !== undefined &&
216 useComposerDraftStore
217 .getState()
218 .markFileUploadMissing(
219 draftTarget,
220 job.image.id,
221 job.environmentId,
222 job.persistedAttachmentId,
223 )
224 ) {
225 clearUploadState(job.image.id);
226 return;
227 }
228 }
229 if (verification.status === "failed" || !job.image.file) {
230 // No `attachmentId` here: a failed state's id marks a pending upload
231 // this queue minted, which retry and release then delete. The persisted
232 // id is the only server copy of a hydrated file, so a transient
233 // verification failure must leave it in place for the next retry.
234 setUploadState(job.image.id, {
235 status: "failed",
236 environmentId: job.environmentId,
237 reason:
238 verification.status === "missing"
239 ? "Uploaded file expired. Remove it and attach it again."
240 : "Uploaded file could not be verified. Retry when the server reconnects.",
241 ...(job.previous ? { previous: job.previous } : {}),
242 });
243 return;
244 }
245 }
246
247 const mimeType =
248 job.image.type === "file"
249 ? job.image.mimeType.toLowerCase()
250 : PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES.find(
251 (supportedMimeType) => supportedMimeType === job.image.mimeType.toLowerCase(),
252 );
253 if (!mimeType) {
254 setUploadState(job.image.id, {
255 status: "failed",
256 environmentId: job.environmentId,
257 reason: "Unsupported image type",
258 ...(job.previous ? { previous: job.previous } : {}),
259 });
260 return;
261 }
262 const file = job.image.file;
263 if (!file) {
264 setUploadState(job.image.id, {
265 status: "failed",
266 environmentId: job.environmentId,
267 reason: "Original file is no longer available",
268 ...(job.previous ? { previous: job.previous } : {}),
269 });
270 return;
271 }
272
273 let lastStep = -1;
274 if (!readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)) {
275 setUploadState(job.image.id, {
276 status: "failed",
277 environmentId: job.environmentId,
278 reason: "This connection cannot upload attachments.",
279 ...(job.previous ? { previous: job.previous } : {}),
280 });
281 return;
282 }
283 const result = await runAttachmentUploadCycle({
284 registry: appAtomRegistry,
285 createUploadUrl: attachmentEnvironment.createUploadUrl,
286 remove: attachmentEnvironment.remove,
287 environmentId: job.environmentId,
288 upload: {
289 ...(job.image.type === "file" ? { type: "file" as const } : {}),
290 name: job.image.name,
291 mimeType,
292 sizeBytes: file.size,
293 },
294 resolveUploadUrl: (relativeUrl) => {
295 if (
296 job.cancelled ||
297 !readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)
298 ) {
299 return null;
300 }
301 const connection = readPreparedConnection(job.environmentId);
302 return connection ? resolveAssetUrl(connection.httpBaseUrl, relativeUrl) : null;
303 },
304 transport: (url) =>
305 uploadBytes({
306 url,
307 file,
308 mimeType,
309 onProgress: (progress) => {
310 const step = Math.floor(progress * 20);
311 if (step === lastStep || job.cancelled) {
312 return;
313 }
314 lastStep = step;
315 setUploadState(job.image.id, {
316 status: "uploading",
317 environmentId: job.environmentId,
318 progress,
319 ...(job.previous ? { previous: job.previous } : {}),
320 });
321 },
322 }),
323 onMinted: (attachmentId) => {
324 if (job.cancelled && readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)) {
325 return "cancel";
326 }
327 job.attachmentId = attachmentId;
328 return "continue";
329 },
330 onTransferStart: (abort) => {
331 job.abort = abort;
332 },
333 });
334 job.abort = null;
335 if (result.status === "cancelled" || job.cancelled) {
336 return;
337 }
338 if (result.status === "uploaded") {
339 setUploadState(job.image.id, {
340 status: "ready",
341 environmentId: job.environmentId,
342 attachmentId: result.attachmentId,
343 });
344 stampDraftFileUpload(job, result.attachmentId);
345 if (job.previous) {
346 deletePendingUpload(job.previous.environmentId, job.previous.attachmentId);
347 }
348 return;
349 }
350 setUploadState(job.image.id, {
351 status: "failed",
352 environmentId: job.environmentId,
353 reason:
354 result.step === "mint"
355 ? "Upload could not start"
356 : result.step === "resolve-url"
357 ? readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)
358 ? "Not connected"
359 : "This connection cannot upload attachments."
360 : result.error instanceof Error
361 ? result.error.message
362 : "Upload failed",
363 ...(result.attachmentId ? { attachmentId: result.attachmentId } : {}),
364 ...(job.previous ? { previous: job.previous } : {}),
365 });
366}
367
368function pumpUploads(): void {
369 for (let index = 0; index < queue.length;) {
370 const job = queue[index]!;
371 const active = activeUploadsByEnvironment.get(job.environmentId) ?? 0;
372 if (active >= MAX_UPLOADS_PER_ENVIRONMENT) {
373 index += 1;
374 continue;
375 }
376
377 queue.splice(index, 1);
378 if (job.cancelled) {
379 continue;
380 }
381 activeUploadsByEnvironment.set(job.environmentId, active + 1);
382 void runUpload(job)
383 .catch(() => {
384 if (!job.cancelled) {
385 setUploadState(job.image.id, {
386 status: "failed",
387 environmentId: job.environmentId,
388 reason: "Upload failed",
389 ...(job.previous ? { previous: job.previous } : {}),
390 });
391 }
392 })
393 .finally(() => {
394 if (
395 jobsByImageId.get(job.image.id) === job &&
396 readAttachmentUpload(job.image.id)?.status !== "failed"
397 ) {
398 job.stopWatchingConnection();
399 jobsByImageId.delete(job.image.id);
400 }
401 const remaining = (activeUploadsByEnvironment.get(job.environmentId) ?? 1) - 1;
402 if (remaining > 0) {
403 activeUploadsByEnvironment.set(job.environmentId, remaining);
404 } else {
405 activeUploadsByEnvironment.delete(job.environmentId);
406 }
407 job.resolveSettled();
408 pumpUploads();
409 });
410 }
411}
412
413export function startAttachmentUpload(input: {
414 readonly environmentId: EnvironmentId;
415 readonly image: ComposerImageAttachment | ComposerFileAttachment;
416 /** Draft that owns the file; lets a background completion persist its ids. */
417 readonly draftTarget?: ComposerThreadTarget;
418}): void {
419 if (!readEnvironmentScope(input.environmentId, AuthOrchestrationOperateScope)) return;
420 const existingJob = jobsByImageId.get(input.image.id);
421 if (existingJob?.environmentId === input.environmentId) {
422 return;
423 }
424
425 const existing = readAttachmentUpload(input.image.id);
426 if (existing?.status === "ready" && existing.environmentId === input.environmentId) {
427 return;
428 }
429 if (existing?.status === "failed" && existing.environmentId === input.environmentId) {
430 return;
431 }
432 if (
433 existing &&
434 "previous" in existing &&
435 existing.previous?.environmentId === input.environmentId
436 ) {
437 cancelAttachmentUpload(input.image.id);
438 if (existing.status === "failed" && existing.attachmentId) {
439 deletePendingUpload(existing.environmentId, existing.attachmentId);
440 }
441 setUploadState(input.image.id, existing.previous);
442 return;
443 }
444
445 if (existingJob) {
446 cancelAttachmentUpload(input.image.id);
447 }
448 const previous = existing?.status === "ready" ? existing : existing?.previous;
449 let resolveSettled: () => void = () => {};
450 const settled = new Promise<void>((resolve) => {
451 resolveSettled = resolve;
452 });
453 const job: UploadJob = {
454 image: input.image,
455 environmentId: input.environmentId,
456 ...(input.draftTarget !== undefined ? { draftTarget: input.draftTarget } : {}),
457 ...(previous ? { previous } : {}),
458 ...(input.image.type === "file" &&
459 input.image.uploadEnvironmentId === input.environmentId &&
460 input.image.uploadedAttachmentId
461 ? { persistedAttachmentId: input.image.uploadedAttachmentId }
462 : {}),
463 settled,
464 resolveSettled,
465 attachmentId: null,
466 cancelled: false,
467 abort: null,
468 stopWatchingConnection: () => {},
469 };
470
471 jobsByImageId.set(input.image.id, job);
472 const connectionAtom = environmentCatalog.stateAtom(job.environmentId);
473 const isConnected = () =>
474 Option.exists(
475 AsyncResult.value(appAtomRegistry.get(connectionAtom)),
476 (state) => state.phase === "connected",
477 );
478 let wasConnected = isConnected();
479 job.stopWatchingConnection = appAtomRegistry.subscribe(connectionAtom, () => {
480 const connected = isConnected();
481 const reconnected = connected && !wasConnected;
482 wasConnected = connected;
483 if (!reconnected) return;
484 // The HTTP failure can arrive after the socket has already reconnected.
485 // Wait for that attempt, then retry only if this job still owns the file.
486 void job.settled.then(() => {
487 if (
488 jobsByImageId.get(job.image.id) === job &&
489 readAttachmentUpload(job.image.id)?.status === "failed" &&
490 isConnected()
491 ) {
492 retryAttachmentUpload(input);
493 }
494 });
495 });
496 queue.push(job);
497 setUploadState(input.image.id, {
498 status: "uploading",
499 environmentId: input.environmentId,
500 progress: 0,
501 ...(previous ? { previous } : {}),
502 });
503 pumpUploads();
504}
505
506/**
507 * Stops the job and deletes only the pending upload it minted itself. A
508 * persisted draft upload survives cancellation (an environment switch cancels
509 * the old job, and the draft still references that server copy).
510 */
511function cancelAttachmentUpload(imageId: string): void {
512 const job = jobsByImageId.get(imageId);
513 if (!job) {
514 return;
515 }
516 job.cancelled = true;
517 job.stopWatchingConnection();
518 jobsByImageId.delete(imageId);
519 const queuedIndex = queue.indexOf(job);
520 if (queuedIndex !== -1) {
521 queue.splice(queuedIndex, 1);
522 }
523 job.abort?.();
524 if (job.attachmentId) {
525 deletePendingUpload(job.environmentId, job.attachmentId);
526 }
527 job.resolveSettled();
528}
529
530export function releaseAttachmentUpload(imageId: string): void {
531 const upload = readAttachmentUpload(imageId);
532 cancelAttachmentUpload(imageId);
533 if (upload?.status === "ready") {
534 deletePendingUpload(upload.environmentId, upload.attachmentId);
535 } else if (upload) {
536 if (upload.status === "failed" && upload.attachmentId) {
537 deletePendingUpload(upload.environmentId, upload.attachmentId);
538 }
539 if (upload.previous) {
540 deletePendingUpload(upload.previous.environmentId, upload.previous.attachmentId);
541 }
542 }
543 clearUploadState(imageId);
544}
545
546export function releasePersistedAttachmentUpload(input: {
547 readonly id: string;
548 readonly environmentId: EnvironmentId;
549 readonly attachmentId: string;
550}): void {
551 const upload = readAttachmentUpload(input.id);
552 if (
553 upload?.status === "ready" &&
554 upload.environmentId === input.environmentId &&
555 upload.attachmentId === input.attachmentId
556 ) {
557 releaseAttachmentUpload(input.id);
558 return;
559 }
560 const job = jobsByImageId.get(input.id);
561 if (
562 job?.environmentId === input.environmentId &&
563 job.persistedAttachmentId === input.attachmentId
564 ) {
565 // Tears down the in-flight verification or re-upload. The queue only
566 // deletes ids it minted, so the persisted id still needs the delete below.
567 releaseAttachmentUpload(input.id);
568 }
569 deletePendingUpload(input.environmentId, input.attachmentId);
570}
571
572export function retryAttachmentUpload(input: {
573 readonly environmentId: EnvironmentId;
574 readonly image: ComposerImageAttachment | ComposerFileAttachment;
575 readonly draftTarget?: ComposerThreadTarget;
576}): void {
577 if (!readEnvironmentScope(input.environmentId, AuthOrchestrationOperateScope)) return;
578 const previous = readAttachmentUpload(input.image.id);
579 cancelAttachmentUpload(input.image.id);
580 // A failed state's `attachmentId` is always one this queue minted, so this
581 // never deletes a persisted draft upload. Retrying a hydrated file whose
582 // verification failed leaves the server copy alone and verifies it again.
583 if (previous?.status === "failed" && previous.attachmentId) {
584 deletePendingUpload(previous.environmentId, previous.attachmentId);
585 }
586 if (previous && "previous" in previous && previous.previous) {
587 setUploadState(input.image.id, previous.previous);
588 } else {
589 clearUploadState(input.image.id);
590 }
591 startAttachmentUpload(input);
592}
593
594/**
595 * Checks that a stashed upload still exists on the server. Pending uploads
596 * are swept after 24 hours, so a stash restore asks first instead of handing
597 * the composer a dead reference.
598 */
599export function verifyStashedAttachmentUpload(input: {
600 readonly environmentId: EnvironmentId;
601 readonly attachmentId: string;
602}): Promise<PersistedAttachmentVerification> {
603 return verifyPersistedAttachmentUpload({
604 registry: appAtomRegistry,
605 createAssetUrl: assetEnvironment.createUrl,
606 environmentId: input.environmentId,
607 attachmentId: input.attachmentId,
608 });
609}
610
611export async function awaitAttachmentUploads(imageIds: ReadonlyArray<string>): Promise<void> {
612 await Promise.all(imageIds.map((imageId) => jobsByImageId.get(imageId)?.settled));
613}
614
615export function getUploadedAttachments(input: {
616 readonly environmentId: EnvironmentId;
617 readonly images: ReadonlyArray<ComposerImageAttachment | ComposerFileAttachment>;
618}): ChatAttachment[] | null {
619 const attachments: ChatAttachment[] = [];
620 for (const image of input.images) {
621 const upload = readAttachmentUpload(image.id);
622 if (upload?.status !== "ready" || upload.environmentId !== input.environmentId) {
623 return null;
624 }
625 attachments.push({
626 type: image.type,
627 id: upload.attachmentId,
628 name: image.name,
629 mimeType: image.mimeType,
630 sizeBytes: image.sizeBytes,
631 ...(image.source ? { source: image.source } : {}),
632 });
633 }
634 return attachments;
635}
636
637/**
638 * The one owner for discarding a draft attachment's server-side upload. The
639 * queue-keyed release only sees in-memory state, so after a reload it finds
640 * nothing and the pending upload leaks. When the draft carries a persisted
641 * `uploadedAttachmentId` (which survives reloads), route through the persisted
642 * release; it still prefers the queue path when the queue owns that same
643 * attachment. Every draft discard path must funnel through here.
644 */
645export function releaseDraftAttachment(
646 attachment: ComposerImageAttachment | ComposerFileAttachment,
647): void {
648 if (
649 attachment.type === "file" &&
650 attachment.uploadedAttachmentId !== undefined &&
651 attachment.uploadEnvironmentId !== undefined
652 ) {
653 releasePersistedAttachmentUpload({
654 id: attachment.id,
655 environmentId: attachment.uploadEnvironmentId,
656 attachmentId: attachment.uploadedAttachmentId,
657 });
658 // A failed re-upload after verification can hold a newer minted
659 // attachment under the queue key. Release whatever is left so neither
660 // copy stays behind. (The pending delete is idempotent server-side.)
661 if (jobsByImageId.has(attachment.id) || readAttachmentUpload(attachment.id)) {
662 releaseAttachmentUpload(attachment.id);
663 }
664 return;
665 }
666 releaseAttachmentUpload(attachment.id);
667}
668
669export function releaseDraftAttachments(
670 attachments: ReadonlyArray<ComposerImageAttachment | ComposerFileAttachment>,
671): void {
672 for (const attachment of attachments) {
673 releaseDraftAttachment(attachment);
674 }
675}