29494915195

Committed 16 Jul 2026 11:33AM UTC coverage: 83.024% (+0.4%) from 82.653%

Build # 29494915195

Build Type

push

github

Committed by

erayhanoglu

Commit Message

1.29.1

Coverage Stats

4118 of 5238 branches covered (78.62%)

Branch coverage included in aggregate %.

34348 of 41093 relevant lines covered (83.59%)

239.47 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.77

/packages/http/src/impl/multipart-reader.ts

import fs from 'node:fs';
import os from 'node:os';
import typeIs from '@browsery/type-is';
import { BadRequestError, HttpMediaType } from '@opra/common';
import fsPromise from 'fs/promises';
import * as MP from 'multipasta/node';
import { AsyncEventEmitter } from 'node-events-async';
import type { StrictOmit } from 'ts-gems';
import { isNotNullish, ValidationError } from 'valgen';
import type { HttpRequest } from '../interfaces/http-request.interface.js';
import { LocalFile } from './local-file.js';

/**
 * MultipartReader is a drop-in replacement for MultipartReader that uses
 * `multipasta` instead of `busboy`. The key advantage: part-level headers
 * (e.g. `Request-Id`) are fully exposed via `item.headers`.
 */
export class MultipartReader extends AsyncEventEmitter {
  protected _started = false;
  protected _streamClosed = false;
  protected _finished = false;
  protected _cancelled = false;
  protected _pendingWrites = 0;
  protected _stream: MP.MultipastaStream;
  protected _items: MultipartReader.Item[] = [];
  protected _stack: MultipartReader.Item[] = [];
  protected tempDirectory: string;
  scope?: string;

  constructor(
    protected request: HttpRequest,
    options?: MultipartReader.Options,
    protected mediaType?: HttpMediaType,
  ) {
    super();
    this.setMaxListeners(1000);
    this.tempDirectory = options?.tempDirectory || os.tmpdir();
    this.scope = options?.scope;

    const stream = MP.make({
      headers: request.headers as Record<string, string>,
      isFile: options?.isFile,
    });
    this._stream = stream;

    stream.once('error', (e: any) => {
      this._cancelled = true;
      this._finished = true;
      if (this.listenerCount('error') > 0) this.emit('error', e);
    });

    // Stream closed means the parser is done, but file writes may still be in
    // progress. _finished is set only when stream closes AND all writes complete.
    stream.on('close', () => {
      this._streamClosed = true;
      this._checkFinished();
    });

    stream.on('field', (part: MP.Field) => {
      const headers = _flattenHeaders(part.info.headers);
      const item: MultipartReader.Field = {
        kind: 'field',
        field: part.info.name,
        value: MP.decodeField(part.info, part.value),
        mimeType: part.info.contentType,
        encoding: part.info.contentTypeParameters['charset'],
        headers,
      };
      this._items.push(item);
      this._stack.push(item);
      this.emit('item', item);
    });

    stream.on('file', (file: MP.FileStream) => {
      const saveTo = LocalFile.tempFilename(
        file.info.filename ?? file.info.name,
        this.tempDirectory,
      );
      const writeStream = fs.createWriteStream(saveTo);
      file.pipe(writeStream);

      // Build a "ready" promise that resolves when the write is fully flushed.
      // buffer() / text() await this so callers always read a complete file.
      const ready = new Promise<void>(resolve =>
        writeStream.once('finish', resolve),
      );

      // Emit the item immediately (preserves part order) — the write may
      // still be in progress, but reading is deferred via ready.
      const headers = _flattenHeaders(file.info.headers);
      const item = new MultipartFile(file.info.name, saveTo, ready, {
        filename: file.info.filename ?? file.info.name,
        type: file.info.contentType,
        encoding: (file.info.contentTypeParameters['charset'] ??
          'utf-8') as BufferEncoding,
        autoDelete: true,
        headers,
      });
      this._items.push(item);
      this._stack.push(item);
      this.emit('item', item);

      this._pendingWrites++;
      ready.then(() => {
        this._pendingWrites--;
        this._checkFinished();
      });
    });
  }

  protected _checkFinished() {
    if (this._streamClosed && this._pendingWrites === 0) {
      this._finished = true;
      this.emit('_finished');
    }
  }

  get items(): MultipartReader.Item[] {
    return this._items;
  }

