fix multipart support for bun http server (#4028)

.changeset/gentle-roses-judge.md

@@ -0,0 +1,8 @@
+---
+"@effect/platform-node": patch
+"@effect/platform-bun": patch
+"@effect/platform": patch
+"effect": patch
+---
+
+fix multipart support for bun http server

packages/platform-bun/package.json

@@ -45,7 +45,8 @@
     "effect": "workspace:^"
   },
   "dependencies": {
-    "@effect/platform-node-shared": "workspace:^"
+    "@effect/platform-node-shared": "workspace:^",
+    "multipasta": "^0.2.5"
   },
   "devDependencies": {
     "@effect/platform": "workspace:^",

packages/platform-bun/src/BunMultipart.ts

@@ -1,8 +1,25 @@
 /**
  * @since 1.0.0
  */
+import type * as FileSystem from "@effect/platform/FileSystem"
+import type * as Multipart from "@effect/platform/Multipart"
+import type * as Path from "@effect/platform/Path"
+import type * as Effect from "effect/Effect"
+import type * as Scope from "effect/Scope"
+import type * as Stream from "effect/Stream"
+import * as internal from "./internal/multipart.js"
 
 /**
  * @since 1.0.0
+ * @category constructors
  */
-export * from "@effect/platform-node-shared/NodeMultipart"
+export const stream: (source: Request) => Stream.Stream<Multipart.Part, Multipart.MultipartError> = internal.stream
+
+/**
+ * @since 1.0.0
+ * @category constructors
+ */
+export const persisted: (
+  source: Request
+) => Effect.Effect<Multipart.Persisted, Multipart.MultipartError, FileSystem.FileSystem | Path.Path | Scope.Scope> =
+  internal.persisted

packages/platform-bun/src/internal/httpServer.ts

@@ -1,4 +1,3 @@
-import * as MultipartNode from "@effect/platform-node-shared/NodeMultipart"
 import * as Cookies from "@effect/platform/Cookies"
 import * as Etag from "@effect/platform/Etag"
 import * as FetchHttpClient from "@effect/platform/FetchHttpClient"
@@ -28,9 +27,9 @@ import type { ReadonlyRecord } from "effect/Record"
 import type * as Runtime from "effect/Runtime"
 import type * as Scope from "effect/Scope"
 import * as Stream from "effect/Stream"
-import { Readable } from "node:stream"
 import * as BunContext from "../BunContext.js"
 import * as Platform from "../BunHttpPlatform.js"
+import * as MultipartBun from "./multipart.js"
 
 /** @internal */
 export const make = (
@@ -343,13 +342,13 @@ class ServerRequestImpl extends Inspectable.Class implements ServerRequest.HttpS
       return this.multipartEffect
     }
     this.multipartEffect = Effect.runSync(Effect.cached(
-      MultipartNode.persisted(Readable.fromWeb(this.source.body! as any), this.headers)
+      MultipartBun.persisted(this.source)
     ))
     return this.multipartEffect
   }
 
   get multipartStream(): Stream.Stream<Multipart.Part, Multipart.MultipartError> {
-    return MultipartNode.stream(Readable.fromWeb(this.source.body! as any), this.headers)
+    return MultipartBun.stream(this.source)
   }
 
   private arrayBufferEffect: Effect.Effect<ArrayBuffer, Error.RequestError> | undefined

packages/platform-bun/src/internal/multipart.ts

