Repository Public
1// Resumable uploads. A session is created with its final size; the client then appends chunks at the offset the2// server reports, each optionally carrying its own SHA-256, and finally asks for completion with the whole-file hash.3import { createHash } from 'node:crypto';4import { createWriteStream } from 'node:fs';5import { open, rm, stat } from 'node:fs/promises';6import { join } from 'node:path';7import type { Readable } from 'node:stream';8import type { FastifyInstance } from 'fastify';9import { MAX_CHUNK_BYTES, safePathSegments, sanitizeName } from '@spd/shared';10import { z } from 'zod';11import { actorOf, requireUser, userOf } from './auth.ts';12import { ensureObject, usedBytes } from './blob.ts';13import { config, notFound, UserError } from './config.ts';14import { sha256File } from './crypto.ts';15import { audit, type Db, lockOwner, sql, tx } from './db.ts';16import { addFile, addVersion, ensurePath, fileJson, getFile, getFolder } from './tree.ts';1718export interface UploadRow { id: string; owner_id: string; actor_user_id: string | null; item_id: string | null; folder_id: string | null; name: string; size: string; received: string; on_conflict: 'fail' | 'rename' | 'version'; status: string; file_id: string | null; expires_at: Date }19export const uploadPath = (u: Pick<UploadRow, 'id' | 'item_id'>) => join(u.item_id ? config.quarantineDir : config.tmpDir, `up-${u.id}`);20export const uploadJson = (u: UploadRow) => ({ id: u.id, name: u.name, size: Number(u.size), received: Number(u.received), status: u.status, fileId: u.file_id });2122/** Quota check under lockOwner: stored bytes + bytes promised to uploads in flight + this one. */23export async function assertQuota(t: Db, ownerId: string, extra: number, exceptUploadId?: string): Promise<void> {24 const [q] = await t<{ limit_bytes: string }[]>`select limit_bytes::text from quotas where user_id = ${ownerId}`;25 const [p] = await t<{ n: string }[]>`select coalesce(sum(size), 0)::text as n from uploads where owner_id = ${ownerId} and status in ('active', 'completing') and id is distinct from ${exceptUploadId ?? null}`;26 const used = await usedBytes(t, ownerId);27 if (used + Number(p!.n) + extra > Number(q?.limit_bytes ?? 0)) throw new UserError('quota_exceeded', 'storage quota exceeded', 413, { used, limit: Number(q?.limit_bytes ?? 0) });28}2930/**31 * Append one chunk. The row lock makes concurrent writers of the same upload take turns; a chunk whose offset is not32 * the current end is refused with the current end, so a client can always resynchronise. `guard` re-checks whatever33 * policy applies (portal still open, not expired…) on every chunk.34 */35export async function appendChunk(id: string, offset: number, body: Readable, chunkSha: string | undefined, guard: (t: Db, u: UploadRow) => Promise<void>): Promise<UploadRow> {36 try {37 return await tx(async (t) => {38 const [u] = await t<UploadRow[]>`select *, size::text, received::text from uploads where id = ${id} for update`;39 if (!u) throw notFound('upload not found');40 await guard(t, u);41 if (u.status !== 'active' || u.expires_at < new Date()) throw new UserError('upload_closed', 'this upload is no longer active', 410);42 const received = Number(u.received), size = Number(u.size);43 if (offset !== received) throw new UserError('offset_mismatch', 'chunk does not start at the current offset', 409, { received });44 const room = Math.min(size - received, MAX_CHUNK_BYTES);45 const path = uploadPath(u);46 const fh = await open(path, received === 0 ? 'w' : 'r+', 0o600);47 let written = 0;48 const hash = createHash('sha256');49 try {50 await fh.truncate(received); // drop bytes of an earlier interrupted chunk51 const out = createWriteStream(path, { fd: fh.fd, start: received, autoClose: false });52 await new Promise<void>((resolve, reject) => {53 body.on('data', (c: Buffer) => {54 written += c.length;55 if (written > room) { body.destroy(); reject(new UserError('chunk_too_large', 'chunk exceeds the declared size or the chunk limit', 413)); return; }56 hash.update(c);57 if (!out.write(c)) { body.pause(); out.once('drain', () => body.resume()); }58 });59 body.on('end', () => out.end(() => resolve()));60 body.on('error', reject); body.on('aborted', () => reject(new Error('aborted'))); out.on('error', reject);61 });62 if (chunkSha && hash.digest('hex') !== chunkSha.toLowerCase()) throw new UserError('chunk_corrupt', 'chunk checksum mismatch, send it again', 422, { received });63 await fh.sync();64 } catch (e) { await fh.truncate(received).catch(() => {}); throw e; } finally { await fh.close(); }65 if (written === 0 && size > 0) throw new UserError('empty_chunk', 'empty chunk', 400);66 const [r] = await t<UploadRow[]>`update uploads set received = received + ${written}, updated_at = now() where id = ${id} returning *, size::text, received::text`;67 return r!;68 });69 } finally { body.resume?.(); }70}7172/** Check the temp file is complete and return its SHA-256 (compared with the client's when given). */73export async function sealCheck(u: UploadRow, clientSha?: string): Promise<string> {74 const size = Number(u.size);75 if (Number(u.received) !== size) throw new UserError('upload_incomplete', 'not all bytes were received', 409, { received: Number(u.received) });76 const path = uploadPath(u);77 if (size === 0) await (await open(path, 'a', 0o600)).close();78 if ((await stat(path)).size !== size) throw new UserError('upload_incomplete', 'stored bytes do not match', 409);79 const sha = await sha256File(path);80 if (clientSha && clientSha.toLowerCase() !== sha) throw new UserError('checksum_mismatch', 'the file received differs from the file sent', 422, { sha256: sha });81 return sha;82}8384const Sha = z.string().regex(/^[0-9a-fA-F]{64}$/);85export function uploadRoutes(app: FastifyInstance): void {86 const authed = { preHandler: requireUser };87 const mine = async (t: Db, userId: string, id: string) => {88 const [u] = await t<UploadRow[]>`select *, size::text, received::text from uploads where id = ${id} and actor_user_id = ${userId}`;89 if (!u) throw notFound('upload not found');90 return u;91 };9293 app.post('/api/v1/uploads', authed, async (req) => {94 const b = z.object({95 folderId: z.string(), name: z.string().min(1).max(1000), size: z.number().int().min(0), relativePath: z.string().max(4000).optional(),96 onConflict: z.enum(['fail', 'rename', 'version']).default('rename'), fileId: z.uuid().optional(),97 }).parse(req.body);98 const u = userOf(req);99 if (b.size > config.maxFileBytes) throw new UserError('file_too_large', 'file exceeds the maximum size', 413, { max: config.maxFileBytes });100 const row = await tx(async (t) => {101 await lockOwner(t, u.id);102 await assertQuota(t, u.id, b.size);103 let folderId: string, name = sanitizeName(b.name), onConflict = b.onConflict;104 if (b.fileId) { // new version of a known file105 const f = await getFile(t, u.id, b.fileId);106 folderId = f.folder_id; name = f.name; onConflict = 'version';107 } else {108 folderId = (await getFolder(t, u.id, b.folderId)).id;109 const segs = b.relativePath ? safePathSegments(b.relativePath).slice(0, -1) : [];110 if (segs.length) folderId = await ensurePath(t, u.id, folderId, segs);111 }112 const [r] = await t<UploadRow[]>`insert into uploads (owner_id, actor_user_id, folder_id, name, size, on_conflict, file_id, expires_at)113 values (${u.id}, ${u.id}, ${folderId}, ${name}, ${b.size}, ${onConflict}, ${b.fileId ?? null}, now() + interval '7 days') returning *, size::text, received::text`;114 return r!;115 });116 return uploadJson(row);117 });118 app.get('/api/v1/uploads/:id', authed, async (req) => {119 const { id } = z.object({ id: z.uuid() }).parse(req.params);120 return uploadJson(await mine(sql, userOf(req).id, id));121 });122 app.patch('/api/v1/uploads/:id', { ...authed, bodyLimit: MAX_CHUNK_BYTES + 4096 }, async (req) => {123 const { id } = z.object({ id: z.uuid() }).parse(req.params);124 const q = z.object({ offset: z.coerce.number().int().min(0) }).parse(req.query);125 const sha = req.headers['x-chunk-sha256'];126 const userId = userOf(req).id;127 const u = await appendChunk(id, q.offset, req.body as Readable, typeof sha === 'string' ? Sha.parse(sha) : undefined,128 async (_t, row) => { if (row.actor_user_id !== userId) throw notFound('upload not found'); });129 return uploadJson(u);130 });131 app.post('/api/v1/uploads/:id/complete', authed, async (req) => {132 const { id } = z.object({ id: z.uuid() }).parse(req.params);133 const b = z.object({ sha256: Sha.optional() }).parse(req.body ?? {});134 const user = userOf(req);135 const [claimed] = await sql<UploadRow[]>`update uploads set status = 'completing', updated_at = now() where id = ${id} and actor_user_id = ${user.id} and status = 'active' returning *, size::text, received::text`;136 if (!claimed) { // unknown, foreign, already done (idempotent answer) or being completed137 const u = await mine(sql, user.id, id);138 if (u.status === 'done' && u.file_id) return { ...uploadJson(u), file: fileJson(await getFile(sql, user.id, u.file_id, { trashed: true })) };139 throw new UserError('upload_closed', 'this upload is no longer active', 409);140 }141 try {142 const sha = await sealCheck(claimed, b.sha256);143 const result = await tx(async (t) => {144 await lockOwner(t, user.id);145 await assertQuota(t, user.id, Number(claimed.size), id);146 const [object] = await ensureObject(t, user.id, uploadPath(claimed), Number(claimed.size), sha);147 let file, created = false;148 if (claimed.file_id) file = await addVersion(t, user.id, claimed.file_id, object, user.id);149 else {150 const folder = await getFolder(t, user.id, claimed.folder_id ?? 'root').catch(() => getFolder(t, user.id, 'root')); // target trashed meanwhile → root151 ({ file, created } = await addFile(t, { ownerId: user.id, folderId: folder.id, name: claimed.name, object, onConflict: claimed.on_conflict, actorId: user.id }));152 }153 const [done] = await t<UploadRow[]>`update uploads set status = 'done', file_id = ${file.id}, updated_at = now() where id = ${id} returning *, size::text, received::text`;154 await audit(t, actorOf(req), created ? 'file.uploaded' : 'file.version_added', { targetType: 'file', targetId: file.id, meta: { size: Number(claimed.size) } });155 return { ...uploadJson(done!), file: fileJson(file), created };156 });157 await rm(uploadPath(claimed), { force: true });158 return result;159 } catch (e) {160 // a refused completion (conflict, quota, checksum) leaves the bytes in place so the client can decide again161 await sql`update uploads set status = 'active' where id = ${id} and status = 'completing'`;162 throw e;163 }164 });165 app.patch('/api/v1/uploads/:id/options', authed, async (req) => {166 const { id } = z.object({ id: z.uuid() }).parse(req.params);167 const b = z.object({ onConflict: z.enum(['fail', 'rename', 'version']) }).parse(req.body);168 const [r] = await sql`update uploads set on_conflict = ${b.onConflict} where id = ${id} and actor_user_id = ${userOf(req).id} and status = 'active' and file_id is null returning id`;169 if (!r) throw notFound('upload not found');170 return { ok: true };171 });172 app.delete('/api/v1/uploads/:id', authed, async (req) => {173 const { id } = z.object({ id: z.uuid() }).parse(req.params);174 const [u] = await sql<UploadRow[]>`update uploads set status = 'aborted', updated_at = now() where id = ${id} and actor_user_id = ${userOf(req).id} and status = 'active' returning *, size::text, received::text`;175 if (!u) throw notFound('upload not found');176 await rm(uploadPath(u), { force: true });177 return { ok: true };178 });179}180