files/nextjs/src/lib/multipart.js

import fs from 'node:fs';
import { Readable, Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

import Busboy from 'busboy';

export async function streamMultipartToPath(request, options) {
  const { fileFieldName = 'file', targetPath, maxFileBytes = 0 } = options;

  if (!request.body) {
    throw new Error('missing-body');
  }

  return new Promise((resolve, reject) => {
    const limits = { files: 1, fields: 20 };
    if (maxFileBytes > 0) {
      limits.fileSize = maxFileBytes;
    }

    const parser = Busboy({ headers: Object.fromEntries(request.headers), limits });
    const fields = {};
    let uploadInfo = null;
    let writeStream = null;
    let writeFinished = false;
    let writeError = null;
    let hitSizeLimit = false;

    function fail(error) {
      if (writeStream) {
        writeStream.destroy();
      }
      fs.promises.rm(targetPath, { force: true }).catch(() => undefined).finally(() => reject(error));
    }

    parser.on('field', (name, value) => {
      fields[name] = value;
    });

    parser.on('file', (name, file, info) => {
      if (name !== fileFieldName || uploadInfo) {
        file.resume();
        return;
      }

      writeStream = fs.createWriteStream(targetPath, { flags: 'wx' });
      uploadInfo = {
        filename: info.filename || '',
        mimeType: info.mimeType || '',
        sizeBytes: 0,
      };

      file.on('data', (chunk) => {
        uploadInfo.sizeBytes += chunk.length;
      });

      file.on('limit', () => {
        hitSizeLimit = true;
      });

      file.on('error', (error) => {
        writeError = error;
      });

      writeStream.on('error', (error) => {
        writeError = error;
      });

      writeStream.on('close', () => {
        writeFinished = true;
      });

      file.pipe(writeStream);
    });

    parser.on('error', (error) => {
      fail(error);
    });

    parser.on('finish', () => {
      const complete = () => {
        if (writeError) {
          fail(writeError);
          return;
        }
        if (hitSizeLimit) {
          fail(new Error('file-too-large'));
          return;
        }
        if (!uploadInfo) {
          fail(new Error('missing-file'));
          return;
        }
        resolve({ fields, file: uploadInfo });
      };

      if (!writeStream) {
        complete();
        return;
      }

      if (writeFinished) {
        complete();
        return;
      }

      writeStream.on('close', complete);
    });

    Readable.fromWeb(request.body).pipe(parser);
  });
}

export async function streamBodyToPath(request, options) {
  const { targetPath, maxBytes = 0 } = options;

  if (!request.body) {
    throw new Error('missing-body');
  }

  let sizeBytes = 0;
  const counter = new Transform({
    transform(chunk, encoding, callback) {
      sizeBytes += chunk.length;

      if (maxBytes > 0 && sizeBytes > maxBytes) {
        callback(new Error('file-too-large'));
        return;
      }

      callback(null, chunk);
    },
  });

  try {
    await pipeline(
      Readable.fromWeb(request.body),
      counter,
      fs.createWriteStream(targetPath, { flags: 'wx' })
    );
  } catch (error) {
    await fs.promises.rm(targetPath, { force: true }).catch(() => undefined);
    throw error;
  }

  if (sizeBytes <= 0) {
    await fs.promises.rm(targetPath, { force: true }).catch(() => undefined);
    throw new Error('missing-file');
  }

  return { sizeBytes };
}