1import { describe, it } from "@effect/vitest"
2import { ByteSize, Effect, ErrorReporter, FileSystem, identity, Path, Schema, Sink, Stream, Unify } from "effect"
3import { HttpClientRequest, HttpIncomingMessage, HttpServerRequest, Multipart, MultipartParser } from "effect/http"
4import * as HttpServerRespondable from "effect/http/HttpServerRespondable"
5import { deepStrictEqual, notStrictEqual, strictEqual } from "node:assert"
6
7describe("Multipart", () => {
8 it.effect("schemaJson applies a JSON reviver", () =>
9 Effect.gen(function*() {
10 const decoded = yield* Multipart.schemaJson(Schema.Struct({ value: Schema.String }), {
11 reviver: (key, value) => key === "value" ? "revived" : value
12 })({ json: "{\"value\":\"original\"}" }, "json")
13
14 deepStrictEqual(decoded, { value: "revived" })
15 }))
16
17 it.effect("parses fields and streams file content as stream parts", () =>
18 Effect.gen(function*() {
19 const data = new globalThis.FormData()
20 data.append("foo", "bar")
21 data.append("test", "ing")
22 data.append("file", new globalThis.File(["A".repeat(1024 * 1024)], "foo.txt", { type: "text/plain" }))
23 const response = new Response(data)
24 const streamed: Array<Multipart.Part> = []
25
26 const parts = yield* Stream.fromReadableStream({
27 evaluate: () => response.body!,
28 onError: identity
29 }).pipe(
30 Stream.pipeThroughChannel(Multipart.makeChannel(Object.fromEntries(response.headers))),
31 Stream.mapEffect((part) => {
32 streamed.push(part)
33 return Unify.unify(
34 part._tag === "File" ?
35 Effect.zip(
36 Effect.succeed(part.name),
37 Stream.mkString(Stream.decodeText(part.content))
38 ) :
39 Effect.succeed([part.key, part.value] as const)
40 )
41 }),
42 Stream.runCollect
43 )
44
45 deepStrictEqual(parts, [
46 ["foo", "bar"],
47 ["test", "ing"],
48 ["foo.txt", "A".repeat(1024 * 1024)]
49 ])
50 deepStrictEqual(
51 streamed.map((part) => [Multipart.isPart(part), Multipart.isStreamPart(part)]),
52 [[true, true], [true, true], [true, true]]
53 )
54 }))
55
56 it.effect("collects file content across pulls and a split trailing boundary", () =>
57 Effect.gen(function*() {
58 const boundary = "----testboundary"
59 const encoder = new TextEncoder()
60 const boundarySplit = 8
61 const chunks = [
62 encoder.encode(
63 `--${boundary}\r\n` +
64 `Content-Disposition: form-data; name="file"; filename="file.txt"\r\n` +
65 `Content-Type: text/plain\r\n\r\n` +
66 "abc"
67 ),
68 encoder.encode(`def\r\n--${boundary.slice(0, boundarySplit)}`),
69 encoder.encode(`${boundary.slice(boundarySplit)}--\r\n`)
70 ]
71
72 const contents = yield* Stream.fromArray(chunks).pipe(
73 Stream.rechunk(1),
74 Stream.pipeThroughChannel(
75 Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` })
76 ),
77 Stream.mapEffect((part) => part._tag === "File" ? part.contentEffect : Effect.die("expected file")),
78 Stream.runCollect
79 )
80
81 deepStrictEqual(contents, [encoder.encode("abcdef")])
82 }))
83
84 it.live("emits a field buffered while reading a file without pulling more input", () =>
85 Effect.gen(function*() {
86 const boundary = "----testboundary"
87 const encoder = new TextEncoder()
88 const body = encoder.encode(
89 `--${boundary}\r\n` +
90 `Content-Disposition: form-data; name="file"; filename="file.txt"\r\n` +
91 `Content-Type: text/plain\r\n\r\n` +
92 "file contents\r\n" +
93 `--${boundary}\r\n` +
94 `Content-Disposition: form-data; name="description"\r\n\r\n` +
95 "test file\r\n" +
96 `--${boundary}--\r\n`
97 )
98 const chunks = [body.subarray(0, 64), body.subarray(64, 128), body.subarray(128)]
99
100 const parts = yield* Stream.fromArray(chunks).pipe(
101 Stream.rechunk(1),
102 Stream.concat(Stream.never),
103 Stream.pipeThroughChannel(
104 Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` })
105 ),
106 Stream.mapEffect((part) =>
107 part._tag === "File"
108 ? part.contentEffect.pipe(Effect.map((content) => [part.key, new TextDecoder().decode(content)] as const))
109 : Effect.succeed([part.key, part.value] as const)
110 ),
111 Stream.take(2),
112 Stream.runCollect,
113 Effect.timeout("2 seconds")
114 )
115
116 deepStrictEqual(parts, [
117 ["file", "file contents"],
118 ["description", "test file"]
119 ])
120 }))
121
122 it.effect("parses non-Latin-1 filenames", () =>
123 Effect.gen(function*() {
124 const data = new globalThis.FormData()
125 data.append("file", new globalThis.File(["content"], "日本語.txt", { type: "text/plain" }))
126 const response = new Response(data)
127
128 const parts = yield* Stream.fromReadableStream({
129 evaluate: () => response.body!,
130 onError: identity
131 }).pipe(
132 Stream.pipeThroughChannel(Multipart.makeChannel(Object.fromEntries(response.headers))),
133 Stream.mapEffect((part) =>
134 Unify.unify(
135 part._tag === "File"
136 ? Stream.runDrain(part.content).pipe(Effect.as([part.key, part.name] as const))
137 : Effect.succeed([part.key, part.value] as const)
138 )
139 ),
140 Stream.runCollect
141 )
142
143 deepStrictEqual(parts, [["file", "日本語.txt"]])
144 }))
145
146 it.effect("fails when a limit is exceeded even if the whole body arrives in one chunk", () =>
147 Effect.gen(function*() {
148 const boundary = "----testboundary"
149 const part = (name: string) =>
150 `--${boundary}\r\n` +
151 `Content-Disposition: form-data; name="${name}"; filename="${name}.txt"\r\n` +
152 `Content-Type: text/plain\r\n\r\n${name}\r\n`
153 const body = part("a") + part("b") + part("c") + `--${boundary}--\r\n`
154
155 const error = yield* Stream.make(new TextEncoder().encode(body)).pipe(
156 Stream.pipeThroughChannel(
157 Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` })
158 ),
159 Stream.runCollect,
160 Effect.provideService(Multipart.MaxParts, 2),
161 Effect.flip
162 )
163
164 strictEqual(error._tag, "MultipartError")
165 strictEqual(error.reason._tag, "TooManyParts")
166 }))
167
168 it.effect("propagates FileTooLarge after a file exceeds the limit mid-stream", () =>
169 Effect.gen(function*() {
170 const boundary = "----testboundary"
171 const encoder = new TextEncoder()
172 let fileParts = 0
173 const fileStart = `--${boundary}\r\n` +
174 `Content-Disposition: form-data; name="file"; filename="file.txt"\r\n` +
175 `Content-Type: text/plain\r\n\r\n` +
176 "a"
177 const fileEnd = `\r\n--${boundary}--\r\n`
178
179 const error = yield* Stream.make(
180 encoder.encode(fileStart),
181 encoder.encode("a".repeat(1024)),
182 encoder.encode(fileEnd)
183 ).pipe(
184 Stream.pipeThroughChannel(
185 Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` })
186 ),
187 Stream.mapEffect((part) => {
188 if (part._tag !== "File") {
189 return Effect.void
190 }
191 fileParts++
192 return Stream.runDrain(part.content)
193 }),
194 Stream.runDrain,
195 Effect.provideService(Multipart.MaxFileSize, ByteSize.bytes(256)),
196 Effect.flip
197 )
198
199 strictEqual(fileParts, 1)
200 strictEqual(error._tag, "MultipartError")
201 strictEqual(error.reason._tag, "FileTooLarge")
202 }))
203
204 it.effect("propagates BodyTooLarge after the total size limit is exceeded mid-file", () =>
205 Effect.gen(function*() {
206 const boundary = "----testboundary"
207 const encoder = new TextEncoder()
208 let fileParts = 0
209 const fileStart = `--${boundary}\r\n` +
210 `Content-Disposition: form-data; name="file"; filename="file.txt"\r\n` +
211 `Content-Type: text/plain\r\n\r\n` +
212 "a"
213 const fileEnd = `\r\n--${boundary}--\r\n`
214
215 const error = yield* Stream.make(
216 encoder.encode(fileStart),
217 encoder.encode("a".repeat(1024)),
218 encoder.encode(fileEnd)
219 ).pipe(
220 Stream.pipeThroughChannel(
221 Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` })
222 ),
223 Stream.mapEffect((part) => {
224 if (part._tag !== "File") {
225 return Effect.void
226 }
227 fileParts++
228 return Stream.runDrain(part.content)
229 }),
230 Stream.runDrain,
231 Effect.provideService(HttpIncomingMessage.MaxBodySize, ByteSize.bytes(256)),
232 Effect.flip
233 )
234
235 strictEqual(fileParts, 1)
236 strictEqual(error._tag, "MultipartError")
237 strictEqual(error.reason._tag, "BodyTooLarge")
238 }))
239
240 const activeFileParts = (error: Multipart.MultipartError) =>
241 Stream.make(
242 new TextEncoder().encode(
243 "--b\r\nContent-Disposition: form-data; name=\"file\"; filename=\"a.txt\"\r\n\r\nhello"
244 )
245 ).pipe(
246 Stream.concat(Stream.fail(error)),
247 Stream.pipeThroughChannel(Multipart.makeChannel({ "content-type": "multipart/form-data; boundary=b" }))
248 )
249
250 it.live("propagates upstream failure while consuming an active file", () =>
251 Effect.gen(function*() {
252 const upstreamError = Multipart.MultipartError.fromReason("InternalError", new Error("body-read-failed"))
253 let bytesRead = 0
254
255 const error = yield* activeFileParts(upstreamError).pipe(
256 Stream.runForEach((part) =>
257 part._tag === "File"
258 ? Stream.runForEach(part.content, (chunk) =>
259 Effect.sync(() => {
260 bytesRead += chunk.length
261 })).pipe(Effect.flatMap(() => Effect.die("content should have failed")))
262 : Effect.die("expected file")
263 ),
264 Effect.timeout("1 second"),
265 Effect.flip
266 )
267
268 strictEqual(bytesRead, 5)
269 strictEqual(error, upstreamError)
270 }))
271
272 it.live("preserves upstream failure when collecting an active file with contentEffect", () =>
273 Effect.gen(function*() {
274 const upstreamError = Multipart.MultipartError.fromReason("InternalError", new Error("body-read-failed"))
275
276 const error = yield* activeFileParts(upstreamError).pipe(
277 Stream.runForEach((part) =>
278 part._tag === "File"
279 ? Effect.flatMap(part.contentEffect, () => Effect.die("contentEffect should have failed"))
280 : Effect.die("expected file")
281 ),
282 Effect.timeout("1 second"),
283 Effect.flip
284 )
285
286 strictEqual(error, upstreamError)
287 }))
288
289 it.live("preserves the upstream error cause when persisting an active file", () =>
290 Effect.gen(function*() {
291 const upstreamError = Multipart.MultipartError.fromReason("InternalError", new Error("body-read-failed"))
292 let bytesWritten = 0
293
294 const error = yield* activeFileParts(upstreamError).pipe(
295 Multipart.toPersisted,
296 Effect.provideService(
297 FileSystem.FileSystem,
298 FileSystem.makeNoop({
299 makeTempDirectoryScoped: () => Effect.succeed("/tmp/multipart-test"),
300 sink: () =>
301 Sink.forEach((chunk: Uint8Array) =>
302 Effect.sync(() => {
303 bytesWritten += chunk.length
304 })
305 )
306 })
307 ),
308 Effect.provide(Path.layer),
309 Effect.scoped,
310 Effect.timeout("1 second"),
311 Effect.flip
312 )
313
314 strictEqual(bytesWritten, 5)
315 strictEqual(error, upstreamError)
316 }))
317
318 it.effect("propagates Parse when the body ends mid-file", () =>
319 Effect.gen(function*() {
320 const boundary = "----testboundary"
321 const encoder = new TextEncoder()
322 let fileParts = 0
323 const fileStart = `--${boundary}\r\n` +
324 `Content-Disposition: form-data; name="file"; filename="file.txt"\r\n` +
325 `Content-Type: text/plain\r\n\r\n` +
326 "a"
327
328 const error = yield* Stream.make(
329 encoder.encode(fileStart),
330 encoder.encode("bbbb")
331 ).pipe(
332 Stream.pipeThroughChannel(
333 Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` })
334 ),
335 Stream.mapEffect((part) => {
336 if (part._tag !== "File") {
337 return Effect.void
338 }
339 fileParts++
340 return Stream.runDrain(part.content)
341 }),
342 Stream.runDrain,
343 Effect.flip
344 )
345
346 strictEqual(fileParts, 1)
347 strictEqual(error._tag, "MultipartError")
348 strictEqual(error.reason._tag, "Parse")
349 }))
350
351 it.each<{
352 description: string
353 options: {
354 readonly maxParts?: number
355 readonly maxFieldSize?: ByteSize.Input
356 readonly maxPartSize?: ByteSize.Input
357 }
358 limit: "MaxParts" | "MaxFieldSize" | "MaxPartSize"
359 expectedFields: Array<string>
360 }>([
361 {
362 description: "maxParts",
363 options: { maxParts: 2 },
364 limit: "MaxParts",
365 expectedFields: ["a", "b"]
366 },
367 {
368 description: "maxFieldSize",
369 options: { maxFieldSize: "1 B" },
370 limit: "MaxFieldSize",
371 expectedFields: []
372 },
373 {
374 description: "maxPartSize",
375 options: { maxPartSize: ByteSize.bytes(1) },
376 limit: "MaxPartSize",
377 expectedFields: []
378 }
379 ])("stops delivering fields when $description is exceeded", ({ expectedFields, limit, options }) => {
380 const boundary = "----testboundary"
381 const encoder = new TextEncoder()
382 const fields: Array<string> = []
383 const errors: Array<MultipartParser.MultipartError> = []
384 const parser = MultipartParser.make({
385 headers: { "content-type": `multipart/form-data; boundary=${boundary}` },
386 ...options,
387 onField(info) {
388 fields.push(info.name)
389 },
390 onFile: () => () => {},
391 onError(error) {
392 errors.push(error)
393 },
394 onDone() {}
395 })
396 const part = (name: string) => `--${boundary}\r\nContent-Disposition: form-data; name="${name}"\r\n\r\nvalue\r\n`
397
398 parser.write(encoder.encode(part("a") + part("b") + part("c") + part("d") + `--${boundary}--\r\n`))
399 parser.end()
400
401 deepStrictEqual(errors, [{ _tag: "ReachedLimit", limit }])
402 deepStrictEqual(fields, expectedFields)
403 })
404
405 it("handles the final boundary delimiter split between the trailing hyphens", () => {
406 const boundary = "----testboundary"
407 const encoder = new TextEncoder()
408 const decoder = new TextDecoder()
409 const errors: Array<MultipartParser.MultipartError> = []
410 const fields: Array<readonly [string, string]> = []
411 let done = false
412 const parser = MultipartParser.make({
413 headers: { "content-type": `multipart/form-data; boundary=${boundary}` },
414 maxTotalSize: Infinity,
415 maxPartSize: Infinity,
416 maxFieldSize: Infinity,
417 onField(info, value) {
418 fields.push([info.name, decoder.decode(value)])
419 },
420 onFile: () => () => {},
421 onError(error) {
422 errors.push(error)
423 },
424 onDone() {
425 done = true
426 }
427 })
428 const body = `--${boundary}\r\nContent-Disposition: form-data; name="field"\r\n\r\nvalue`
429
430 parser.write(encoder.encode(`${body}\r\n--${boundary}-`))
431 parser.write(encoder.encode("-\r\n"))
432 strictEqual(done, true)
433 parser.end()
434
435 deepStrictEqual(fields, [["field", "value"]])
436 deepStrictEqual(errors, [])
437 })
438
439 it.effect("returns distinct persisted file paths as non-stream parts", () =>
440 Effect.gen(function*() {
441 const formData = new FormData()
442 formData.append("first", new File(["one"], "same.txt"))
443 formData.append("second", new File(["two"], "same.txt"))
444 const request = HttpServerRequest.fromClientRequest(
445 HttpClientRequest.bodyFormData(HttpClientRequest.post("https://example.com"), formData)
446 )
447 const writes: Array<string> = []
448 const persisted = yield* Multipart.toPersisted(
449 request.multipartStream,
450 (path) => Effect.sync(() => writes.push(path))
451 ).pipe(
452 Effect.provideService(
453 FileSystem.FileSystem,
454 FileSystem.makeNoop({
455 makeTempDirectoryScoped: () => Effect.succeed("/tmp/audit")
456 })
457 ),
458 Effect.provide(Path.layer)
459 )
460 const first = (persisted.first as Array<Multipart.PersistedFile>)[0]
461 const second = (persisted.second as Array<Multipart.PersistedFile>)[0]
462 strictEqual(Multipart.isPersistedFile(first), true)
463 strictEqual(Multipart.isPart(first), true)
464 strictEqual(Multipart.isStreamPart(first), false)
465 strictEqual(first.path, "/tmp/audit/same.txt")
466 notStrictEqual(first.path, second.path)
467 deepStrictEqual(writes, [first.path, second.path])
468 }))
469
470 it.effect("responds based on the reason and is ignored by the ErrorReporter", () =>
471 Effect.gen(function*() {
472 const cases = [
473 ["FileTooLarge", 413],
474 ["FieldTooLarge", 413],
475 ["BodyTooLarge", 413],
476 ["TooManyParts", 413],
477 ["InternalError", 500],
478 ["Parse", 400]
479 ] as const
480
481 for (const [reason, status] of cases) {
482 const error = Multipart.MultipartError.fromReason(reason)
483 const response = yield* HttpServerRespondable.toResponse(error)
484
485 strictEqual(response.status, status)
486 strictEqual(ErrorReporter.isIgnored(error), true)
487 }
488 }))
489
490 describe("FileSchema", () => {
491 it("toJsonSchema", () => {
492 const document = Schema.toJsonSchemaDocument(Multipart.PersistedFileSchema)
493 deepStrictEqual(document, {
494 dialect: "draft-2020-12",
495 schema: {
496 "type": "object",
497 "properties": {
498 "key": {
499 "type": "string"
500 },
501 "name": {
502 "type": "string"
503 },
504 "contentType": {
505 "type": "string",
506 "contentEncoding": "binary"
507 },
508 "path": {
509 "type": "string"
510 }
511 },
512 "required": ["key", "name", "contentType", "path"],
513 "additionalProperties": true
514 },
515 definitions: {}
516 })
517 })
518 })
519
520 it("identifies multipart errors", () => {
521 strictEqual(Multipart.isMultipartError(Multipart.MultipartError.fromReason("Parse")), true)
522 strictEqual(Multipart.isMultipartError({ _tag: "MultipartError" }), false)
523 strictEqual(Multipart.isMultipartError(null), false)
524 })
525})