Use decodeReplyFromBusboy in node runtime (#48686)

When sending a multipart form data, we currently wait until the entire update finishes before we decode it via `decodeReply`. This way is universal in both runtimes (as we don't have `decodeReplyFromBusboy` in Edge), but also not efficient.

This PR changes it to use `decodeReplyFromBusboy` in the Node runtime, which can decode the payload during streaming as well as file update support.

packages/next/src/server/app-render/action-handler.ts

@@ -38,10 +38,24 @@ export async function handleAction({
     let bound = []
 
     const workerName = 'app' + pathname
-    const { decodeReply } = ComponentMod
+    const serverModuleMap = new Proxy(
+      {},
+      {
+        get: (_, id: string) => {
+          return {
+            id: serverActionsManifest.node[id].workers[workerName],
+            name: id,
+            chunks: [],
+          }
+        },
+      }
+    )
 
     try {
       if (process.env.NEXT_RUNTIME === 'edge') {
+        // Use react-server-dom-webpack/server.edge
+        const { decodeReply } = ComponentMod
+
         const webRequest = req as unknown as Request
         if (!webRequest.body) {
           throw new Error('invariant: Missing request body.')
@@ -72,44 +86,22 @@ export async function handleAction({
             formData.delete('$$id')
             bound = [formData]
           } else {
-            bound = await decodeReply(actionData)
+            bound = await decodeReply(actionData, serverModuleMap)
           }
         }
       } else {
-        if (isMultipartAction) {
-          const formFields = new FormData()
+        // Use react-server-dom-webpack/server.node which supports streaming
+        const {
+          decodeReply,
+          decodeReplyFromBusboy,
+        } = require('next/dist/compiled/react-server-dom-webpack/server.node')
 
+        if (isMultipartAction) {
           const busboy = require('busboy')
           const bb = busboy({ headers: req.headers })
-          let innerResolvor: () => void, innerRejector: (e: any) => void
-          const promise = new Promise<void>((resolve, reject) => {
-            innerResolvor = resolve
-            innerRejector = reject
-          })
-          bb.on('file', () =>
-            innerRejector(new Error('File upload is not supported.'))
-          )
-          bb.on('error', (err: any) => innerRejector(err))
-          bb.on('field', (id: any, val: any) => formFields.append(id, val))
-          bb.on('finish', () => innerResolvor())
           req.pipe(bb)
-          await promise
-
-          bound = await decodeReply(
-            formFields,
-            new Proxy(
-              {},
-              {
-                get: (_, id: string) => {
-                  return {
-                    id: serverActionsManifest.node[id].workers[workerName],
-                    name: id,
-                    chunks: [],
-                  }
-                },
-              }
-            )
-          )
+
+          bound = await decodeReplyFromBusboy(bb, serverModuleMap)
         } else {
           const { parseBody } =
             require('../api-utils/node') as typeof import('../api-utils/node')
@@ -124,7 +116,7 @@ export async function handleAction({
             formData.delete('$$id')
             bound = [formData]
           } else {
-            bound = await decodeReply(actionData)
+            bound = await decodeReply(actionData, serverModuleMap)
           }
         }
       }