1import {
2 PROVIDER_SEND_TURN_MAX_FILE_BYTES,
3 WS_METHODS,
4 type AttachmentCreateUploadUrlInput,
5 type AttachmentCreateUploadUrlResult,
6 type AttachmentDeleteInput,
7 type EnvironmentId,
8} from "@t3tools/contracts";
9import type { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
10
11import type { EnvironmentRegistry } from "../connection/registry.ts";
12import {
13 createEnvironmentRpcCommand,
14 executeAtomQuery,
15 runAtomCommand,
16 squashAtomCommandFailure,
17 type AtomCommand,
18} from "./runtime.ts";
19
20export function createAttachmentEnvironmentAtoms<R, E>(
21 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
22) {
23 return {
24 createUploadUrl: createEnvironmentRpcCommand(runtime, {
25 label: "environment-command:attachments:create-upload-url",
26 tag: WS_METHODS.attachmentsCreateUploadUrl,
27 }),
28 remove: createEnvironmentRpcCommand(runtime, {
29 label: "environment-command:attachments:delete",
30 tag: WS_METHODS.attachmentsDelete,
31 }),
32 };
33}
34
35/**
36 * Pending uploads expire, so this is the signal to upload the bytes again rather than retry the
37 * lookup. A structural `_tag` check rather than a schema check: the squashed cause of a failed
38 * RPC is not guaranteed to be a decoded error class instance, only a tagged value.
39 */
40export function isAssetAttachmentNotFoundFailure(error: unknown): boolean {
41 return (
42 typeof error === "object" &&
43 error !== null &&
44 "_tag" in error &&
45 error._tag === "AssetAttachmentNotFoundError"
46 );
47}
48
49export type PersistedAttachmentVerification =
50 | { readonly status: "verified" }
51 | { readonly status: "missing" }
52 | { readonly status: "failed"; readonly error: unknown };
53
54export async function verifyPersistedAttachmentUpload<A, E>(input: {
55 readonly registry: AtomRegistry.AtomRegistry;
56 readonly createAssetUrl: (query: {
57 readonly environmentId: EnvironmentId;
58 readonly input: {
59 readonly resource: { readonly _tag: "attachment"; readonly attachmentId: string };
60 };
61 }) => Atom.Atom<AsyncResult.AsyncResult<A, E>>;
62 readonly environmentId: EnvironmentId;
63 readonly attachmentId: string;
64}): Promise<PersistedAttachmentVerification> {
65 const result = await executeAtomQuery(
66 input.registry,
67 input.createAssetUrl({
68 environmentId: input.environmentId,
69 input: { resource: { _tag: "attachment", attachmentId: input.attachmentId } },
70 }),
71 // `refresh` forces a server round trip: the asset URL query atom caches
72 // results (SWR), so a retry right after a transient failure would
73 // otherwise re-observe the cached failure and never ask the server.
74 { reportFailure: false, reportDefect: false, refresh: true },
75 );
76 if (result._tag === "Success") {
77 return { status: "verified" };
78 }
79 const error = squashAtomCommandFailure(result);
80 return isAssetAttachmentNotFoundFailure(error)
81 ? { status: "missing" }
82 : { status: "failed", error };
83}
84
85type AttachmentCreateUploadUrlCommand<E> = AtomCommand<
86 { readonly environmentId: EnvironmentId; readonly input: AttachmentCreateUploadUrlInput },
87 AttachmentCreateUploadUrlResult,
88 E
89>;
90
91type AttachmentRemoveCommand<E> = AtomCommand<
92 { readonly environmentId: EnvironmentId; readonly input: AttachmentDeleteInput },
93 unknown,
94 E
95>;
96
97export function deletePendingAttachmentUpload<E>(input: {
98 readonly registry: AtomRegistry.AtomRegistry;
99 readonly remove: AttachmentRemoveCommand<E>;
100 readonly environmentId: EnvironmentId;
101 readonly attachmentId: string;
102}): void {
103 void runAtomCommand(
104 input.registry,
105 input.remove,
106 { environmentId: input.environmentId, input: { attachmentId: input.attachmentId } },
107 { reportFailure: false, reportDefect: false },
108 );
109}
110
111export interface AttachmentByteUpload {
112 readonly done: Promise<void>;
113 readonly abort: () => void;
114}
115
116export type AttachmentUploadCycleResult =
117 | { readonly status: "uploaded"; readonly attachmentId: string }
118 | { readonly status: "cancelled"; readonly attachmentId: string | null }
119 | {
120 readonly status: "failed";
121 readonly step: "mint" | "resolve-url" | "transfer";
122 readonly attachmentId: string | null;
123 readonly error: unknown;
124 };
125
126/**
127 * The cycle never deletes the minted pending upload on failure: callers keep the returned
128 * `attachmentId` and decide between retry and release. The one exception is `onMinted`
129 * returning `"cancel"`, where the caller already abandoned the upload and the fresh mint is
130 * deleted before returning.
131 */
132export async function runAttachmentUploadCycle<E, RE>(input: {
133 readonly registry: AtomRegistry.AtomRegistry;
134 readonly createUploadUrl: AttachmentCreateUploadUrlCommand<E>;
135 readonly remove: AttachmentRemoveCommand<RE>;
136 readonly environmentId: EnvironmentId;
137 readonly upload: AttachmentCreateUploadUrlInput;
138 readonly resolveUploadUrl: (relativeUrl: string) => string | null;
139 readonly transport: (url: string) => AttachmentByteUpload;
140 readonly onMinted?: (attachmentId: string) => "continue" | "cancel";
141 readonly onTransferStart?: (abort: () => void) => void;
142}): Promise<AttachmentUploadCycleResult> {
143 const minted = await runAtomCommand(
144 input.registry,
145 input.createUploadUrl,
146 { environmentId: input.environmentId, input: input.upload },
147 { reportFailure: false },
148 );
149 if (minted._tag !== "Success") {
150 return {
151 status: "failed",
152 step: "mint",
153 attachmentId: null,
154 error: squashAtomCommandFailure(minted),
155 };
156 }
157 const attachmentId = minted.value.attachmentId;
158 if (input.onMinted?.(attachmentId) === "cancel") {
159 deletePendingAttachmentUpload({
160 registry: input.registry,
161 remove: input.remove,
162 environmentId: input.environmentId,
163 attachmentId,
164 });
165 return { status: "cancelled", attachmentId };
166 }
167
168 const url = input.resolveUploadUrl(minted.value.relativeUrl);
169 if (!url) {
170 return {
171 status: "failed",
172 step: "resolve-url",
173 attachmentId,
174 error: new Error("The environment is not connected."),
175 };
176 }
177
178 const transfer = input.transport(url);
179 input.onTransferStart?.(transfer.abort);
180 try {
181 await transfer.done;
182 } catch (error) {
183 return { status: "failed", step: "transfer", attachmentId, error };
184 }
185 return { status: "uploaded", attachmentId };
186}
187
188/**
189 * The contract caps what a turn may reference, so a larger advertised value must not admit
190 * files the send would then refuse.
191 */
192export function clampFileAttachmentUploadBytes(advertisedMaxUploadBytes: number): number {
193 return Math.min(advertisedMaxUploadBytes, PROVIDER_SEND_TURN_MAX_FILE_BYTES);
194}
195
196export function formatAttachmentSize(sizeBytes: number): string {
197 const kilobytes = Math.max(1, Math.ceil(sizeBytes / 1024));
198 // Pick the unit after rounding, so 1023.5 KB reads "1.0 MB", not "1024 KB".
199 return kilobytes >= 1024 ? `${(sizeBytes / (1024 * 1024)).toFixed(1)} MB` : `${kilobytes} KB`;
200}
201
202export function fileAttachmentTooLargeMessage(name: string, maxUploadBytes: number): string {
203 const maxUploadSize =
204 maxUploadBytes >= 1024 * 1024 && maxUploadBytes % (1024 * 1024) === 0
205 ? `${maxUploadBytes / (1024 * 1024)} MB`
206 : maxUploadBytes >= 1024 && maxUploadBytes % 1024 === 0
207 ? `${maxUploadBytes / 1024} KB`
208 : `${maxUploadBytes} ${maxUploadBytes === 1 ? "byte" : "bytes"}`;
209 return `'${name}' exceeds the ${maxUploadSize} attachment limit.`;
210}