Skip to content

Commit 715fcd0

Browse files
antfubotantfubot
committed
refactor(rpc): extract transport-agnostic rpc core and reusable ws peer hooks
createContextRpcServer owns everything about serving RPC that is independent of how peers connect (auth wiring, session resolver, auto-trust shim); createWsRpcPeerHooks shapes the per-peer lifecycle for any crossws adapter. startHttpAndWs behavior is unchanged — it now composes the two, so other transports (fetch-upgrade runtimes) can reuse the same wiring.
1 parent 74fe0ea commit 715fcd0

3 files changed

Lines changed: 272 additions & 180 deletions

File tree

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

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,148 @@
1+
import type { BirpcGroup, EventOptions } from 'birpc'
2+
import type { Peer } from 'crossws'
3+
import type { DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
4+
import type { DevframeAuthHandler } from './auth'
5+
import type { RpcFunctionsHostImpl } from './host-functions'
6+
import { AsyncLocalStorage } from 'node:async_hooks'
7+
import { createRpcServer } from 'devframe/rpc/server'
8+
import { diagnostics } from './diagnostics'
9+
10+
export interface CreateContextRpcServerOptions {
11+
context: DevframeNodeContext
12+
/** See `StartHttpAndWsOptions.auth` — same contract, transport-agnostic. */
13+
auth?: boolean | DevframeAuthHandler
14+
/** See `StartHttpAndWsOptions.authorize`. */
15+
authorize?: (methodName: string, session: DevframeNodeRpcSession) => boolean
16+
/** See `StartHttpAndWsOptions.onPeerConnect`. */
17+
onPeerConnect?: (peer: Peer, session: DevframeNodeRpcSession) => void
18+
/** See `StartHttpAndWsOptions.rpcOptions`. */
19+
rpcOptions?: Pick<
20+
EventOptions<DevframeRpcClientFunctions, DevframeRpcServerFunctions, false>,
21+
'onFunctionError' | 'onGeneralError'
22+
>
23+
}
24+
25+
export interface ContextRpcServer {
26+
rpcGroup: BirpcGroup<DevframeRpcClientFunctions, DevframeRpcServerFunctions, false>
27+
/** The resolved auth handler when `auth` was passed as one. */
28+
authHandler?: DevframeAuthHandler
29+
/**
30+
* Peer lifecycle handlers to wire into a WS transport
31+
* (`attachWsRpcTransport`'s `onConnected` / `onDisconnected`, or any other
32+
* crossws adapter's peer hooks via `createWsRpcPeerHooks`).
33+
*/
34+
onConnected?: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
35+
onDisconnected: (peer: Peer, meta: DevframeNodeRpcSessionMeta) => void
36+
}
37+
38+
/**
39+
* Bind a devframe context's registered RPC functions to a birpc group,
40+
* transport-agnostically — the shared core under `startHttpAndWs` (Node
41+
* http + WS) and the Bun fetch-upgrade tier of `createHandler`.
42+
*
43+
* Owns everything about serving RPC that is independent of *how* peers
44+
* connect: the auth handler's function registration, the
45+
* `AsyncLocalStorage`-based session resolver (so
46+
* `ctx.rpc.getCurrentRpcSession()` works inside handlers), the
47+
* `authorize` gate, and the `auth: false` auto-trust handshake shim.
48+
*/
49+
export function createContextRpcServer(options: CreateContextRpcServerOptions): ContextRpcServer {
50+
const { context } = options
51+
const rpcHost = context.rpc as unknown as RpcFunctionsHostImpl
52+
53+
const asyncStorage = new AsyncLocalStorage<DevframeNodeRpcSession>()
54+
55+
// A full auth handler (e.g. from `createInteractiveAuth`) registers its own
56+
// RPC functions and supplies both the resolver gate and the connect-time
57+
// trust hook. `authorize`/`onPeerConnect` are the lower-level escape
58+
// hatches for callers not using a full handler.
59+
const authHandler: DevframeAuthHandler | undefined = typeof options.auth === 'object' ? options.auth : undefined
60+
const effectiveAuthorize = options.authorize ?? authHandler?.authorize
61+
62+
if (authHandler) {
63+
for (const fn of authHandler.rpcFunctions) {
64+
if (!rpcHost.definitions.has(fn.name))
65+
rpcHost.register(fn)
66+
}
67+
}
68+
69+
const rpcGroup = createRpcServer<DevframeRpcClientFunctions, DevframeRpcServerFunctions>(
70+
rpcHost.functions,
71+
{
72+
rpcOptions: {
73+
// Forwarded as-is so a host with its own structured diagnostics
74+
// keeps seeing RPC failures; see `StartHttpAndWsOptions.rpcOptions`.
75+
onFunctionError: options.rpcOptions?.onFunctionError,
76+
onGeneralError: options.rpcOptions?.onGeneralError,
77+
// Wrap each RPC handler in an AsyncLocalStorage context so
78+
// `ctx.rpc.getCurrentRpcSession()` works inside handlers (used
79+
// by streaming subscribe/unsubscribe/cancel and shared-state
80+
// sync), and — when an `authorize` gate is configured — reject
81+
// the call before it ever reaches the handler. Mirrors
82+
// `packages/core/src/node/ws.ts`'s resolver.
83+
resolver(name, fn) {
84+
// eslint-disable-next-line ts/no-this-alias
85+
const rpc = this
86+
if (!fn)
87+
return undefined
88+
return async function (this: any, ...args) {
89+
const meta = rpc.$meta as DevframeNodeRpcSessionMeta
90+
if (effectiveAuthorize && !effectiveAuthorize(name, { meta, rpc: rpc as any }))
91+
throw diagnostics.DF0036({ name })
92+
return await asyncStorage.run({
93+
rpc,
94+
meta,
95+
}, async () => {
96+
return (await fn).apply(this, args)
97+
})
98+
}
99+
},
100+
},
101+
},
102+
)
103+
104+
;(rpcHost as any)._rpcGroup = rpcGroup
105+
;(rpcHost as any)._asyncStorage = asyncStorage
106+
;(rpcHost as any)._authDisabled = options.auth === false
107+
108+
// The browser client unconditionally calls `anonymous:devframe:auth` on
109+
// connect (see `client/rpc-ws.ts`). When `auth: false` is set on the
110+
// standalone server, register a noop handler that auto-trusts so the
111+
// client's hardcoded handshake succeeds. A host passing a full
112+
// `DevframeAuthHandler` already registered the real handler above, and
113+
// never opts into `auth: false`, so the two paths never overlap.
114+
if (options.auth === false && !rpcHost.definitions.has('anonymous:devframe:auth')) {
115+
rpcHost.register({
116+
name: 'anonymous:devframe:auth',
117+
type: 'action',
118+
handler: () => {
119+
const session = rpcHost.getCurrentRpcSession()
120+
if (session)
121+
session.meta.isTrusted = true
122+
return { isTrusted: true }
123+
},
124+
})
125+
}
126+
127+
const onConnected = (authHandler || options.onPeerConnect)
128+
? (peer: Peer, meta: DevframeNodeRpcSessionMeta) => {
129+
const session: DevframeNodeRpcSession = {
130+
meta,
131+
rpc: rpcGroup.clients.find(client => (client as any).$meta === meta) as any,
132+
}
133+
authHandler?.onConnect(peer, session)
134+
options.onPeerConnect?.(peer, session)
135+
}
136+
: undefined
137+
138+
const onDisconnected = (_peer: Peer, meta: DevframeNodeRpcSessionMeta): void => {
139+
rpcHost._emitSessionDisconnected(meta)
140+
}
141+
142+
return {
143+
rpcGroup,
144+
authHandler,
145+
onConnected,
146+
onDisconnected,
147+
}
148+
}

‎packages/devframe/src/node/server.ts‎

Lines changed: 14 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,15 @@ import type { BirpcGroup, EventOptions } from 'birpc'
22
import type { Peer } from 'crossws'
33
import type { NodeAdapter } from 'crossws/adapters/node'
44
import type { WsOriginRegistry } from 'devframe/rpc/transports/ws-server'
5-
import type { ConnectionMeta, DevframeNodeContext, DevframeNodeRpcSession, DevframeNodeRpcSessionMeta, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
5+
import type { ConnectionMeta, DevframeNodeContext, DevframeNodeRpcSession, DevframeRpcClientFunctions, DevframeRpcServerFunctions } from 'devframe/types'
66
import type { Server as NodeHttpServer } from 'node:http'
77
import type { DevframeAuthHandler } from './auth'
88
import type { RpcFunctionsHostImpl } from './host-functions'
9-
import { AsyncLocalStorage } from 'node:async_hooks'
109
import { createServer } from 'node:http'
11-
import { createRpcServer } from 'devframe/rpc/server'
1210
import { attachWsRpcTransport } from 'devframe/rpc/transports/ws-server'
1311
import { H3, toNodeHandler } from 'h3'
14-
import { diagnostics } from './diagnostics'
1512
import { getInternalContext } from './hub-internals/context'
13+
import { createContextRpcServer } from './rpc-core'
1614
import { formatHostForUrl, normalizeHttpServerUrl } from './utils'
1715

1816
export interface StartHttpAndWsOptions {
@@ -26,7 +24,7 @@ export interface StartHttpAndWsOptions {
2624
*/
2725
app?: H3
2826
/**
29-
* Bind the WS endpoint to a single upgrade route (e.g. `/__devframe_ws`) instead of
27+
* Bind the WS endpoint to a single upgrade route (e.g. `/__ws`) instead of
3028
* claiming every upgrade on the port. This lets the socket share a server
3129
* with other upgrade handlers (Vite HMR, a host framework's own sockets)
3230
* and is what the SPA's `__connection.json` points at. When omitted, the WS
@@ -146,56 +144,15 @@ export async function startHttpAndWs(options: StartHttpAndWsOptions): Promise<St
146144
const httpServer = options.server ?? createServer(toNodeHandler(app))
147145
const rpcHost = context.rpc as unknown as RpcFunctionsHostImpl
148146

149-
const asyncStorage = new AsyncLocalStorage<DevframeNodeRpcSession>()
150-
151-
// A full auth handler (e.g. from `createInteractiveAuth`) registers its own
152-
// RPC functions and supplies both the resolver gate and the connect-time
153-
// trust hook. `authorize`/`onPeerConnect` are the lower-level escape
154-
// hatches for callers not using a full handler.
155-
const authHandler: DevframeAuthHandler | undefined = typeof options.auth === 'object' ? options.auth : undefined
156-
const effectiveAuthorize = options.authorize ?? authHandler?.authorize
157-
158-
if (authHandler) {
159-
for (const fn of authHandler.rpcFunctions) {
160-
if (!rpcHost.definitions.has(fn.name))
161-
rpcHost.register(fn)
162-
}
163-
}
164-
165-
const rpcGroup = createRpcServer<DevframeRpcClientFunctions, DevframeRpcServerFunctions>(
166-
rpcHost.functions,
167-
{
168-
rpcOptions: {
169-
// Forwarded as-is so a host with its own structured diagnostics
170-
// keeps seeing RPC failures; see `StartHttpAndWsOptions.rpcOptions`.
171-
onFunctionError: options.rpcOptions?.onFunctionError,
172-
onGeneralError: options.rpcOptions?.onGeneralError,
173-
// Wrap each RPC handler in an AsyncLocalStorage context so
174-
// `ctx.rpc.getCurrentRpcSession()` works inside handlers (used
175-
// by streaming subscribe/unsubscribe/cancel and shared-state
176-
// sync), and — when an `authorize` gate is configured — reject
177-
// the call before it ever reaches the handler. Mirrors
178-
// `packages/core/src/node/ws.ts`'s resolver.
179-
resolver(name, fn) {
180-
// eslint-disable-next-line ts/no-this-alias
181-
const rpc = this
182-
if (!fn)
183-
return undefined
184-
return async function (this: any, ...args) {
185-
const meta = rpc.$meta as DevframeNodeRpcSessionMeta
186-
if (effectiveAuthorize && !effectiveAuthorize(name, { meta, rpc: rpc as any }))
187-
throw diagnostics.DF0036({ name })
188-
return await asyncStorage.run({
189-
rpc,
190-
meta,
191-
}, async () => {
192-
return (await fn).apply(this, args)
193-
})
194-
}
195-
},
196-
},
197-
},
198-
)
147+
// Transport-agnostic RPC core: auth wiring, session resolver, and the
148+
// peer lifecycle handlers the WS transport below plugs into.
149+
const { rpcGroup, onConnected, onDisconnected } = createContextRpcServer({
150+
context,
151+
auth: options.auth,
152+
authorize: options.authorize,
153+
onPeerConnect: options.onPeerConnect,
154+
rpcOptions: options.rpcOptions,
155+
})
199156

200157
// A dedicated WS port (the "different port" scenario) only applies when we
201158
// own the HTTP server — a shared host server already dictates the port.
@@ -214,44 +171,10 @@ export async function startHttpAndWs(options: StartHttpAndWsOptions): Promise<St
214171
// other sockets, so leave non-matching upgrades for them.
215172
destroyUnmatched: ownsHttpServer,
216173
allowedOrigins: options.allowedOrigins,
217-
onConnected: (authHandler || options.onPeerConnect)
218-
? (peer, meta) => {
219-
const session: DevframeNodeRpcSession = {
220-
meta,
221-
rpc: rpcGroup.clients.find(client => (client as any).$meta === meta) as any,
222-
}
223-
authHandler?.onConnect(peer, session)
224-
options.onPeerConnect?.(peer, session)
225-
}
226-
: undefined,
227-
onDisconnected: (_peer, meta) => {
228-
rpcHost._emitSessionDisconnected(meta)
229-
},
174+
onConnected,
175+
onDisconnected,
230176
})
231177

232-
;(rpcHost as any)._rpcGroup = rpcGroup
233-
;(rpcHost as any)._asyncStorage = asyncStorage
234-
;(rpcHost as any)._authDisabled = options.auth === false
235-
236-
// The browser client unconditionally calls `anonymous:devframe:auth` on
237-
// connect (see `client/rpc-ws.ts`). When `auth: false` is set on the
238-
// standalone server, register a noop handler that auto-trusts so the
239-
// client's hardcoded handshake succeeds. A host passing a full
240-
// `DevframeAuthHandler` already registered the real handler above, and
241-
// never opts into `auth: false`, so the two paths never overlap.
242-
if (options.auth === false && !rpcHost.definitions.has('anonymous:devframe:auth')) {
243-
rpcHost.register({
244-
name: 'anonymous:devframe:auth',
245-
type: 'action',
246-
handler: () => {
247-
const session = rpcHost.getCurrentRpcSession()
248-
if (session)
249-
session.meta.isTrusted = true
250-
return { isTrusted: true }
251-
},
252-
})
253-
}
254-
255178
// Only start listening on a server we created. A shared server is already
256179
// (or about to be) listening under the caller's control.
257180
if (ownsHttpServer) {

0 commit comments

Comments
 (0)