import crypto from 'node:crypto'; import fs from 'node:fs'; import fsp from 'node:fs/promises'; import path from 'node:path'; import { Transform } from 'node:stream'; import { pipeline } from 'node:stream/promises'; import { HttpError, badRequest, conflict, notFound } from './http-error.js'; export const MAX_CLIPS = 200; export const MAX_CLIP_BYTES = 4 * 1024 ** 3; export const MAX_CHUNK_BYTES = 16 * 1024 * 1024; const ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; const TIMESTAMP_PATTERN = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$/; const DATE_PATTERN = /^\d{4}-\d{2}-\d{2}$/; const VIDEO_EXTENSIONS = new Set(['mp4', 'mov', 'm4v', '3gp', 'avi', 'mkv', 'webm']); function extensionOf(name) { return name.split('.').pop().toLowerCase(); } function optionalString(value, maxLength) { if (value === null || value === undefined) return null; if (typeof value !== 'string') throw badRequest('Ungültige Angabe.'); return value.slice(0, maxLength); } function validateClip(input, index) { const file = typeof input?.file === 'string' ? input.file : ''; if (!file || file.length > 255 || !VIDEO_EXTENSIONS.has(extensionOf(file))) { throw badRequest(`Clip ${index + 1}: kein unterstütztes Video.`); } if (!Number.isSafeInteger(input.size) || input.size < 1 || input.size > MAX_CLIP_BYTES) { throw badRequest(`Clip ${index + 1}: ungültige Dateigröße.`); } if (typeof input.timestamp !== 'string' || !TIMESTAMP_PATTERN.test(input.timestamp)) { throw badRequest(`Clip ${index + 1}: ungültiger Zeitstempel.`); } return { index, file, size: input.size, timestamp: input.timestamp, sender: optionalString(input.sender, 200), caption: optionalString(input.caption, 2000) ?? '', status: 'pending', }; } function validateRange(range) { const from = range?.from ?? null; const to = range?.to ?? null; for (const value of [from, to]) { if (value !== null && (typeof value !== 'string' || !DATE_PATTERN.test(value))) { throw badRequest('Ungültiger Zeitraum.'); } } return { from, to }; } // Bricht ab, sobald mehr Bytes kommen als erlaubt. function byteLimit(limit) { let count = 0; return new Transform({ transform(chunk, encoding, callback) { count += chunk.length; if (count > limit) return callback(new HttpError(413, 'Zu viele Daten.')); callback(null, chunk); }, }); } async function fileSize(file) { try { return (await fsp.stat(file)).size; } catch (error) { if (error.code === 'ENOENT') return 0; throw error; } } export function createProjectStore({ dataDir, probe }) { const root = path.join(dataDir, 'projects'); const queues = new Map(); const busyClips = new Set(); function projectDir(id) { if (typeof id !== 'string' || !ID_PATTERN.test(id)) throw notFound(); return path.join(root, id); } const projectFile = (id) => path.join(projectDir(id), 'project.json'); const clipFile = (id, clip) => path.join(projectDir(id), 'clips', `${clip.index}.${extensionOf(clip.file)}`); async function read(id) { try { return JSON.parse(await fsp.readFile(projectFile(id), 'utf8')); } catch (error) { if (error.code === 'ENOENT') throw notFound(); throw error; } } async function write(project) { const file = projectFile(project.id); const temp = `${file}.tmp`; await fsp.writeFile(temp, JSON.stringify(project, null, 2)); await fsp.rename(temp, file); } // Änderungen an einem Projekt laufen nacheinander, damit keine verloren geht. function update(id, change) { const previous = queues.get(id) ?? Promise.resolve(); const next = previous.then(async () => { const project = await read(id); await change(project); project.updatedAt = new Date().toISOString(); await write(project); return project; }); queues.set(id, next.catch(() => {})); return next; } function getClip(project, index) { const clip = project.clips[Number(index)]; if (!/^\d+$/.test(String(index)) || !clip) throw notFound('Clip nicht gefunden.'); return clip; } async function withProgress(project) { const clips = await Promise.all( project.clips.map(async (clip) => ({ ...clip, received: clip.status === 'pending' ? await fileSize(`${clipFile(project.id, clip)}.part`) : clip.size, })), ); return { ...project, clips }; } return { async create({ range, clips }) { if (!Array.isArray(clips) || clips.length === 0) throw badRequest('Keine Clips ausgewählt.'); if (clips.length > MAX_CLIPS) throw badRequest(`Höchstens ${MAX_CLIPS} Clips.`); const now = new Date().toISOString(); const project = { id: crypto.randomUUID(), createdAt: now, updatedAt: now, range: validateRange(range), clips: clips.map(validateClip), }; await fsp.mkdir(path.join(projectDir(project.id), 'clips'), { recursive: true }); await write(project); return withProgress(project); }, async get(id) { return withProgress(await read(id)); }, async list() { let entries = []; try { entries = await fsp.readdir(root); } catch (error) { if (error.code !== 'ENOENT') throw error; } const projects = []; for (const entry of entries) { if (!ID_PATTERN.test(entry)) continue; try { const project = await read(entry); projects.push({ id: project.id, createdAt: project.createdAt, range: project.range, clipCount: project.clips.length, readyCount: project.clips.filter((clip) => clip.status === 'ready').length, }); } catch { // Unvollständig angelegte Projekte überspringen } } return projects.sort((a, b) => b.createdAt.localeCompare(a.createdAt)); }, async remove(id) { const dir = projectDir(id); await read(id); await fsp.rm(dir, { recursive: true, force: true }); }, clipPath(project, index) { const clip = getClip(project, index); if (clip.status !== 'ready') throw notFound('Clip ist noch nicht hochgeladen.'); return clipFile(project.id, clip); }, // Hängt ein Stück an. Die Größe der Teildatei ist der maßgebliche Stand: // bricht eine Übertragung ab, bleibt ein gültiger Anfang liegen. async appendChunk(id, index, offset, stream) { const project = await read(id); const clip = getClip(project, index); if (clip.status !== 'pending') return { received: clip.size, status: clip.status }; const key = `${id}/${clip.index}`; if (busyClips.has(key)) throw conflict('Upload läuft bereits.', { received: null }); busyClips.add(key); try { const final = clipFile(id, clip); const part = `${final}.part`; const received = await fileSize(part); if (offset !== received) throw conflict('Upload-Stand weicht ab.', { received }); // Zu viel Gesendetes wird gar nicht erst geschrieben; der Anfang bleibt gültig. const allowed = Math.min(MAX_CHUNK_BYTES, clip.size - received); await pipeline(stream, byteLimit(allowed), fs.createWriteStream(part, { flags: 'a' })); const total = await fileSize(part); if (total < clip.size) return { received: total, status: 'pending' }; await fsp.rename(part, final); let media = null; let problem = null; try { media = await probe(final); } catch (error) { problem = error.message; } const updated = await update(id, (current) => { const target = current.clips[clip.index]; target.status = media ? 'ready' : 'error'; target.media = media; if (problem) target.error = problem; }); return { received: total, status: updated.clips[clip.index].status }; } finally { busyClips.delete(key); } }, }; }