Skip to content

Commit 6611125

Browse files
antfubotantfubot
committed
refactor!: transport-neutral RPC sessions and connection meta
Introduce a transport-neutral `DevframeRpcConnection` (id, transport kind, connect-time request, send/close, with the crossws `Peer` kept as a WS-specific escape hatch) and route every connection lifecycle hook through it — `onPeerConnect` / `onPeerDisconnect` options, `DevframeAuthHandler.onConnect`, and the transport binding's `onConnected` / `onDisconnected` — instead of the raw crossws `Peer`. Session meta minting moves to a shared `rpc/transports/session` module so future transports share one id space. Also widens `ConnectionMeta` ahead of the SSE transport: `backend` gains 'sse' / 'none', and an `sse` endpoint field (`ConnectionMetaSse`) mirrors the proxy-safe `websocket` shape. BREAKING CHANGE: `onPeerConnect` / `onPeerDisconnect` callbacks and `DevframeAuthHandler.onConnect` now receive a `DevframeRpcConnection` instead of a crossws `Peer`; reach the peer via `connection.peer`.
1 parent 5dc9b97 commit 6611125

20 files changed

Lines changed: 234 additions & 90 deletions

File tree

‎packages/devframe/src/adapters/dev.ts‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import type { Peer } from 'crossws'
1+
import type { DevframeRpcConnection } from 'devframe/rpc/transports/ws-server'
22
import type { DevframeAuthHandler } from '../node/auth/handler'
33
import type { StartedServer } from '../node/instance-shell'
44
import type { DevframeDefinition, DevframeWsOptions, McpRouteOptions } from '../types/devframe'
@@ -82,16 +82,16 @@ export interface CreateDevServerOptions {
8282
*/
8383
mcp?: boolean | McpRouteOptions
8484
/**
85-
* Called once per new WS connection, right after its session is created.
86-
* Forwarded verbatim to the underlying WS transport binding.
85+
* Called once per new RPC connection, right after its session is created.
86+
* Forwarded verbatim to the underlying transport binding.
8787
*/
88-
onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
88+
onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
8989
/**
90-
* Called once per closed WS connection, right after its session's
90+
* Called once per closed RPC connection, right after its session's
9191
* disconnect bookkeeping runs. Forwarded verbatim to the underlying
92-
* the underlying WS transport binding.
92+
* transport binding.
9393
*/
94-
onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
94+
onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
9595
/**
9696
* Called once the WS server is bound. Devframe stays headless
9797
* otherwise — wire this if you want a startup banner.

‎packages/devframe/src/adapters/initiate.ts‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,4 @@
1-
import type { Peer } from 'crossws'
2-
import type { WsOriginRegistry } from 'devframe/rpc/transports/ws-server'
1+
import type { DevframeRpcConnection, WsOriginRegistry } from 'devframe/rpc/transports/ws-server'
32
import type { ConnectionMeta, DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeStorageScope } from 'devframe/types'
43
import type { Buffer } from 'node:buffer'
54
import type { IncomingMessage, Server as NodeHttpServer, ServerResponse } from 'node:http'
@@ -129,15 +128,15 @@ export interface InitDevframeOptions {
129128
*/
130129
destroyUnmatchedUpgrades?: boolean
131130
/**
132-
* Called once per new WS connection, right after its session is created.
131+
* Called once per new RPC connection, right after its session is created.
133132
* Forwarded verbatim to the underlying transport.
134133
*/
135-
onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
134+
onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
136135
/**
137-
* Called once per closed WS connection, right after the transport's own
136+
* Called once per closed RPC connection, right after the transport's own
138137
* disconnect bookkeeping runs.
139138
*/
140-
onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
139+
onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
141140
}
142141

