28585340608

Committed 02 Jul 2026 09:58AM UTC coverage: 82.799% (-0.3%) from 83.077%

Build # 28585340608

Build Type

push

github

Committed by

erayhanoglu

Commit Message

Feat: Added bundle support for HTTP protocol
Refactor: Renamed HttpIncoming to HttpRequest, HttpOutgoing to HttpResponse
Refactor: Refactored MultipartReader to support multipart/mixed

Coverage Stats

3763 of 4860 branches covered (77.43%)

Branch coverage included in aggregate %.

1755 of 1981 new or added lines in 54 files covered. (88.59%)

94 existing lines in 4 files now uncovered.

33673 of 40353 relevant lines covered (83.45%)

220.83 hits per line

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

88.72

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

panates / opra / 28585340608

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