1/**
2 * Node.js streams adapter for the low-level multipart parser.
3 *
4 * @since 4.0.0
5 */
6// oxlint-disable typescript/no-unsafe-declaration-merging
7/// <reference types="node" />
8import type { BaseConfig, MultipartError, Parser, PartInfo } from "effect/http/MultipartParser"
9import { make as makeParser } from "effect/http/MultipartParser"
10import type { IncomingHttpHeaders } from "node:http"
11import { Duplex, Readable } from "node:stream"
12
13/**
14 * A part emitted by the Node.js multipart parser.
15 *
16 * @category models
17 * @since 4.0.0
18 */
19export type Part = Field | FileStream
20
21/**
22 * A parsed multipart field.
23 *
24 * @category models
25 * @since 4.0.0
26 */
27export interface Field {
28 readonly _tag: "Field"
29 readonly info: PartInfo
30 readonly value: Uint8Array
31}
32
33/**
34 * A Node.js duplex stream that emits parsed multipart parts.
35 *
36 * @category models
37 * @since 4.0.0
38 */
39export interface MultipartStream extends Duplex {
40 [Symbol.asyncIterator](): NodeJS.AsyncIterator<Part>
41
42 on(event: "field", listener: (field: Field) => void): this
43 on(event: "file", listener: (file: FileStream) => void): this
44 on(event: "close", listener: () => void): this
45 on(event: "data", listener: (part: Part) => void): this
46 on(event: "drain", listener: () => void): this
47 on(event: "end", listener: () => void): this
48 on(event: "error", listener: (err: MultipartError) => void): this
49 on(event: "finish", listener: () => void): this
50 on(event: "pause", listener: () => void): this
51 on(event: "pipe", listener: (src: Readable) => void): this
52 on(event: "readable", listener: () => void): this
53 on(event: "resume", listener: () => void): this
54 on(event: "unpipe", listener: (src: Readable) => void): this
55 on(event: string | symbol, listener: (...args: Array<any>) => void): this
56
57 read(size?: number): Part | null
58}
59
60/**
61 * Configuration for the Node.js multipart parser.
62 *
63 * @category models
64 * @since 4.0.0
65 */
66export type NodeConfig = Omit<BaseConfig, "headers"> & {
67 readonly headers: IncomingHttpHeaders
68}
69
70/**
71 * A Node.js duplex stream that parses multipart input.
72 *
73 * @category models
74 * @since 4.0.0
75 */
76export class MultipartStream extends Duplex {
77 private _parser: Parser
78 _canWrite = true
79 private _writeCallback: (() => void) | undefined
80
81 constructor(config: NodeConfig) {
82 super({ readableObjectMode: true })
83 let currentError: MultipartError | undefined
84 let currentFile: FileStream | undefined
85 this._parser = makeParser({
86 ...(config as any),
87 onField: (info, value) => {
88 if (currentError !== undefined) return
89 const field: Field = { _tag: "Field", info, value }
90 this.push(field)
91 this.emit("field", field)
92 },
93 onFile: (info) => {
94 if (currentError !== undefined) return (_) => {}
95 const file = new FileStream(info, this)
96 currentFile = file
97 this.push(file)
98 this.emit("file", file)
99 return (chunk) => {
100 if (currentError !== undefined) return
101 this._canWrite = file.push(chunk)
102 if (chunk === null && !this._canWrite) {
103 currentFile = undefined
104 this._resume()
105 }
106 }
107 },
108 onError: (error) => {
109 this.emit("error", error)
110 currentFile?.emit("error", error)
111 currentError = error
112 },
113 onDone: () => {
114 this.push(null)
115 }
116 })
117 }
118
119 _resume() {
120 this._canWrite = true
121 if (this._writeCallback !== undefined) {
122 const callback = this._writeCallback
123 this._writeCallback = undefined
124 callback()
125 }
126 }
127
128 override _read(_size: number) {}
129
130 override _write(
131 chunk: any,
132 encoding: BufferEncoding,
133 callback: (error?: Error | null | undefined) => void
134 ): void {
135 this._parser.write(
136 chunk instanceof Uint8Array ? chunk : Buffer.from(chunk, encoding)
137 )
138 if (this._canWrite) {
139 callback()
140 } else {
141 this._writeCallback = callback
142 }
143 }
144
145 override _final(callback: (error?: Error | null | undefined) => void): void {
146 this._parser.end()
147 callback()
148 }
149}
150
151/**
152 * Creates a Node.js multipart parser stream.
153 *
154 * @category constructors
155 * @since 4.0.0
156 */
157export const make = (config: NodeConfig): MultipartStream => new MultipartStream(config)
158
159/**
160 * A readable stream containing a parsed multipart file.
161 *
162 * @category models
163 * @since 4.0.0
164 */
165export class FileStream extends Readable {
166 readonly _tag = "File"
167 readonly filename: string | undefined
168 readonly info: PartInfo
169 private _parent: MultipartStream
170 constructor(
171 info: PartInfo,
172 parent: MultipartStream
173 ) {
174 super()
175 this.info = info
176 this._parent = parent
177 this.filename = info.filename
178 }
179 override _read(_size: number) {
180 if (this._parent._canWrite === false) {
181 this._parent._resume()
182 }
183 }
184}