refactor(standard-server-peer): dedupe + improve (#733)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Added a utility function to efficiently convert Blob, File, Response,
or Request objects into buffers.
* Introduced a helper to standardize server peer message handling across
adapters.
  * Added a utility to detect event stream headers.

* **Improvements**
* Broadened supported message types to include Uint8Array for encoded
messages and message ports.
* Unified and simplified message handling logic in server adapters,
delegating processing to a shared utility.
* Enhanced body and header serialization/deserialization for messages,
improving consistency.
* Updated buffer handling in client and server WebSocket adapters for
better compatibility.

* **Bug Fixes**
* Improved compatibility for buffer extraction from Blobs and similar
objects.

* **Documentation**
* Expanded test coverage for new utilities and updated tests to reflect
changes in buffer types and message handling.

* **Chores**
* Updated ESLint rules to discourage unsupported `.bytes` property usage
and recommend the new utility.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: dinwwwh

eslint.config.js

@@ -15,6 +15,10 @@ export default antfu({
         name: ['JSON', 'stringify'],
         message: 'JSON.stringify can return undefined, use stringifyJSON instead',
       },
+      {
+        name: ['*', 'bytes'],
+        message: 'Request/Blob/Response/... .bytes is not widely supported, use readAsBuffer instead',
+      },
     ],
     'no-restricted-imports': ['error', {
       patterns: [{

packages/client/src/adapters/message-port/message-port.ts

@@ -24,7 +24,7 @@ export type SupportedMessagePort = Pick<MessagePort, 'addEventListener' | 'postM
 /**
  *  Message port can support [The structured clone algorithm](https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm)
  */
-export type SupportedMessagePortData = string | ArrayBufferLike
+export type SupportedMessagePortData = string | ArrayBufferLike | Uint8Array
 
 export function postMessagePortMessage(port: SupportedMessagePort, data: SupportedMessagePortData): void {
   port.postMessage(data)

packages/client/src/adapters/websocket/index.ts

@@ -1,2 +1,2 @@
-export * from './link-websocket-client'
+export * from './link-client'
 export * from './rpc-link'

…pters/websocket/link-websocket-client.ts …nt/src/adapters/websocket/link-client.tspackages/client/src/adapters/websocket/link-websocket-client.ts renamed to packages/client/src/adapters/websocket/link-client.ts packages/client/src/adapters/websocket/link-websocket-client.ts renamed to packages/client/src/adapters/websocket/link-client.ts

@@ -1,6 +1,7 @@
 import type { StandardLazyResponse, StandardRequest } from '@orpc/standard-server'
 import type { ClientContext, ClientOptions } from '../../types'
 import type { StandardLinkClient } from '../standard'
+import { readAsBuffer } from '@orpc/shared'
 import { ClientPeer } from '@orpc/standard-server-peer'
 
 export interface experimental_LinkWebsocketClientOptions {
@@ -29,7 +30,7 @@ export class experimental_LinkWebsocketClient<T extends ClientContext> implement
 
     options.websocket.addEventListener('message', async (event) => {
       const message = event.data instanceof Blob
-        ? await event.data.arrayBuffer()
+        ? await readAsBuffer(event.data)
         : event.data
 
       this.peer.message(message)

packages/client/src/adapters/websocket/rpc-link.ts

@@ -1,8 +1,8 @@
 import type { ClientContext } from '../../types'
 import type { StandardRPCLinkOptions } from '../standard'
-import type { experimental_LinkWebsocketClientOptions as LinkWebsocketClientOptions } from './link-websocket-client'
+import type { experimental_LinkWebsocketClientOptions as LinkWebsocketClientOptions } from './link-client'
 import { StandardRPCLink } from '../standard'
-import { experimental_LinkWebsocketClient as LinkWebsocketClient } from './link-websocket-client'
+import { experimental_LinkWebsocketClient as LinkWebsocketClient } from './link-client'
 
 export interface experimental_RPCLinkOptions<T extends ClientContext>
   extends Omit<StandardRPCLinkOptions<T>, 'url' | 'headers' | 'method' | 'fallbackMethod' | 'maxUrlLength'>, LinkWebsocketClientOptions {}

packages/server/package.json

@@ -35,6 +35,11 @@
         "import": "./dist/adapters/standard/index.mjs",
         "default": "./dist/adapters/standard/index.mjs"
       },
+      "./standard-peer": {
+        "types": "./dist/adapters/standard-peer/index.d.mts",
+        "import": "./dist/adapters/standard-peer/index.mjs",
+        "default": "./dist/adapters/standard-peer/index.mjs"
+      },
       "./fetch": {
         "types": "./dist/adapters/fetch/index.d.mts",
         "import": "./dist/adapters/fetch/index.mjs",
@@ -82,6 +87,7 @@
     "./plugins": "./src/plugins/index.ts",
     "./hibernation": "./src/hibernation/index.ts",
     "./standard": "./src/adapters/standard/index.ts",
+    "./standard-peer": "./src/adapters/standard-peer/index.ts",
     "./fetch": "./src/adapters/fetch/index.ts",
     "./node": "./src/adapters/node/index.ts",
     "./aws-lambda": "./src/adapters/aws-lambda/index.ts",

packages/server/src/adapters/bun-ws/handler.ts

@@ -1,13 +1,17 @@
 import type { MaybeOptionalOptions } from '@orpc/shared'
 import type { Context } from '../../context'
 import type { StandardHandler } from '../standard'
-import type { FriendlyStandardHandleOptions } from '../standard/utils'
+import type {
+  experimental_HandleStandardServerPeerMessageOptions as HandleStandardServerPeerMessageOptions,
+} from '../standard-peer'
 import { resolveMaybeOptionalOptions } from '@orpc/shared'
 import { ServerPeer } from '@orpc/standard-server-peer'
-import { resolveFriendlyStandardHandleOptions } from '../standard/utils'
+import {
+  experimental_handleStandardServerPeerMessage as handleStandardServerPeerMessage,
+} from '../standard-peer'
 
 export interface ServerWebSocket {
-  send(message: string | ArrayBufferLike): number
+  send(message: string | ArrayBufferLike | Uint8Array): number
 }
 
 export class experimental_BunWsHandler<T extends Context> {
@@ -20,8 +24,8 @@ export class experimental_BunWsHandler<T extends Context> {
 
   async message(
     ws: ServerWebSocket,
-    message: string | { buffer: ArrayBufferLike },
-    ...rest: MaybeOptionalOptions<Omit<FriendlyStandardHandleOptions<T>, 'prefix'>>
+    message: string | ArrayBufferView,
+    ...rest: MaybeOptionalOptions<HandleStandardServerPeerMessageOptions<T>>
   ): Promise<void> {
     let peer = this.peers.get(ws)
 
@@ -31,17 +35,16 @@ export class experimental_BunWsHandler<T extends Context> {
       }))
     }
 
-    const [id, request] = await peer.message(typeof message === 'string' ? message : message.buffer)
+    const encodedMessage = typeof message === 'string'
+      ? message
+      : new Uint8Array(message.buffer, message.byteOffset, message.byteLength)
 
-    if (!request) {
-      return
-    }
-
-    const options = resolveFriendlyStandardHandleOptions(resolveMaybeOptionalOptions(rest))
-
-    const { response } = await this.standardHandler.handle({ ...request, body: () => Promise.resolve(request.body) }, options)
-
-    await peer.response(id, response ?? { status: 404, headers: {}, body: 'No procedure matched' })
+    await handleStandardServerPeerMessage(
+      this.standardHandler,
+      peer,
+      encodedMessage,
+      resolveMaybeOptionalOptions(rest),
+    )
   }
 
   close(ws: ServerWebSocket): void {

packages/server/src/adapters/bun-ws/rpc-handler.test.ts

@@ -1,4 +1,4 @@
-import { decodeResponseMessage, encodeRequestMessage, MessageType } from '@orpc/standard-server-peer'
+import { encodeRequestMessage, MessageType } from '@orpc/standard-server-peer'
 import { os } from '../../builder'
 import { experimental_RPCHandler as RPCHandler } from './rpc-handler'
 
@@ -16,92 +16,37 @@ describe('rpcHandler', async () => {
       await new Promise(resolve => setTimeout(resolve, 10))
       return 'pong'
     }),
-
-    file: os.handler(async ({ signal: _signal }) => {
-      signal = _signal!
-      await new Promise(resolve => setTimeout(resolve, 10))
-      return new Blob(['pong'])
-    }),
   })
 
   const wss = {
     send: vi.fn(),
   }
 
-  const ping_request_message = await encodeRequestMessage('19', MessageType.REQUEST, {
+  const string_request_message = await encodeRequestMessage('19', MessageType.REQUEST, {
     url: new URL('orpc:/ping'),
     body: { json: 'input' },
     headers: {},
     method: 'POST',
   }) as string
 
-  const file_request_message = {
-    buffer: new TextEncoder().encode(await encodeRequestMessage('19', MessageType.REQUEST, {
-      url: new URL('orpc:/file'),
-      body: { json: 'input' },
-      headers: {},
-      method: 'POST',
-    }) as string),
-  }
-
-  const not_found_request_message = await encodeRequestMessage('19', MessageType.REQUEST, {
-    url: new URL('orpc:/not-found'),
-    body: { json: 'input' },
-    headers: {},
-    method: 'POST',
-  }) as string
-
-  const abort_message = await encodeRequestMessage('19', MessageType.ABORT_SIGNAL, undefined) as string
+  const buffer_request_message = {
+    buffer: new TextEncoder().encode(string_request_message),
+  } as any
 
-  it('on success', async () => {
-    handler.message(wss, ping_request_message)
+  it('work with string event', async () => {
+    handler.message(wss, string_request_message)
 
     await vi.waitFor(() => expect(wss.send).toHaveBeenCalledTimes(1))
-
-    const [id,, payload] = (await decodeResponseMessage(wss.send.mock.calls[0]![0]))
-
-    expect(id).toBeTypeOf('string')
-    expect(payload).toEqual({
-      status: 200,
-      headers: {},
-      body: { json: 'pong' },
-    })
   })
 
-  it('on success with buffer data', async () => {
-    handler.message(wss, file_request_message)
+  it('work with buffer event', async () => {
+    handler.message(wss, buffer_request_message)
 
     await vi.waitFor(() => expect(wss.send).toHaveBeenCalledTimes(1))
-    const [id, , payload] = (await decodeResponseMessage(wss.send.mock.calls[0]![0]))
-
-    expect(id).toBeTypeOf('string')
-    expect(payload).toEqual({
-      status: 200,
-      headers: {
-        'content-type': expect.any(String),
-      },
-      body: expect.any(FormData),
-    })
-
-    expect(await (payload as any).body.get('0').text()).toBe('pong')
-  })
-
-  it('on abort signal', async () => {
-    handler.message(wss, ping_request_message)
-
-    await new Promise(resolve => setTimeout(resolve, 0))
-
-    expect(signal.aborted).toBe(false)
-    expect(wss.send).not.toHaveBeenCalled()
-
-    handler.message(wss, abort_message)
-
-    await vi.waitFor(() => expect(signal.aborted).toBe(true))
-    expect(wss.send).not.toHaveBeenCalled()
   })
 
-  it('on close', async () => {
-    handler.message(wss, ping_request_message)
+  it('abort on close', async () => {
+    handler.message(wss, string_request_message)
 
     await new Promise(resolve => setTimeout(resolve, 0))
 
@@ -112,19 +57,4 @@ describe('rpcHandler', async () => {
     await vi.waitFor(() => expect(signal.aborted).toBe(true))
     expect(wss.send).not.toHaveBeenCalled()
   })
-
-  it('on no procedure matched', async () => {
-    handler.message(wss, not_found_request_message)
-
-    await new Promise(resolve => setTimeout(resolve, 0))
-
-    const [id,, payload] = (await decodeResponseMessage(wss.send.mock.calls[0]![0]))
-
-    expect(id).toBeTypeOf('string')
-    expect(payload).toEqual({
-      status: 404,
-      headers: {},
-      body: 'No procedure matched',
-    })
-  })
 })

packages/server/src/adapters/crossws/handler.ts

@@ -2,10 +2,14 @@ import type { MaybeOptionalOptions } from '@orpc/shared'
 import type { Message, Peer } from 'crossws'
 import type { Context } from '../../context'
 import type { StandardHandler } from '../standard'
-import type { FriendlyStandardHandleOptions } from '../standard/utils'
+import type {
+  experimental_HandleStandardServerPeerMessageOptions as HandleStandardServerPeerMessageOptions,
+} from '../standard-peer'
 import { resolveMaybeOptionalOptions } from '@orpc/shared'
 import { ServerPeer } from '@orpc/standard-server-peer'
-import { resolveFriendlyStandardHandleOptions } from '../standard/utils'
+import {
+  experimental_handleStandardServerPeerMessage as handleStandardServerPeerMessage,
+} from '../standard-peer'
 
 export class experimental_CrosswsHandler<T extends Context> {
   private readonly peers: WeakMap<Pick<Peer, 'send'>, ServerPeer> = new WeakMap()
@@ -18,7 +22,7 @@ export class experimental_CrosswsHandler<T extends Context> {
   async message(
     ws: Pick<Peer, 'send'>,
     message: Pick<Message, 'rawData' | 'uint8Array'>,
-    ...rest: MaybeOptionalOptions<Omit<FriendlyStandardHandleOptions<T>, 'prefix'>>
+    ...rest: MaybeOptionalOptions<HandleStandardServerPeerMessageOptions<T>>
   ): Promise<void> {
     let peer = this.peers.get(ws)
 
@@ -28,17 +32,14 @@ export class experimental_CrosswsHandler<T extends Context> {
       }))
     }
 
-    const [id, request] = await peer.message(typeof message.rawData === 'string' ? message.rawData : message.uint8Array())
+    const encodedMessage = typeof message.rawData === 'string' ? message.rawData : message.uint8Array()
 
-    if (!request) {
-      return
-    }
-
-    const options = resolveFriendlyStandardHandleOptions(resolveMaybeOptionalOptions(rest))
-
-    const { response } = await this.standardHandler.handle({ ...request, body: () => Promise.resolve(request.body) }, options)
-
-    await peer.response(id, response ?? { status: 404, headers: {}, body: 'No procedure matched' })
+    await handleStandardServerPeerMessage(
+      this.standardHandler,
+      peer,
+      encodedMessage,
+      resolveMaybeOptionalOptions(rest),
+    )
   }
 
   close(ws: Pick<Peer, 'send'>): void {