1import { settlePromise, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime";
2import { useAtomValue } from "@effect/atom-react";
3import { AuthOrchestrationOperateScope, type EnvironmentId } from "@t3tools/contracts";
4import * as Schema from "effect/Schema";
5import { Atom, AsyncResult } from "effect/reactivity";
6
7import { prepareTurnAttachments } from "../lib/attachmentUpload";
8import { retainComposerAttachmentFileForPreview } from "../lib/composerAttachmentPreviewRetention";
9import {
10 isFileBackedComposerAttachment,
11 type DraftComposerAttachment,
12} from "../lib/composerImages";
13import {
14 composerAttachmentUploadKey,
15 composerDraftEnvironmentId,
16 canUploadComposerAttachment,
17 createComposerAttachmentUploadQueue,
18 type ComposerAttachmentUploadState,
19} from "../lib/composerAttachmentUploadQueue";
20import { appAtomRegistry } from "./atom-registry";
21import { environmentServerConfigsAtom } from "./server";
22import { flattenQueuedThreadMessages, threadOutboxManager } from "./thread-outbox";
23import {
24 ComposerDraftPersistenceError,
25 type ComposerDraft,
26 composerDraftsAtom,
27 composerDraftUploadSourcesAtom,
28 ensureComposerDraftsLoaded,
29 flushComposerDrafts,
30 setComposerDraftAttachmentUpload,
31} from "./use-composer-drafts";
32import { environmentSession, readEnvironmentScope, sessionHasScope } from "./session";
33import { workspaceConnections } from "./workspace";
34
35const isComposerDraftPersistenceError = Schema.is(ComposerDraftPersistenceError);
36
37export {
38 composerAttachmentUploadBlockReason,
39 composerAttachmentsStillUploading,
40} from "../lib/composerAttachmentUploadQueue";
41
42export const composerAttachmentUploadsAtom = Atom.make<
43 Readonly<Record<string, ComposerAttachmentUploadState>>
44>({}).pipe(Atom.keepAlive);
45const uploadStateAtom = Atom.family((key: string) =>
46 Atom.map(composerAttachmentUploadsAtom, (states) => states[key]),
47);
48let uploadQueue: ReturnType<typeof createComposerAttachmentUploadQueue> | null = null;
49
50export function useComposerAttachmentUploadState(
51 environmentId: EnvironmentId | undefined,
52 attachmentId: string,
53) {
54 return useAtomValue(
55 uploadStateAtom(environmentId ? composerAttachmentUploadKey(environmentId, attachmentId) : ""),
56 );
57}
58
59export function retryComposerAttachmentUpload(environmentId: EnvironmentId, attachmentId: string) {
60 if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return;
61 uploadQueue?.retry(environmentId, attachmentId);
62}
63
64type ComposerAttachmentUploadRequest = {
65 readonly environmentId: EnvironmentId;
66 readonly attachment: DraftComposerAttachment;
67};
68
69export function composerAttachmentUploadRequests(input: {
70 readonly drafts: Readonly<Record<string, Pick<ComposerDraft, "attachments" | "project">>>;
71 readonly queuedMessagesByThreadKey: Parameters<typeof flattenQueuedThreadMessages>[0];
72 /** Every environment the client knows, so a draft key never resolves to a shorter connected prefix. */
73 readonly knownEnvironmentIds: Iterable<EnvironmentId>;
74 readonly connectedEnvironmentIds: ReadonlySet<EnvironmentId>;
75 readonly operableEnvironmentIds: ReadonlySet<EnvironmentId>;
76 readonly serverConfigs: ReadonlyMap<
77 EnvironmentId,
78 Parameters<typeof canUploadComposerAttachment>[1]
79 >;
80}): ReadonlyArray<ComposerAttachmentUploadRequest> {
81 const queued = flattenQueuedThreadMessages(input.queuedMessagesByThreadKey);
82 return Object.entries(input.drafts).flatMap(([key, draft]) => {
83 const environmentId = composerDraftEnvironmentId(key, input.knownEnvironmentIds, queued, draft);
84 if (
85 environmentId === null ||
86 !input.connectedEnvironmentIds.has(environmentId) ||
87 !input.operableEnvironmentIds.has(environmentId)
88 )
89 return [];
90 return draft.attachments
91 .filter((attachment) =>
92 canUploadComposerAttachment(attachment, input.serverConfigs.get(environmentId)),
93 )
94 .map((attachment) => ({ environmentId, attachment }));
95 });
96}
97
98// Each connected environment's session is read here so a granted or revoked
99// operate scope re-syncs the queue.
100const composerAttachmentUploadRequestsAtom = Atom.make((get) => {
101 const environments = get(workspaceConnections.environmentsAtom);
102 const connected = environments.filter(
103 (environment) => environment.connectionState === "connected",
104 );
105 const operableEnvironmentIds = new Set<EnvironmentId>();
106 for (const { environmentId } of connected) {
107 if (
108 sessionHasScope(
109 get(environmentSession.sessionStateAtom(environmentId)),
110 AuthOrchestrationOperateScope,
111 )
112 ) {
113 operableEnvironmentIds.add(environmentId);
114 }
115 }
116 return composerAttachmentUploadRequests({
117 drafts: get(composerDraftUploadSourcesAtom),
118 queuedMessagesByThreadKey: get(threadOutboxManager.queuedMessagesByThreadKeyAtom),
119 knownEnvironmentIds: environments.map((environment) => environment.environmentId),
120 connectedEnvironmentIds: new Set(connected.map((environment) => environment.environmentId)),
121 operableEnvironmentIds,
122 serverConfigs: get(environmentServerConfigsAtom),
123 });
124}).pipe(Atom.withLabel("mobile:composer-attachment-upload-requests"));
125
126/**
127 * Uploads draft attachments to their environment as soon as it is connected and
128 * operable. The queue follows the requests atom for its whole lifetime.
129 */
130export function createComposerAttachmentUploadWorker(
131 requestsAtom: Atom.Atom<
132 ReadonlyArray<ComposerAttachmentUploadRequest>
133 > = composerAttachmentUploadRequestsAtom,
134) {
135 ensureComposerDraftsLoaded();
136 const queue = createComposerAttachmentUploadQueue({
137 onChange: (states) => appAtomRegistry.set(composerAttachmentUploadsAtom, states),
138 upload: async ({ environmentId, attachment }, signal, onProgress) => {
139 const release = isFileBackedComposerAttachment(attachment)
140 ? retainComposerAttachmentFileForPreview(attachment)
141 : undefined;
142 const prepared = await settlePromise(() =>
143 prepareTurnAttachments({
144 environmentId,
145 attachments: [attachment],
146 supportsImageUploads: true,
147 signal,
148 onUploadProgress: (_, progress) => onProgress(progress),
149 persistUploadedReferences: async ([uploaded]) => {
150 if (signal.aborted || !uploaded) return "abandon";
151 const queued = flattenQueuedThreadMessages(
152 appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom),
153 );
154 const knownEnvironmentIds = appAtomRegistry
155 .get(workspaceConnections.environmentsAtom)
156 .map((environment) => environment.environmentId);
157 let retained = false;
158 for (const [key, draft] of Object.entries(appAtomRegistry.get(composerDraftsAtom))) {
159 if (
160 composerDraftEnvironmentId(key, knownEnvironmentIds, queued, draft) ===
161 environmentId &&
162 draft.attachments.some((candidate) => candidate.id === attachment.id)
163 ) {
164 retained = setComposerDraftAttachmentUpload(key, uploaded) || retained;
165 }
166 }
167 if (!retained) return "abandon";
168 // Draft-save failure only costs re-upload after restart; the outbox saves references independently, so preserve accepted uploads.
169 await flushComposerDrafts().catch((error: unknown) => {
170 if (!isComposerDraftPersistenceError(error)) throw error;
171 console.warn("[composer-attachments] could not save the uploaded reference", error);
172 });
173 return "persisted";
174 },
175 }),
176 );
177 release?.();
178 if (AsyncResult.isFailure(prepared)) throw squashAtomCommandFailure(prepared);
179 return prepared.value.status === "ready";
180 },
181 });
182 uploadQueue = queue;
183 const unsubscribe = appAtomRegistry.subscribe(requestsAtom, (requests) => queue.sync(requests), {
184 immediate: true,
185 });
186 return {
187 queue,
188 dispose: () => {
189 unsubscribe();
190 queue.dispose();
191 if (uploadQueue === queue) uploadQueue = null;
192 },
193 };
194}
195
196export function installComposerAttachmentUploads(): () => void {
197 return createComposerAttachmentUploadWorker().dispose;
198}