patx/youtube-clone

import { createWriteStream } from 'node:fs';
import { unlink } from 'node:fs/promises';
import { randomBytes } from 'node:crypto';
import { join, extname } from 'node:path';
import { HttpError } from './http.js';

const DASH_DASH = Buffer.from('--');
const CRLF = Buffer.from('\r\n');
const HEADER_END = Buffer.from('\r\n\r\n');

/**
 * Streaming multipart/form-data parser.
 *
 * Text fields are collected in memory; any part with a filename is streamed
 * straight to `dir` so a large upload never has to be buffered whole.
 * Returns { fields, files } where each file is
 * { field, filename, contentType, path, storedName, size }.
 */
export async function parseMultipart(req, { dir, maxFileBytes = 512 * 1024 * 1024, maxFieldBytes = 64 * 1024 }) {
  const type = req.headers['content-type'] || '';
  const match = /boundary=(?:"([^"]+)"|([^;]+))/i.exec(type);
  if (!/^multipart\/form-data/i.test(type) || !match) {
    throw new HttpError(400, 'Expected a multipart/form-data upload.');
  }
  const boundary = Buffer.from(`--${(match[1] || match[2]).trim()}`);

  const fields = Object.create(null);
  const files = [];
  const cleanup = async () => {
    await Promise.all(files.map((f) => unlink(f.path).catch(() => {})));
  };

  let buf = Buffer.alloc(0);
  let state = 'preamble'; // preamble -> headers -> body
  let part = null;
  let finished = false;

  const closePart = async (trailing) => {
    if (!part) return;
    if (part.file) {
      part.file.size += trailing.length;
      if (part.file.size > maxFileBytes) throw new HttpError(413, 'That file is too large.');
      await new Promise((resolve, reject) => {
        part.stream.end(trailing, (err) => (err ? reject(err) : resolve()));
      });
    } else {
      part.chunks.push(trailing);
      const value = Buffer.concat(part.chunks).toString('utf8');
      if (value.length > maxFieldBytes) throw new HttpError(413, 'A form field was too large.');
      fields[part.name] = value;
    }
    part = null;
  };

  try {
    for await (const chunk of req) {
      buf = buf.length ? Buffer.concat([buf, chunk]) : chunk;

      // Keep consuming complete structures out of the buffer.
      for (;;) {
        if (finished) break;

        if (state === 'preamble' || state === 'headers') {
          if (state === 'preamble') {
            const at = buf.indexOf(boundary);
            if (at < 0) break;
            buf = buf.subarray(at + boundary.length);
            if (buf.length < 2) break;
            if (buf.subarray(0, 2).equals(DASH_DASH)) { finished = true; break; }
            if (!buf.subarray(0, 2).equals(CRLF)) break;
            buf = buf.subarray(2);
            state = 'headers';
          }
          const end = buf.indexOf(HEADER_END);
          if (end < 0) break;
          const headerText = buf.subarray(0, end).toString('utf8');
          buf = buf.subarray(end + HEADER_END.length);
          part = startPart(headerText, dir);
          if (part.file) files.push(part.file);
          state = 'body';
          continue;
        }

        // state === 'body': flush everything that cannot contain the boundary.
        const at = buf.indexOf(boundary);
        if (at < 0) {
          const keep = boundary.length + 4; // room for a split boundary + CRLF
          if (buf.length > keep) {
            await writeChunk(part, buf.subarray(0, buf.length - keep), maxFileBytes);
            buf = buf.subarray(buf.length - keep);
          }
          break;
        }
        // A body ends with the CRLF that precedes its boundary line.
        const bodyEnd = at >= 2 && buf.subarray(at - 2, at).equals(CRLF) ? at - 2 : at;
        await closePart(buf.subarray(0, bodyEnd));
        buf = buf.subarray(at + boundary.length);
        if (buf.length >= 2 && buf.subarray(0, 2).equals(DASH_DASH)) { finished = true; break; }
        state = 'preamble';
        buf = Buffer.concat([boundary, buf]); // re-scan this same boundary as a delimiter
      }
    }

    if (part) await closePart(Buffer.alloc(0));
    return { fields, files, cleanup };
  } catch (err) {
    if (part?.stream) part.stream.destroy();
    await cleanup();
    throw err;
  }
}

function startPart(headerText, dir) {
  const headers = Object.create(null);
  for (const line of headerText.split('\r\n')) {
    const i = line.indexOf(':');
    if (i > 0) headers[line.slice(0, i).toLowerCase().trim()] = line.slice(i + 1).trim();
  }
  const disposition = headers['content-disposition'] || '';
  const name = /name="([^"]*)"/i.exec(disposition)?.[1] ?? '';
  const filename = /filename="([^"]*)"/i.exec(disposition)?.[1];

  if (filename === undefined || filename === '') {
    return { name, chunks: [], file: null };
  }
  const ext = extname(filename).slice(0, 10).replace(/[^A-Za-z0-9.]/g, '') || '.bin';
  const storedName = `${Date.now().toString(36)}-${randomBytes(6).toString('hex')}${ext}`;
  const path = join(dir, storedName);
  const file = {
    field: name,
    filename,
    contentType: headers['content-type'] || 'application/octet-stream',
    path,
    storedName,
    size: 0,
  };
  return { name, file, stream: createWriteStream(path) };
}

async function writeChunk(part, data, maxFileBytes) {
  if (!part) return;
  if (!part.file) { part.chunks.push(data); return; }
  part.file.size += data.length;
  if (part.file.size > maxFileBytes) throw new HttpError(413, 'That file is too large.');
  if (!part.stream.write(data)) {
    await new Promise((resolve) => part.stream.once('drain', resolve));
  }
}