143142
export interface DevframeInstance {

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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
1-
import type { Peer } from 'crossws'
21
import type { RpcFunctionDefinitionAny } from 'devframe/rpc'
2+
import type { DevframeRpcConnection } from 'devframe/rpc/transports/ws-server'
33
import type { DevframeNodeRpcSession } from 'devframe/types'
44

55
/**
@@ -28,12 +28,12 @@ export interface DevframeAuthHandler {
2828
*/
2929
authorize: (methodName: string, session: DevframeNodeRpcSession) => boolean
3030
/**
31-
* Connect-time trust: reads a bearer token off the peer's upgrade request
32-
* (an `Authorization: Bearer <token>` header, or a static/pre-shared
33-
* token from `clientAuthTokens`) and, when valid, marks the session
31+
* Connect-time trust: reads a bearer token off the connection's initial
32+
* request (a static/pre-shared token from `clientAuthTokens`, or a token
33+
* minted by the code exchange) and, when valid, marks the session
3434
* trusted immediately — before the client's own handshake call.
3535
*/
36-
onConnect: (peer: Peer, session: DevframeNodeRpcSession) => void
36+
onConnect: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
3737
/**
3838
* Print the current one-time code and its magic-link URL. Devframe stays
3939
* headless — call this yourself once the server is listening. Safe to

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

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import type { BirpcGroup } from 'birpc'
2-
import type { Peer } from 'crossws'
32
import type { NodeAdapter } from 'crossws/adapters/node'
4-
import type { WsOriginRegistry, WsRpcTransport } from 'devframe/rpc/transports/ws-server'
3+
import type { DevframeRpcConnection, WsOriginRegistry, WsRpcTransport } from 'devframe/rpc/transports/ws-server'
54
import type { H3 } from 'h3'
65
import type { Buffer } from 'node:buffer'
76
import type { IncomingMessage, Server as NodeHttpServer, ServerResponse } from 'node:http'
@@ -56,8 +55,8 @@ interface BindHttpAndWsOptions {
5655
auth?: boolean | DevframeAuthHandler
5756
allowedOrigins?: readonly string[] | WsOriginRegistry | false
5857
destroyUnmatched?: boolean
59-
onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
60-
onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
58+
onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
59+
onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
6160
}
6261

6362
/**
@@ -212,8 +211,8 @@ export interface CreateInstanceShellOptions<TContext extends DevframeNodeContext
212211
allowedOrigins?: readonly string[] | WsOriginRegistry | false
213212
/** Destroy off-route upgrades on a shared `server`. */
214213
destroyUnmatchedUpgrades?: boolean
215-
onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
216-
onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
214+
onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
215+
onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
217216
/**
218217
* Advertise the WS route as a base-absolute path (`<base>__ws`) instead of
219218
* the base-relative default. A hub serves one meta document from several

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

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import type { BirpcGroup, EventOptions } from 'birpc'
2-
import type { Peer } from 'crossws'
2+
import type { DevframeRpcConnection } from 'devframe/rpc/transports/ws-server'
33
import type { DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
44
import type { DevframeAuthHandler } from './auth'
55
import type { RpcFunctionsHostImpl } from './host-functions'
@@ -16,10 +16,10 @@ export interface CreateContextRpcServerOptions {
1616
auth?: boolean | DevframeAuthHandler
1717
/** Lower-level per-call gate by method name and session, without a full handler. */
1818
authorize?: (methodName: string, session: DevframeNodeRpcSession) => boolean
19-
/** Called once per new WS connection, right after its session is created. */
20-
onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
21-
/** Called once per closed WS connection, after the transport's disconnect bookkeeping. */
22-
onPeerDisconnect?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
19+
/** Called once per new RPC connection, right after its session is created. */
20+
onPeerConnect?: (connection: DevframeRpcConnection, session: DevframeNodeRpcSession) => void
21+
/** Called once per closed RPC connection, after the transport's disconnect bookkeeping. */
22+
onPeerDisconnect?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
2323
/** Forwarded verbatim to birpc's `rpcOptions` so a host keeps seeing RPC failures. */
2424
rpcOptions?: Pick<
2525
EventOptions<DevframeRpcClientFunctions, DevframeRpcServerFunctions, false>,
@@ -32,12 +32,12 @@ export interface ContextRpcServer {
3232
/** The resolved auth handler when `auth` was passed as one. */
3333
authHandler?: DevframeAuthHandler
3434
/**
35-
* Peer lifecycle handlers to wire into a WS transport
35+
* Connection lifecycle handlers to wire into a transport binding
3636
* (`attachWsRpcTransport`'s `onConnected` / `onDisconnected`, or any other
3737
* crossws adapter's peer hooks via `createWsRpcPeerHooks`).
3838
*/
39-
onConnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
40-
onDisconnected: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
39+
onConnected?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
40+
onDisconnected: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
4141
}
4242

4343
/**
@@ -131,18 +131,18 @@ export function createContextRpcServer(options: CreateContextRpcServerOptions):
131131
}
132132

133133
const onConnected = (authHandler || options.onPeerConnect)
134-
? (peer: Peer, meta: DevframeNodeRpcSessionMeta) => {
134+
? (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => {
135135
const session: DevframeNodeRpcSession = {
136136
meta,
137137
rpc: rpcGroup.clients.find(client => (client as any).$meta === meta) as any,
138138
}
139-
authHandler?.onConnect(peer, session)
140-
options.onPeerConnect?.(peer, session)
139+
authHandler?.onConnect(connection, session)
140+
options.onPeerConnect?.(connection, session)
141141
}
142142
: undefined
143143

144-
const onDisconnected = (peer: Peer, meta: DevframeNodeRpcSessionMeta): void => {
145-
options.onPeerDisconnect?.(peer, meta)
144+
const onDisconnected = (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta): void => {
145+
options.onPeerDisconnect?.(connection, meta)
146146
rpcHost._emitSessionDisconnected(meta)
147147
}
148148

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -150,18 +150,18 @@ export function createInteractiveAuth(
150150
}
151151

152152
function onConnect(
153-
peer: { request?: { url?: string, headers?: { get?: (name: string) => string | null } } },
153+
connection: { request?: { url?: string, headers?: { get?: (name: string) => string | null | undefined } } },
154154
session: DevframeNodeRpcSession,
155155
): void {
156156
let token: string | undefined
157157
let requestOrigin: string | undefined
158158
try {
159-
const url = new URL(peer.request?.url ?? '', 'http://localhost')
159+
const url = new URL(connection.request?.url ?? '', 'http://localhost')
160160
token = url.searchParams.get(DEVFRAME_AUTH_TOKEN_QUERY_PARAM) ?? undefined
161161
}
162162
catch {}
163163
try {
164-
requestOrigin = peer.request?.headers?.get?.('origin') ?? undefined
164+
requestOrigin = connection.request?.headers?.get?.('origin') ?? undefined
165165
}
166166
catch {}
167167
if (!token)

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

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
import type { Peer } from 'crossws'
2+
3+
/**
4+
* Which wire transport produced an RPC connection. Every transport speaks
5+
* the same birpc channel protocol; the kind only matters to code that needs
6+
* transport-specific behavior (e.g. reaching the WS escape hatch on
7+
* {@link DevframeRpcConnection.peer}).
8+
*/
9+
export type DevframeRpcTransportKind = 'websocket' | 'sse'
10+
11+
/**
12+
* Structural view of the connect-time HTTP request behind an RPC connection
13+
* — the WS upgrade request, or the request opening an SSE stream. Shaped to
14+
* match both the web `Request` a crossws peer exposes and a plain
15+
* `node:http` request wrapper, so auth hooks can read the bearer-token
16+
* query param and the `Origin` header without caring which transport (or
17+
* runtime) produced the connection.
18+
*/
19+
export interface DevframeRpcConnectionRequest {
20+
/** Request URL (may be path-only, e.g. `/__ws?devframe_auth_token=…`). */
21+
url?: string
22+
/** Header lookup, `Headers`-style. */
23+
headers?: { get: (name: string) => string | null | undefined }
24+
}
25+
26+
/**
27+
* A live RPC connection, independent of the transport that carries it. One
28+
* exists per connected client; transport bindings construct it alongside the
29+
* session meta and hand both to the connect/disconnect hooks
30+
* (`onPeerConnect` / `onPeerDisconnect`, {@link DevframeAuthHandler.onConnect}).
31+
*/
32+
export interface DevframeRpcConnection {
33+
/** Session id — the same value as the session meta's `id`. */
34+
id: number
35+
/** The transport carrying this connection. */
36+
transport: DevframeRpcTransportKind
37+
/** The connect-time HTTP request (upgrade request / stream request). */
38+
request?: DevframeRpcConnectionRequest
39+
/** Send a raw wire frame to this client. Prefer the birpc channel. */
40+
send?: (data: string) => void
41+
/** Terminate the connection from the server side. */
42+
close?: (code?: number, reason?: string) => void
43+
/**
44+
* The crossws peer backing a `websocket` connection — the WS-specific
45+
* escape hatch (pub/sub, raw socket access). Absent on other transports.
46+
*/
47+
peer?: Peer
48+
}
49+
50+
export interface DevframeNodeRpcSessionMeta {
51+
id: number
52+
/** The crossws peer backing this session's socket (WS transport only). */
53+
peer?: Peer
54+
clientAuthToken?: string
55+
isTrusted?: boolean
56+
subscribedStates: Set<string>
57+
/**
58+
* Streams this session has subscribed to via
59+
* `rpc.streaming.subscribe(channel, id)`. Tracked here for O(1) cleanup
60+
* on disconnect; the wire format is `${channel}\x1F${id}`.
61+
*/
62+
subscribedStreams?: Set<string>
63+
/**
64+
* Inbound streams this session is currently uploading to (via
65+
* `rpc.streaming.upload(channel, id)`). Tracked for cleanup on
66+
* disconnect; same wire format as `subscribedStreams`.
67+
*/
68+
uploadingStreams?: Set<string>
69+
}
70+
71+
let sessionId = 0
72+
73+
/**
74+
* Mint the per-connection session meta every transport binding shares —
75+
* one id space across transports, so session bookkeeping (streaming
76+
* subscriptions, shared-state sync, auth trust) never collides between a
77+
* WS peer and an SSE session on the same server.
78+
*/
79+
export function createRpcSessionMeta(): DevframeNodeRpcSessionMeta {
80+
return {
81+
id: sessionId++,
82+
subscribedStates: new Set(),
83+
}
84+
}

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

Lines changed: 22 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -7,34 +7,22 @@ import type { Server as HttpsServer, ServerOptions as HttpsServerOptions } from
77
import type { AddressInfo } from 'node:net'
88
import type { Duplex } from 'node:stream'
99
import type { RpcFunctionDefinitionAny } from '../types'
10+
import type { DevframeNodeRpcSessionMeta, DevframeRpcConnection } from './session'
1011
import { createServer as createHttpServer } from 'node:http'
1112
import { createServer as createHttpsServer } from 'node:https'
1213
import crossws from 'crossws/adapters/node'
1314
import { DEVFRAME_VIEWER_ORIGIN_QUERY_PARAM, DEVFRAME_VIEWER_ORIGIN_TOKEN_QUERY_PARAM } from 'devframe/constants'
1415
import { randomToken, timingSafeEqual } from 'devframe/utils/crypto-token'
1516
import { structuredCloneParse, structuredCloneStringify } from 'devframe/utils/structured-clone'
1617
import { strictJsonStringify, STRUCTURED_CLONE_PREFIX } from '../serialization'
18+
import { createRpcSessionMeta } from './session'
1719

18-
export interface DevframeNodeRpcSessionMeta {
19-
id: number
20-
/** The crossws peer backing this session's socket. */
21-
peer?: Peer
22-
clientAuthToken?: string
23-
isTrusted?: boolean
24-
subscribedStates: Set<string>
25-
/**
26-
* Streams this session has subscribed to via
27-
* `rpc.streaming.subscribe(channel, id)`. Tracked here for O(1) cleanup
28-
* on disconnect; the wire format is `${channel}\x1F${id}`.
29-
*/
30-
subscribedStreams?: Set<string>
31-
/**
32-
* Inbound streams this session is currently uploading to (via
33-
* `rpc.streaming.upload(channel, id)`). Tracked for cleanup on
34-
* disconnect; same wire format as `subscribedStreams`.
35-
*/
36-
uploadingStreams?: Set<string>
37-
}
20+
export type {
21+
DevframeNodeRpcSessionMeta,
22+
DevframeRpcConnection,
23+
DevframeRpcConnectionRequest,
24+
DevframeRpcTransportKind,
25+
} from './session'
3826

3927
export interface WsRpcTransportOptions {
4028
/**
@@ -94,8 +82,8 @@ export interface WsRpcTransportOptions {
9482
* loses dev-time validation for `jsonSerializable: true` declarations.
9583
*/
9684
definitions?: ReadonlyMap<string, Pick<RpcFunctionDefinitionAny, 'jsonSerializable'>>
97-
onConnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
98-
onDisconnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
85+
onConnected?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
86+
onDisconnected?: (connection: DevframeRpcConnection, meta: DevframeNodeRpcSessionMeta) => void
9987
/** Override the default per-call serializer. Most callers should leave this unset. */
10088
serialize?: ChannelOptions['serialize']
10189
/** Override the default per-call deserializer. Most callers should leave this unset. */
@@ -208,8 +196,6 @@ export interface WsRpcTransport {
208196
close: () => Promise<void>
209197
}
210198

211-
let sessionId = 0
212-
213199
const EMPTY_DEFS: ReadonlyMap<string, Pick<RpcFunctionDefinitionAny, 'jsonSerializable'>> = new Map()
214200

215201
function NOOP() {}
@@ -342,6 +328,7 @@ export function createWsRpcPeerHooks<
342328

343329
interface PeerState {
344330
meta: DevframeNodeRpcSessionMeta
331+
connection: DevframeRpcConnection
345332
channel: ChannelOptions
346333
/** birpc's inbound-message handler, registered via the channel's `on`. */
347334
onMessage?: (data: string) => void
@@ -350,18 +337,23 @@ export function createWsRpcPeerHooks<
350337

351338
return {
352339
open: (peer) => {
353-
const meta: DevframeNodeRpcSessionMeta = {
354-
id: sessionId++,
340+
const meta = createRpcSessionMeta()
341+
meta.peer = peer
342+
const connection: DevframeRpcConnection = {
343+
id: meta.id,
344+
transport: 'websocket',
345+
request: peer.request,
346+
send: data => peer.send(data),
347+
close: (code, reason) => peer.close(code, reason),
355348
peer,
356-
subscribedStates: new Set(),
357349
}
358350

359351
// Per-connection state: maps an incoming request id to its method
360352
// name so the matching outgoing response can look the method back
361353
// up in `definitions` and pick the right encoder. One map per
362354
// session — request-id spaces don't collide across sessions.
363355
const pendingRequestMethods = new Map<string, string>()
364-
const state: PeerState = { meta, channel: undefined as unknown as ChannelOptions }
356+
const state: PeerState = { meta, connection, channel: undefined as unknown as ChannelOptions }
365357
const channel: ChannelOptions = {
366358
post: (data) => {
367359
peer.send(data)
@@ -404,7 +396,7 @@ export function createWsRpcPeerHooks<
404396
rpcGroup.updateChannels((channels) => {
405397
channels.push(channel)
406398
})
407-
onConnected(peer, meta)
399+
onConnected(connection, meta)
408400
},
409401
message: (peer, message) => {
410402
states.get(peer)?.onMessage?.(message.text())
@@ -419,7 +411,7 @@ export function createWsRpcPeerHooks<
419411
if (index >= 0)
420412
channels.splice(index, 1)
421413
})
422-
onDisconnected(peer, state.meta)
414+
onDisconnected(state.connection, state.meta)
423415
},
424416
}
425417
}

0 commit comments

Comments
 (0)