1// @effect-diagnostics nodeBuiltinImport:off -- exclusive copy and streaming hashes are filesystem boundaries missing from Effect's API.
2import * as NodeCrypto from "node:crypto";
3import * as NodeFSP from "node:fs/promises";
4import {
5 ATTACHMENT_UPLOAD_URL_TTL_MS,
6 type AttachmentCreateUploadUrlInput,
7 AttachmentUploadSigningKeyError,
8} from "@t3tools/contracts";
9import { isSvgAttachment } from "@t3tools/shared/image";
10import * as Clock from "effect/Clock";
11import * as SecureRandom from "effect/crypto/SecureRandom";
12import * as Effect from "effect/Effect";
13import * as FileSystem from "effect/FileSystem";
14import * as Option from "effect/Option";
15import * as Path from "effect/Path";
16import * as PlatformError from "effect/PlatformError";
17import * as Schema from "effect/Schema";
18import * as Stream from "effect/Stream";
19import type * as HttpServerRequest from "effect/http/HttpServerRequest";
20
21import {
22 attachmentFileExtension,
23 createPendingAttachmentId,
24 parseThreadSegmentFromAttachmentId,
25 PENDING_ATTACHMENT_THREAD_SEGMENT,
26 resolveAttachmentPathById,
27 sweepStalePendingAttachments,
28} from "../attachmentStore.ts";
29import { resolveAttachmentRelativePath } from "../attachmentPaths.ts";
30import { base64UrlDecodeUtf8, base64UrlEncode } from "../auth/utils.ts";
31import * as AssetSigner from "./AssetSigner.ts";
32import * as ServerConfig from "../config.ts";
33import { inferImageExtension } from "../imageMime.ts";
34import * as DocumentText from "../provider/DocumentText.ts";
35
36export const ATTACHMENT_UPLOAD_ROUTE_PREFIX = "/api/attachments/upload";
37
38const PENDING_ATTACHMENT_SWEEP_INTERVAL_MS = 15 * 60_000;
39const lastPendingSweepByDirectory = new Map<string, number>();
40
41const AttachmentUploadClaims = Schema.Struct({
42 version: Schema.Literal(1),
43 kind: Schema.Literal("attachment-upload"),
44 type: Schema.Literals(["image", "file"]).pipe(
45 Schema.withDecodingDefault(Effect.succeed("image" as const)),
46 ),
47 attachmentId: Schema.String,
48 name: Schema.String,
49 mimeType: Schema.String,
50 sizeBytes: Schema.Number,
51 expiresAt: Schema.Number,
52});
53export type AttachmentUploadClaims = typeof AttachmentUploadClaims.Type;
54
55const attachmentUploadClaimsJson = Schema.fromJsonString(AttachmentUploadClaims);
56const decodeAttachmentUploadClaims = Schema.decodeUnknownOption(attachmentUploadClaimsJson);
57const encodeAttachmentUploadClaims = Schema.encodeSync(attachmentUploadClaimsJson);
58
59function decodeClaims(encodedPayload: string): AttachmentUploadClaims | null {
60 try {
61 return Option.getOrNull(decodeAttachmentUploadClaims(base64UrlDecodeUtf8(encodedPayload)));
62 } catch {
63 return null;
64 }
65}
66
67export const issueAttachmentUploadUrl = Effect.fn("AttachmentUpload.issueUrl")(function* (
68 input: AttachmentCreateUploadUrlInput,
69) {
70 const config = yield* ServerConfig.ServerConfig;
71 const nowMs = yield* Clock.currentTimeMillis;
72 const previousSweep = lastPendingSweepByDirectory.get(config.attachmentsDir);
73 if (
74 previousSweep === undefined ||
75 nowMs - previousSweep >= PENDING_ATTACHMENT_SWEEP_INTERVAL_MS
76 ) {
77 lastPendingSweepByDirectory.set(config.attachmentsDir, nowMs);
78 const swept = sweepStalePendingAttachments({
79 attachmentsDir: config.attachmentsDir,
80 nowMs,
81 });
82 if (swept.deleted > 0) {
83 yield* Effect.logInfo("Removed expired attachment uploads.", { deleted: swept.deleted });
84 }
85 }
86
87 const attachmentType = input.type ?? "image";
88 const name =
89 attachmentType === "file" && isSvgAttachment(input) && !/\.svg$/i.test(input.name)
90 ? `${input.name}.svg`
91 : input.name;
92 const attachmentId = createPendingAttachmentId(
93 attachmentType === "file" ? attachmentFileExtension(name) : undefined,
94 );
95 const expiresAt = nowMs + ATTACHMENT_UPLOAD_URL_TTL_MS;
96 const encodedPayload = base64UrlEncode(
97 encodeAttachmentUploadClaims({
98 version: 1,
99 kind: "attachment-upload",
100 type: attachmentType,
101 attachmentId,
102 name,
103 mimeType: input.mimeType,
104 sizeBytes: input.sizeBytes,
105 expiresAt,
106 }),
107 );
108
109 const signer = yield* AssetSigner.AssetSigner;
110 const signature = yield* signer
111 .sign(encodedPayload)
112 .pipe(Effect.mapError((cause) => new AttachmentUploadSigningKeyError({ cause })));
113 return {
114 attachmentId,
115 relativeUrl: `${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/${encodedPayload}.${signature}`,
116 expiresAt,
117 };
118});
119
120export const validateAttachmentUploadToken = Effect.fn("AttachmentUpload.validateToken")(function* (
121 token: string,
122) {
123 const [encodedPayload, signature, unexpectedSegment] = token.split(".");
124 if (!encodedPayload || !signature || unexpectedSegment) {
125 return null;
126 }
127
128 const signer = yield* AssetSigner.AssetSigner;
129 const verified = yield* signer.verify(encodedPayload, signature).pipe(
130 Effect.tapError((cause) =>
131 Effect.logError("Failed to load the attachment upload signing key.", { cause }),
132 ),
133 Effect.orElseSucceed(() => false),
134 );
135 if (!verified) {
136 return null;
137 }
138
139 const claims = decodeClaims(encodedPayload);
140 if (!claims || claims.expiresAt <= (yield* Clock.currentTimeMillis)) {
141 return null;
142 }
143 return claims;
144});
145
146export type StoreAttachmentUploadResult =
147 | { readonly ok: true }
148 | { readonly ok: false; readonly status: number; readonly detail: string };
149
150export const storeAttachmentUpload = Effect.fn("AttachmentUpload.store")(function* (
151 claims: AttachmentUploadClaims,
152 body: Uint8Array | HttpServerRequest.HttpServerRequest["stream"],
153) {
154 if (body instanceof Uint8Array && body.byteLength !== claims.sizeBytes) {
155 return {
156 ok: false,
157 status: 400,
158 detail: `Body was ${body.byteLength} bytes, expected ${claims.sizeBytes}.`,
159 } satisfies StoreAttachmentUploadResult;
160 }
161
162 const config = yield* ServerConfig.ServerConfig;
163 const extension =
164 claims.type === "file"
165 ? attachmentFileExtension(claims.name)
166 : inferImageExtension({ mimeType: claims.mimeType, fileName: claims.name });
167 const relativePath = `${claims.attachmentId}${extension}`;
168 const finalPath = resolveAttachmentRelativePath({
169 attachmentsDir: config.attachmentsDir,
170 relativePath,
171 });
172 const random = yield* SecureRandom.SecureRandom;
173 const partId = yield* random.uuidV4();
174 const partPath = resolveAttachmentRelativePath({
175 attachmentsDir: config.attachmentsDir,
176 relativePath: `${relativePath}.${partId}.part`,
177 });
178 if (!finalPath || !partPath) {
179 return { ok: false, status: 500, detail: "Failed to resolve attachment path." };
180 }
181
182 const fileSystem = yield* FileSystem.FileSystem;
183 const path = yield* Path.Path;
184 let receivedBytes = 0;
185 const bodyStream = body instanceof Uint8Array ? Stream.make(body) : body;
186 return yield* Effect.gen(function* () {
187 yield* fileSystem.makeDirectory(path.dirname(finalPath), { recursive: true });
188 yield* Stream.run(
189 bodyStream.pipe(
190 Stream.takeWhile((chunk) => {
191 receivedBytes += chunk.byteLength;
192 return receivedBytes <= claims.sizeBytes;
193 }),
194 ),
195 fileSystem.sink(partPath),
196 );
197 if (receivedBytes !== claims.sizeBytes) {
198 return {
199 ok: false,
200 status: 400,
201 detail: `Body was ${receivedBytes} bytes, expected ${claims.sizeBytes}.`,
202 } satisfies StoreAttachmentUploadResult;
203 }
204 // Publishing without replacement keeps an upload and its cached text immutable.
205 const published = yield* fileSystem.link(partPath, finalPath).pipe(
206 Effect.as(true),
207 Effect.catch((error) => {
208 if (error.reason._tag === "AlreadyExists") return Effect.succeed(false);
209 return Effect.tryPromise({
210 try: () => NodeFSP.copyFile(partPath, finalPath, NodeFSP.constants.COPYFILE_EXCL),
211 catch: (cause) =>
212 PlatformError.systemError({
213 _tag:
214 cause instanceof Error && "code" in cause && cause.code === "EEXIST"
215 ? "AlreadyExists"
216 : "Unknown",
217 module: "FileSystem",
218 method: "copyFile",
219 pathOrDescriptor: finalPath,
220 cause,
221 }),
222 }).pipe(
223 Effect.as(true),
224 Effect.catchIf(
225 (error) => error.reason._tag === "AlreadyExists",
226 () => Effect.succeed(false),
227 ),
228 );
229 }),
230 );
231 if (!published) {
232 if (Number((yield* fileSystem.stat(finalPath)).size) !== receivedBytes) {
233 return { ok: false, status: 409, detail: "Attachment already uploaded." };
234 }
235 const hashes = yield* Effect.forEach([partPath, finalPath], (file) =>
236 fileSystem.stream(file).pipe(
237 Stream.runFold(
238 () => NodeCrypto.createHash("sha256"),
239 (hash, chunk) => hash.update(chunk),
240 ),
241 Effect.map((hash) => hash.digest("hex")),
242 ),
243 );
244 if (hashes[0] !== hashes[1]) {
245 return { ok: false, status: 409, detail: "Attachment already uploaded." };
246 }
247 }
248 const documents = yield* DocumentText.DocumentText;
249 yield* documents.ensure(finalPath);
250 return { ok: true } satisfies StoreAttachmentUploadResult;
251 }).pipe(
252 Effect.catch((cause) =>
253 Effect.logError("Failed to persist attachment upload.", {
254 attachmentId: claims.attachmentId,
255 cause,
256 }).pipe(
257 Effect.as({
258 ok: false,
259 status: 500,
260 detail: "Failed to persist upload.",
261 } satisfies StoreAttachmentUploadResult),
262 ),
263 ),
264 Effect.ensuring(
265 fileSystem.remove(partPath, { force: true }).pipe(Effect.orElseSucceed(() => undefined)),
266 ),
267 );
268});
269
270export const deletePendingAttachment = Effect.fn("AttachmentUpload.deletePending")(function* (
271 attachmentId: string,
272) {
273 if (parseThreadSegmentFromAttachmentId(attachmentId) !== PENDING_ATTACHMENT_THREAD_SEGMENT) {
274 return;
275 }
276
277 const config = yield* ServerConfig.ServerConfig;
278 const attachmentPath = resolveAttachmentPathById({
279 attachmentsDir: config.attachmentsDir,
280 attachmentId,
281 });
282 if (!attachmentPath) {
283 return;
284 }
285
286 const fileSystem = yield* FileSystem.FileSystem;
287 yield* fileSystem.remove(attachmentPath, { force: true }).pipe(Effect.orElseSucceed(() => {}));
288 yield* fileSystem.remove(`${attachmentPath}.txt`, { force: true }).pipe(Effect.ignore);
289});