refactor!: transport-neutral RPC sessions and connection meta (#193)

Author: antfubot

packages/devframe/src/adapters/dev.ts

@@ -1,4 +1,4 @@
-import type { Peer } from 'crossws'
+import type { DevframeRpcConnection } from 'devframe/rpc/transports/ws-server'
 import type { DevframeAuthHandler } from '../node/auth/handler'
 import type { StartedServer } from '../node/instance-shell'
 import type { DevframeDefinition, DevframeWsOptions, McpRouteOptions } from '../types/devframe'
@@ -82,16 +82,16 @@ export interface CreateDevServerOptions {
    */
   mcp?: boolean | McpRouteOptions
   /**
-   * Called once per new WS connection, right after its session is created.
-   * Forwarded verbatim to the underlying WS transport binding.
+   * Called once per new RPC connection, right after its session is created.
+   * Forwarded verbatim to the underlying transport binding.
    */
-  onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
+  onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
   /**
-   * Called once per closed WS connection, right after its session's
+   * Called once per closed RPC connection, right after its session's
    * disconnect bookkeeping runs. Forwarded verbatim to the underlying
-   * the underlying WS transport binding.
+   * transport binding.
    */
-  onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
   /**
    * Called once the WS server is bound. Devframe stays headless
    * otherwise — wire this if you want a startup banner.

packages/devframe/src/adapters/initiate.ts

@@ -1,5 +1,4 @@
-import type { Peer } from 'crossws'
-import type { WsOriginRegistry } from 'devframe/rpc/transports/ws-server'
+import type { DevframeRpcConnection, WsOriginRegistry } from 'devframe/rpc/transports/ws-server'
 import type { ConnectionMeta, DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeStorageScope } from 'devframe/types'
 import type { Buffer } from 'node:buffer'
 import type { IncomingMessage, Server as NodeHttpServer, ServerResponse } from 'node:http'
@@ -129,15 +128,15 @@ export interface InitDevframeOptions {
    */
   destroyUnmatchedUpgrades?: boolean
   /**
-   * Called once per new WS connection, right after its session is created.
+   * Called once per new RPC connection, right after its session is created.
    * Forwarded verbatim to the underlying transport.
    */
-  onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
+  onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
   /**
-   * Called once per closed WS connection, right after the transport's own
+   * Called once per closed RPC connection, right after the transport's own
    * disconnect bookkeeping runs.
    */
-  onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
 }
 
 export interface DevframeInstance {

packages/devframe/src/node/auth/handler.ts

@@ -1,5 +1,5 @@
-import type { Peer } from 'crossws'
 import type { RpcFunctionDefinitionAny } from 'devframe/rpc'
+import type { DevframeRpcConnection } from 'devframe/rpc/transports/ws-server'
 import type { DevframeNodeRpcSession } from 'devframe/types'
 
 /**
@@ -28,12 +28,12 @@ export interface DevframeAuthHandler {
    */
   authorize: (methodName: string, session: DevframeNodeRpcSession) => boolean
   /**
-   * Connect-time trust: reads a bearer token off the peer's upgrade request
-   * (an `Authorization: Bearer <token>` header, or a static/pre-shared
-   * token from `clientAuthTokens`) and, when valid, marks the session
+   * Connect-time trust: reads a bearer token off the connection's initial
+   * request (a static/pre-shared token from `clientAuthTokens`, or a token
+   * minted by the code exchange) and, when valid, marks the session
    * trusted immediately — before the client's own handshake call.
    */
-  onConnect: (peer: Peer, session: DevframeNodeRpcSession) => void
+  onConnect: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
   /**
    * Print the current one-time code and its magic-link URL. Devframe stays
    * headless — call this yourself once the server is listening. Safe to

packages/devframe/src/node/instance-shell.ts

@@ -1,7 +1,6 @@
 import type { BirpcGroup } from 'birpc'
-import type { Peer } from 'crossws'
 import type { NodeAdapter } from 'crossws/adapters/node'
-import type { WsOriginRegistry, WsRpcTransport } from 'devframe/rpc/transports/ws-server'
+import type { DevframeRpcConnection, WsOriginRegistry, WsRpcTransport } from 'devframe/rpc/transports/ws-server'
 import type { H3 } from 'h3'
 import type { Buffer } from 'node:buffer'
 import type { IncomingMessage, Server as NodeHttpServer, ServerResponse } from 'node:http'
@@ -56,8 +55,8 @@ interface BindHttpAndWsOptions {
   auth?: boolean | DevframeAuthHandler
   allowedOrigins?: readonly string[] | WsOriginRegistry | false
   destroyUnmatched?: boolean
-  onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
-  onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
+  onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
 }
 
 /**
@@ -212,8 +211,8 @@ export interface CreateInstanceShellOptions<TContext extends DevframeNodeContext
   allowedOrigins?: readonly string[] | WsOriginRegistry | false
   /** Destroy off-route upgrades on a shared `server`. */
   destroyUnmatchedUpgrades?: boolean
-  onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
-  onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
+  onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
   /**
    * Advertise the WS route as a base-absolute path (`<base>__ws`) instead of
    * the base-relative default. A hub serves one meta document from several

packages/devframe/src/node/rpc-core.ts

@@ -1,5 +1,5 @@
 import type { BirpcGroup, EventOptions } from 'birpc'
-import type { Peer } from 'crossws'
+import type { DevframeRpcConnection } from 'devframe/rpc/transports/ws-server'
 import type { DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
 import type { DevframeAuthHandler } from './auth'
 import type { RpcFunctionsHostImpl } from './host-functions'
@@ -16,10 +16,10 @@ export interface CreateContextRpcServerOptions {
   auth?: boolean | DevframeAuthHandler
   /** Lower-level per-call gate by method name and session, without a full handler. */
   authorize?: (methodName: string, session: DevframeNodeRpcSession) => boolean
-  /** Called once per new WS connection, right after its session is created. */
-  onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
-  /** Called once per closed WS connection, after the transport's disconnect bookkeeping. */
-  onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  /** Called once per new RPC connection, right after its session is created. */
+  onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
+  /** Called once per closed RPC connection, after the transport's disconnect bookkeeping. */
+  onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
   /** Forwarded verbatim to birpc's `rpcOptions` so a host keeps seeing RPC failures. */
   rpcOptions?: Pick<
     EventOptions<DevframeRpcClientFunctions, DevframeRpcServerFunctions, false>,
@@ -32,12 +32,12 @@ export interface ContextRpcServer {
   /** The resolved auth handler when `auth` was passed as one. */
   authHandler?: DevframeAuthHandler
   /**
-   * Peer lifecycle handlers to wire into a WS transport
+   * Connection lifecycle handlers to wire into a transport binding
    * (`attachWsRpcTransport`'s `onConnected` / `onDisconnected`, or any other
    * crossws adapter's peer hooks via `createWsRpcPeerHooks`).
    */
-  onConnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
-  onDisconnected: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  onConnected?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
+  onDisconnected: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
 }
 
 /**
@@ -131,18 +131,18 @@ export function createContextRpcServer(options: CreateContextRpcServerOptions):
   }
 
   const onConnected = (authHandler || options.onPeerConnect)
-    ? (peer: Peer, meta: DevframeNodeRpcSessionMeta) => {
+    ? (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => {
         const session: DevframeNodeRpcSession = {
           meta,
           rpc: rpcGroup.clients.find(client => (client as any).$meta === meta) as any,
         }
-        authHandler?.onConnect(peer, session)
-        options.onPeerConnect?.(peer, session)
+        authHandler?.onConnect(connection, session)
+        options.onPeerConnect?.(connection, session)
       }
     : undefined
 
-  const onDisconnected = (peer: Peer, meta: DevframeNodeRpcSessionMeta): void => {
-    options.onPeerDisconnect?.(peer, meta)
+  const onDisconnected = (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta): void => {
+    options.onPeerDisconnect?.(connection, meta)
     rpcHost._emitSessionDisconnected(meta)
   }
 

packages/devframe/src/recipes/interactive-auth.ts

@@ -150,18 +150,18 @@ export function createInteractiveAuth(
   }
 
   function onConnect(
-    peer: { request?: { url?: string, headers?: { get?: (name: string) => string | null } } },
+    connection: { request?: { url?: string, headers?: { get?: (name: string) => string | null | undefined } } },
     session: DevframeNodeRpcSession,
   ): void {
     let token: string | undefined
     let requestOrigin: string | undefined
     try {
-      const url = new URL(peer.request?.url ?? '', 'http://localhost')
+      const url = new URL(connection.request?.url ?? '', 'http://localhost')
       token = url.searchParams.get(DEVFRAME_AUTH_TOKEN_QUERY_PARAM) ?? undefined
     }
     catch {}
     try {
-      requestOrigin = peer.request?.headers?.get?.('origin') ?? undefined
+      requestOrigin = connection.request?.headers?.get?.('origin') ?? undefined
     }
     catch {}
     if (!token)

packages/devframe/src/rpc/transports/session.ts

@@ -0,0 +1,84 @@
+import type { Peer } from 'crossws'
+
+/**
+ * Which wire transport produced an RPC connection. Every transport speaks
+ * the same birpc channel protocol; the kind only matters to code that needs
+ * transport-specific behavior (e.g. reaching the WS escape hatch on
+ * {@link DevframeRpcConnection.peer}).
+ */
+export type DevframeRpcTransportKind = 'websocket' | 'sse'
+
+/**
+ * Structural view of the connect-time HTTP request behind an RPC connection
+ * — the WS upgrade request, or the request opening an SSE stream. Shaped to
+ * match both the web `Request` a crossws peer exposes and a plain
+ * `node:http` request wrapper, so auth hooks can read the bearer-token
+ * query param and the `Origin` header without caring which transport (or
+ * runtime) produced the connection.
+ */
+export interface DevframeRpcConnectionRequest {
+  /** Request URL (may be path-only, e.g. `/__ws?devframe_auth_token=…`). */
+  url?: string
+  /** Header lookup, `Headers`-style. */
+  headers?: { get: (name: string) => string | null | undefined }
+}
+
+/**
+ * A live RPC connection, independent of the transport that carries it. One
+ * exists per connected client; transport bindings construct it alongside the
+ * session meta and hand both to the connect/disconnect hooks
+ * (`onPeerConnect` / `onPeerDisconnect`, {@link DevframeAuthHandler.onConnect}).
+ */
+export interface DevframeRpcConnection {
+  /** Session id — the same value as the session meta's `id`. */
+  id: number
+  /** The transport carrying this connection. */
+  transport: DevframeRpcTransportKind
+  /** The connect-time HTTP request (upgrade request / stream request). */
+  request?: DevframeRpcConnectionRequest
+  /** Send a raw wire frame to this client. Prefer the birpc channel. */
+  send?: (data: string) => void
+  /** Terminate the connection from the server side. */
+  close?: (code?: number, reason?: string) => void
+  /**
+   * The crossws peer backing a `websocket` connection — the WS-specific
+   * escape hatch (pub/sub, raw socket access). Absent on other transports.
+   */
+  peer?: Peer
+}
+
+export interface DevframeNodeRpcSessionMeta {
+  id: number
+  /** The crossws peer backing this session's socket (WS transport only). */
+  peer?: Peer
+  clientAuthToken?: string
+  isTrusted?: boolean
+  subscribedStates: Set<string>
+  /**
+   * Streams this session has subscribed to via
+   * `rpc.streaming.subscribe(channel, id)`. Tracked here for O(1) cleanup
+   * on disconnect; the wire format is `${channel}\x1F${id}`.
+   */
+  subscribedStreams?: Set<string>
+  /**
+   * Inbound streams this session is currently uploading to (via
+   * `rpc.streaming.upload(channel, id)`). Tracked for cleanup on
+   * disconnect; same wire format as `subscribedStreams`.
+   */
+  uploadingStreams?: Set<string>
+}
+
+let sessionId = 0
+
+/**
+ * Mint the per-connection session meta every transport binding shares —
+ * one id space across transports, so session bookkeeping (streaming
+ * subscriptions, shared-state sync, auth trust) never collides between a
+ * WS peer and an SSE session on the same server.
+ */
+export function createRpcSessionMeta(): DevframeNodeRpcSessionMeta {
+  return {
+    id: sessionId++,
+    subscribedStates: new Set(),
+  }
+}

packages/devframe/src/rpc/transports/ws-server.ts

@@ -7,34 +7,22 @@ import type { Server as HttpsServer, ServerOptions as HttpsServerOptions } from
 import type { AddressInfo } from 'node:net'
 import type { Duplex } from 'node:stream'
 import type { RpcFunctionDefinitionAny } from '../types'
+import type { DevframeNodeRpcSessionMeta, DevframeRpcConnection } from './session'
 import { createServer as createHttpServer } from 'node:http'
 import { createServer as createHttpsServer } from 'node:https'
 import crossws from 'crossws/adapters/node'
 import { DEVFRAME_VIEWER_ORIGIN_QUERY_PARAM, DEVFRAME_VIEWER_ORIGIN_TOKEN_QUERY_PARAM } from 'devframe/constants'
 import { randomToken, timingSafeEqual } from 'devframe/utils/crypto-token'
 import { structuredCloneParse, structuredCloneStringify } from 'devframe/utils/structured-clone'
 import { strictJsonStringify, STRUCTURED_CLONE_PREFIX } from '../serialization'
+import { createRpcSessionMeta } from './session'
 
-export interface DevframeNodeRpcSessionMeta {
-  id: number
-  /** The crossws peer backing this session's socket. */
-  peer?: Peer
-  clientAuthToken?: string
-  isTrusted?: boolean
-  subscribedStates: Set<string>
-  /**
-   * Streams this session has subscribed to via
-   * `rpc.streaming.subscribe(channel, id)`. Tracked here for O(1) cleanup
-   * on disconnect; the wire format is `${channel}\x1F${id}`.
-   */
-  subscribedStreams?: Set<string>
-  /**
-   * Inbound streams this session is currently uploading to (via
-   * `rpc.streaming.upload(channel, id)`). Tracked for cleanup on
-   * disconnect; same wire format as `subscribedStreams`.
-   */
-  uploadingStreams?: Set<string>
-}
+export type {
+  DevframeNodeRpcSessionMeta,
+  DevframeRpcConnection,
+  DevframeRpcConnectionRequest,
+  DevframeRpcTransportKind,
+} from './session'
 
 export interface WsRpcTransportOptions {
   /**
@@ -94,8 +82,8 @@ export interface WsRpcTransportOptions {
    * loses dev-time validation for `jsonSerializable: true` declarations.
    */
   definitions?: ReadonlyMap<string, Pick<RpcFunctionDefinitionAny, 'jsonSerializable'>>
-  onConnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
-  onDisconnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
+  onConnected?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
+  onDisconnected?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
   /** Override the default per-call serializer. Most callers should leave this unset. */
   serialize?: ChannelOptions['serialize']
   /** Override the default per-call deserializer. Most callers should leave this unset. */
@@ -208,8 +196,6 @@ export interface WsRpcTransport {
   close: () => Promise<void>
 }
 
-let sessionId = 0
-
 const EMPTY_DEFS: ReadonlyMap<string, Pick<RpcFunctionDefinitionAny, 'jsonSerializable'>> = new Map()
 
 function NOOP() {}
@@ -342,6 +328,7 @@ export function createWsRpcPeerHooks<
 
   interface PeerState {
     meta: DevframeNodeRpcSessionMeta
+    connection: DevframeRpcConnection
     channel: ChannelOptions
     /** birpc's inbound-message handler, registered via the channel's `on`. */
     onMessage?: (data: string) => void
@@ -350,18 +337,23 @@ export function createWsRpcPeerHooks<
 
   return {
     open: (peer) => {
-      const meta: DevframeNodeRpcSessionMeta = {
-        id: sessionId++,
+      const meta = createRpcSessionMeta()
+      meta.peer = peer
+      const connection: DevframeRpcConnection = {
+        id: meta.id,
+        transport: 'websocket',
+        request: peer.request,
+        send: data => peer.send(data),
+        close: (code, reason) => peer.close(code, reason),
         peer,
-        subscribedStates: new Set(),
       }
 
       // Per-connection state: maps an incoming request id to its method
       // name so the matching outgoing response can look the method back
       // up in `definitions` and pick the right encoder. One map per
       // session — request-id spaces don't collide across sessions.
       const pendingRequestMethods = new Map<string, string>()
-      const state: PeerState = { meta, channel: undefined as unknown as ChannelOptions }
+      const state: PeerState = { meta, connection, channel: undefined as unknown as ChannelOptions }
       const channel: ChannelOptions = {
         post: (data) => {
           peer.send(data)
@@ -404,7 +396,7 @@ export function createWsRpcPeerHooks<
       rpcGroup.updateChannels((channels) => {
         channels.push(channel)
       })
-      onConnected(peer, meta)
+      onConnected(connection, meta)
     },
     message: (peer, message) => {
       states.get(peer)?.onMessage?.(message.text())
@@ -419,7 +411,7 @@ export function createWsRpcPeerHooks<
         if (index >= 0)
           channels.splice(index, 1)
       })
-      onDisconnected(peer, state.meta)
+      onDisconnected(state.connection, state.meta)
     },
   }
 }