From 1b429049c200b0f33c46ee343258eff6ae87d3a1 Mon Sep 17 00:00:00 2001 From: Nicolas Date: Thu, 10 Sep 2026 18:25:27 +0200 Subject: [PATCH] =?UTF-8?q?Upload=20fragment=C3=A9=20et=20reprenable=20pou?= =?UTF-8?q?r=20les=20gros=20fichiers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Serveur : - POST /upload/init -> session { id, chunkSize (16 Mo), chunkCount } - PUT /upload/:id/:index -> écrit le morceau à son offset (octet-stream brut, bufferisé, parser dédié) ; renvoie { received, total } - POST /upload/:id/complete -> vérifie tous les morceaux + la taille, renomme le .mediadrop-part vers le chemin final ; 201 + métadonnées - DELETE /upload/:id -> abandon, supprime le fichier partiel - src/lib/upload-sessions.ts : registre en mémoire, purge après 30 min d'inactivité, 100 sessions max - src/lib/targets.ts : resolveTarget / resolvePlacement extraits et partagés - writeChunkAt / createEmptyFile / finalizeFile / removeFile dans storage.ts - limites, allowDirectories, jokers MIME et confinement dans la cible s'appliquent aussi au chemin fragmenté Page : - bascule automatique en fragmenté au-delà de 32 Mo : 3 morceaux en parallèle, 3 réessais par morceau, progression et annulation agrégées - makeReporter partagé (débit + ETA) ; job.cancel unifié (simple / fragmenté / en file d'attente) Tests : +tests/upload-chunked.test.ts (9), +writeChunkAt (storage) ; 73 tests. Vérifié en live : 40 Mio -> 3 morceaux // -> octets exacts, DELETE nettoie. Co-Authored-By: Claude Sonnet 5 --- README.md | 46 ++++--- public/index.html | 253 +++++++++++++++++++++++++++-------- src/app.ts | 22 ++- src/errors.ts | 14 ++ src/fastify.d.ts | 3 + src/lib/storage.ts | 70 ++++++++-- src/lib/targets.ts | 38 ++++++ src/lib/upload-sessions.ts | 90 +++++++++++++ src/routes/upload-chunked.ts | 198 +++++++++++++++++++++++++++ src/routes/upload.ts | 38 +----- tests/storage.test.ts | 39 +++++- tests/upload-chunked.test.ts | 195 +++++++++++++++++++++++++++ 12 files changed, 887 insertions(+), 119 deletions(-) create mode 100644 src/lib/targets.ts create mode 100644 src/lib/upload-sessions.ts create mode 100644 src/routes/upload-chunked.ts create mode 100644 tests/upload-chunked.test.ts diff --git a/README.md b/README.md index f18f43d..e3015c0 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,11 @@ pouvant écrire les paquets de l'organisation. | `POST /auth/logout` | session | Ferme la session (+ déconnexion RP si disponible) | | `GET /auth/me` | non | `{ "authenticated": bool, "name"?: string }` | | `GET /targets` | session | Cibles : `[{ "name", "label", "allowDirectories" }]` (sans chemin) | -| `POST /upload?target=[&path=

]` | session | Dépôt d'un fichier `multipart/form-data`, champ `file` | +| `POST /upload?target=[&path=

]` | session | Dépôt direct d'un fichier `multipart/form-data`, champ `file` | +| `POST /upload/init` | session | Démarre un upload fragmenté → `{ id, chunkSize, chunkCount }` | +| `PUT /upload/:id/:index` | session | Envoie un morceau (corps brut `application/octet-stream`) | +| `POST /upload/:id/complete` | session | Assemble et finalise → `201` + métadonnées | +| `DELETE /upload/:id` | session | Abandonne (supprime le fichier partiel) | `POST /upload` — `?target=` choisit la cible (absent → **première** cible). `?path=` (URL-encodé) dépose dans un **sous-dossier** recréé sous @@ -148,17 +152,26 @@ La page gère aussi le **glisser-déposer d'un dossier entier** et un bouton > Le dépôt passe **uniquement par la session** établie via la page. Il n'y a pas > d'accès par jeton Bearer pour les scripts. -### Très gros fichiers +### Gros fichiers — upload fragmenté -`maxFileSizeMb` accepte de grandes valeurs (`10240` = 10 Go). Les fichiers sont -écrits en streaming (mémoire constante) et le serveur n'impose **aucun timeout** -de requête. Derrière un reverse proxy, penser à : +La page bascule automatiquement en **upload fragmenté** au-delà de **32 Mo** : +morceaux de 16 Mo, **3 en parallèle**, chaque morceau ré-essayé jusqu'à 3 fois. +Aucune configuration ; `maxFileSizeMb` reste la limite (`10240` = 10 Go). -- **nginx** : `client_max_body_size 11g;`, `proxy_request_buffering off;`, - `proxy_read_timeout`/`proxy_send_timeout` élevés -- **Caddy / Traefik** : limites de taille de corps et timeouts de lecture +- Écriture en place à l'offset du morceau, mémoire constante ; assemblage sans + seconde passe. Fichier partiel en `.mediadrop-part`, renommé à la + finalisation. +- Sessions d'upload en mémoire, purgées après **30 min** d'inactivité (le + `.part` est alors supprimé) ; 100 uploads simultanés au maximum. +- **Chaque requête reste petite (≤ 16 Mo)** : plus besoin de régler + `client_max_body_size` ni les timeouts du reverse proxy pour les gros fichiers. -Prévoir aussi l'espace disque sur le volume de la cible. +Le chemin **direct** (`POST /upload`, fichiers ≤ 32 Mo ou clients tiers) envoie +tout le corps d'un coup — le serveur n'impose alors aucun timeout, mais un +reverse proxy peut nécessiter un `client_max_body_size` adapté et +`proxy_request_buffering off`. + +Prévoir l'espace disque sur le volume de la cible. ## Configuration @@ -230,13 +243,16 @@ src/ errors.ts classes d'erreur (AppError + dérivées) error-handler.ts gestionnaire d'erreurs global routes/ - auth.ts GET /auth/login, /auth/callback, /auth/me ; POST /auth/logout - health.ts GET /health - upload.ts GET /targets ; POST /upload?target=... (session requise) + auth.ts GET /auth/login, /auth/callback, /auth/me ; POST /auth/logout + health.ts GET /health + upload.ts GET /targets ; POST /upload (dépôt direct) + upload-chunked.ts POST /upload/init, PUT /upload/:id/:index, .../complete, DELETE lib/ - storage.ts écriture disque dans la cible, nom de fichier sûr, plafond de taille - media-types.ts liste blanche des types + extensions - paths.ts assainissement des chemins relatifs (dépôt de dossier) + storage.ts écriture disque, nom sûr, plafond, morceaux (writeChunkAt) + upload-sessions.ts registre en mémoire des uploads fragmentés (+ purge TTL) + targets.ts résolution cible / chemin relatif (partagé par les deux routes) + media-types.ts liste blanche des types + extensions + jokers + paths.ts assainissement des chemins relatifs (dépôt de dossier) public/ index.html page de dépôt (login = lien vers /auth/login, sans build) ``` diff --git a/public/index.html b/public/index.html index f16372d..14c4af0 100644 --- a/public/index.html +++ b/public/index.html @@ -262,7 +262,10 @@ const queueCancel = document.getElementById('queueCancel'); const list = document.getElementById('list'); - const MAX_PARALLEL = 3; + const MAX_PARALLEL = 2; + const CHUNK_THRESHOLD = 32 * 1024 * 1024; // au-delà : upload fragmenté et reprenable + const CHUNK_PARALLEL = 3; + const CHUNK_RETRIES = 3; let authenticated = false; let currentTarget = null; @@ -417,106 +420,242 @@ updateQueueRow(); } - function startUpload(job) { - const { file, relativePath, ui } = job; - - const params = new URLSearchParams(); - if (currentTarget) params.set('target', currentTarget); - if (relativePath) params.set('path', relativePath); - const query = params.toString(); - - const form = new FormData(); - form.append('file', file); - - const xhr = new XMLHttpRequest(); - job.xhr = xhr; - active.add(job); - xhr.open('POST', query ? '/upload?' + query : '/upload'); - - let startedAt = 0; + function makeReporter(ui, total) { + let started = 0; let lastAt = 0; let lastLoaded = 0; let speed = 0; - - xhr.upload.addEventListener('progress', (event) => { - if (!event.lengthComputable) return; + return function report(loaded) { const now = performance.now(); - if (!startedAt) { - startedAt = now; + if (!started) { + started = now; lastAt = now; } const dt = (now - lastAt) / 1000; if (dt >= 0.25) { - const inst = (event.loaded - lastLoaded) / dt; + const inst = (loaded - lastLoaded) / dt; speed = speed ? speed * 0.7 + inst * 0.3 : inst; lastAt = now; - lastLoaded = event.loaded; + lastLoaded = loaded; } - ui.bar.style.width = (event.loaded / event.total) * 100 + '%'; - const pct = Math.floor((event.loaded / event.total) * 100); - const eta = speed > 0 ? formatEta((event.total - event.loaded) / speed) : ''; + const pct = total ? Math.floor((loaded / total) * 100) : 0; + ui.bar.style.width = pct + '%'; + const eta = speed > 0 && total ? formatEta((total - loaded) / speed) : ''; ui.progress.textContent = - formatSize(event.loaded) + + formatSize(loaded) + ' / ' + - formatSize(event.total) + + formatSize(total) + ' · ' + pct + ' %' + (speed > 0 ? ' · ' + formatSize(speed) + '/s' : '') + (eta ? ' · ' + eta : ''); - }); - - const finish = () => { - active.delete(job); - ui.cancelBtn.hidden = true; - pump(); }; + } + function uploadQuery(relativePath) { + const p = new URLSearchParams(); + if (currentTarget) p.set('target', currentTarget); + if (relativePath) p.set('path', relativePath); + const s = p.toString(); + return s ? '?' + s : ''; + } + + function finish(job) { + active.delete(job); + job.ui.cancelBtn.hidden = true; + pump(); + } + + function applyResult(job, status, body) { + const ui = job.ui; + if (status >= 200 && status < 300) { + ui.bar.style.width = '100%'; + ui.progress.textContent = formatSize(job.file.size); + setStatus(ui, 'Déposé : ' + (body.storedName || 'OK'), 'ok'); + } else if (status === 401) { + render({ authenticated: false }); + showBanner('Session expirée — reconnectez-vous.'); + setStatus(ui, 'Non authentifié', 'err'); + } else { + setStatus(ui, 'Erreur ' + status + ' : ' + (body.message || 'échec'), 'err'); + } + } + + function startUpload(job) { + active.add(job); + job.ui.cancelBtn.hidden = false; + job.ui.cancelBtn.onclick = () => job.cancel && job.cancel(); + setStatus(job.ui, 'Envoi…', 'muted'); + if (job.file.size > CHUNK_THRESHOLD) runChunked(job); + else runSimple(job); + } + + function runSimple(job) { + const { file, relativePath, ui } = job; + const report = makeReporter(ui, file.size); + const form = new FormData(); + form.append('file', file); + const xhr = new XMLHttpRequest(); + job.cancel = () => xhr.abort(); + xhr.open('POST', '/upload' + uploadQuery(relativePath)); + xhr.upload.addEventListener('progress', (e) => { + if (e.lengthComputable) report(e.loaded); + }); xhr.addEventListener('load', () => { - finish(); + finish(job); let body = {}; try { body = JSON.parse(xhr.responseText); } catch { body = {}; } - if (xhr.status >= 200 && xhr.status < 300) { - ui.bar.style.width = '100%'; - ui.progress.textContent = formatSize(file.size); - setStatus(ui, 'Déposé : ' + (body.storedName || 'OK'), 'ok'); - } else if (xhr.status === 401) { - render({ authenticated: false }); - showBanner('Session expirée — reconnectez-vous.'); - setStatus(ui, 'Non authentifié', 'err'); - } else { - setStatus(ui, 'Erreur ' + xhr.status + ' : ' + (body.message || 'échec'), 'err'); - } + applyResult(job, xhr.status, body); }); - xhr.addEventListener('error', () => { - finish(); + finish(job); setStatus(ui, 'Erreur réseau', 'err'); }); - xhr.addEventListener('abort', () => { - finish(); + finish(job); ui.progress.textContent = ''; setStatus(ui, 'Annulé', 'err'); }); - - ui.cancelBtn.hidden = false; - ui.cancelBtn.onclick = () => xhr.abort(); - setStatus(ui, 'Envoi…', 'muted'); xhr.send(form); } + async function runChunked(job) { + const { file, relativePath, ui } = job; + const report = makeReporter(ui, file.size); + const inflight = new Set(); + let aborted = false; + let failed = false; + let sessionId = null; + + const dropSession = () => { + if (sessionId) fetch('/upload/' + sessionId, { method: 'DELETE' }).catch(() => {}); + }; + job.cancel = () => { + aborted = true; + for (const x of inflight) x.abort(); + dropSession(); + finish(job); + ui.progress.textContent = ''; + setStatus(ui, 'Annulé', 'err'); + }; + + let meta; + try { + const res = await fetch('/upload/init' + uploadQuery(relativePath), { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + name: file.name, + size: file.size, + type: file.type || 'application/octet-stream', + }), + }); + meta = await res.json().catch(() => ({})); + if (!res.ok) { + finish(job); + applyResult(job, res.status, meta); + return; + } + } catch { + finish(job); + setStatus(ui, 'Erreur réseau (init)', 'err'); + return; + } + if (aborted) return; + sessionId = meta.id; + const chunkSize = meta.chunkSize; + const chunkCount = meta.chunkCount; + const loadedPer = new Array(chunkCount).fill(0); + const emit = () => report(loadedPer.reduce((a, b) => a + b, 0)); + let nextIndex = 0; + + const sendOnce = (index) => + new Promise((resolve) => { + const start = index * chunkSize; + const end = Math.min(start + chunkSize, file.size); + const xhr = new XMLHttpRequest(); + inflight.add(xhr); + xhr.open('PUT', '/upload/' + sessionId + '/' + index); + xhr.setRequestHeader('content-type', 'application/octet-stream'); + xhr.upload.addEventListener('progress', (e) => { + if (e.lengthComputable) { + loadedPer[index] = e.loaded; + emit(); + } + }); + const settle = (r) => { + inflight.delete(xhr); + resolve(r); + }; + xhr.addEventListener('load', () => { + if (xhr.status >= 200 && xhr.status < 300) { + loadedPer[index] = end - start; + emit(); + settle({ ok: true }); + } else { + settle({ ok: false, status: xhr.status }); + } + }); + xhr.addEventListener('error', () => settle({ ok: false, status: 0 })); + xhr.addEventListener('abort', () => settle({ ok: false, aborted: true })); + xhr.send(file.slice(start, end)); + }); + + async function worker() { + while (!aborted && !failed && nextIndex < chunkCount) { + const index = nextIndex; + nextIndex += 1; + let r = { ok: false }; + for (let attempt = 0; attempt <= CHUNK_RETRIES; attempt += 1) { + r = await sendOnce(index); + if (r.ok || r.aborted || r.status === 401) break; + await new Promise((res) => setTimeout(res, 400 * (attempt + 1))); + if (aborted) return; + } + if (r.ok) continue; + if (r.aborted) return; + failed = true; + if (r.status === 401) applyResult(job, 401, {}); + else + setStatus(ui, 'Erreur morceau ' + index + ' (' + (r.status || 'réseau') + ')', 'err'); + return; + } + } + + const workers = []; + for (let i = 0; i < Math.min(CHUNK_PARALLEL, chunkCount); i += 1) workers.push(worker()); + await Promise.all(workers); + + if (aborted) return; + if (failed) { + finish(job); + dropSession(); + return; + } + + try { + const res = await fetch('/upload/' + sessionId + '/complete', { method: 'POST' }); + const body = await res.json().catch(() => ({})); + finish(job); + applyResult(job, res.status, body); + } catch { + finish(job); + setStatus(ui, 'Erreur réseau (finalisation)', 'err'); + } + } + function cancelAll() { while (queue.length) { const job = queue.shift(); job.ui.cancelBtn.hidden = true; setStatus(job.ui, 'Annulé', 'err'); } - for (const job of Array.from(active)) job.xhr.abort(); + for (const job of Array.from(active)) if (job.cancel) job.cancel(); updateQueueRow(); } diff --git a/src/app.ts b/src/app.ts index bd4c5eb..00fb4d9 100644 --- a/src/app.ts +++ b/src/app.ts @@ -7,9 +7,11 @@ import secureSession from '@fastify/secure-session'; import type { AppConfig } from './config.js'; import { registerErrorHandler } from './error-handler.js'; -import { ensureUploadDir } from './lib/storage.js'; +import { ensureUploadDir, removeFile } from './lib/storage.js'; +import { createUploadSessionStore } from './lib/upload-sessions.js'; import { createOidcClient, type OidcClient } from './oidc.js'; import { authRoutes } from './routes/auth.js'; +import { chunkedUploadRoutes, CHUNK_BODY_LIMIT } from './routes/upload-chunked.js'; import { healthRoutes } from './routes/health.js'; import { uploadRoutes } from './routes/upload.js'; @@ -46,8 +48,25 @@ export async function buildApp( connectionTimeout: 0, }); + const uploadSessions = createUploadSessionStore((session) => { + void removeFile(session.partPath); + }); + app.decorate('appConfig', config); app.decorate('oidcClient', oidcClient); + app.decorate('uploadSessions', uploadSessions); + app.addHook('onClose', () => { + uploadSessions.stop(); + }); + + // Corps brut (bufferisé) pour les morceaux d'upload fragmenté. + app.addContentTypeParser( + 'application/octet-stream', + { parseAs: 'buffer', bodyLimit: CHUNK_BODY_LIMIT }, + (_request, body, taskDone) => { + taskDone(null, body); + }, + ); await app.register(secureSession, { key: config.session.key, @@ -81,6 +100,7 @@ export async function buildApp( await app.register(authRoutes); await app.register(healthRoutes); await app.register(uploadRoutes); + await app.register(chunkedUploadRoutes); return app; } diff --git a/src/errors.ts b/src/errors.ts index 2f07a72..b4ef790 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -49,3 +49,17 @@ export class PayloadTooLargeError extends AppError { super(message, 413); } } + +/** Session d'upload introuvable ou expirée. */ +export class UploadSessionNotFoundError extends AppError { + public constructor(message = "Session d'upload inconnue ou expirée") { + super(message, 404); + } +} + +/** Trop de sessions d'upload simultanées. */ +export class TooManyUploadsError extends AppError { + public constructor(message = "Trop d'uploads en cours, réessayez plus tard") { + super(message, 429); + } +} diff --git a/src/fastify.d.ts b/src/fastify.d.ts index 43cf8f4..2177f5a 100644 --- a/src/fastify.d.ts +++ b/src/fastify.d.ts @@ -1,6 +1,7 @@ import type { AppConfig } from './config.js'; import type { OidcClient } from './oidc.js'; import type { LoginTransaction, SessionUser } from './session.js'; +import type { UploadSessionStore } from './lib/upload-sessions.js'; declare module 'fastify' { interface FastifyInstance { @@ -8,6 +9,8 @@ declare module 'fastify' { appConfig: AppConfig; /** Façade OIDC (découverte effectuée), injectée par `buildApp`. */ oidcClient: OidcClient; + /** Registre des uploads fragmentés en cours. */ + uploadSessions: UploadSessionStore; } } diff --git a/src/lib/storage.ts b/src/lib/storage.ts index d9049dd..60ecce5 100644 --- a/src/lib/storage.ts +++ b/src/lib/storage.ts @@ -1,5 +1,5 @@ import { createWriteStream } from 'node:fs'; -import { mkdir, rm } from 'node:fs/promises'; +import { mkdir, open, rename, rm } from 'node:fs/promises'; import path from 'node:path'; import { randomUUID } from 'node:crypto'; import { Transform, type Readable } from 'node:stream'; @@ -67,7 +67,8 @@ function sanitizeExtension(originalName: string): string { return /^\.[a-z0-9]{1,12}$/.test(ext) ? ext : '.bin'; } -function buildStoredName(originalName: string, mimeType: string): string { +/** Nom de fichier unique dérivé côté serveur (`nom-timestamp-uuid.ext`). */ +export function buildStoredName(originalName: string, mimeType: string): string { const ext = extensionForMime(mimeType) ?? sanitizeExtension(originalName); return `${sanitizeBase(originalName)}-${Date.now()}-${randomUUID().slice(0, 8)}${ext}`; } @@ -92,21 +93,34 @@ function wasTruncated(stream: Readable): boolean { return (stream as { truncated?: unknown }).truncated === true; } -interface ResolvedDestination { +export interface ResolvedDestination { + /** Nom simple, ou chemin relatif `dossier/sous/fichier.ext`. */ storedName: string; + /** Chemin absolu du fichier final. */ destination: string; } -async function resolveDestination(params: SaveFileParams): Promise { - if (params.placement === undefined) { - const storedName = buildStoredName(params.originalName, params.mimeType); - return { storedName, destination: path.join(params.targetDir, storedName) }; +export interface DestinationInput { + targetDir: string; + originalName: string; + mimeType: string; + placement?: FilePlacement; +} + +/** + * Résout le chemin final (crée le sous-dossier au besoin) et garantit qu'il reste + * sous `targetDir`. + */ +export async function resolveDestination(input: DestinationInput): Promise { + if (input.placement === undefined) { + const storedName = buildStoredName(input.originalName, input.mimeType); + return { storedName, destination: path.join(input.targetDir, storedName) }; } - const { segments, fileName } = params.placement; - const directory = path.join(params.targetDir, ...segments); + const { segments, fileName } = input.placement; + const directory = path.join(input.targetDir, ...segments); - const root = path.resolve(params.targetDir); + const root = path.resolve(input.targetDir); const resolved = path.resolve(directory); if (resolved !== root && !resolved.startsWith(root + path.sep)) { throw new ValidationError('Chemin de dépôt hors de la cible'); @@ -162,3 +176,39 @@ export async function saveFile(params: SaveFileParams): Promise { return { storedName, originalName, size: bytesWritten, mimeType, path: destination }; } + +/** Crée (ou remet à zéro) un fichier vide, ainsi que son dossier parent. */ +export async function createEmptyFile(filePath: string): Promise { + await mkdir(path.dirname(filePath), { recursive: true }); + const handle = await open(filePath, 'w'); + await handle.close(); +} + +/** + * Écrit `data` dans `filePath` à partir de l'octet `offset` (le fichier doit + * exister), en une passe positionnée : sûr pour des écritures concurrentes sur + * des zones disjointes du même fichier. + */ +export async function writeChunkAt(params: { + data: Buffer; + filePath: string; + offset: number; +}): Promise { + const handle = await open(params.filePath, 'r+'); + try { + await handle.write(params.data, 0, params.data.length, params.offset); + } finally { + await handle.close(); + } + return params.data.length; +} + +/** Renomme le fichier partiel vers son chemin final (écrase si présent). */ +export async function finalizeFile(partPath: string, finalPath: string): Promise { + await rename(partPath, finalPath); +} + +/** Supprime un fichier s'il existe (jamais d'erreur si absent). */ +export async function removeFile(filePath: string): Promise { + await rm(filePath, { force: true }); +} diff --git a/src/lib/targets.ts b/src/lib/targets.ts new file mode 100644 index 0000000..6d3140d --- /dev/null +++ b/src/lib/targets.ts @@ -0,0 +1,38 @@ +import type { DropTarget, DropTargets } from '../config.js'; +import { ValidationError } from '../errors.js'; +import { parseRelativePath, type SafeRelativePath } from './paths.js'; + +/** Résout la cible demandée (`?target=`), ou la première par défaut ; 400 si inconnue. */ +export function resolveTarget(targets: DropTargets, requested: unknown): DropTarget { + if (requested === undefined || requested === '') { + return targets[0]; + } + if (typeof requested !== 'string') { + throw new ValidationError('Paramètre « target » invalide'); + } + const match = targets.find((target) => target.name === requested); + if (match === undefined) { + throw new ValidationError(`Cible de dépôt inconnue : ${requested}`); + } + return match; +} + +/** + * Interprète un chemin relatif client : `undefined` -> dépôt simple (nom dérivé) ; + * sinon dépôt de dossier, refusé (400) si la cible ne l'autorise pas. + */ +export function resolvePlacement( + target: DropTarget, + requested: unknown, +): SafeRelativePath | undefined { + if (requested === undefined || requested === '') { + return undefined; + } + if (typeof requested !== 'string') { + throw new ValidationError('Paramètre « path » invalide'); + } + if (!target.allowDirectories) { + throw new ValidationError(`La cible « ${target.name} » n'autorise pas le dépôt de dossiers`); + } + return parseRelativePath(requested); +} diff --git a/src/lib/upload-sessions.ts b/src/lib/upload-sessions.ts new file mode 100644 index 0000000..cf25ead --- /dev/null +++ b/src/lib/upload-sessions.ts @@ -0,0 +1,90 @@ +import { randomUUID } from 'node:crypto'; + +import type { DropTarget } from '../config.js'; +import { TooManyUploadsError, UploadSessionNotFoundError } from '../errors.js'; + +/** Un upload fragmenté en cours : morceaux écrits dans `partPath`, finalisé en `finalPath`. */ +export interface UploadSession { + id: string; + target: DropTarget; + /** Nom simple, ou chemin relatif `dossier/sous/fichier.ext` (mode dossier). */ + storedName: string; + mimeType: string; + partPath: string; + finalPath: string; + totalSize: number; + chunkSize: number; + chunkCount: number; + received: Set; + lastActivity: number; +} + +export type NewUploadSession = Omit; + +const IDLE_TTL_MS = 30 * 60_000; +const SWEEP_INTERVAL_MS = 60_000; +const MAX_SESSIONS = 100; + +export interface UploadSessionStore { + create(input: NewUploadSession): UploadSession; + /** Récupère et rafraîchit la session ; lève `UploadSessionNotFoundError` si absente. */ + use(id: string): UploadSession; + remove(id: string): UploadSession | undefined; + stop(): void; +} + +/** + * Registre en mémoire des uploads fragmentés. Les sessions inactives depuis plus + * de 30 min sont purgées et `onExpire` est appelé (pour supprimer le `.part`). + */ +export function createUploadSessionStore( + onExpire: (session: UploadSession) => void, +): UploadSessionStore { + const sessions = new Map(); + + const timer = setInterval(() => { + const now = Date.now(); + for (const session of sessions.values()) { + if (now - session.lastActivity > IDLE_TTL_MS) { + sessions.delete(session.id); + onExpire(session); + } + } + }, SWEEP_INTERVAL_MS); + timer.unref(); + + return { + create(input: NewUploadSession): UploadSession { + if (sessions.size >= MAX_SESSIONS) { + throw new TooManyUploadsError(); + } + const session: UploadSession = { + ...input, + id: randomUUID(), + received: new Set(), + lastActivity: Date.now(), + }; + sessions.set(session.id, session); + return session; + }, + + use(id: string): UploadSession { + const session = sessions.get(id); + if (session === undefined) { + throw new UploadSessionNotFoundError(); + } + session.lastActivity = Date.now(); + return session; + }, + + remove(id: string): UploadSession | undefined { + const session = sessions.get(id); + sessions.delete(id); + return session; + }, + + stop(): void { + clearInterval(timer); + }, + }; +} diff --git a/src/routes/upload-chunked.ts b/src/routes/upload-chunked.ts new file mode 100644 index 0000000..2aedd10 --- /dev/null +++ b/src/routes/upload-chunked.ts @@ -0,0 +1,198 @@ +import { stat } from 'node:fs/promises'; + +import type { FastifyPluginCallback, FastifyReply } from 'fastify'; + +import { PayloadTooLargeError, ValidationError } from '../errors.js'; +import { isAllowedMedia } from '../lib/media-types.js'; +import { + createEmptyFile, + finalizeFile, + removeFile, + resolveDestination, + writeChunkAt, +} from '../lib/storage.js'; +import { resolvePlacement, resolveTarget } from '../lib/targets.js'; +import { requireSession } from '../session.js'; + +const MIN_CHUNK = 1024 * 1024; +const MAX_CHUNK = 32 * 1024 * 1024; +const DEFAULT_CHUNK = 16 * 1024 * 1024; +const PART_SUFFIX = '.mediadrop-part'; + +/** Limite de corps pour un morceau (à appliquer au parser octet-stream). */ +export const CHUNK_BODY_LIMIT = MAX_CHUNK + 4096; + +interface InitBody { + target?: unknown; + path?: unknown; + name?: unknown; + size?: unknown; + type?: unknown; + chunkSize?: unknown; +} + +interface InitResponse { + id: string; + chunkSize: number; + chunkCount: number; +} + +interface ChunkAck { + received: number; + total: number; +} + +interface CompleteResponse { + target: string; + storedName: string; + size: number; + mimeType: string; +} + +function positiveInt(value: unknown, name: string): number { + if (typeof value !== 'number' || !Number.isInteger(value) || value <= 0) { + throw new ValidationError(`${name} doit être un entier positif`); + } + return value; +} + +function nonEmptyString(value: unknown, fallback: string): string { + return typeof value === 'string' && value.length > 0 ? value : fallback; +} + +function clampChunkSize(value: unknown): number { + if (value === undefined) { + return DEFAULT_CHUNK; + } + return Math.min(MAX_CHUNK, Math.max(MIN_CHUNK, positiveInt(value, 'chunkSize'))); +} + +/** + * Upload fragmenté et reprenable : + * - `POST /upload/init` crée une session, renvoie `{ id, chunkSize, chunkCount }` + * - `PUT /upload/:id/:index` envoie un morceau (corps brut octet-stream) + * - `POST /upload/:id/complete` assemble et finalise, renvoie 201 + métadonnées + * - `DELETE /upload/:id` abandonne (supprime le fichier partiel) + */ +export const chunkedUploadRoutes: FastifyPluginCallback = (app, _options, done) => { + const cfg = app.appConfig; + + app.post( + '/upload/init', + { preHandler: requireSession }, + async (request, reply): Promise => { + const body = (request.body ?? {}) as InitBody; + const target = resolveTarget(cfg.targets, body.target); + const placement = resolvePlacement(target, body.path); + + const mimeType = nonEmptyString(body.type, 'application/octet-stream'); + const totalSize = positiveInt(body.size, 'size'); + if (totalSize > cfg.maxFileSizeBytes) { + throw new PayloadTooLargeError(); + } + if (!isAllowedMedia(mimeType, cfg.allowedMimeTypes)) { + throw new ValidationError(`Type de média non autorisé : ${mimeType}`); + } + + const chunkSize = clampChunkSize(body.chunkSize); + const chunkCount = Math.ceil(totalSize / chunkSize); + + const { storedName, destination } = await resolveDestination({ + targetDir: target.dir, + originalName: nonEmptyString(body.name, 'fichier'), + mimeType, + ...(placement !== undefined ? { placement } : {}), + }); + const partPath = destination + PART_SUFFIX; + await createEmptyFile(partPath); + + const session = app.uploadSessions.create({ + target, + storedName, + mimeType, + partPath, + finalPath: destination, + totalSize, + chunkSize, + chunkCount, + }); + + reply.status(201); + return { id: session.id, chunkSize, chunkCount }; + }, + ); + + app.put( + '/upload/:id/:index', + { preHandler: requireSession, bodyLimit: CHUNK_BODY_LIMIT }, + async (request): Promise => { + const { id, index } = request.params as { id: string; index: string }; + const session = app.uploadSessions.use(id); + + const chunkIndex = Number(index); + if (!Number.isInteger(chunkIndex) || chunkIndex < 0 || chunkIndex >= session.chunkCount) { + throw new ValidationError('Index de morceau hors limites'); + } + + const data = request.body; + if (!Buffer.isBuffer(data)) { + throw new ValidationError('Corps du morceau attendu (application/octet-stream)'); + } + if (data.length > session.chunkSize) { + throw new PayloadTooLargeError('Morceau plus grand que la taille annoncée'); + } + + await writeChunkAt({ + data, + filePath: session.partPath, + offset: chunkIndex * session.chunkSize, + }); + session.received.add(chunkIndex); + + return { received: session.received.size, total: session.chunkCount }; + }, + ); + + app.post( + '/upload/:id/complete', + { preHandler: requireSession }, + async (request, reply): Promise => { + const { id } = request.params as { id: string }; + const session = app.uploadSessions.use(id); + + if (session.received.size !== session.chunkCount) { + throw new ValidationError('Upload incomplet : des morceaux manquent'); + } + const { size } = await stat(session.partPath); + if (size !== session.totalSize) { + throw new ValidationError('Taille finale incohérente'); + } + + await finalizeFile(session.partPath, session.finalPath); + app.uploadSessions.remove(id); + + reply.status(201); + return { + target: session.target.name, + storedName: session.storedName, + size: session.totalSize, + mimeType: session.mimeType, + }; + }, + ); + + app.delete( + '/upload/:id', + { preHandler: requireSession }, + async (request, reply): Promise => { + const { id } = request.params as { id: string }; + const session = app.uploadSessions.remove(id); + if (session !== undefined) { + await removeFile(session.partPath); + } + return reply.status(204).send(); + }, + ); + + done(); +}; diff --git a/src/routes/upload.ts b/src/routes/upload.ts index 61748d8..2d970d1 100644 --- a/src/routes/upload.ts +++ b/src/routes/upload.ts @@ -1,11 +1,11 @@ import type { FastifyPluginCallback } from 'fastify'; -import type { DropTarget, DropTargets } from '../config.js'; +import type { DropTarget } from '../config.js'; import { ValidationError } from '../errors.js'; import { isAllowedMedia } from '../lib/media-types.js'; -import { parseRelativePath } from '../lib/paths.js'; +import { resolvePlacement, resolveTarget } from '../lib/targets.js'; import { requireSession } from '../session.js'; -import { saveFile, type FilePlacement, type StoredFile } from '../lib/storage.js'; +import { saveFile, type StoredFile } from '../lib/storage.js'; const FIELD_NAME = 'file'; @@ -28,38 +28,6 @@ interface UploadQuery { path?: unknown; } -/** Résout la cible demandée (`?target=`), ou la première par défaut ; 400 si inconnue. */ -function resolveTarget(targets: DropTargets, requested: unknown): DropTarget { - if (requested === undefined || requested === '') { - return targets[0]; - } - if (typeof requested !== 'string') { - throw new ValidationError('Paramètre « target » invalide'); - } - const match = targets.find((target) => target.name === requested); - if (match === undefined) { - throw new ValidationError(`Cible de dépôt inconnue : ${requested}`); - } - return match; -} - -/** - * Interprète `?path=` : `undefined` -> dépôt simple (nom dérivé) ; sinon dépôt de - * dossier, refusé (400) si la cible ne l'autorise pas. - */ -function resolvePlacement(target: DropTarget, requested: unknown): FilePlacement | undefined { - if (requested === undefined || requested === '') { - return undefined; - } - if (typeof requested !== 'string') { - throw new ValidationError('Paramètre « path » invalide'); - } - if (!target.allowDirectories) { - throw new ValidationError(`La cible « ${target.name} » n'autorise pas le dépôt de dossiers`); - } - return parseRelativePath(requested); -} - function toResponse(target: DropTarget, stored: StoredFile): UploadResponse { return { target: target.name, diff --git a/tests/storage.test.ts b/tests/storage.test.ts index 50a85a2..c9adcd2 100644 --- a/tests/storage.test.ts +++ b/tests/storage.test.ts @@ -1,9 +1,16 @@ import { readFile, readdir } from 'node:fs/promises'; import path from 'node:path'; +import { randomBytes } from 'node:crypto'; import { Readable } from 'node:stream'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { saveFile } from '../src/lib/storage.js'; +import { + createEmptyFile, + finalizeFile, + removeFile, + saveFile, + writeChunkAt, +} from '../src/lib/storage.js'; import { PayloadTooLargeError } from '../src/errors.js'; import { makeTempDir, removeDir, TINY_PNG } from './helpers.js'; @@ -104,3 +111,33 @@ describe('saveFile — mode dossier (placement)', () => { expect(await readdir(path.join(uploadDir, 'dir'))).toEqual(['a.bin']); }); }); + +describe('writeChunkAt — réassemblage de morceaux', () => { + it('écrit des morceaux à des offsets disjoints puis finalise, octets exacts', async () => { + const chunkSize = 64 * 1024; + const data = randomBytes(chunkSize * 2 + 777); + const partPath = path.join(uploadDir, 'sub', 'video.bin.part'); + await createEmptyFile(partPath); + + // Dans le désordre pour vérifier l'indépendance des écritures positionnées. + for (const index of [2, 0, 1]) { + const slice = data.subarray(index * chunkSize, (index + 1) * chunkSize); + const written = await writeChunkAt({ + data: slice, + filePath: partPath, + offset: index * chunkSize, + }); + expect(written).toBe(slice.length); + } + + const finalPath = path.join(uploadDir, 'sub', 'video.bin'); + await finalizeFile(partPath, finalPath); + + expect(await readFile(finalPath)).toEqual(data); + expect(await readdir(path.join(uploadDir, 'sub'))).toEqual(['video.bin']); + }); + + it('removeFile ne lève pas si le fichier est absent', async () => { + await expect(removeFile(path.join(uploadDir, 'absent'))).resolves.toBeUndefined(); + }); +}); diff --git a/tests/upload-chunked.test.ts b/tests/upload-chunked.test.ts new file mode 100644 index 0000000..4ea1660 --- /dev/null +++ b/tests/upload-chunked.test.ts @@ -0,0 +1,195 @@ +import { readFile, readdir } from 'node:fs/promises'; +import { join } from 'node:path'; +import { randomBytes } from 'node:crypto'; + +import type { FastifyInstance } from 'fastify'; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from 'vitest'; + +import { buildApp } from '../src/app.js'; +import type { OidcConfig } from '../src/config.js'; +import { makeConfig, makeTempDir, removeDir, startStubIdp, type StubIdp } from './helpers.js'; + +const MIB = 1024 * 1024; + +let idp: StubIdp; +let cookie: string; +let app: FastifyInstance; +let baseUrl: string; +let dir: string; + +function oidcOverride(ref: StubIdp): OidcConfig { + return { + issuer: ref.issuer, + clientId: ref.clientId, + clientSecret: ref.clientSecret, + scopes: ['openid', 'profile', 'email'], + }; +} + +function sessionCookie(res: Response): string { + return (res.headers.get('set-cookie') ?? '').split(';')[0] ?? ''; +} + +/** Déroule la connexion OIDC via de vraies requêtes HTTP ; renvoie l'en-tête Cookie. */ +async function loginViaHttp(): Promise { + const start = await fetch(`${baseUrl}/auth/login`, { redirect: 'manual' }); + const txn = sessionCookie(start); + const authUrl = new URL(start.headers.get('location') ?? ''); + idp.setNextIdToken({ nonce: authUrl.searchParams.get('nonce') ?? '' }); + const state = authUrl.searchParams.get('state') ?? ''; + const cb = await fetch(`${baseUrl}/auth/callback?code=x&state=${encodeURIComponent(state)}`, { + redirect: 'manual', + headers: { cookie: txn }, + }); + return sessionCookie(cb); +} + +beforeAll(async () => { + idp = await startStubIdp(); +}); + +afterAll(async () => { + await idp.close(); +}); + +beforeEach(async () => { + dir = await makeTempDir(); + app = await buildApp( + makeConfig({ + targets: [ + { name: 'flat', label: 'Plat', dir, allowDirectories: false }, + { name: 'tree', label: 'Arbo', dir, allowDirectories: true }, + ], + maxFileSizeBytes: 8 * MIB, + allowedMimeTypes: ['video/*', 'image/png'], + oidc: oidcOverride(idp), + }), + { logger: false }, + ); + await app.listen({ port: 0, host: '127.0.0.1' }); + const address = app.server.address(); + if (address === null || typeof address === 'string') throw new Error('adresse inattendue'); + baseUrl = `http://127.0.0.1:${String(address.port)}`; + cookie = await loginViaHttp(); +}); + +afterEach(async () => { + await app.close(); + await removeDir(dir); +}); + +function initBody(over: Record = {}): Record { + return { target: 'flat', name: 'clip.mp4', size: 3, type: 'video/mp4', ...over }; +} + +interface InitResult { + status: number; + body: { id?: string; chunkSize?: number; chunkCount?: number; message?: string }; +} + +async function init(over: Record = {}, withCookie = true): Promise { + const res = await fetch(`${baseUrl}/upload/init`, { + method: 'POST', + headers: { 'content-type': 'application/json', ...(withCookie ? { cookie } : {}) }, + body: JSON.stringify(initBody(over)), + }); + return { status: res.status, body: (await res.json().catch(() => ({}))) as InitResult['body'] }; +} + +async function putChunk(id: string, index: number, data: Buffer): Promise { + const res = await fetch(`${baseUrl}/upload/${id}/${String(index)}`, { + method: 'PUT', + headers: { 'content-type': 'application/octet-stream', cookie }, + body: new Uint8Array(data), + }); + return res.status; +} + +async function complete(id: string): Promise<{ status: number; body: unknown }> { + const res = await fetch(`${baseUrl}/upload/${id}/complete`, { + method: 'POST', + headers: { cookie }, + }); + return { status: res.status, body: await res.json().catch(() => ({})) }; +} + +describe('upload fragmenté', () => { + it('init -> morceau -> complete : le fichier est finalisé au bon endroit', async () => { + const data = randomBytes(200 * 1024); + const started = await init({ + size: data.length, + chunkSize: MIB, + path: 'a/b/clip.mp4', + target: 'tree', + }); + expect(started.status).toBe(201); + const { id, chunkCount } = started.body; + expect(chunkCount).toBe(1); + + const ack = await fetch(`${baseUrl}/upload/${id!}/0`, { + method: 'PUT', + headers: { 'content-type': 'application/octet-stream', cookie }, + body: new Uint8Array(data), + }); + expect(ack.status).toBe(200); + expect(await ack.json()).toEqual({ received: 1, total: 1 }); + + const done = await complete(id!); + expect(done.status).toBe(201); + expect(done.body).toMatchObject({ storedName: 'a/b/clip.mp4', size: data.length }); + + expect(await readFile(join(dir, 'a', 'b', 'clip.mp4'))).toEqual(data); + expect(await readdir(join(dir, 'a', 'b'))).toEqual(['clip.mp4']); + }); + + it('init refuse une taille au-dessus de la limite (413)', async () => { + expect((await init({ size: 9 * MIB })).status).toBe(413); + }); + + it('init refuse un type non autorisé (400)', async () => { + expect((await init({ type: 'application/zip' })).status).toBe(400); + }); + + it('init refuse un chemin sur une cible sans allowDirectories (400)', async () => { + expect((await init({ target: 'flat', path: 'x/clip.mp4' })).status).toBe(400); + }); + + it('init exige une session (401)', async () => { + expect((await init({}, false)).status).toBe(401); + }); + + it('PUT refuse un index hors limites (400)', async () => { + const { body } = await init({ size: MIB, chunkSize: MIB }); + expect(await putChunk(body.id!, 5, Buffer.alloc(10))).toBe(400); + }); + + it('complete refuse un upload incomplet (400) et ne finalise rien', async () => { + const data = randomBytes(2 * MIB + 10); + const { body } = await init({ size: data.length, chunkSize: MIB }); + expect(body.chunkCount).toBe(3); + expect(await putChunk(body.id!, 0, data.subarray(0, MIB))).toBe(200); + + expect((await complete(body.id!)).status).toBe(400); + const files = await readdir(dir); + expect(files).toHaveLength(1); + expect(files[0]).toMatch(/\.mediadrop-part$/); + }); + + it('404 sur une session inconnue', async () => { + expect(await putChunk('inexistante', 0, Buffer.alloc(10))).toBe(404); + expect((await complete('inexistante')).status).toBe(404); + }); + + it('DELETE abandonne : le fichier partiel disparaît, la session aussi', async () => { + const { body } = await init({ size: MIB, chunkSize: MIB }); + expect(await putChunk(body.id!, 0, Buffer.alloc(1000))).toBe(200); + + const del = await fetch(`${baseUrl}/upload/${body.id!}`, { + method: 'DELETE', + headers: { cookie }, + }); + expect(del.status).toBe(204); + expect(await readdir(dir)).toHaveLength(0); + expect(await putChunk(body.id!, 0, Buffer.alloc(10))).toBe(404); + }); +});