1import { useAtomValue } from "@effect/atom-react";
2import { AuthOrchestrationOperateScope, type EnvironmentId } from "@t3tools/contracts";
3import { Atom } from "effect/reactivity";
4import { useEffect, useRef } from "react";
5
6import { prepareTurnAttachments } from "../lib/attachmentUpload";
7import { retainComposerAttachmentFileForPreview } from "../lib/composerAttachmentPreviewRetention";
8import { isFileBackedComposerAttachment } from "../lib/composerImages";
9import {
10 composerAttachmentUploadKey,
11 composerDraftEnvironmentId,
12 canUploadComposerAttachment,
13 createComposerAttachmentUploadQueue,
14 type ComposerAttachmentUploadState,
15} from "../lib/composerAttachmentUploadQueue";
16import { appAtomRegistry } from "./atom-registry";
17import { useServerConfigs } from "./entities";
18import { flattenQueuedThreadMessages, threadOutboxManager } from "./thread-outbox";
19import { useThreadOutboxMessages } from "./use-thread-outbox";
20import {
21 composerDraftsAtom,
22 ensureComposerDraftsLoaded,
23 flushComposerDrafts,
24 setComposerDraftAttachmentUpload,
25} from "./use-composer-drafts";
26import { useRemoteConnectionStatus } from "./use-remote-environment-registry";
27import { readEnvironmentScope, useEnvironmentsWithScope } from "./session";
28
29export {
30 composerAttachmentUploadBlockReason,
31 composerAttachmentsStillUploading,
32} from "../lib/composerAttachmentUploadQueue";
33
34export const composerAttachmentUploadsAtom = Atom.make<
35 Readonly<Record<string, ComposerAttachmentUploadState>>
36>({}).pipe(Atom.keepAlive);
37const uploadStateAtom = Atom.family((key: string) =>
38 Atom.map(composerAttachmentUploadsAtom, (states) => states[key]),
39);
40let uploadQueue: ReturnType<typeof createComposerAttachmentUploadQueue> | null = null;
41
42export function useComposerAttachmentUploadState(
43 environmentId: EnvironmentId | undefined,
44 attachmentId: string,
45) {
46 return useAtomValue(
47 uploadStateAtom(environmentId ? composerAttachmentUploadKey(environmentId, attachmentId) : ""),
48 );
49}
50
51export function retryComposerAttachmentUpload(environmentId: EnvironmentId, attachmentId: string) {
52 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return;
53 uploadQueue?.retry(environmentId, attachmentId);
54}
55
56/** Runs outside mounted composers so a transfer can finish after navigation. */
57export function useComposerAttachmentUploadWorker() {
58 const drafts = useAtomValue(composerDraftsAtom);
59 const queuedMessages = useThreadOutboxMessages();
60 const serverConfigs = useServerConfigs();
61 const { connectedEnvironments } = useRemoteConnectionStatus();
62 const operableEnvironments = useEnvironmentsWithScope(
63 connectedEnvironments,
64 AuthOrchestrationOperateScope,
65 );
66 const queueRef = useRef<ReturnType<typeof createComposerAttachmentUploadQueue> | null>(null);
67
68 useEffect(() => {
69 ensureComposerDraftsLoaded();
70 const queue = createComposerAttachmentUploadQueue({
71 onChange: (states) => appAtomRegistry.set(composerAttachmentUploadsAtom, states),
72 upload: async ({ environmentId, attachment }, signal, onProgress) => {
73 const release = isFileBackedComposerAttachment(attachment)
74 ? retainComposerAttachmentFileForPreview(attachment)
75 : undefined;
76 try {
77 const result = await prepareTurnAttachments({
78 environmentId,
79 attachments: [attachment],
80 supportsImageUploads: true,
81 signal,
82 onUploadProgress: (_, progress) => onProgress(progress),
83 persistUploadedReferences: async ([uploaded]) => {
84 if (signal.aborted || !uploaded) return "abandon";
85 const queued = flattenQueuedThreadMessages(
86 appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom),
87 );
88 let retained = false;
89 for (const [key, draft] of Object.entries(appAtomRegistry.get(composerDraftsAtom))) {
90 if (
91 composerDraftEnvironmentId(key, queued, draft) === environmentId &&
92 draft.attachments.some((candidate) => candidate.id === attachment.id)
93 ) {
94 retained = setComposerDraftAttachmentUpload(key, uploaded) || retained;
95 }
96 }
97 if (!retained) return "abandon";
98 await flushComposerDrafts();
99 return "persisted";
100 },
101 });
102 return result.status === "ready";
103 } finally {
104 release?.();
105 }
106 },
107 });
108 queueRef.current = queue;
109 uploadQueue = queue;
110 return () => {
111 queue.dispose();
112 if (uploadQueue === queue) uploadQueue = null;
113 queueRef.current = null;
114 };
115 }, []);
116
117 useEffect(() => {
118 const queued = flattenQueuedThreadMessages(queuedMessages);
119 const connected = new Set(
120 connectedEnvironments
121 .filter((environment) => environment.connectionState === "connected")
122 .map((environment) => environment.environmentId),
123 );
124 const requests = Object.entries(drafts).flatMap(([key, draft]) => {
125 const environmentId = composerDraftEnvironmentId(key, queued, draft);
126 if (
127 environmentId === null ||
128 !connected.has(environmentId) ||
129 !operableEnvironments.has(environmentId)
130 )
131 return [];
132 return draft.attachments
133 .filter((attachment) =>
134 canUploadComposerAttachment(attachment, serverConfigs.get(environmentId)),
135 )
136 .map((attachment) => ({ environmentId, attachment }));
137 });
138 queueRef.current?.sync(requests);
139 }, [connectedEnvironments, drafts, operableEnvironments, queuedMessages, serverConfigs]);
140}