Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions packages/devframe/src/adapters/dev.ts
Original file line number Diff line number Diff line change
@@ -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'
Expand Down Expand Up @@ -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.
Expand Down
11 changes: 5 additions & 6 deletions packages/devframe/src/adapters/initiate.ts
Original file line number Diff line number Diff line change
@@ -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'
Expand Down Expand Up @@ -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 {
Expand Down
10 changes: 5 additions & 5 deletions packages/devframe/src/node/auth/handler.ts
Original file line number Diff line number Diff line change
@@ -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'

/**
Expand Down Expand Up @@ -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
Expand Down
11 changes: 5 additions & 6 deletions packages/devframe/src/node/instance-shell.ts
Original file line number Diff line number Diff line change
@@ -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'
Expand Down Expand Up @@ -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
}

/**
Expand Down Expand Up @@ -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
Expand Down
26 changes: 13 additions & 13 deletions packages/devframe/src/node/rpc-core.ts
Original file line number Diff line number Diff line change
@@ -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'
Expand All @@ -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>,
Expand All @@ -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
}

/**
Expand Down Expand Up @@ -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)
}

Expand Down
6 changes: 3 additions & 3 deletions packages/devframe/src/recipes/interactive-auth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
84 changes: 84 additions & 0 deletions packages/devframe/src/rpc/transports/session.ts
Original file line number Diff line number Diff line change
@@ -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(),
}
}
52 changes: 22 additions & 30 deletions packages/devframe/src/rpc/transports/ws-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
/**
Expand Down Expand Up @@ -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. */
Expand Down Expand Up @@ -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() {}
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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())
Expand All @@ -419,7 +411,7 @@ export function createWsRpcPeerHooks<
if (index >= 0)
channels.splice(index, 1)
})
onDisconnected(peer, state.meta)
onDisconnected(state.connection, state.meta)
},
}
}
Expand Down
Loading
Loading