  /**
   * Retrieves the next item (field or file) from the multipart stream.
   */
  async getNext(): Promise<MultipartReader.Item | undefined> {
    let item = this._stack.shift();
    if (!item && !this._finished) {
      this.resume();
      item = await new Promise<any>((resolve, reject) => {
        let resolved = false;
        if (this._stack.length) return resolve(this._stack.shift());
        if (this._finished) return resolve(this._stack.shift());
        const onDone = () => {
          if (resolved) return;
          resolved = true;
          resolve(this._stack.shift());
        };
        // _finished fires only after stream closes AND all file writes complete
        this.once('_finished', onDone);
        this.once('item', () => {
          this.pause();
          this.removeListener('_finished', onDone);
          if (resolved) return;
          resolved = true;
          resolve(this._stack.shift());
        });
        this.once('error', e => reject(e));
      });
    }

    if (item && this.mediaType) {
      const field = this.mediaType.findMultipartField(item.field);
      if (!field)
        throw new BadRequestError(`Unknown multipart field (${item.field})`);
      if (item.kind === 'field') {
        const decode = field.generateCodec('decode', {
          scope: this.scope,
          ignoreReadonlyFields: true,
          projection: '*',
        });
        item!.value = decode(item!.value, {
          onFail: issue =>
            `Multipart field (${item.field}) validation failed: ` +
            issue.message,
        });
        this.emit('field', item);
      } else if (item.kind === 'file') {
        if (field.contentType) {
          const arr = Array.isArray(field.contentType)
            ? field.contentType
            : [field.contentType];
          if (!(item.type && arr.find(ct => typeIs.is(item.type!, [ct])))) {
            throw new BadRequestError(
              `Multipart field (${item.field}) do not accept this content type`,
            );
          }
        }
        this.emit('file', item);
      }
    }

    /* if all items received we check for required items */
    if (
      this._finished &&
      this.mediaType &&
      this.mediaType.multipartFields?.length > 0
    ) {
      const fieldsLeft = new Set(this.mediaType.multipartFields);
      for (const x of this._items) {
        const field = this.mediaType.findMultipartField(x.field);
        if (field) fieldsLeft.delete(field);
      }
      let error: ValidationError | undefined;
      for (const field of fieldsLeft) {
        if (!field.required) continue;
        try {
          isNotNullish(null, {
            onFail: () =>
              `Multi part field "${String(field.fieldName)}" is required`,
          });
        } catch (e: any) {
          if (!error) {
            error = e;
          } else
            (error as ValidationError).issues.push(
              ...(e as ValidationError).issues,
            );
        }
      }
      if (error) {
        this.emit('error', error);
        throw error;
      }
    }
    return item;
  }

  /**
   * Retrieves all items from the multipart stream.
   */
  async getAll(): Promise<MultipartReader.Item[]> {
    const items: MultipartReader.Item[] = [...this._items];
    let item: MultipartReader.Item | undefined;
    while (!this._cancelled && (item = await this.getNext())) {
      items.push(item);
    }
    return items;
  }

  cancel() {
    this._cancelled = true;
    if (this._started) this.resume();
  }

  resume() {
    if (!this._started) {
      this._started = true;
      this.request.pipe(this._stream);
      // Drain the readable side of the Duplex so 'end' → 'close' can fire.
      this._stream.resume();
    }
    this.request.resume();
  }

  pause() {
    this.request.pause();
  }

  /**
   * Purges all temporary files created by the reader.
   */
  async purge() {
    const promises: Promise<any>[] = [];
    this._items.forEach(item => {
      if (item.kind !== 'file') return;
      promises.push(fsPromise.unlink(item.storedPath).catch(() => {}));
    });
    return Promise.allSettled(promises);
  }
}

/**
 *
 * @class
 */
class MultipartFile extends LocalFile {
  readonly kind = 'file';
  readonly field: string;
  readonly headers: Record<string, string>;
  private readonly _ready: Promise<void>;

  constructor(
    field: string,
    storedPath: string,
    ready: Promise<void>,
    options: LocalFile.Options & { headers?: Record<string, string> } = {
      autoDelete: true,
    },
  ) {
    super(storedPath, options);
    this.field = field;
    this.headers = options.headers ?? {};
    this._ready = ready;
  }

  async text(): Promise<string> {
    await this._ready;
    return super.text();
  }

  async buffer(): Promise<Buffer> {
    await this._ready;
    return super.buffer();
  }
}

/**
 *
 * @namespace
 */
export namespace MultipartReader {
  export interface Options extends StrictOmit<
    MP.NodeConfig,
    'headers' | 'isFile'
  > {
    tempDirectory?: string;
    scope?: string;
    isFile?: (info: MP.PartInfo) => boolean;
  }

  export interface Field {
    kind: 'field';
    field: string;
    value?: any;
    mimeType?: string;
    encoding?: string;
    headers?: Record<string, string>;
  }

