Open History to see the last commit
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 /** Original owner; completion resolves the current owner because uploads can outlive draft moves. */
48 readonly draftTarget?: ComposerThreadTarget;
49 readonly previous?: ReadyAttachmentUpload;
50 /** Draft-owned; delete only through releasePersistedAttachmentUpload, never queue cancel/retry. */
51 readonly persistedAttachmentId?: string;
52 readonly settled: Promise<void>;
53 resolveSettled: () => void;
54 attachmentId: string | null;
55 cancelled: boolean;
56 abort: (() => void) | null;
57 stopWatchingConnection: () => void;
58}
59
60type DraftFileOwner = Pick<UploadJob, "image" | "environmentId" | "draftTarget">;
61
62const jobsByImageId = new Map<string, UploadJob>();
63const queue: UploadJob[] = [];
64const activeUploadsByEnvironment = new Map<EnvironmentId, number>();
65
66function setUploadState(imageId: string, upload: AttachmentUploadState): void {
67 useAttachmentUploadStore.setState((state) => ({
68 uploadsByImageId: { ...state.uploadsByImageId, [imageId]: upload },
69 }));
70}
71
72function clearUploadState(imageId: string): void {
73 useAttachmentUploadStore.setState((state) => {
74 if (!(imageId in state.uploadsByImageId)) {
75 return state;
76 }
77 const uploadsByImageId = { ...state.uploadsByImageId };
78 delete uploadsByImageId[imageId];
79 return { uploadsByImageId };
80 });
81}
82
83export function readAttachmentUpload(imageId: string): AttachmentUploadState | undefined {
84 return useAttachmentUploadStore.getState().uploadsByImageId[imageId];
85}
86
87function resolveCurrentFileDraftTarget(job: DraftFileOwner): ComposerThreadTarget | undefined {
88 if (job.draftTarget === undefined || job.image.type !== "file") {
89 return undefined;
90 }
91 const store = useComposerDraftStore.getState();
92 for (const [key, draft] of Object.entries(store.draftsByThreadKey)) {
93 if (!draft.files.some((file) => file.id === job.image.id)) {
94 continue;
95 }
96 const draftSession = store.draftThreadsByThreadKey[key];
97 if (draftSession !== undefined) {
98 if (draftSession.environmentId === job.environmentId) {
99 return DraftId.make(key);
100 }
101 continue;
102 }
103 if (typeof job.draftTarget === "string" && job.draftTarget === key) {
104 return DraftId.make(key);
105 }
106 const threadRef = parseScopedThreadKey(key);
107 if (threadRef?.environmentId === job.environmentId) {
108 return threadRef;
109 }
110 }
111 return undefined;
112}
113
114/** Persist background completions even when no composer is mounted. */
115function stampDraftFileUpload(job: DraftFileOwner, attachmentId: string): void {
116 const draftTarget = resolveCurrentFileDraftTarget(job);
117 if (draftTarget === undefined) {
118 return;
119 }
120 useComposerDraftStore
121 .getState()
122 .setFileUpload(draftTarget, job.image.id, job.environmentId, attachmentId);
123}
124
125function deletePendingUpload(environmentId: EnvironmentId, attachmentId: string): void {
126 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return;
127 deletePendingAttachmentUpload({
128 registry: appAtomRegistry,
129 remove: attachmentEnvironment.remove,
130 environmentId,
131 attachmentId,
132 });
133}
134
135function uploadBytes(input: {
136 readonly url: string;
137 readonly file: File;
138 readonly mimeType: string;
139 readonly onProgress: (progress: number) => void;
140}): { readonly done: Promise<void>; readonly abort: () => void } {
141 const xhr = new XMLHttpRequest();
142 const done = new Promise<void>((resolve, reject) => {
143 xhr.open("POST", input.url, true);
144 xhr.timeout = UPLOAD_TIMEOUT_MS;
145 xhr.setRequestHeader("Content-Type", input.mimeType);
146 xhr.upload.addEventListener("progress", (event) => {
147 if (event.lengthComputable && event.total > 0) {
148 input.onProgress(event.loaded / event.total);
149 }
150 });
151 xhr.addEventListener("load", () => {
152 if (xhr.status >= 200 && xhr.status < 300) {
153 resolve();
154 } else {
155 reject(new Error(`Upload rejected (${xhr.status})`));
156 }
157 });
158 xhr.addEventListener("error", () => reject(new Error("Upload failed")));
159 xhr.addEventListener("timeout", () => reject(new Error("Upload timed out")));
160 xhr.addEventListener("abort", () => reject(new Error("Upload cancelled")));
161 xhr.send(input.file);
162 });
163
164 return { done, abort: () => xhr.abort() };
165}
166
167async function runUpload(job: UploadJob): Promise<void> {
168 if (!readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)) {
169 setUploadState(job.image.id, {
170 status: "failed",
171 environmentId: job.environmentId,
172 reason: "This connection cannot upload attachments.",
173 ...(job.previous ? { previous: job.previous } : {}),
174 });
175 return;
176 }
177 if (job.persistedAttachmentId) {
178 const verification = await verifyPersistedAttachmentUpload({
179 registry: appAtomRegistry,
180 createAssetUrl: assetEnvironment.createUrl,
181 environmentId: job.environmentId,
182 attachmentId: job.persistedAttachmentId,
183 });
184 if (job.cancelled) {
185 return;
186 }
187 if (verification.status === "verified") {
188 setUploadState(job.image.id, {
189 status: "ready",
190 environmentId: job.environmentId,
191 attachmentId: job.persistedAttachmentId,
192 });
193 stampDraftFileUpload(job, job.persistedAttachmentId);
194 return;
195 }
196 if (verification.status === "missing" && !job.image.file && job.image.type === "file") {
197 const draftTarget = resolveCurrentFileDraftTarget(job);
198 if (
199 draftTarget !== undefined &&
200 useComposerDraftStore
201 .getState()
202 .markFileUploadMissing(
203 draftTarget,
204 job.image.id,
205 job.environmentId,
206 job.persistedAttachmentId,
207 )
208 ) {
209 clearUploadState(job.image.id);
210 return;
211 }
212 }
213 if (verification.status === "failed" || !job.image.file) {
214 // Failed-state ids get deleted on retry/release; exclude persisted ids after transient verification failure.
215 setUploadState(job.image.id, {
216 status: "failed",
217 environmentId: job.environmentId,
218 reason:
219 verification.status === "missing"
220 ? "Uploaded file expired. Remove it and attach it again."
221 : "Uploaded file could not be verified. Retry when the server reconnects.",
222 ...(job.previous ? { previous: job.previous } : {}),
223 });
224 return;
225 }
226 }
227
228 const mimeType =
229 job.image.type === "file"
230 ? job.image.mimeType.toLowerCase()
231 : PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES.find(
232 (supportedMimeType) => supportedMimeType === job.image.mimeType.toLowerCase(),
233 );
234 if (!mimeType) {
235 setUploadState(job.image.id, {
236 status: "failed",
237 environmentId: job.environmentId,
238 reason: "Unsupported image type",
239 ...(job.previous ? { previous: job.previous } : {}),
240 });
241 return;
242 }
243 const file = job.image.file;
244 if (!file) {
245 setUploadState(job.image.id, {
246 status: "failed",
247 environmentId: job.environmentId,
248 reason: "Original file is no longer available",
249 ...(job.previous ? { previous: job.previous } : {}),
250 });
251 return;
252 }
253
254 let lastStep = -1;
255 if (!readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)) {
256 setUploadState(job.image.id, {
257 status: "failed",
258 environmentId: job.environmentId,
259 reason: "This connection cannot upload attachments.",
260 ...(job.previous ? { previous: job.previous } : {}),
261 });
262 return;
263 }
264 const result = await runAttachmentUploadCycle({
265 registry: appAtomRegistry,
266 createUploadUrl: attachmentEnvironment.createUploadUrl,
267 remove: attachmentEnvironment.remove,
268 environmentId: job.environmentId,
269 upload: {
270 ...(job.image.type === "file" ? { type: "file" as const } : {}),
271 name: job.image.name,
272 mimeType,
273 sizeBytes: file.size,
274 },
275 resolveUploadUrl: (relativeUrl) => {
276 if (
277 job.cancelled ||
278 !readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)
279 ) {
280 return null;
281 }
282 const connection = readPreparedConnection(job.environmentId);
283 return connection ? resolveAssetUrl(connection.httpBaseUrl, relativeUrl) : null;
284 },
285 transport: (url) =>
286 uploadBytes({
287 url,
288 file,
289 mimeType,
290 onProgress: (progress) => {
291 const step = Math.floor(progress * 20);
292 if (step === lastStep || job.cancelled) {
293 return;
294 }
295 lastStep = step;
296 setUploadState(job.image.id, {
297 status: "uploading",
298 environmentId: job.environmentId,
299 progress,
300 ...(job.previous ? { previous: job.previous } : {}),
301 });
302 },
303 }),
304 onMinted: (attachmentId) => {
305 if (job.cancelled && readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)) {
306 return "cancel";
307 }
308 job.attachmentId = attachmentId;
309 return "continue";
310 },
311 onTransferStart: (abort) => {
312 job.abort = abort;
313 },
314 });
315 job.abort = null;
316 if (result.status === "cancelled" || job.cancelled) {
317 return;
318 }
319 if (result.status === "uploaded") {
320 setUploadState(job.image.id, {
321 status: "ready",
322 environmentId: job.environmentId,
323 attachmentId: result.attachmentId,
324 });
325 stampDraftFileUpload(job, result.attachmentId);
326 if (job.previous) {
327 deletePendingUpload(job.previous.environmentId, job.previous.attachmentId);
328 }
329 return;
330 }
331 setUploadState(job.image.id, {
332 status: "failed",
333 environmentId: job.environmentId,
334 reason:
335 result.step === "mint"
336 ? "Upload could not start"
337 : result.step === "resolve-url"
338 ? readEnvironmentScope(job.environmentId, AuthOrchestrationOperateScope)
339 ? "Not connected"
340 : "This connection cannot upload attachments."
341 : result.error instanceof Error
342 ? result.error.message
343 : "Upload failed",
344 ...(result.attachmentId ? { attachmentId: result.attachmentId } : {}),
345 ...(job.previous ? { previous: job.previous } : {}),
346 });
347}
348
349function pumpUploads(): void {
350 for (let index = 0; index < queue.length;) {
351 const job = queue[index]!;
352 const active = activeUploadsByEnvironment.get(job.environmentId) ?? 0;
353 if (active >= MAX_UPLOADS_PER_ENVIRONMENT) {
354 index += 1;
355 continue;
356 }
357
358 queue.splice(index, 1);
359 if (job.cancelled) {
360 continue;
361 }
362 activeUploadsByEnvironment.set(job.environmentId, active + 1);
363 void runUpload(job)
364 .catch(() => {
365 if (!job.cancelled) {
366 setUploadState(job.image.id, {
367 status: "failed",
368 environmentId: job.environmentId,
369 reason: "Upload failed",
370 ...(job.previous ? { previous: job.previous } : {}),
371 });
372 }
373 })
374 .finally(() => {
375 if (
376 jobsByImageId.get(job.image.id) === job &&
377 readAttachmentUpload(job.image.id)?.status !== "failed"
378 ) {
379 job.stopWatchingConnection();
380 jobsByImageId.delete(job.image.id);
381 }
382 const remaining = (activeUploadsByEnvironment.get(job.environmentId) ?? 1) - 1;
383 if (remaining > 0) {
384 activeUploadsByEnvironment.set(job.environmentId, remaining);
385 } else {
386 activeUploadsByEnvironment.delete(job.environmentId);
387 }
388 job.resolveSettled();
389 pumpUploads();
390 });
391 }
392}
393
394export function startAttachmentUpload(input: {
395 readonly environmentId: EnvironmentId;
396 readonly image: ComposerImageAttachment | ComposerFileAttachment;
397 readonly draftTarget?: ComposerThreadTarget;
398}): void {
399 if (!readEnvironmentScope(input.environmentId, AuthOrchestrationOperateScope)) return;
400 const existingJob = jobsByImageId.get(input.image.id);
401 if (existingJob?.environmentId === input.environmentId) {
402 return;
403 }
404
405 const existing = readAttachmentUpload(input.image.id);
406 if (existing?.status === "ready" && existing.environmentId === input.environmentId) {
407 // A file restored into a draft after its upload finished (stash, failed send) missed the
408 // completion stamp; stamping here keeps it from re-uploading after a reload.
409 stampDraftFileUpload(input, existing.attachmentId);
410 return;
411 }
412 if (existing?.status === "failed" && existing.environmentId === input.environmentId) {
413 return;
414 }
415 if (
416 existing &&
417 "previous" in existing &&
418 existing.previous?.environmentId === input.environmentId
419 ) {
420 cancelAttachmentUpload(input.image.id);
421 if (existing.status === "failed" && existing.attachmentId) {
422 deletePendingUpload(existing.environmentId, existing.attachmentId);
423 }
424 setUploadState(input.image.id, existing.previous);
425 return;
426 }
427
428 if (existingJob) {
429 cancelAttachmentUpload(input.image.id);
430 }
431 const previous = existing?.status === "ready" ? existing : existing?.previous;
432 let resolveSettled: () => void = () => {};
433 const settled = new Promise<void>((resolve) => {
434 resolveSettled = resolve;
435 });
436 const job: UploadJob = {
437 image: input.image,
438 environmentId: input.environmentId,
439 ...(input.draftTarget !== undefined ? { draftTarget: input.draftTarget } : {}),
440 ...(previous ? { previous } : {}),
441 ...(input.image.type === "file" &&
442 input.image.uploadEnvironmentId === input.environmentId &&
443 input.image.uploadedAttachmentId
444 ? { persistedAttachmentId: input.image.uploadedAttachmentId }
445 : {}),
446 settled,
447 resolveSettled,
448 attachmentId: null,
449 cancelled: false,
450 abort: null,
451 stopWatchingConnection: () => {},
452 };
453
454 jobsByImageId.set(input.image.id, job);
455 const connectionAtom = environmentCatalog.stateAtom(job.environmentId);
456 const isConnected = () =>
457 Option.exists(
458 AsyncResult.value(appAtomRegistry.get(connectionAtom)),
459 (state) => state.phase === "connected",
460 );
461 let wasConnected = isConnected();
462 job.stopWatchingConnection = appAtomRegistry.subscribe(connectionAtom, () => {
463 const connected = isConnected();
464 const reconnected = connected && !wasConnected;
465 wasConnected = connected;
466 if (!reconnected) return;
467 // HTTP failures can follow reconnection; await the attempt and retry only if this job still owns the file.
468 void job.settled.then(() => {
469 if (
470 jobsByImageId.get(job.image.id) === job &&
471 readAttachmentUpload(job.image.id)?.status === "failed" &&
472 isConnected()
473 ) {
474 retryAttachmentUpload(input);
475 }
476 });
477 });
478 queue.push(job);
479 setUploadState(input.image.id, {
480 status: "uploading",
481 environmentId: input.environmentId,
482 progress: 0,
483 ...(previous ? { previous } : {}),
484 });
485 pumpUploads();
486}
487
488function cancelAttachmentUpload(imageId: string): void {
489 const job = jobsByImageId.get(imageId);
490 if (!job) {
491 return;
492 }
493 job.cancelled = true;
494 job.stopWatchingConnection();
495 jobsByImageId.delete(imageId);
496 const queuedIndex = queue.indexOf(job);
497 if (queuedIndex !== -1) {
498 queue.splice(queuedIndex, 1);
499 }
500 job.abort?.();
501 if (job.attachmentId) {
502 deletePendingUpload(job.environmentId, job.attachmentId);
503 }
504 job.resolveSettled();
505}
506
507export function releaseAttachmentUpload(imageId: string): void {
508 const upload = readAttachmentUpload(imageId);
509 cancelAttachmentUpload(imageId);
510 if (upload?.status === "ready") {
511 deletePendingUpload(upload.environmentId, upload.attachmentId);
512 } else if (upload) {
513 if (upload.status === "failed" && upload.attachmentId) {
514 deletePendingUpload(upload.environmentId, upload.attachmentId);
515 }
516 if (upload.previous) {
517 deletePendingUpload(upload.previous.environmentId, upload.previous.attachmentId);
518 }
519 }
520 clearUploadState(imageId);
521}
522
523export function releasePersistedAttachmentUpload(input: {
524 readonly id: string;
525 readonly environmentId: EnvironmentId;
526 readonly attachmentId: string;
527}): void {
528 const upload = readAttachmentUpload(input.id);
529 if (
530 upload?.status === "ready" &&
531 upload.environmentId === input.environmentId &&
532 upload.attachmentId === input.attachmentId
533 ) {
534 releaseAttachmentUpload(input.id);
535 return;
536 }
537 const job = jobsByImageId.get(input.id);
538 if (
539 job?.environmentId === input.environmentId &&
540 job.persistedAttachmentId === input.attachmentId
541 ) {
542 releaseAttachmentUpload(input.id);
543 }
544 deletePendingUpload(input.environmentId, input.attachmentId);
545}
546
547export function retryAttachmentUpload(input: {
548 readonly environmentId: EnvironmentId;
549 readonly image: ComposerImageAttachment | ComposerFileAttachment;
550 readonly draftTarget?: ComposerThreadTarget;
551}): void {
552 if (!readEnvironmentScope(input.environmentId, AuthOrchestrationOperateScope)) return;
553 const previous = readAttachmentUpload(input.image.id);
554 cancelAttachmentUpload(input.image.id);
555 if (previous?.status === "failed" && previous.attachmentId) {
556 deletePendingUpload(previous.environmentId, previous.attachmentId);
557 }
558 if (previous && "previous" in previous && previous.previous) {
559 setUploadState(input.image.id, previous.previous);
560 } else {
561 clearUploadState(input.image.id);
562 }
563 startAttachmentUpload(input);
564}
565
566/** Verifies uploads before stash restore; pending uploads expire after 24 hours. */
567export function verifyStashedAttachmentUpload(input: {
568 readonly environmentId: EnvironmentId;
569 readonly attachmentId: string;
570}): Promise<PersistedAttachmentVerification> {
571 return verifyPersistedAttachmentUpload({
572 registry: appAtomRegistry,
573 createAssetUrl: assetEnvironment.createUrl,
574 environmentId: input.environmentId,
575 attachmentId: input.attachmentId,
576 });
577}
578
579export async function awaitAttachmentUploads(imageIds: ReadonlyArray<string>): Promise<void> {
580 await Promise.all(imageIds.map((imageId) => jobsByImageId.get(imageId)?.settled));
581}
582
583export function getUploadedAttachments(input: {
584 readonly environmentId: EnvironmentId;
585 readonly images: ReadonlyArray<ComposerImageAttachment | ComposerFileAttachment>;
586}): ChatAttachment[] | null {
587 const attachments: ChatAttachment[] = [];
588 for (const image of input.images) {
589 const upload = readAttachmentUpload(image.id);
590 if (upload?.status !== "ready" || upload.environmentId !== input.environmentId) {
591 return null;
592 }
593 attachments.push({
594 type: image.type,
595 id: upload.attachmentId,
596 name: image.name,
597 mimeType: image.mimeType,
598 sizeBytes: image.sizeBytes,
599 ...(image.source ? { source: image.source } : {}),
600 });
601 }
602 return attachments;
603}
604
605/** In-memory queue releases miss uploads after reload; every draft discard must use this path. */
606export function releaseDraftAttachment(
607 attachment: ComposerImageAttachment | ComposerFileAttachment,
608): void {
609 if (
610 attachment.type === "file" &&
611 attachment.uploadedAttachmentId !== undefined &&
612 attachment.uploadEnvironmentId !== undefined
613 ) {
614 releasePersistedAttachmentUpload({
615 id: attachment.id,
616 environmentId: attachment.uploadEnvironmentId,
617 attachmentId: attachment.uploadedAttachmentId,
618 });
619 if (jobsByImageId.has(attachment.id) || readAttachmentUpload(attachment.id)) {
620 releaseAttachmentUpload(attachment.id);
621 }
622 return;
623 }
624 releaseAttachmentUpload(attachment.id);
625}
626
627export function releaseDraftAttachments(
628 attachments: ReadonlyArray<ComposerImageAttachment | ComposerFileAttachment>,
629): void {
630 for (const attachment of attachments) {
631 releaseDraftAttachment(attachment);
632 }
633}