@@ -0,0 +1,142 @@
+import * as Multipart from "@effect/platform/Multipart"
+import * as Effect from "effect/Effect"
+import { pipe } from "effect/Function"
+import * as Inspectable from "effect/Inspectable"
+import * as Stream from "effect/Stream"
+import type { MultipartError, PartInfo } from "multipasta"
+import { decodeField } from "multipasta"
+import * as MP from "multipasta/web"
+
+/** @internal */
+export const stream = (source: Request): Stream.Stream<Multipart.Part, Multipart.MultipartError> =>
+  pipe(
+    Multipart.makeConfig({}),
+    Effect.map((config) => {
+      const parser = MP.make({
+        ...config,
+        headers: source.headers
+      })
+      return Stream.fromReadableStream(
+        () => source.body!.pipeThrough(parser),
+        (cause) => convertError(cause as MultipartError)
+      )
+    }),
+    Stream.unwrap,
+    Stream.map(convertPart)
+  )
+
+/** @internal */
+export const persisted = (source: Request) =>
+  Multipart.toPersisted(stream(source), (path, file) =>
+    Effect.tryPromise({
+      try: async () => {
+        const fileImpl = file as FileImpl
+        const writer = Bun.file(path).writer()
+        const reader = fileImpl.file.readable.getReader()
+        try {
+          while (true) {
+            const { done, value } = await reader.readMany()
+            if (done) break
+            for (const chunk of value) {
+              writer.write(chunk)
+            }
+            await writer.flush()
+          }
+        } finally {
+          reader.cancel()
+          await writer.end()
+        }
+      },
+      catch: (cause) => new Multipart.MultipartError({ reason: "InternalError", cause })
+    }))
+
+const convertPart = (part: MP.Part): Multipart.Part =>
+  part._tag === "Field" ? new FieldImpl(part.info, part.value) : new FileImpl(part)
+
+abstract class PartBase extends Inspectable.Class {
+  readonly [Multipart.TypeId]: Multipart.TypeId
+  constructor() {
+    super()
+    this[Multipart.TypeId] = Multipart.TypeId
+  }
+}
+
+class FieldImpl extends PartBase implements Multipart.Field {
+  readonly _tag = "Field"
+  readonly key: string
+  readonly contentType: string
+  readonly value: string
+
+  constructor(
+    info: PartInfo,
+    value: Uint8Array
+  ) {
+    super()
+    this.key = info.name
+    this.contentType = info.contentType
+    this.value = decodeField(info, value)
+  }
+
+  toJSON(): unknown {
+    return {
+      _id: "@effect/platform/Multipart/Part",
+      _tag: "Field",
+      key: this.key,
+      value: this.value,
+      contentType: this.contentType
+    }
+  }
+}
+
+class FileImpl extends PartBase implements Multipart.File {
+  readonly _tag = "File"
+  readonly key: string
+  readonly name: string
+  readonly contentType: string
+  readonly content: Stream.Stream<Uint8Array, Multipart.MultipartError>
+
+  constructor(readonly file: MP.File) {
+    super()
+    this.key = file.info.name
+    this.name = file.info.filename ?? file.info.name
+    this.contentType = file.info.contentType
+    this.content = Stream.fromReadableStream(
+      () => file.readable,
+      (cause) => new Multipart.MultipartError({ reason: "InternalError", cause })
+    )
+  }
+
+  toJSON(): unknown {
+    return {
+      _id: "@effect/platform/Multipart/Part",
+      _tag: "File",
+      key: this.key,
+      name: this.name,
+      contentType: this.contentType
+    }
+  }
+}
+
+function convertError(cause: MultipartError): Multipart.MultipartError {
+  switch (cause._tag) {
+    case "ReachedLimit": {
+      switch (cause.limit) {
+        case "MaxParts": {
+          return new Multipart.MultipartError({ reason: "TooManyParts", cause })
+        }
+        case "MaxFieldSize": {
+          return new Multipart.MultipartError({ reason: "FieldTooLarge", cause })
+        }
+        case "MaxPartSize": {
+          return new Multipart.MultipartError({ reason: "FileTooLarge", cause })
+        }
+        case "MaxTotalSize": {
+          return new Multipart.MultipartError({ reason: "BodyTooLarge", cause })
+        }
+      }
+    }
+    default: {
+      return new Multipart.MultipartError({ reason: "Parse", cause })
+    }
+  }
+}

packages/platform/src/internal/multipart.ts

@@ -408,6 +408,8 @@ export const toPersisted = (
           ;(persisted[part.key] as Array<string>).push(part.value)
         }
         return Effect.void
+      } else if (part.name === "") {
+        return Effect.void
       }
       const file = part
       const path = path_.join(dir, path_.basename(file.name).slice(-128))