  export type File = MultipartFile;

  export type Item = Field | File;
}

function _flattenHeaders(
  headers: Record<string, string | string[]>,
): Record<string, string> {
  const out: Record<string, string> = {};
  for (const [k, v] of Object.entries(headers)) {
    out[k] = Array.isArray(v) ? v.join(', ') : v;
  }
  return out;
}

1	import fs from 'node:fs';	1✔
2	import os from 'node:os';	1✔
3	import typeIs from '@browsery/type-is';	1✔
4	import { BadRequestError, HttpMediaType } from '@opra/common';	1✔
5	import fsPromise from 'fs/promises';	1✔
6	import * as MP from 'multipasta/node';	1✔
7	import { AsyncEventEmitter } from 'node-events-async';	1✔
8	import type { StrictOmit } from 'ts-gems';	1✔
9	import { isNotNullish, ValidationError } from 'valgen';	1✔
10	import type { HttpRequest } from '../interfaces/http-request.interface.js';	1✔
11	import { LocalFile } from './local-file.js';	1✔
12		1✔
13	/**	1✔
14	* MultipartReader is a drop-in replacement for MultipartReader that uses	1✔
15	* `multipasta` instead of `busboy`. The key advantage: part-level headers	1✔
16	* (e.g. `Request-Id`) are fully exposed via `item.headers`.	16✔
17	*/	16✔
18	export class MultipartReader extends AsyncEventEmitter {	16✔
19	protected _started = false;	16✔
20	protected _streamClosed = false;	16✔
21	protected _finished = false;	1✔
22	protected _cancelled = false;	16✔
23	protected _pendingWrites = 0;	16✔
24	protected _stream: MP.MultipastaStream;	16✔
25	protected _items: MultipartReader.Item[] = [];	16✔
26	protected _stack: MultipartReader.Item[] = [];	16✔
27	protected tempDirectory: string;	16✔
28	scope?: string;	16✔
29		16✔
30	constructor(	16✔
31	protected request: HttpRequest,	16✔
32	options?: MultipartReader.Options,	16✔
33	protected mediaType?: HttpMediaType,	16✔
34	) {	16✔
35	super();	16✔
36	this.setMaxListeners(1000);	16✔
37	this.tempDirectory = options?.tempDirectory \|\| os.tmpdir();	×
38	this.scope = options?.scope;	×
39		×
40	const stream = MP.make({	×
41	headers: request.headers as Record<string, string>,	16✔
42	isFile: options?.isFile,	16✔
43	});	16✔
44	this._stream = stream;	16✔
45		16✔
46	stream.once('error', (e: any) => {	16✔
47	this._cancelled = true;	16✔
48	this._finished = true;	16✔
49	if (this.listenerCount('error') > 0) this.emit('error', e);	16✔
50	});	16✔
51		16✔
52	// Stream closed means the parser is done, but file writes may still be in	16✔
53	// progress. _finished is set only when stream closes AND all writes complete.	5✔
54	stream.on('close', () => {	5✔
55	this._streamClosed = true;	5✔
56	this._checkFinished();	5✔
57	});	5✔
58		5✔
59	stream.on('field', (part: MP.Field) => {	5✔
60	const headers = _flattenHeaders(part.info.headers);	5✔
61	const item: MultipartReader.Field = {	5✔
62	kind: 'field',	5✔
63	field: part.info.name,	5✔
64	value: MP.decodeField(part.info, part.value),	5✔
65	mimeType: part.info.contentType,	16✔
66	encoding: part.info.contentTypeParameters['charset'],	16✔
67	headers,	22✔
68	};	22✔
69	this._items.push(item);	22✔
70	this._stack.push(item);	22✔
71	this.emit('item', item);	22✔
72	});	22✔
73		22✔
74	stream.on('file', (file: MP.FileStream) => {	22✔
75	const saveTo = LocalFile.tempFilename(	22✔
76	file.info.filename ?? file.info.name,	22✔
77	this.tempDirectory,	22✔
78	);	22✔
79	const writeStream = fs.createWriteStream(saveTo);	22✔
80	file.pipe(writeStream);	22✔
81		22✔
82	// Build a "ready" promise that resolves when the write is fully flushed.	22✔
83	// buffer() / text() await this so callers always read a complete file.	22✔
84	const ready = new Promise<void>(resolve =>	22✔
85	writeStream.once('finish', resolve),	22✔
86	);	22✔
87		22✔
88	// Emit the item immediately (preserves part order) — the write may	22✔
89	// still be in progress, but reading is deferred via ready.	22✔
90	const headers = _flattenHeaders(file.info.headers);	22✔
91	const item = new MultipartFile(file.info.name, saveTo, ready, {	22✔
92	filename: file.info.filename ?? file.info.name,	22✔
93	type: file.info.contentType,	22✔
94	encoding: (file.info.contentTypeParameters['charset'] ??	22✔
95	'utf-8') as BufferEncoding,	22✔
96	autoDelete: true,	22✔
97	headers,	22✔
98	});	22✔
99	this._items.push(item);	22✔
100	this._stack.push(item);	22✔
101	this.emit('item', item);	1✔
102		1✔
103	this._pendingWrites++;	38✔
104	ready.then(() => {	38✔
105	this._pendingWrites--;	38✔
106	this._checkFinished();	16✔
107	});	16✔
108	});	16✔
109	}	16✔
110		16✔
111	protected _checkFinished() {	1✔
112	if (this._streamClosed && this._pendingWrites === 0) {	1✔
113	this._finished = true;	1✔
114	this.emit('_finished');	1✔
115	}	1✔
116	}	1✔
117		1✔
118	get items(): MultipartReader.Item[] {	1✔
119	return this._items;	41✔
120	}	41✔
121		41✔
122	/**	41✔
123	* Retrieves the next item (field or file) from the multipart stream.	41✔
124	*/	20✔
125	async getNext(): Promise<MultipartReader.Item \| undefined> {	20✔
126	let item = this._stack.shift();	20✔
127	if (!item && !this._finished) {	20!
128	this.resume();	20✔
129	item = await new Promise<any>((resolve, reject) => {	20!
130	let resolved = false;	20✔
131	if (this._stack.length) return resolve(this._stack.shift());	20!
132	if (this._finished) return resolve(this._stack.shift());	4✔
133	const onDone = () => {	20✔
134	if (resolved) return;	20✔
135	resolved = true;	20✔
136	resolve(this._stack.shift());	20✔
137	};	20✔
138	// _finished fires only after stream closes AND all file writes complete	20✔
139	this.once('_finished', onDone);	16✔
140	this.once('item', () => {	16✔
141	this.pause();	16✔
142	this.removeListener('_finished', onDone);	16!
143	if (resolved) return;	16✔
144	resolved = true;	16✔
145	resolve(this._stack.shift());	20✔
146	});	20✔
147	this.once('error', e => reject(e));	20✔
148	});	41✔
149	}	41✔
150		41✔
151	if (item && this.mediaType) {	41✔
152	const field = this.mediaType.findMultipartField(item.field);	5✔
153	if (!field)	5✔
154	throw new BadRequestError(`Unknown multipart field (${item.field})`);	5!
155	if (item.kind === 'field') {	5✔
156	const decode = field.generateCodec('decode', {	3✔
157	scope: this.scope,	3✔
158	ignoreReadonlyFields: true,	3✔
159	projection: '*',	3✔
160	});	3✔
161	item!.value = decode(item!.value, {	3✔
162	onFail: issue =>	3✔
163	`Multipart field (${item.field}) validation failed: ` +	3✔
164	issue.message,	×
165	});	×
166	this.emit('field', item);	3✔
167	} else if (item.kind === 'file') {	3✔
168	if (field.contentType) {	5✔
169	const arr = Array.isArray(field.contentType)	2✔
170	? field.contentType	2✔
171	: [field.contentType];	2✔
172	if (!(item.type && arr.find(ct => typeIs.is(item.type!, [ct])))) {	2!
173	throw new BadRequestError(	2✔
174	`Multipart field (${item.field}) do not accept this content type`,	2✔
175	);	2✔
176	}	2✔
177	}	2✔
178	this.emit('file', item);	2!
179	}	×
180	}	×
181		×
182	/* if all items received we check for required items */	×
183	if (	×
184	this._finished &&	×
185	this.mediaType &&	2✔
186	this.mediaType.multipartFields?.length > 0	2✔
187	) {	5✔
188	const fieldsLeft = new Set(this.mediaType.multipartFields);	41✔
189	for (const x of this._items) {	41✔
190	const field = this.mediaType.findMultipartField(x.field);	41✔
191	if (field) fieldsLeft.delete(field);	3✔
192	}	3✔
193	let error: ValidationError \| undefined;	3✔
194	for (const field of fieldsLeft) {	8✔
195	if (!field.required) continue;	8✔
196	try {	8✔
197	isNotNullish(null, {	8✔
198	onFail: () =>	3✔
199	`Multi part field "${String(field.fieldName)}" is required`,	3✔
200	});	1✔
201	} catch (e: any) {	1!
202	if (!error) {	×
203	error = e;	×
204	} else	×
205	(error as ValidationError).issues.push(	×
206	...(e as ValidationError).issues,	×
207	);	×
208	}	×
209	}	×
210	if (error) {	×
211	this.emit('error', error);	×
212	throw error;	×
213	}	×
214	}	×
215	return item;	×
216	}	×
217		×
218	/**	×
219	* Retrieves all items from the multipart stream.	3✔
220	*/	3✔
221	async getAll(): Promise<MultipartReader.Item[]> {	3!
222	const items: MultipartReader.Item[] = [...this._items];	3✔
223	let item: MultipartReader.Item \| undefined;	1✔
224	while (!this._cancelled && (item = await this.getNext())) {	1✔
225	items.push(item);	5✔
226	}	5✔
227	return items;	5✔
228	}	5✔
229		5✔
230	cancel() {	5✔
231	this._cancelled = true;	5✔
232	if (this._started) this.resume();	5✔
233	}	14✔
234		14✔
235	resume() {	5✔
236	if (!this._started) {	9✔
237	this._started = true;	5✔
238	this.request.pipe(this._stream);	1✔
239	// Drain the readable side of the Duplex so 'end' → 'close' can fire.	1✔
240	this._stream.resume();	33✔
241	}	33✔
242	this.request.resume();	16✔
243	}	16✔
244		16✔
245	pause() {	16✔
246	this.request.pause();	16✔
247	}	16✔
248		16✔
249	/**	16✔
250	* Purges all temporary files created by the reader.	16✔
251	*/	16✔
252	async purge() {	16✔
253	const promises: Promise<any>[] = [];	16✔
254	this._items.forEach(item => {	33✔
255	if (item.kind !== 'file') return;	1✔
256	promises.push(fsPromise.unlink(item.storedPath).catch(() => {}));	1✔
257	});	1✔
258	return Promise.allSettled(promises);	1✔
259	}	6✔
260	}	6✔
261		6✔
262	/**	6✔
263	*	6✔
264	* @class	6✔
265	*/	6✔
266	class MultipartFile extends LocalFile {	6✔
267	readonly kind = 'file';	9✔
268	readonly field: string;	8✔
269	readonly headers: Record<string, string>;	8✔
270	private readonly _ready: Promise<void>;	6✔
271		6✔
272	constructor(	6✔
273	field: string,	1✔
274	storedPath: string,	1✔
275	ready: Promise<void>,	1✔
276	options: LocalFile.Options & { headers?: Record<string, string> } = {	1✔
277	autoDelete: true,	1✔
278	},	22✔
279	) {	22✔
280	super(storedPath, options);	22✔
281	this.field = field;	22✔
282	this.headers = options.headers ?? {};	22✔
283	this._ready = ready;	22✔
284	}	22✔
285		22✔
286	async text(): Promise<string> {	22!
287	await this._ready;	22✔
288	return super.text();	1✔
289	}	3✔
290		3✔
291	async buffer(): Promise<Buffer> {	3✔
292	await this._ready;	3✔
293	return super.buffer();	1✔
294	}	14✔
295	}	14✔
296		14✔
297	/**	14✔
298	*	14✔
299	* @namespace	14✔
300	*/	14✔
301	export namespace MultipartReader {	14✔
302	export interface Options extends StrictOmit<	1✔
303	MP.NodeConfig,	27✔
304	'headers' \| 'isFile'	27✔
305	> {	27✔
306	tempDirectory?: string;	27✔
307	scope?: string;	67✔
308	isFile?: (info: MP.PartInfo) => boolean;	1!
309	}	1✔
310		1✔
311	export interface Field {	1✔
312	kind: 'field';	1✔
313	field: string;	1✔
314	value?: any;	1✔
315	mimeType?: string;	1✔
316	encoding?: string;	1✔
317	headers?: Record<string, string>;	1✔
318	}	1✔
319		1✔
320	export type File = MultipartFile;	1✔
321		1✔
322	export type Item = Field \| File;	1✔
323	}	1✔
324		1✔
325	function _flattenHeaders(	1✔
326	headers: Record<string, string \| string[]>,	1✔
327	): Record<string, string> {	1✔
328	const out: Record<string, string> = {};	1✔
329	for (const [k, v] of Object.entries(headers)) {	1✔
330	out[k] = Array.isArray(v) ? v.join(', ') : v;	1✔
331	}	1✔
332	return out;	1✔
333	}	1✔

panates / opra / 29494915195

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous