diff --git a/.gitignore b/.gitignore index 6e80429..5a891a2 100644 --- a/.gitignore +++ b/.gitignore @@ -41,3 +41,4 @@ __pycache__/ # Compiled test scripts packages/kkrpc/__tests__/scripts/*.js +references/ diff --git a/docs/superpowers/plans/2026-06-05-browser-lite-serialization-implementation-plan.md b/docs/superpowers/plans/2026-06-05-browser-lite-serialization-implementation-plan.md new file mode 100644 index 0000000..0d13fd5 --- /dev/null +++ b/docs/superpowers/plans/2026-06-05-browser-lite-serialization-implementation-plan.md @@ -0,0 +1,1158 @@ +# Browser Lite Serialization Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add a `kkrpc/browser-lite` entrypoint that keeps the existing `new RPCChannel(...)` facade while avoiding any static SuperJSON dependency. + +**Architecture:** Split the current serialization/channel coupling into shared protocol types, a JSON-only serializer, a full SuperJSON serializer, and a shared `RPCChannelCore` that receives its serializer runtime from thin public wrappers. Keep `kkrpc` and `kkrpc/browser` fully compatible; add `kkrpc/browser-lite` as an opt-in JSON-only browser build. + +**Tech Stack:** TypeScript, Bun tests, tsdown build, pnpm workspace, kkrpc `IoInterface` adapters, existing package export verification. + +--- + +## Commit Policy + +This plan includes checkpoint steps for small commits because the implementation is broad. Only run the commit commands if the user explicitly requests commits. Otherwise, stop at verification and leave changes unstaged or staged as appropriate for review. + +## File Structure + +Create or modify these files: + +- Create: `packages/kkrpc/src/serialization-types.ts` +- Create: `packages/kkrpc/src/serialization-json.ts` +- Create: `packages/kkrpc/src/serialization-full.ts` +- Modify: `packages/kkrpc/src/serialization.ts` +- Create: `packages/kkrpc/src/channel-core.ts` +- Modify: `packages/kkrpc/src/channel.ts` +- Create: `packages/kkrpc/src/channel-lite.ts` +- Create: `packages/kkrpc/browser-lite-mod.ts` +- Modify: `packages/kkrpc/tsdown.config.ts` +- Modify: `packages/kkrpc/package.json` +- Modify: `packages/kkrpc/deno.json` +- Modify: `packages/kkrpc/__tests__/serialization.test.ts` +- Create: `packages/kkrpc/__tests__/browser-lite.test.ts` +- Create: `packages/kkrpc/scripts/check-browser-lite-bundle.ts` +- Modify: `packages/kkrpc/package.json` scripts +- Modify: `README.md` + +Boundary decisions: + +- `serialization-types.ts` owns protocol types only. It imports nothing from SuperJSON or transfer modules. +- `serialization-json.ts` owns JSON/string encoding, structured envelope encoding, transfer slot processing, and error serialization. It imports no SuperJSON. +- `serialization-full.ts` owns SuperJSON support and full default behavior. +- `serialization.ts` remains a compatibility barrel for existing imports. +- `channel-core.ts` owns the RPC state machine and imports protocol types from `serialization-types.ts`, not `serialization.ts`. +- `channel.ts` and `channel-lite.ts` are thin public wrappers that export the same `RPCChannel` class facade. + +### Task 1: Extract Protocol Types And JSON Serializer + +**Files:** +- Create: `packages/kkrpc/src/serialization-types.ts` +- Create: `packages/kkrpc/src/serialization-json.ts` +- Modify: `packages/kkrpc/__tests__/serialization.test.ts` + +- [ ] **Step 1: Add failing JSON serializer tests** + +Append these tests to `packages/kkrpc/__tests__/serialization.test.ts` before creating the new modules: + +```ts +import { + decodeJsonMessage, + encodeJsonMessage, + jsonSerializationRuntime, + type Message +} from "../src/serialization-json.ts" +``` + +Add this `describe` block after the existing `describe("Serializer", ...)` block: + +```ts +describe("JSON-only serialization", () => { + test("round-trips JSON messages without SuperJSON", async () => { + const message: Message = { + id: "json-1", + method: "echo", + args: ["hello", new Uint8Array([1, 2, 3])], + type: "request" + } + + const serialized = encodeJsonMessage(message) + const deserialized = await decodeJsonMessage(serialized) + + expect(deserialized).toEqual({ ...message, version: "json" } as Message) + }) + + test("rejects SuperJSON-looking strings with a clear lite error", async () => { + const superjsonLike = '{"json":{"id":"1","method":"echo","args":[],"type":"request"}}\n' + + await expect(decodeJsonMessage(superjsonLike)).rejects.toThrow( + "Received a SuperJSON-encoded kkrpc message" + ) + }) + + test("encodes structured envelopes without SuperJSON", async () => { + const message: Message = { + id: "structured-1", + method: "echo", + args: ["hello"], + type: "request" + } + + const encoded = jsonSerializationRuntime.encodeMessage(message, {}, true) + + expect(encoded.mode).toBe("structured") + if (encoded.mode !== "structured") { + expect.unreachable("expected structured encoding") + } + expect(encoded.data.payload).toEqual(message) + expect(await jsonSerializationRuntime.decodeMessage(encoded.data)).toEqual(message) + }) +}) +``` + +- [ ] **Step 2: Run the focused test and verify it fails** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/serialization.test.ts +``` + +Expected: FAIL because `../src/serialization-json.ts` does not exist. + +- [ ] **Step 3: Create `serialization-types.ts`** + +Create `packages/kkrpc/src/serialization-types.ts` with this content: + +```ts +/** + * Shared wire protocol types for kkrpc serializers and channels. + * This file has no runtime dependencies so lite entrypoints can reuse the + * protocol shape without importing SuperJSON or full serialization helpers. + */ + +export interface Message { + id: string + method: string + args: T + type: + | "request" + | "response" + | "callback" + | "get" + | "set" + | "construct" + | "stream-chunk" + | "stream-end" + | "stream-error" + | "stream-cancel" + callbackIds?: string[] + version?: "json" | "superjson" + meta?: RPCMessageMetadata + path?: string[] + value?: unknown + transferSlots?: TransferSlot[] +} + +/** + * Optional out-of-band metadata carried with an RPC message. + */ +export interface RPCMessageMetadata { + traceparent?: string + tracestate?: string + baggage?: string + requestId?: string + sessionId?: string + runtime?: Record + [key: string]: unknown +} + +export interface Response { + result?: T + error?: string | EnhancedError +} + +export interface EnhancedError { + name: string + message: string + stack?: string + cause?: unknown + [key: string]: unknown +} + +export interface SerializationOptions { + version?: "json" | "superjson" +} + +export const TRANSFER_SLOT_PREFIX = "__kkrpc_transfer_" + +export interface TransferSlot { + type: "raw" | "handler" + handlerName?: string + metadata?: unknown + /** Random per-slot token that proves a placeholder was generated for this message. */ + token?: string +} + +export interface WireEnvelope { + version: 2 + payload: Message + transferSlots?: TransferSlot[] + encoding: "object" + __transferredValues?: unknown[] +} + +export type WireV1 = string +export type WireFormat = WireV1 | WireEnvelope + +export type EncodedMessage = + | { mode: "string"; data: string } + | { mode: "structured"; data: WireEnvelope } + +export interface RPCSerializationRuntime { + encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues?: unknown[] + ): EncodedMessage + decodeMessage(raw: WireFormat): Promise> + serializeError(error: Error): EnhancedError + deserializeError(error: EnhancedError): Error +} +``` + +- [ ] **Step 4: Create `serialization-json.ts` from the existing JSON-safe logic** + +Create `packages/kkrpc/src/serialization-json.ts` by moving the JSON-safe logic out of `serialization.ts`. The file must start with this module comment and must not import `superjson`: + +```ts +/** + * JSON-only kkrpc serialization runtime. + * Browser-lite imports this module to keep SuperJSON out of the dependency graph + * while preserving JSON string messages, structured envelopes, transfer slots, + * and rich error serialization. + */ +``` + +The exports must include: + +```ts +export { + TRANSFER_SLOT_PREFIX, + type EncodedMessage, + type EnhancedError, + type Message, + type RPCMessageMetadata, + type RPCSerializationRuntime, + type Response, + type SerializationOptions, + type TransferSlot, + type WireEnvelope, + type WireFormat, + type WireV1 +} from "./serialization-types.ts" +``` + +Implement these public functions using the current code from `serialization.ts`: + +```ts +export function serializeError(error: Error): EnhancedError +export function deserializeError(enhanced: EnhancedError): Error +export function encodeJsonMessage(message: Message): string +export function decodeJsonMessage(message: string): Promise> +export function encodeStructuredMessage( + message: Message, + transferredValues?: unknown[] +): WireEnvelope +export function decodeStructuredMessage(raw: WireEnvelope): Promise> +export function encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues?: unknown[] +): EncodedMessage +export async function decodeMessage(raw: WireFormat): Promise> +export function processValueForTransfer(...): unknown +export function reconstructValueFromTransfer(...): unknown +export const jsonSerializationRuntime: RPCSerializationRuntime +``` + +Use the existing `replacer`, `reviver`, transfer placeholder, `processValueForTransfer`, and `reconstructValueFromTransfer` logic from `serialization.ts`. Keep the existing tagged placeholder object behavior from `origin/main`. + +`decodeJsonMessage` must reject SuperJSON-looking strings before calling `JSON.parse`: + +```ts +if (message.trimStart().startsWith('{"json":')) { + throw new Error( + 'Received a SuperJSON-encoded kkrpc message, but this entrypoint is JSON-only. Use kkrpc/browser or configure both endpoints with serialization.version = "json".' + ) +} +``` + +`encodeMessage` in this module must reject SuperJSON options: + +```ts +if (options.version === "superjson") { + throw new Error( + 'SuperJSON serialization is not available in kkrpc/browser-lite. Use kkrpc/browser or configure both endpoints with serialization.version = "json".' + ) +} +``` + +`encodeStructuredMessage` and `decodeStructuredMessage` must not inspect `options.version`; they operate on object envelopes and are shared by the full serializer for transfer-capable transports. + +- [ ] **Step 5: Run focused JSON serialization tests** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/serialization.test.ts +``` + +Expected: the new `JSON-only serialization` tests pass. Existing tests may still pass or may fail if `serialization.ts` has not yet been refactored; do not proceed until the new module compiles. + +- [ ] **Step 6: Checkpoint** + +If commits are explicitly requested: + +```bash +git add packages/kkrpc/src/serialization-types.ts packages/kkrpc/src/serialization-json.ts packages/kkrpc/__tests__/serialization.test.ts +git commit -m "refactor(kkrpc): extract json serialization runtime" +``` + +### Task 2: Add Full Serializer And Preserve `serialization.ts` + +**Files:** +- Create: `packages/kkrpc/src/serialization-full.ts` +- Modify: `packages/kkrpc/src/serialization.ts` +- Modify: `packages/kkrpc/__tests__/serialization.test.ts` + +- [ ] **Step 1: Add compatibility test for existing `serialization.ts` imports** + +Ensure `packages/kkrpc/__tests__/serialization.test.ts` still imports existing APIs from `../src/serialization.ts`: + +```ts +import superjson from "superjson" +import { deserializeMessage, serializeMessage, type Message } from "../src/serialization.ts" +``` + +Add this test inside `describe("Serializer", ...)`: + +```ts +test("default compatibility serializer still uses SuperJSON", async () => { + const message: Message = { + id: "superjson-default", + method: "date.echo", + args: [new Date("2026-06-05T00:00:00.000Z")], + type: "request" + } + + const serialized = serializeMessage(message) + expect(serialized.startsWith('{"json":')).toBe(true) + + const deserialized = await deserializeMessage(serialized) + expect(deserialized.args[0]).toBeInstanceOf(Date) +}) +``` + +- [ ] **Step 2: Run compatibility test and verify current behavior** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/serialization.test.ts +``` + +Expected: PASS before the refactor, or FAIL only because Task 1 imports are not yet wired. After Task 2 implementation this test must pass. + +- [ ] **Step 3: Create `serialization-full.ts`** + +Create `packages/kkrpc/src/serialization-full.ts`: + +```ts +/** + * Full kkrpc serialization runtime with SuperJSON support. + * Existing package entrypoints import this module to preserve the current + * default behavior for Date, Map, Set, BigInt, Uint8Array, and legacy messages. + */ + +import superjson from "superjson" +import { + decodeMessage as decodeJsonWireMessage, + decodeStructuredMessage, + deserializeError, + encodeJsonMessage, + encodeStructuredMessage, + processValueForTransfer, + reconstructValueFromTransfer, + serializeError +} from "./serialization-json.ts" +import type { + EncodedMessage, + EnhancedError, + Message, + RPCSerializationRuntime, + SerializationOptions, + WireFormat +} from "./serialization-types.ts" + +export type { + EncodedMessage, + EnhancedError, + Message, + RPCMessageMetadata, + RPCSerializationRuntime, + Response, + SerializationOptions, + TransferSlot, + WireEnvelope, + WireFormat, + WireV1 +} from "./serialization-types.ts" +export { + deserializeError, + processValueForTransfer, + reconstructValueFromTransfer, + serializeError +} + +export function serializeMessage( + message: Message, + options: SerializationOptions = {} +): string { + const version = options.version || "superjson" + const msgWithVersion = { ...message, version } + return version === "json" ? encodeJsonMessage(message) : superjson.stringify(msgWithVersion) + "\n" +} + +export function deserializeMessage(message: string): Promise> { + return new Promise((resolve, reject) => { + try { + if (message.trimStart().startsWith('{"json":')) { + resolve(superjson.parse>(message)) + return + } + decodeJsonWireMessage(message).then(resolve, reject) + } catch (error) { + console.error("failed to parse message", typeof message, message, error) + reject(error) + } + }) +} + +export function encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues: unknown[] = [] +): EncodedMessage { + if (!withTransfers) { + return { + mode: "string", + data: serializeMessage(message, options) + } + } + return { + mode: "structured", + data: encodeStructuredMessage(message, transferredValues) + } +} + +export async function decodeMessage(raw: WireFormat): Promise> { + if (typeof raw === "string") { + return deserializeMessage(raw) + } + return decodeStructuredMessage(raw) +} + +export const fullSerializationRuntime: RPCSerializationRuntime = { + encodeMessage, + decodeMessage, + serializeError, + deserializeError +} +``` + +This full serializer shares structured-envelope helpers with the JSON serializer, but it must not call the JSON serializer's generic `encodeMessage` with `version: "superjson"`. + +- [ ] **Step 4: Replace `serialization.ts` with a compatibility barrel** + +Replace `packages/kkrpc/src/serialization.ts` with: + +```ts +/** + * Compatibility exports for kkrpc's full serialization runtime. + * Existing users import protocol types and helpers from this path, so it remains + * SuperJSON-enabled. Browser-lite must import `serialization-json.ts` or + * `serialization-types.ts` directly instead of this barrel. + */ + +export * from "./serialization-types.ts" +export * from "./serialization-full.ts" +``` + +- [ ] **Step 5: Run serialization tests** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/serialization.test.ts +``` + +Expected: PASS. The default serializer still preserves `Date`; JSON-only serializer still rejects SuperJSON-looking strings. + +- [ ] **Step 6: Run transfer tests** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/transfer.test.ts +``` + +Expected: PASS. Existing imports from `../src/serialization.ts` continue to expose transfer processing helpers. + +- [ ] **Step 7: Checkpoint** + +If commits are explicitly requested: + +```bash +git add packages/kkrpc/src/serialization-full.ts packages/kkrpc/src/serialization.ts packages/kkrpc/__tests__/serialization.test.ts +git commit -m "refactor(kkrpc): keep full serialization compatibility" +``` + +### Task 3: Extract Channel Core And Add Full/Lite Wrappers + +**Files:** +- Create: `packages/kkrpc/src/channel-core.ts` +- Modify: `packages/kkrpc/src/channel.ts` +- Create: `packages/kkrpc/src/channel-lite.ts` +- Create: `packages/kkrpc/__tests__/browser-lite.test.ts` + +- [ ] **Step 1: Add failing lite RPC integration tests** + +Create `packages/kkrpc/__tests__/browser-lite.test.ts`: + +```ts +/** + * Browser-lite integration tests. + * These tests exercise the JSON-only RPCChannel wrapper without importing the + * full SuperJSON-enabled channel facade. + */ + +import { describe, expect, test } from "bun:test" +import { WebSocketServer } from "ws" +import { WebSocketClientIO, WebSocketServerIO } from "../src/adapters/websocket.ts" +import { RPCChannel } from "../src/channel-lite.ts" +import type { IoInterface } from "../src/interface.ts" + +interface TestAPI { + echo(message: string): Promise + add(a: number, b: number): Promise +} + +const testApi: TestAPI = { + echo: async (message) => message, + add: async (a, b) => a + b +} + +describe("browser-lite RPCChannel", () => { + test("calls remote methods with JSON serialization", async () => { + const port = 3095 + const wss = new WebSocketServer({ port }) + wss.on("connection", (ws) => { + const serverIO = new WebSocketServerIO(ws) + new RPCChannel(serverIO, { expose: testApi }) + }) + + const clientIO = new WebSocketClientIO({ url: `ws://localhost:${port}` }) + const rpc = new RPCChannel<{}, TestAPI, IoInterface>(clientIO) + const api = rpc.getAPI() + + try { + expect(await api.echo("hello")).toBe("hello") + expect(await api.add(2, 5)).toBe(7) + } finally { + clientIO.destroy() + wss.close() + } + }) + + test("rejects explicit SuperJSON serialization option", () => { + const io = { + name: "closed-test-io", + read: async () => null, + write: async () => {}, + on: () => {}, + off: () => {} + } satisfies IoInterface + + expect( + () => + new RPCChannel<{}, TestAPI, IoInterface>(io, { + serialization: { version: "superjson" } + }) + ).toThrow("SuperJSON serialization is not available") + }) +}) +``` + +- [ ] **Step 2: Run lite tests and verify they fail** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/browser-lite.test.ts +``` + +Expected: FAIL because `../src/channel-lite.ts` does not exist. + +- [ ] **Step 3: Copy current channel implementation to `channel-core.ts`** + +Run: + +```bash +cp packages/kkrpc/src/channel.ts packages/kkrpc/src/channel-core.ts +``` + +Then edit the top of `packages/kkrpc/src/channel-core.ts`: + +```ts +/** + * Shared RPC channel state machine. + * Public entrypoints wrap this class with a serialization runtime so lite + * browser builds can avoid importing SuperJSON while full builds keep the + * existing serialization behavior. + */ +``` + +Replace the serialization import with type imports from `serialization-types.ts` and runtime helpers only from the injected serializer: + +```ts +import type { + EnhancedError, + Message, + RPCSerializationRuntime, + Response, + SerializationOptions, + TransferSlot +} from "./serialization-types.ts" +``` + +Keep these value imports from `serialization-json.ts` because transfer value processing is JSON/runtime-neutral and has no SuperJSON dependency: + +```ts +import { processValueForTransfer, reconstructValueFromTransfer } from "./serialization-json.ts" +``` + +- [ ] **Step 4: Add shared channel options type** + +In `channel-core.ts`, add this exported interface before the class: + +```ts +export interface RPCChannelOptions> { + expose?: LocalAPI + serialization?: SerializationOptions + enableTransfer?: boolean + /** Optional validators for the exposed API. Validates inputs/outputs on the receiving side. */ + validators?: RPCValidators + /** Interceptors that wrap handler invocation on the receiving side. */ + interceptors?: RPCInterceptor[] + /** Timeout in ms for outgoing RPC calls. Default: 0 (no timeout). */ + timeout?: number +} +``` + +Change the class name and constructor: + +```ts +export class RPCChannelCore< + LocalAPI extends Record, + RemoteAPI extends Record, + Io extends IoInterface = IoInterface +> { + constructor( + private io: Io, + options: RPCChannelOptions | undefined, + private serializationRuntime: RPCSerializationRuntime + ) { + this.apiImplementation = options?.expose + this.validators = options?.validators + this.interceptors = options?.interceptors ?? [] + this.timeout = options?.timeout ?? 0 + this.serializationOptions = options?.serialization || {} + this.structuredClone = io.capabilities?.structuredClone === true + if ( + this.structuredClone && + io.capabilities?.transfer === true && + options?.enableTransfer !== false + ) { + this.supportsTransfer = true + } + this.listen() + } +} +``` + +- [ ] **Step 5: Replace direct serializer calls inside `channel-core.ts`** + +Replace calls as follows: + +```ts +decodeMessage(payload) +``` + +becomes: + +```ts +this.serializationRuntime.decodeMessage(payload) +``` + +```ts +decodeMessage(messageStr) +``` + +becomes: + +```ts +this.serializationRuntime.decodeMessage(messageStr) +``` + +```ts +encodeMessage(message, this.serializationOptions, this.supportsTransfer, transferredValues) +``` + +becomes: + +```ts +this.serializationRuntime.encodeMessage( + message, + this.serializationOptions, + this.supportsTransfer, + transferredValues +) +``` + +```ts +serializeError(error) +deserializeError(error) +``` + +become: + +```ts +this.serializationRuntime.serializeError(error) +this.serializationRuntime.deserializeError(error) +``` + +Do not change request/response/callback logic in this task. + +- [ ] **Step 6: Recreate `channel.ts` as the full wrapper** + +Replace `packages/kkrpc/src/channel.ts` with: + +```ts +/** + * Full kkrpc RPCChannel facade. + * This entry keeps existing behavior by using the SuperJSON-enabled + * serialization runtime. + */ + +export { + isRPCTimeoutError, + RPCTimeoutError, + type RPCChannelOptions +} from "./channel-core.ts" +import { RPCChannelCore, type RPCChannelOptions } from "./channel-core.ts" +import type { IoInterface } from "./interface.ts" +import { fullSerializationRuntime } from "./serialization-full.ts" + +export class RPCChannel< + LocalAPI extends Record, + RemoteAPI extends Record, + Io extends IoInterface = IoInterface +> extends RPCChannelCore { + constructor(io: Io, options?: RPCChannelOptions) { + super(io, options, fullSerializationRuntime) + } +} +``` + +- [ ] **Step 7: Add `channel-lite.ts` wrapper** + +Create `packages/kkrpc/src/channel-lite.ts`: + +```ts +/** + * Browser-lite RPCChannel facade. + * This wrapper preserves the public `new RPCChannel(...)` API while using the + * JSON-only serialization runtime so SuperJSON stays out of lite bundles. + */ + +export { + isRPCTimeoutError, + RPCTimeoutError, + type RPCChannelOptions +} from "./channel-core.ts" +import { RPCChannelCore, type RPCChannelOptions } from "./channel-core.ts" +import type { IoInterface } from "./interface.ts" +import { jsonSerializationRuntime } from "./serialization-json.ts" + +export class RPCChannel< + LocalAPI extends Record, + RemoteAPI extends Record, + Io extends IoInterface = IoInterface +> extends RPCChannelCore { + constructor(io: Io, options?: RPCChannelOptions) { + if (options?.serialization?.version === "superjson") { + throw new Error( + 'SuperJSON serialization is not available in kkrpc/browser-lite. Use kkrpc/browser or configure both endpoints with serialization.version = "json".' + ) + } + super(io, options, jsonSerializationRuntime) + } +} +``` + +- [ ] **Step 8: Run lite and core tests** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/browser-lite.test.ts __tests__/serialization.test.ts __tests__/middleware.test.ts __tests__/timeout.test.ts +``` + +Expected: PASS. If TypeScript complains about `RPCChannelCore` generics, update the wrapper generic constraints to exactly match the current `RPCChannel` constraints. + +- [ ] **Step 9: Checkpoint** + +If commits are explicitly requested: + +```bash +git add packages/kkrpc/src/channel-core.ts packages/kkrpc/src/channel.ts packages/kkrpc/src/channel-lite.ts packages/kkrpc/__tests__/browser-lite.test.ts +git commit -m "refactor(kkrpc): split channel core from serializer facade" +``` + +### Task 4: Add Browser Lite Entrypoint And Package Exports + +**Files:** +- Create: `packages/kkrpc/browser-lite-mod.ts` +- Modify: `packages/kkrpc/tsdown.config.ts` +- Modify: `packages/kkrpc/package.json` +- Modify: `packages/kkrpc/deno.json` + +- [ ] **Step 1: Add `browser-lite-mod.ts`** + +Create `packages/kkrpc/browser-lite-mod.ts`: + +```ts +/** + * @module @kunkun/kkrpc/browser-lite + * @description Browser-only kkrpc entrypoint that avoids static SuperJSON imports. + */ + +export * from "./src/adapters/worker.ts" +export * from "./src/adapters/iframe.ts" +export * from "./src/adapters/websocket.ts" +export * from "./src/adapters/tauri.ts" +export * from "./src/interface.ts" +export * from "./src/channel-lite.ts" +export * from "./src/utils.ts" +export * from "./src/serialization-json.ts" +export * from "./src/serialization-types.ts" +export * from "./src/transfer.ts" +export * from "./src/transfer-handlers.ts" +export * from "./src/standard-schema.ts" +export * from "./src/validation.ts" +export * from "./src/middleware.ts" +``` + +- [ ] **Step 2: Add tsdown entry** + +Modify `packages/kkrpc/tsdown.config.ts` entry array: + +```ts +entry: [ + "./mod.ts", + "./browser-mod.ts", + "./browser-lite-mod.ts", + "./http.ts", + "./deno-mod.ts", + "./chrome-extension.ts", + "./socketio.ts", + "./rabbitmq.ts", + "./kafka.ts", + "./redis-streams.ts", + "./nats.ts", + "./electron.ts", + "./electron-ipc.ts", + "./inspector.ts" +] +``` + +- [ ] **Step 3: Add package export** + +In `packages/kkrpc/package.json`, add this export object after `./browser`: + +```json +"./browser-lite": { + "import": { + "types": "./dist/browser-lite-mod.d.ts", + "default": "./dist/browser-lite-mod.js" + }, + "require": { + "types": "./dist/browser-lite-mod.d.cts", + "default": "./dist/browser-lite-mod.cjs" + } +} +``` + +- [ ] **Step 4: Add Deno export** + +In `packages/kkrpc/deno.json`, add: + +```json +"./browser-lite": "./browser-lite-mod.ts" +``` + +Keep the JSON object valid with correct commas. + +- [ ] **Step 5: Build package** + +Run: + +```bash +pnpm --filter kkrpc build +``` + +Expected: PASS and generates: + +```txt +packages/kkrpc/dist/browser-lite-mod.js +packages/kkrpc/dist/browser-lite-mod.cjs +packages/kkrpc/dist/browser-lite-mod.d.ts +packages/kkrpc/dist/browser-lite-mod.d.cts +``` + +- [ ] **Step 6: Checkpoint** + +If commits are explicitly requested: + +```bash +git add packages/kkrpc/browser-lite-mod.ts packages/kkrpc/tsdown.config.ts packages/kkrpc/package.json packages/kkrpc/deno.json +git commit -m "feat(kkrpc): add browser-lite entrypoint" +``` + +### Task 5: Add Bundle Verification Script Using tsdown Output + +**Files:** +- Create: `packages/kkrpc/scripts/check-browser-lite-bundle.ts` +- Modify: `packages/kkrpc/package.json` + +- [ ] **Step 1: Create bundle check script** + +Create `packages/kkrpc/scripts/check-browser-lite-bundle.ts`: + +```ts +/** + * Verifies that the tsdown-built browser-lite artifacts do not include + * SuperJSON or its known dependency strings. This uses the canonical dist + * output instead of adding another bundler to the project. + */ + +import { readFile } from "node:fs/promises" +import { join } from "node:path" + +const files = [ + "dist/browser-lite-mod.js", + "dist/browser-lite-mod.cjs", + "dist/browser-lite-mod.d.ts", + "dist/browser-lite-mod.d.cts" +] + +const forbidden = ["superjson", "copy-anything", "is-what"] +let failed = false + +for (const file of files) { + const path = join(import.meta.dir, "..", file) + const contents = await readFile(path, "utf8") + for (const token of forbidden) { + if (contents.toLowerCase().includes(token)) { + console.error(`[browser-lite-bundle] Found forbidden token ${token} in ${file}`) + failed = true + } + } +} + +if (failed) { + process.exit(1) +} + +console.log("[browser-lite-bundle] No SuperJSON dependency strings found") +``` + +- [ ] **Step 2: Add package script** + +In `packages/kkrpc/package.json`, add: + +```json +"check:browser-lite-bundle": "pnpm build && bun run scripts/check-browser-lite-bundle.ts" +``` + +Keep scripts sorted near existing `build`/`check-types` scripts if possible. + +- [ ] **Step 3: Run bundle check** + +Run: + +```bash +pnpm --filter kkrpc check:browser-lite-bundle +``` + +Expected: + +```txt +[browser-lite-bundle] No SuperJSON dependency strings found +``` + +- [ ] **Step 4: Checkpoint** + +If commits are explicitly requested: + +```bash +git add packages/kkrpc/scripts/check-browser-lite-bundle.ts packages/kkrpc/package.json +git commit -m "test(kkrpc): verify browser-lite bundle excludes superjson" +``` + +### Task 6: Document Browser Lite Usage + +**Files:** +- Modify: `README.md` + +- [ ] **Step 1: Add README section** + +In `README.md`, add this section near the browser import guidance: + +```md +### Browser Lite Entry + +For browser-only apps that do not need SuperJSON, use the lite entrypoint: + +```ts +import { RPCChannel, WorkerParentIO } from "kkrpc/browser-lite" +``` + +`kkrpc/browser-lite` keeps the same `RPCChannel` facade as `kkrpc/browser`, but uses JSON-only string serialization and structured-clone envelopes for transports that support object messages. It does not statically import SuperJSON, which keeps browser bundles smaller. + +Use `kkrpc/browser` instead when you need SuperJSON-specific value preservation over string transports, such as Date, Map, Set, BigInt, or richer non-JSON values. + +When mixing full and lite endpoints over string transports, configure the full endpoint with JSON serialization: + +```ts +new RPCChannel(io, { + serialization: { version: "json" } +}) +``` + +Lite endpoints reject `serialization: { version: "superjson" }` with a clear runtime error. +``` + +Ensure nested code fences are valid Markdown. If this section is placed inside an existing fenced block, close that block first. + +- [ ] **Step 2: Run markdown formatting** + +Run: + +```bash +pnpm format -- README.md +``` + +Expected: README formatting completes without errors. + +- [ ] **Step 3: Checkpoint** + +If commits are explicitly requested: + +```bash +git add README.md +git commit -m "docs(kkrpc): document browser-lite entrypoint" +``` + +### Task 7: Full Verification + +**Files:** +- Verify entire `packages/kkrpc` package + +- [ ] **Step 1: Run focused tests** + +Run: + +```bash +pnpm --filter kkrpc test -- __tests__/serialization.test.ts __tests__/transfer.test.ts __tests__/browser-lite.test.ts __tests__/middleware.test.ts __tests__/timeout.test.ts +``` + +Expected: PASS. + +- [ ] **Step 2: Run package typecheck** + +Run: + +```bash +pnpm --filter kkrpc check-types +``` + +Expected: PASS. + +- [ ] **Step 3: Run package build** + +Run: + +```bash +pnpm --filter kkrpc build +``` + +Expected: PASS. + +- [ ] **Step 4: Run browser-lite bundle check** + +Run: + +```bash +pnpm --filter kkrpc check:browser-lite-bundle +``` + +Expected: + +```txt +[browser-lite-bundle] No SuperJSON dependency strings found +``` + +- [ ] **Step 5: Inspect git diff** + +Run: + +```bash +git status --short +git diff --stat +git diff -- packages/kkrpc/src/channel.ts packages/kkrpc/src/channel-core.ts packages/kkrpc/src/channel-lite.ts packages/kkrpc/src/serialization.ts packages/kkrpc/src/serialization-json.ts packages/kkrpc/src/serialization-full.ts packages/kkrpc/browser-lite-mod.ts +``` + +Expected: only intended files changed. Confirm `browser-lite` dependency path does not import `serialization.ts`, `serialization-full.ts`, or `superjson`. + +- [ ] **Step 6: Final checkpoint** + +If commits are explicitly requested: + +```bash +git add packages/kkrpc README.md +git commit -m "feat(kkrpc): add superjson-free browser lite entry" +``` + +Do not commit generated `dist/` or `docs/` Typedoc output unless the repository convention explicitly requires generated artifacts for release. + +## Self-Review Checklist + +- Spec coverage: Implements serializer split, full compatibility, browser-lite export, tsdown verification, docs, and tests. +- Placeholder scan: No unresolved placeholder steps; broad edits use concrete copy/replace instructions and exact file paths. +- Type consistency: `RPCSerializationRuntime`, `SerializationOptions`, `Message`, `EnhancedError`, and `RPCChannelOptions` names are consistent across tasks. +- Scope: Focused on issue #24 and the agreed architecture. Does not split validation/middleware/transfer helpers beyond necessary exports. diff --git a/docs/superpowers/reports/2026-06-06-browser-lite-size-analysis-and-optimization.md b/docs/superpowers/reports/2026-06-06-browser-lite-size-analysis-and-optimization.md new file mode 100644 index 0000000..903e2ac --- /dev/null +++ b/docs/superpowers/reports/2026-06-06-browser-lite-size-analysis-and-optimization.md @@ -0,0 +1,433 @@ +# Browser Lite Bundle Size Analysis And Optimization Notes + +Date: 2026-06-06 + +Related implementation commit: `fde53eb feat(kkrpc): add superjson-free browser lite entry` + +## Purpose + +This document records the bundle-size measurements taken after adding `kkrpc/browser-lite`, compares the result with `comctx@1.6.1`, and outlines concrete optimization space for a future session. + +The current `browser-lite` entry is intentionally not a micro RPC runtime. It keeps the full `RPCChannel` facade and most kkrpc protocol features, while removing the static SuperJSON dependency from the browser dependency graph. + +## Current Browser Lite Implementation Summary + +The implementation split serialization and channel wiring into: + +- `packages/kkrpc/src/serialization-types.ts`: protocol-only shared types. +- `packages/kkrpc/src/serialization-json.ts`: JSON-only runtime used by browser-lite. +- `packages/kkrpc/src/serialization-full.ts`: SuperJSON-enabled full runtime. +- `packages/kkrpc/src/serialization.ts`: compatibility barrel that remains SuperJSON-enabled. +- `packages/kkrpc/src/channel-core.ts`: shared RPC state machine, parameterized by serialization runtime. +- `packages/kkrpc/src/channel.ts`: full `RPCChannel` wrapper. +- `packages/kkrpc/src/channel-lite.ts`: JSON-only `RPCChannel` wrapper. +- `packages/kkrpc/browser-lite-mod.ts`: public browser-lite package entry. + +Verified behavior: + +- `kkrpc/browser` keeps existing SuperJSON behavior. +- `kkrpc/browser-lite` preserves `new RPCChannel(io, options)` but uses JSON serialization. +- `browser-lite` rejects `serialization: { version: "superjson" }` with a clear runtime error. +- The bundle-check script walks the static import graph and rejects `superjson`, `copy-anything`, `is-what`, `serialization-full`, or `serialization.ts` compatibility chunks in the lite dependency path. + +## Verification Already Run + +Run from `packages/kkrpc` unless noted: + +```bash +bun test __tests__/serialization.test.ts __tests__/transfer.test.ts __tests__/browser-lite.test.ts __tests__/middleware.test.ts __tests__/timeout.test.ts +``` + +Result: + +```txt +40 pass +0 fail +101 expect() calls +``` + +Run from the parent workspace root: + +```bash +pnpm --filter kkrpc check-types +pnpm --filter kkrpc check:browser-lite-bundle +``` + +Result: + +- Typecheck passed. +- `check:browser-lite-bundle` passed. +- Build completed. Typedoc emitted existing warning-style output, but no errors. + +## Measurement Setup + +Temporary benchmark directory used during the session: + +```txt +/var/folders/n6/tx2574_56h33v09wbc2hk_nm0000gn/T/opencode/kkrpc-bundle-compare +``` + +The examples expose a factory on `globalThis` so Bun does not dead-code-eliminate the entire import path. Measurements used: + +```bash +bun build .js --target=browser --minify --outfile=.js +``` + +Compressed sizes were calculated with Node/Bun `zlib`: + +```ts +gzipSync(data, { level: 9 }) +brotliCompressSync(data) +``` + +`comctx@1.6.1` was installed in the temp directory with: + +```bash +pnpm add comctx --registry=https://registry.npmmirror.com --config.strict-ssl=false +``` + +The mirror and SSL override were only used because direct npm registry access failed in the environment during this session. + +## Bundle Size Results + +### Public Browser Entries + +These use the public package browser entries and `WorkerParentIO`: + +```ts +import { RPCChannel, WorkerParentIO } from "kkrpc/browser" +import { RPCChannel, WorkerParentIO } from "kkrpc/browser-lite" +``` + +Measured via local built dist files: + +| Bundle | Modules | Raw minified | Gzip | Brotli | +| --- | ---: | ---: | ---: | ---: | +| `kkrpc/browser` | 25 | 32.40 KB | 9.94 KB | 8.82 KB | +| `kkrpc/browser-lite` | 6 | 21.07 KB | 6.16 KB | 5.42 KB | +| Difference | -19 | -11.32 KB | -3.78 KB | -3.40 KB | + +Unminified reference: + +| Bundle | Raw | Gzip | Brotli | +| --- | ---: | ---: | ---: | +| `kkrpc/browser` | 66.06 KB | 14.33 KB | 12.57 KB | +| `kkrpc/browser-lite` | 39.65 KB | 8.38 KB | 7.32 KB | +| Difference | -26.41 KB | -5.95 KB | -5.25 KB | + +Interpretation: + +- `browser-lite` removes roughly 3.8 KB gzip from a minimal browser Worker bundle. +- The raw minified reduction is larger because SuperJSON and its helpers are compressible. + +### Custom Tiny Adapter Entries + +To determine whether adapters or the public browser re-export shape caused most of the size, the session also bundled direct imports of the channel wrappers with a tiny local Worker-like adapter. + +| Demo | Raw minified | Gzip | Brotli | Modules | +| --- | ---: | ---: | ---: | ---: | +| `comctx@1.6.1` custom Worker adapter | 4.79 KB | 2.02 KB | 1.78 KB | 2 | +| `kkrpc-lite` direct channel + custom adapter | 20.83 KB | 5.96 KB | 5.25 KB | 10 | +| `kkrpc-superjson` direct channel + custom adapter | 32.16 KB | 9.73 KB | 8.65 KB | 27 | +| `kkrpc/browser-lite` public Worker example | 21.07 KB | 6.16 KB | 5.42 KB | 6 | +| `kkrpc/browser` public Worker example | 32.40 KB | 9.94 KB | 8.82 KB | 25 | + +Interpretation: + +- Choosing a smaller adapter does not materially change kkrpc size. +- Public `WorkerParentIO` vs custom tiny adapter changes lite gzip by only about 0.20 KB. +- The main size driver is `channel-core.ts`, not adapter code. +- `comctx` is much smaller because its core runtime has a narrower feature set. + +## Module Contribution Breakdown + +### kkrpc Lite Custom Adapter Bundle + +Bun metafile summary for `kkrpc-lite` direct channel + custom adapter: + +| Module | Minified contribution | Notes | +| --- | ---: | --- | +| `src/channel-core.ts` | 15.16 KB | Main size source. Full RPC state machine. | +| `src/serialization-json.ts` | 3.63 KB | JSON serializer, structured envelopes, transfer slots, error serialization. | +| custom adapter demo | 0.76 KB | Benchmark fixture, not package code. | +| `src/validation.ts` | 0.66 KB | Internal Standard Schema validation helpers, not Zod. | +| `src/channel-lite.ts` | 0.25 KB | Thin wrapper. | +| `src/utils.ts` | 0.12 KB | UUID helper. | +| `src/transfer.ts` | 0.11 KB | Transfer marker. | +| `src/middleware.ts` | 0.10 KB | Interceptor runner. | +| `src/transfer-handlers.ts` | 0.01 KB | Registry. | + +Important: no Zod or Valibot is bundled. `validation.ts` is kkrpc's Standard Schema glue and is small. + +### kkrpc SuperJSON Custom Adapter Bundle + +Bun metafile summary for `kkrpc-superjson` direct channel + custom adapter: + +| Module | Minified contribution | Notes | +| --- | ---: | --- | +| `src/channel-core.ts` | 15.18 KB | Same main core size. | +| `src/serialization-json.ts` | 3.28 KB | Shared JSON and structured-envelope support. | +| `superjson/dist/transformer.js` | 3.25 KB | Largest external SuperJSON module. | +| `superjson/dist/plainer.js` | 1.68 KB | SuperJSON internals. | +| `superjson/dist/index.js` | 1.67 KB | SuperJSON public runtime. | +| `superjson/dist/accessDeep.js` | 1.22 KB | SuperJSON internals. | +| `superjson/dist/is.js` | 0.83 KB | SuperJSON internals. | +| `src/validation.ts` | 0.66 KB | Internal validation glue. | +| `src/serialization-full.ts` | 0.56 KB | Thin SuperJSON wrapper. | +| `copy-anything` | 0.53 KB | SuperJSON dependency. | +| `is-what` visible contribution | about 0.13 KB | SuperJSON dependency. | + +Interpretation: + +- Browser-lite successfully removes the external SuperJSON graph. +- The remaining gap with comctx is not dependency bloat. It is kkrpc's feature-rich channel implementation. + +### comctx Bundle + +Bun metafile summary for `comctx@1.6.1` custom Worker adapter: + +| Module | Minified contribution | Notes | +| --- | ---: | --- | +| `comctx/core/dist/index.js` | 4.42 KB | Nearly the entire bundle. | +| custom adapter demo | 0.34 KB | Benchmark fixture. | + +Comctx bundle size: + +- Raw minified: 4.79 KB. +- Gzip: 2.02 KB. +- Brotli: 1.78 KB. + +## Why `channel-core.ts` Is Large + +`channel-core.ts` contains the complete kkrpc state machine. Bundlers generally cannot drop unused class methods when `new RPCChannel(...)` and dynamic protocol dispatch are used, because any method can be reached at runtime through incoming messages or proxy traps. + +Feature areas currently included in `channel-core.ts`: + +| Feature area | Representative functions | +| --- | --- | +| Transport read loop and close handling | `listen`, `closeFromTransport`, `rejectPendingRequests` | +| String transport newline buffering | `bufferString`, `handleMessageStr` | +| Structured-clone transport messages | `handleIncomingMessage`, `sendMessage` | +| Method call request/response | `callMethod`, `handleRequest`, `handleResponse`, `sendResponse`, `sendError` | +| Callback function arguments | `invokeCallback`, `handleCallback`, `callbacks`, `callbackCache` | +| Remote property get/set | `getProperty`, `setProperty`, `handleGet`, `handleSet` | +| Remote constructor calls | `callConstructor`, `handleConstruct` | +| AsyncIterable streaming | `streamResult`, `createStreamIterable`, `handleStreamChunk`, `handleStreamEnd`, `handleStreamError`, `handleStreamCancel` | +| Validation and middleware hooks | `runValidation`, `runInterceptors`, `mergeValidatedArgs` | +| Transfer support | calls into `processValueForTransfer`, `reconstructValueFromTransfer` | +| Timeouts | `startTimeout`, `clearTimeout`, `RPCTimeoutError` | +| Nested proxy facade | `createNestedProxy`, proxy traps for `get`, `set`, `apply`, `construct` | +| Broadcast transport tolerance | `shouldIgnoreBroadcastRequestWithoutApi`, `shouldIgnoreBroadcastResolutionMiss` | + +This explains why a smaller adapter barely changes the result: the core class is carrying most of the functionality. + +## comctx Comparison Caveats + +The comctx comparison is useful but not fully apples-to-apples. + +Comctx provides: + +- Method RPC across custom adapters. +- Callback support. +- Optional transfer extraction. +- Heartbeat and provider/injector roles. +- A small proxy model based on user-defined `defineProxy` namespaces. + +kkrpc full/lite additionally provides or keeps support for: + +- Multiple cross-runtime transport families through a common `IoInterface`. +- String transports and newline framing for stdio-style protocols. +- Structured-clone envelopes and transfer slot reconstruction. +- Remote nested property get and set. +- Remote constructor calls. +- AsyncIterable streaming protocol. +- Runtime validation hooks via Standard Schema. +- Middleware/interceptor hooks. +- Timeout errors and cleanup of pending requests. +- Rich error serialization and deserialization. +- Broadcast transport tolerance for missing APIs or paths. + +Therefore, matching comctx's size likely requires a new smaller kkrpc entrypoint with fewer features, not just optimizing the current `browser-lite` path. + +## Optimization Space + +### Low-Risk Follow-Ups For Current Browser Lite + +These can be considered without changing public semantics: + +1. Add a reproducible bundle-size benchmark script. + - Example path: `packages/kkrpc/scripts/compare-browser-bundle-size.ts`. + - Use `Bun.build` or shell out to `bun build`. + - Generate raw/gzip/brotli output for `browser`, `browser-lite`, and any future micro entry. + - Keep examples small but prevent dead-code elimination by exporting or assigning factories to `globalThis`. + +2. Keep the existing static graph check. + - `scripts/check-browser-lite-bundle.ts` should remain focused on forbidden imports, not arbitrary string scanning. + - Do not fail on user-facing strings that mention SuperJSON. + +3. Inspect whether optional validation and middleware imports can be injected. + - Current size contribution is small, about 0.76 KB combined for `validation.ts` and `middleware.ts` in the lite custom benchmark. + - This is not the highest-priority optimization. + +4. Inspect `serialization-json.ts` for a transfer-free variant. + - Current contribution is about 3.63 KB minified. + - A `serialization-json-basic.ts` with no transfer slots and less enhanced error support could be materially smaller. + - This would require a new entrypoint or explicit option, because existing transfer behavior must remain intact. + +### Medium-Risk Refactors + +These may reduce size but need careful API design: + +1. Split `channel-core.ts` by feature into smaller core variants. + - A file split alone will not reduce bundle size if the public class still imports and dispatches every feature. + - The split must be paired with separate entrypoints/classes that do not import unused features. + +2. Create a `RPCMethodChannel` or similar internal implementation. + - Method calls only. + - Optional callbacks. + - No property get/set. + - No constructor calls. + - No streaming. + - No validation or middleware by default. + - No string newline buffering if targeting browser structured-clone only. + +3. Make streaming an opt-in entrypoint. + - Streaming currently pulls in producer and consumer state machines. + - Removing it from current `RPCChannel` would be breaking. + - A separate `browser-method` or `browser-micro` entry can omit it safely. + +4. Make remote property and constructor support opt-in. + - These are mostly enabled by `createNestedProxy` and message handlers. + - A method-only proxy can be much smaller. + - This changes behavior for `await api.foo`, `api.foo = value`, and `new api.Ctor()`. + +### High-Impact New Entry: `kkrpc/browser-micro` + +Most promising path if size is the priority. + +Goal: + +- Compete more directly with comctx for browser-only RPC size. +- Keep the full `kkrpc/browser-lite` entry as the complete, SuperJSON-free option. + +Possible scope: + +- Browser-only, structured-clone/postMessage style transports. +- Method calls only: `api.method(...args)`. +- Optional callback arguments. +- Optional transfer support, or a separate `browser-micro-transfer` entry. +- JSON/string fallback only if necessary, ideally omitted. +- No property get/set. +- No constructor calls. +- No AsyncIterable streaming. +- No validation or middleware. +- Minimal error serialization. +- No broadcast transport special cases. +- No stdio newline buffering. + +Potential user-facing import: + +```ts +import { RPCChannel, WorkerParentIO } from "kkrpc/browser-micro" +``` + +or, to avoid implying exact parity with full `RPCChannel`: + +```ts +import { MethodRPCChannel, WorkerParentIO } from "kkrpc/browser-micro" +``` + +API compatibility decision: + +- If it exports `RPCChannel`, document that only method calls are supported. +- If it exports `MethodRPCChannel`, the limitation is clearer and avoids surprising users. + +Expected effect: + +- The current lite custom benchmark is about 5.96 KB gzip. +- Comctx custom benchmark is about 2.02 KB gzip. +- A method-only structured-clone kkrpc micro entry could plausibly get closer to comctx, but this has not been prototyped or measured. +- Do not promise a target until a prototype is bundled. + +## Recommended Next-Session Plan + +### Step 1: Add A Reproducible Size Benchmark + +Create a script under `packages/kkrpc/scripts/` that: + +- Builds minimal examples for: + - `kkrpc/browser` public Worker path. + - `kkrpc/browser-lite` public Worker path. + - direct `channel-lite` + custom tiny adapter diagnostic path. + - optional `comctx` comparison only if `comctx` is installed or can be fetched. +- Reports raw, gzip, and brotli sizes. +- Emits Bun metafile markdown or JSON for module contribution analysis. + +Important benchmark details: + +- Prevent DCE by exporting a function and assigning it to `globalThis`. +- Use equivalent example behavior across libraries. +- Separate public-entry benchmarks from internal diagnostic-entry benchmarks. +- Keep temp outputs outside tracked source or under ignored paths. + +### Step 2: Prototype Method-Only Browser Core + +Create a throwaway prototype first, not a polished API: + +- `src/channel-method-core.ts` or temp file outside package source. +- Minimal message types: `request`, `response`, maybe `callback`. +- Minimal proxy: `get` returns a function proxy; `apply` sends a method call. +- No property await behavior. +- No set trap. +- No construct trap. +- No stream messages. +- No validators/interceptors. +- Structured messages only for browser transports if possible. + +Benchmark the prototype before integrating it. + +### Step 3: Decide Product Shape + +Choose one: + +1. `browser-lite` remains full-featured SuperJSON-free runtime, and a new `browser-micro` handles size-sensitive browser apps. +2. `browser-lite` becomes smaller and loses features. This is not recommended because the current implementation has already established lite as same-facade/full-feature minus SuperJSON. +3. Add option flags to `RPCChannel`. This is less likely to help bundle size unless options are compile-time and separate entrypoints avoid importing disabled features. + +Recommendation: keep `browser-lite` as-is and add `browser-micro` or `browser-method` for size-sensitive use cases. + +### Step 4: Add Tests For Any New Micro Entry + +Minimum tests: + +- Can call a remote method over a browser-like adapter. +- Can propagate remote errors. +- Can clean up pending calls on destroy/EOF if supported. +- If callbacks are included, can pass and invoke callback args. +- If transfer is included, can transfer `ArrayBuffer` over structured clone. +- Bundle check proves the micro entry excludes `serialization.ts`, `serialization-full.ts`, `superjson`, and any intentionally omitted feature modules. + +### Step 5: Document Limitations Clearly + +If a micro entry is added, README should include a decision table: + +| Need | Entry | +| --- | --- | +| Full kkrpc browser behavior plus SuperJSON | `kkrpc/browser` | +| Full kkrpc browser behavior without SuperJSON | `kkrpc/browser-lite` | +| Smallest browser method-call runtime | `kkrpc/browser-micro` or `kkrpc/browser-method` | + +## Risks And Constraints + +- Do not break existing `RPCChannel` behavior for `kkrpc` or `kkrpc/browser`. +- Do not make `browser-lite` import `serialization.ts`, `serialization-full.ts`, or `superjson`. +- Do not assume file splitting alone reduces bundle size. Bundle measurements must prove it. +- Be careful with class-based APIs: methods on a class are often retained even if a specific app never calls them. +- Be explicit about unsupported features in any micro entry. +- Keep bundle-size checks based on static import graph, not raw string tokens. + +## Current Best Interpretation + +`kkrpc/browser-lite` achieved its intended goal: preserve the full kkrpc browser API shape while removing static SuperJSON dependency. It reduced a minimal browser Worker bundle by about 3.8 KB gzip compared with `kkrpc/browser`. + +The next optimization target is not adapter size and not Zod-like dependencies. The next target is `channel-core.ts`. To make meaningful progress, introduce a smaller browser-only method-channel entrypoint with a deliberately reduced feature set, then benchmark it against comctx and the existing lite entry. diff --git a/docs/superpowers/specs/2026-06-05-browser-lite-serialization-design.md b/docs/superpowers/specs/2026-06-05-browser-lite-serialization-design.md new file mode 100644 index 0000000..78e4da0 --- /dev/null +++ b/docs/superpowers/specs/2026-06-05-browser-lite-serialization-design.md @@ -0,0 +1,235 @@ +# Browser Lite Serialization Design + +Date: 2026-06-05 + +## Context + +Issue #24 requests a minimal browser bundle that does not include SuperJSON. The current browser entrypoint imports `src/channel.ts`, which imports `src/serialization.ts`, which statically imports `superjson`. As a result, browser bundlers can include SuperJSON even when users only need JSON or structured-clone messaging. + +The current package already supports `serialization: { version: "json" }`, but the dependency graph still reaches SuperJSON at import time. The improved design should remove that coupling without changing the public `new RPCChannel(...)` API. + +The feature branch is `feat/browser-lite`, based on `origin/main`, with the existing metadata propagation commit reapplied on top. + +## Goals + +- Add a browser-focused entrypoint that avoids any static SuperJSON import. +- Keep the existing `RPCChannel` facade and constructor shape unchanged. +- Avoid duplicating the full channel state machine. +- Preserve existing `kkrpc`, `kkrpc/browser`, and SuperJSON behavior. +- Keep Worker/iframe structured envelope and transfer support available in the lite entrypoint. +- Verify the browser-lite output through the existing `tsdown` build pipeline. + +## Non-Goals + +- Do not replace the public `RPCChannel` constructor with a user-facing factory function. +- Do not copy `channel.ts` into a large `channel-minimal.ts` fork. +- Do not introduce esbuild or another bundler solely for bundle assertions. +- Do not split validation, middleware, or transfer helpers unless later measurement shows they are material bundle-size contributors. + +## Recommended Architecture + +Keep one shared channel implementation and move the serialization choice behind thin entrypoint-specific wrappers. + +Proposed source layout: + +```txt +packages/kkrpc/ + browser-lite-mod.ts + src/ + channel-core.ts + channel.ts + channel-lite.ts + serialization.ts + serialization-types.ts + serialization-json.ts + serialization-full.ts +``` + +Responsibilities: + +| File | Responsibility | +|---|---| +| `channel-core.ts` | Shared `RPCChannelCore` state machine. It does not import SuperJSON. | +| `channel.ts` | Existing full `RPCChannel` wrapper. It imports the full serializer. | +| `channel-lite.ts` | Lite `RPCChannel` wrapper. It imports only the JSON serializer. | +| `serialization.ts` | Compatibility barrel for the existing full serialization API. It may re-export from `serialization-full.ts` and `serialization-types.ts`. Browser-lite must not import it. | +| `serialization-types.ts` | Shared protocol types such as `Message`, `Response`, `WireEnvelope`, metadata, and transfer slots. | +| `serialization-json.ts` | JSON string serialization, structured envelope encode/decode, transfer processing, and error serialization. No SuperJSON import. | +| `serialization-full.ts` | Full serializer with SuperJSON support and existing compatibility behavior. | + +The main dependency invariant is: + +```txt +browser-lite-mod.ts -> channel-lite.ts -> channel-core.ts -> serialization-json.ts +``` + +This path must not import `serialization-full.ts`, `serialization.ts`, or `superjson`. + +Shared runtime files used by both entries should import protocol types from `serialization-types.ts`, not from the compatibility `serialization.ts` barrel. + +The existing full path remains: + +```txt +browser-mod.ts -> channel.ts -> channel-core.ts -> serialization-full.ts -> superjson +``` + +## Serializer Runtime Contract + +`channel-core.ts` receives a serializer runtime from the wrapper class. The interface is internal and should stay small: + +```ts +interface RPCSerializationRuntime { + encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues?: unknown[] + ): EncodedMessage + + decodeMessage(raw: WireFormat): Promise> + serializeError(error: Error): EnhancedError + deserializeError(error: EnhancedError): Error +} +``` + +`channel.ts` exports the existing class name: + +```ts +export class RPCChannel extends RPCChannelCore { + constructor(io: Io, options?: RPCChannelOptions) { + super(io, options, fullSerializationRuntime) + } +} +``` + +`channel-lite.ts` exports the same public class name: + +```ts +export class RPCChannel extends RPCChannelCore { + constructor(io: Io, options?: RPCChannelOptions) { + super(io, options, jsonSerializationRuntime) + } +} +``` + +If needed, shared constructor option types can move from `channel.ts` into `channel-core.ts` or a small `channel-types.ts` file. + +## Public API + +Add a new subpath export: + +```ts +import { RPCChannel, WorkerParentIO } from "kkrpc/browser-lite" +``` + +Existing imports stay valid: + +```ts +import { RPCChannel } from "kkrpc/browser" +import { RPCChannel } from "kkrpc" +``` + +Runtime defaults: + +| Entry | Default serialization | SuperJSON dependency | +|---|---|---| +| `kkrpc` | `superjson` | Yes | +| `kkrpc/browser` | `superjson` | Yes | +| `kkrpc/browser-lite` | `json` | No | + +## Compatibility Behavior + +| Case | Behavior | +|---|---| +| Lite to Lite over Worker/iframe | Works. Uses structured envelope when transfers are enabled. | +| Lite to Lite over string transport | Works with JSON serialization. | +| Full to Full | Existing behavior unchanged. | +| Full JSON to Lite JSON | Works if both sides use JSON serialization. | +| Full default SuperJSON to Lite over string transport | Not supported. Lite should reject with a clear error. | +| Lite with `serialization.version = "superjson"` | Throw a clear runtime error. | + +The lite decoder must not import SuperJSON to parse legacy SuperJSON strings. If it receives a SuperJSON-looking wire string, it should throw an actionable error: + +```txt +Received a SuperJSON-encoded kkrpc message, but this entrypoint is JSON-only. Use kkrpc/browser or configure both endpoints with serialization.version = "json". +``` + +Structured envelope messages remain object payloads and do not require SuperJSON. + +## Export And Build Changes + +Add `browser-lite-mod.ts` with browser-safe exports: + +```ts +export * from "./src/adapters/worker.ts" +export * from "./src/adapters/iframe.ts" +export * from "./src/adapters/websocket.ts" +export * from "./src/adapters/tauri.ts" +export * from "./src/interface.ts" +export * from "./src/channel-lite.ts" +export * from "./src/utils.ts" +export * from "./src/serialization-json.ts" +export * from "./src/serialization-types.ts" +export * from "./src/transfer.ts" +export * from "./src/transfer-handlers.ts" +export * from "./src/standard-schema.ts" +export * from "./src/validation.ts" +export * from "./src/middleware.ts" +``` + +Update: + +- `packages/kkrpc/tsdown.config.ts`: add `./browser-lite-mod.ts` to entries. +- `packages/kkrpc/package.json`: add `./browser-lite` export for ESM, CJS, and types. +- `packages/kkrpc/deno.json`: add `./browser-lite` export. +- README or docs: document when to use `browser-lite` versus `browser`. + +The existing `tsdown` build is the canonical bundle verification path. Do not add esbuild. + +## Tests And Verification + +Unit and integration tests: + +| Test | Purpose | +|---|---| +| JSON serializer round-trip | `serialization-json.ts` serializes and deserializes messages without SuperJSON. | +| Lite rejects SuperJSON | Lite decoder rejects SuperJSON-looking wire strings with a clear error. | +| Lite RPC integration | `channel-lite.ts` can call a simple API through a test `IoInterface`. | +| Structured envelope regression | Existing transfer/envelope path still reconstructs transferred values. | +| Full serializer regression | Existing SuperJSON behavior remains unchanged. | + +Build verification: + +```bash +pnpm --filter kkrpc build +rg "superjson|copy-anything|is-what" packages/kkrpc/dist/browser-lite-mod.* +``` + +The search should return no matches for the lite bundle. This can be formalized as a package script using Bun and the generated `dist/` files if recurring release checks need it. + +## Implementation Order + +1. Extract protocol and wire types into `serialization-types.ts`. +2. Extract JSON/string/envelope/error/transfer logic into `serialization-json.ts`. +3. Create `serialization-full.ts` that wraps or extends JSON behavior with SuperJSON support. +4. Keep `serialization.ts` as a compatibility full-serialization barrel for existing imports. +5. Move the channel implementation into `channel-core.ts` and replace direct serialization imports with the runtime contract. +6. Recreate `channel.ts` as the full wrapper and add `channel-lite.ts` as the JSON-only wrapper. +7. Add `browser-lite-mod.ts` and package/JSR/tsdown exports. +8. Add tests and run typecheck/build/test verification. +9. Update docs for `browser-lite` usage and limitations. + +## Risks + +- The channel extraction is a broad move, so preserve behavior with small commits and focused tests. +- Full and lite wrappers must export the same public class name to avoid facade API churn. +- Type exports must remain source-compatible for existing users importing protocol types from `serialization.ts`. +- Lite/full interop over string transports requires explicit JSON configuration on the full side. + +## Success Criteria + +- Existing `kkrpc` and `kkrpc/browser` behavior remains compatible. +- `kkrpc/browser-lite` builds and exposes `RPCChannel` with the same constructor facade. +- `browser-lite` dependency path does not statically import SuperJSON. +- `tsdown` output for `browser-lite-mod` contains no SuperJSON dependency strings. +- Existing tests pass, and new lite serialization/RPC tests cover the new path. diff --git a/packages/kkrpc/README.md b/packages/kkrpc/README.md index 3e1b183..f369945 100644 --- a/packages/kkrpc/README.md +++ b/packages/kkrpc/README.md @@ -166,6 +166,28 @@ class RPCChannel< > {} ``` +### Browser Lite Entry + +For browser-only apps that do not need SuperJSON, use the lite entrypoint: + +```ts +import { RPCChannel, WorkerParentIO } from "kkrpc/browser-lite" +``` + +`kkrpc/browser-lite` keeps the same `RPCChannel` facade as `kkrpc/browser`, but uses JSON-only string serialization and structured-clone envelopes for transports that support object messages. It does not statically import SuperJSON, which keeps browser bundles smaller. + +Use `kkrpc/browser` instead when you need SuperJSON-specific value preservation over string transports, such as Date, Map, Set, BigInt, or richer non-JSON values. + +When mixing full and lite endpoints over string transports, configure the full endpoint with JSON serialization: + +```ts +new RPCChannel(io, { + serialization: { version: "json" } +}) +``` + +Lite endpoints reject `serialization: { version: "superjson" }` with a clear runtime error. + ## Serialization kkrpc supports two serialization formats for message transmission: diff --git a/packages/kkrpc/__tests__/browser-lite.test.ts b/packages/kkrpc/__tests__/browser-lite.test.ts new file mode 100644 index 0000000..14afb55 --- /dev/null +++ b/packages/kkrpc/__tests__/browser-lite.test.ts @@ -0,0 +1,60 @@ +/** + * Browser-lite integration tests. + * These tests exercise the JSON-only RPCChannel wrapper without importing the + * full SuperJSON-enabled channel facade. + */ +import { describe, expect, test } from "bun:test" +import { WebSocketServer } from "ws" +import { WebSocketClientIO, WebSocketServerIO } from "../src/adapters/websocket.ts" +import { RPCChannel } from "../src/channel-lite.ts" +import type { IoInterface } from "../src/interface.ts" + +interface TestAPI { + echo(message: string): Promise + add(a: number, b: number): Promise +} + +const testApi: TestAPI = { + echo: async (message) => message, + add: async (a, b) => a + b +} + +describe("browser-lite RPCChannel", () => { + test("calls remote methods with JSON serialization", async () => { + const port = 3095 + const wss = new WebSocketServer({ port }) + wss.on("connection", (ws) => { + const serverIO = new WebSocketServerIO(ws) + new RPCChannel(serverIO, { expose: testApi }) + }) + + const clientIO = new WebSocketClientIO({ url: `ws://localhost:${port}` }) + const rpc = new RPCChannel<{}, TestAPI, IoInterface>(clientIO) + const api = rpc.getAPI() + + try { + expect(await api.echo("hello")).toBe("hello") + expect(await api.add(2, 5)).toBe(7) + } finally { + clientIO.destroy() + wss.close() + } + }) + + test("rejects explicit SuperJSON serialization option", () => { + const io = { + name: "closed-test-io", + read: async () => null, + write: async () => {}, + on: () => {}, + off: () => {} + } satisfies IoInterface + + expect( + () => + new RPCChannel<{}, TestAPI, IoInterface>(io, { + serialization: { version: "superjson" } + }) + ).toThrow("SuperJSON serialization is not available") + }) +}) diff --git a/packages/kkrpc/__tests__/metadata.test.ts b/packages/kkrpc/__tests__/metadata.test.ts new file mode 100644 index 0000000..a5c56d2 --- /dev/null +++ b/packages/kkrpc/__tests__/metadata.test.ts @@ -0,0 +1,138 @@ +/** + * RPC metadata propagation tests. + * These verify callers can attach out-of-band metadata that receiving + * interceptors observe without coupling kkrpc to any application-specific type. + */ +import { describe, expect, test } from "bun:test" +import { RPCChannel } from "../src/channel.ts" +import type { IoInterface, IoMessage } from "../src/interface.ts" +import type { RPCInterceptor } from "../src/middleware.ts" +import type { RPCMessageMetadata } from "../src/serialization-types.ts" + +class MemoryIO implements IoInterface { + private peer?: MemoryIO + private messageQueue: Array = [] + private resolveRead: ((value: string | IoMessage | null) => void) | null = null + + constructor(public readonly name: string) {} + + connect(peer: MemoryIO): void { + this.peer = peer + } + + on(_event: "message", _listener: (message: string | IoMessage) => void): void + on(_event: "error", _listener: (error: Error) => void): void + on(_event: "message" | "error", _listener: Function): void {} + + off(_event: "message" | "error", _listener: Function): void {} + + read(): Promise { + if (this.messageQueue.length > 0) { + return Promise.resolve(this.messageQueue.shift() ?? null) + } + + return new Promise((resolve) => { + this.resolveRead = resolve + }) + } + + write(message: string | IoMessage): Promise { + this.peer?.receive(message) + return Promise.resolve() + } + + destroy(): void { + this.receive(null) + } + + receive(message: string | IoMessage | null): void { + if (this.resolveRead) { + this.resolveRead(message) + this.resolveRead = null + return + } + + this.messageQueue.push(message) + } +} + +function createMemoryPair(): [MemoryIO, MemoryIO] { + const left = new MemoryIO("metadata-left") + const right = new MemoryIO("metadata-right") + left.connect(right) + right.connect(left) + return [left, right] +} + +function pendingRequestCount(channel: object): number { + const pendingRequests = Reflect.get(channel, "pendingRequests") + if (!pendingRequests || typeof pendingRequests !== "object") { + throw new Error("RPCChannel pendingRequests state is unavailable") + } + return Object.keys(pendingRequests).length +} + +interface ServerAPI { + echo(message: string): Promise +} + +describe("RPC metadata", () => { + test("outgoing metadata provider reaches the receiving interceptor", async () => { + let seenMeta: RPCMessageMetadata | undefined + const [clientIO, serverIO] = createMemoryPair() + const interceptor: RPCInterceptor = async (ctx, next) => { + seenMeta = ctx.meta + return next() + } + const serverApi: ServerAPI = { + echo: async (message) => message + } + + new RPCChannel(serverIO, { + expose: serverApi, + interceptors: [interceptor] + }) + const clientRpc = new RPCChannel<{}, ServerAPI>(clientIO, { + getMetadata: () => ({ + activity: { activityId: "activity-1", operationId: "operation-1" } + }) + }) + + try { + expect(await clientRpc.getAPI().echo("hello")).toBe("hello") + expect(seenMeta).toEqual({ + activity: { activityId: "activity-1", operationId: "operation-1" } + }) + } finally { + clientIO.destroy() + serverIO.destroy() + } + }) + + test("throwing metadata provider rejects without retaining a pending request", async () => { + let handlerCalled = false + const [clientIO, serverIO] = createMemoryPair() + const serverApi: ServerAPI = { + echo: async (message) => { + handlerCalled = true + return message + } + } + + new RPCChannel(serverIO, { expose: serverApi }) + const clientRpc = new RPCChannel<{}, ServerAPI>(clientIO, { + getMetadata: () => { + throw new Error("metadata unavailable") + } + }) + + try { + await expect(clientRpc.getAPI().echo("hello")).rejects.toThrow("metadata unavailable") + expect(handlerCalled).toBe(false) + expect(pendingRequestCount(clientRpc)).toBe(0) + } finally { + clientIO.destroy() + serverIO.destroy() + } + }) +}) diff --git a/packages/kkrpc/__tests__/serialization.test.ts b/packages/kkrpc/__tests__/serialization.test.ts index 73da704..c18e34d 100644 --- a/packages/kkrpc/__tests__/serialization.test.ts +++ b/packages/kkrpc/__tests__/serialization.test.ts @@ -1,6 +1,12 @@ import { describe, expect, test } from "bun:test" import superjson from "superjson" import { deserializeMessage, serializeMessage, type Message } from "../src/serialization.ts" +import { + decodeJsonMessage, + encodeJsonMessage, + jsonSerializationRuntime, + type Message as JsonMessage +} from "../src/serialization-json.ts" describe("Serializer", () => { test("should serialize and deserialize a message", async () => { @@ -44,4 +50,61 @@ describe("Serializer", () => { const deserialized = superjson.parse(serialized) expect(deserialized).toEqual(message as any) }) + + test("default compatibility serializer still uses SuperJSON", async () => { + const message: Message = { + id: "superjson-default", + method: "date.echo", + args: [new Date("2026-06-05T00:00:00.000Z")], + type: "request" + } + + const serialized = serializeMessage(message) + expect(serialized.startsWith('{"json":')).toBe(true) + + const deserialized = await deserializeMessage(serialized) + expect(deserialized.args[0]).toBeInstanceOf(Date) + }) +}) + +describe("JSON-only serialization", () => { + test("round-trips JSON messages without SuperJSON", async () => { + const message: JsonMessage = { + id: "json-1", + method: "echo", + args: ["hello", new Uint8Array([1, 2, 3])], + type: "request" + } + + const serialized = encodeJsonMessage(message) + const deserialized = await decodeJsonMessage(serialized) + + expect(deserialized).toEqual({ ...message, version: "json" } as JsonMessage) + }) + + test("rejects SuperJSON-looking strings with a clear lite error", async () => { + const superjsonLike = '{"json":{"id":"1","method":"echo","args":[],"type":"request"}}\n' + + await expect(decodeJsonMessage(superjsonLike)).rejects.toThrow( + "Received a SuperJSON-encoded kkrpc message" + ) + }) + + test("encodes structured envelopes without SuperJSON", async () => { + const message: JsonMessage = { + id: "structured-1", + method: "echo", + args: ["hello"], + type: "request" + } + + const encoded = jsonSerializationRuntime.encodeMessage(message, {}, true) + + expect(encoded.mode).toBe("structured") + if (encoded.mode !== "structured") { + expect.unreachable("expected structured encoding") + } + expect(encoded.data.payload).toEqual(message) + expect(await jsonSerializationRuntime.decodeMessage(encoded.data)).toEqual(message) + }) }) diff --git a/packages/kkrpc/browser-lite-mod.ts b/packages/kkrpc/browser-lite-mod.ts new file mode 100644 index 0000000..91605fc --- /dev/null +++ b/packages/kkrpc/browser-lite-mod.ts @@ -0,0 +1,19 @@ +/** + * @module @kunkun/kkrpc/browser-lite + * @description Browser-only kkrpc entrypoint that avoids static SuperJSON imports. + */ + +export * from "./src/adapters/worker.ts" +export * from "./src/adapters/iframe.ts" +export * from "./src/adapters/websocket.ts" +export * from "./src/adapters/tauri.ts" +export * from "./src/interface.ts" +export * from "./src/channel-lite.ts" +export * from "./src/utils.ts" +export * from "./src/serialization-json.ts" +export * from "./src/serialization-types.ts" +export * from "./src/transfer.ts" +export * from "./src/transfer-handlers.ts" +export * from "./src/standard-schema.ts" +export * from "./src/validation.ts" +export * from "./src/middleware.ts" diff --git a/packages/kkrpc/deno.json b/packages/kkrpc/deno.json index bebd9de..a87142b 100644 --- a/packages/kkrpc/deno.json +++ b/packages/kkrpc/deno.json @@ -34,6 +34,7 @@ "exports": { ".": "./mod.ts", "./browser": "./browser-mod.ts", + "./browser-lite": "./browser-lite-mod.ts", "./http": "./http.ts", "./chrome-extension": "./chrome-extension.ts", "./deno": "./deno-mod.ts", diff --git a/packages/kkrpc/package.json b/packages/kkrpc/package.json index 576dbf6..a6cd7c5 100644 --- a/packages/kkrpc/package.json +++ b/packages/kkrpc/package.json @@ -40,6 +40,7 @@ "docs": "typedoc --out docs", "check-types": "tsc --noEmit", "build": "tsdown", + "check:browser-lite-bundle": "pnpm build && bun run scripts/check-browser-lite-bundle.ts", "postbuild": "pnpm run docs && verify-package-export verify", "dev": "tsdown --watch" }, @@ -66,6 +67,16 @@ "default": "./dist/browser-mod.cjs" } }, + "./browser-lite": { + "import": { + "types": "./dist/browser-lite-mod.d.ts", + "default": "./dist/browser-lite-mod.js" + }, + "require": { + "types": "./dist/browser-lite-mod.d.cts", + "default": "./dist/browser-lite-mod.cjs" + } + }, "./http": { "import": { "types": "./dist/http.d.ts", diff --git a/packages/kkrpc/scripts/check-browser-lite-bundle.ts b/packages/kkrpc/scripts/check-browser-lite-bundle.ts new file mode 100644 index 0000000..ec888c4 --- /dev/null +++ b/packages/kkrpc/scripts/check-browser-lite-bundle.ts @@ -0,0 +1,120 @@ +/** + * Verifies that the tsdown-built browser-lite artifacts do not statically import + * SuperJSON, its known dependencies, or the full serializer runtime. This uses + * the canonical dist output instead of adding another bundler to the project. + */ +import { access, readFile } from "node:fs/promises" +import { basename, dirname, join, normalize } from "node:path" + +const entryFiles = [ + "dist/browser-lite-mod.js", + "dist/browser-lite-mod.cjs", + "dist/browser-lite-mod.d.ts", + "dist/browser-lite-mod.d.cts" +] + +const packageRoot = join(import.meta.dir, "..") +const forbiddenPackages = ["superjson", "copy-anything", "is-what"] +const staticSpecifierPattern = + /\b(?:import|export)(?:[^"'`]*?\bfrom)?\s*["'`]([^"'`]+)["'`]|\bimport\(\s*["'`]([^"'`]+)["'`]\s*\)|\brequire\(\s*["'`]([^"'`]+)["'`]\s*\)/g +const pending = [...entryFiles] +const visited = new Set() +const failures: string[] = [] + +function normalizePath(file: string): string { + return normalize(file).replaceAll("\\", "/") +} + +function getForbiddenSpecifierReason(specifier: string): string | null { + for (const packageName of forbiddenPackages) { + if (specifier === packageName || specifier.startsWith(`${packageName}/`)) { + return `package import ${packageName}` + } + } + + if (!specifier.startsWith(".")) { + return null + } + + const fileName = basename(specifier) + if (fileName.startsWith("serialization-full")) { + return "serialization-full runtime chunk" + } + if ( + fileName === "serialization.js" || + fileName === "serialization.cjs" || + /^serialization-(?!json-|types-)/.test(fileName) + ) { + return "serialization.ts compatibility chunk" + } + + return null +} + +async function fileExists(file: string): Promise { + try { + await access(join(packageRoot, file)) + return true + } catch { + return false + } +} + +async function resolveLocalImport(currentFile: string, specifier: string): Promise { + if (!specifier.startsWith(".")) { + return null + } + + const resolved = normalizePath(join(dirname(currentFile), specifier)) + const candidates = [resolved] + if (currentFile.endsWith(".d.ts") && resolved.endsWith(".js")) { + candidates.push(`${resolved.slice(0, -3)}.d.ts`) + } + if (currentFile.endsWith(".d.cts") && resolved.endsWith(".cjs")) { + candidates.push(`${resolved.slice(0, -4)}.d.cts`) + } + + for (const candidate of candidates) { + if (await fileExists(candidate)) { + return candidate + } + } + + return null +} + +while (pending.length > 0) { + const file = pending.shift() + if (!file || visited.has(file)) { + continue + } + visited.add(file) + + const filePath = join(packageRoot, file) + const contents = await readFile(filePath, "utf8") + for (const match of contents.matchAll(staticSpecifierPattern)) { + const specifier = match[1] ?? match[2] ?? match[3] + if (!specifier) { + continue + } + + const reason = getForbiddenSpecifierReason(specifier) + if (reason) { + failures.push(`[browser-lite-bundle] ${file} imports ${specifier} (${reason})`) + } + + const resolved = await resolveLocalImport(file, specifier) + if (resolved) { + pending.push(resolved) + } + } +} + +if (failures.length > 0) { + for (const failure of failures) { + console.error(failure) + } + process.exit(1) +} + +console.log("[browser-lite-bundle] No forbidden SuperJSON dependency imports found") diff --git a/packages/kkrpc/src/channel-core.ts b/packages/kkrpc/src/channel-core.ts new file mode 100644 index 0000000..c9baf8e --- /dev/null +++ b/packages/kkrpc/src/channel-core.ts @@ -0,0 +1,1364 @@ +/** + * Shared RPC channel state machine. + * Public entrypoints wrap this class with a serialization runtime so lite + * browser builds can avoid importing SuperJSON while full builds keep the + * existing serialization behavior. + */ +import type { IoInterface, IoMessage } from "./interface.ts" +import { runInterceptors, type RPCInterceptor } from "./middleware.ts" +import { processValueForTransfer, reconstructValueFromTransfer } from "./serialization-json.ts" +import { + type EnhancedError, + type Message, + type RPCMessageMetadata, + type RPCSerializationRuntime, + type Response, + type SerializationOptions, + type TransferSlot +} from "./serialization-types.ts" +import { generateUUID } from "./utils.ts" +import { + lookupValidator, + RPCValidationError, + runValidation, + type RPCValidators +} from "./validation.ts" + +// --------------------------------------------------------------------------- +// RPCTimeoutError +// --------------------------------------------------------------------------- + +/** + * Error thrown when an RPC call exceeds the configured timeout. + * + * Custom properties (`method`, `timeoutMs`) survive kkrpc's error + * serialization round-trip (same mechanism + * as `RPCValidationError`). + */ +export class RPCTimeoutError extends Error { + public readonly method: string + public readonly timeoutMs: number + + constructor(method: string, timeoutMs: number) { + super(`RPC call to "${method}" timed out after ${timeoutMs}ms`) + this.name = "RPCTimeoutError" + this.method = method + this.timeoutMs = timeoutMs + } +} + +/** + * Type guard for RPCTimeoutError. + * + * Uses `.name` check (not `instanceof`) so it works on deserialized + * errors received over the wire. + */ +export function isRPCTimeoutError(error: unknown): error is RPCTimeoutError { + return error instanceof Error && error.name === "RPCTimeoutError" +} + +interface PendingRequest { + resolve: (result: any) => void + reject: (error: any) => void +} + +interface CallbackFunction { + (...args: unknown[]): void +} + +type MessageWithTransferredValues = Message & { + __transferredValues?: unknown[] +} + +type StreamMarker = { + __stream: true +} + +function isObjectLike(value: unknown): value is object { + return value !== null && typeof value === "object" +} + +function isRecord(value: unknown): value is Record { + return isObjectLike(value) +} + +function getTransferredValues(message: Message): unknown[] { + const values = (message as MessageWithTransferredValues).__transferredValues + return Array.isArray(values) ? values : [] +} + +function getMessageArgs(message: Message): Record { + return isRecord(message.args) ? message.args : {} +} + +function isStreamMarker(value: unknown): value is StreamMarker { + return isRecord(value) && value.__stream === true +} + +function isEnhancedError(value: unknown): value is EnhancedError { + return isRecord(value) && typeof value.name === "string" && typeof value.message === "string" +} + +// --------------------------------------------------------------------------- +// Streaming internals +// --------------------------------------------------------------------------- + +/** + * Consumer-side state for an active stream. + * + * Chunks arrive asynchronously from the producer. The consumer reads them + * via `for await...of`. We bridge these with a simple queue: chunks that + * arrive before the consumer calls `next()` are buffered; `next()` calls + * that arrive before a chunk are parked as pending resolvers. + */ +interface StreamConsumerState { + /** Buffered iterator results (chunks, end, or error) that arrived before the consumer called next(). */ + buffer: IteratorResult[] + /** Error received from the producer (stream-error). Stored so subsequent next() calls reject. */ + error?: Error + /** Whether the stream has ended (stream-end or stream-error received). */ + done: boolean + /** Pending next() resolve/reject callbacks waiting for the next chunk. */ + waiters: Array<{ + resolve: (result: IteratorResult) => void + reject: (error: Error) => void + }> +} + +/** + * Producer-side state for an active stream. + */ +interface StreamProducerState { + /** AbortController to signal cancellation to the streaming loop. */ + abortController: AbortController +} + +/** Helper to detect AsyncIterable objects. */ +function isAsyncIterable(value: unknown): value is AsyncIterable { + return ( + value !== null && + value !== undefined && + typeof value === "object" && + Symbol.asyncIterator in value + ) +} + +export interface RPCChannelOptions> { + expose?: LocalAPI + serialization?: SerializationOptions + enableTransfer?: boolean + /** Optional validators for the exposed API. Validates inputs/outputs on the receiving side. */ + validators?: RPCValidators + /** Interceptors that wrap handler invocation on the receiving side. */ + interceptors?: RPCInterceptor[] + /** Timeout in ms for outgoing RPC calls. Default: 0 (no timeout). */ + timeout?: number + /** Optional metadata provider for outgoing request messages. */ + getMetadata?: () => RPCMessageMetadata | undefined +} + +/** + * A bidirectional Stdio IPC channel in RPC style. + * This allows 2 JS/TS processes to call each other's API like using libraries in RPC style, + * without needing to deal with `argv`, `stdin`, `stdout` directly. + */ +export class RPCChannelCore< + LocalAPI extends Record, + RemoteAPI extends Record, + Io extends IoInterface = IoInterface +> { + private pendingRequests: Record = {} + private pendingTimers: Record> = {} + private callbacks: Record = {} + private callbackCache: Map = new Map() + /** Producer-side: active outgoing streams keyed by request id. */ + private activeStreams: Map = new Map() + /** Consumer-side: active incoming streams keyed by request id. */ + private streamConsumers: Map = new Map() + private count: number = 0 + private messageStr = "" + private apiImplementation?: LocalAPI + /** + * Optional validators map that mirrors the LocalAPI shape. + * When set, incoming RPC calls are validated before the handler runs (input) + * and after it returns (output). Validation only applies to the locally + * exposed API — kkrpc is bidirectional, so each side validates its own API. + */ + private validators?: RPCValidators + /** Interceptor chain that wraps handler invocation in handleRequest. */ + private interceptors: ReadonlyArray + private serializationOptions: SerializationOptions + private supportsTransfer = false + private structuredClone = false + /** Timeout in ms for outgoing RPC calls. 0 or Infinity means no timeout. */ + private timeout: number + /** Optional metadata provider for outgoing request messages. */ + private getMetadata: (() => RPCMessageMetadata | undefined) | undefined + /** Set after EOF/destroy so new outbound calls reject immediately instead of timing out. */ + private isClosed = false + + constructor( + private io: Io, + options: RPCChannelOptions | undefined, + private serializationRuntime: RPCSerializationRuntime + ) { + this.apiImplementation = options?.expose + this.validators = options?.validators + this.interceptors = options?.interceptors ?? [] + this.timeout = options?.timeout ?? 0 + this.getMetadata = options?.getMetadata + this.serializationOptions = options?.serialization || {} + this.structuredClone = io.capabilities?.structuredClone === true + if ( + this.structuredClone && + io.capabilities?.transfer === true && + options?.enableTransfer !== false + ) { + this.supportsTransfer = true + } + this.listen() + } + + /** + * Exposes a local API implementation that can be called remotely + * @param api The local API implementation to expose + */ + expose(api: LocalAPI, validators?: RPCValidators) { + this.apiImplementation = api + if (validators) this.validators = validators + } + + /** + * Returns the IO interface used by this channel + * @returns The IO interface instance + */ + getIO(): Io { + return this.io + } + + /** + * Listens for incoming messages on the IO interface and dispatches them. + */ + private async listen(): Promise { + while (!this.isClosed) { + // Check adapters that expose an isDestroyed flag before blocking in read(). + if ("isDestroyed" in this.io && (this.io as Record).isDestroyed) { + this.closeFromTransport(new Error("RPC transport closed")) + break + } + + try { + const incoming = await this.io.read() + // null is the IoInterface EOF signal. Treat it as a hard close and reject waiters. + if (incoming === null) { + this.closeFromTransport(new Error("RPC transport closed")) + break + } + if (incoming === undefined) { + continue + } + await this.handleIncomingMessage(incoming) + } catch (error: unknown) { + // If the error indicates the adapter is destroyed, stop listening + if (error instanceof Error && error.message.includes("destroyed")) { + this.closeFromTransport(error) + break + } + console.error("kkrpc: failed to handle incoming message", error) + } + } + } + + private async handleIncomingMessage(raw: string | IoMessage): Promise { + if (typeof raw === "string") { + if (raw.trim().length === 0) { + return + } + this.bufferString(raw) + return + } + + const payload = raw.data + if (typeof payload === "string") { + await this.handleIncomingMessage(payload) + return + } + + if (!payload || typeof payload !== "object") { + return + } + + if (!payload.__transferredValues && raw.transfers && raw.transfers.length > 0) { + payload.__transferredValues = raw.transfers + } + + const message = await this.serializationRuntime.decodeMessage(payload) + await this.processDecodedMessage(message) + } + + private bufferString(chunk: string): void { + this.messageStr += chunk + const lastChar = this.messageStr[this.messageStr.length - 1] + const msgsSplit = this.messageStr.split("\n") + const msgs = lastChar === "\n" ? msgsSplit : msgsSplit.slice(0, -1) + this.messageStr = lastChar === "\n" ? "" : (msgsSplit.at(-1) ?? "") + + for (const msgStr of msgs.map((msg) => msg.trim()).filter(Boolean)) { + if (msgStr.startsWith("{")) { + void this.handleMessageStr(msgStr) + } else { + console.log(`(kkrpc stdout passthrough):`, msgStr) + } + } + } + + private async handleMessageStr(messageStr: string): Promise { + this.count++ + try { + const parsedMessage = await this.serializationRuntime.decodeMessage(messageStr) + await this.processDecodedMessage(parsedMessage) + } catch (error) { + console.error("failed to parse message", typeof messageStr, messageStr, error) + } + } + + private async processDecodedMessage(parsedMessage: Message): Promise { + if (parsedMessage.type === "response") { + this.handleResponse(parsedMessage as Message>) + } else if (parsedMessage.type === "request") { + this.handleRequest(parsedMessage) + } else if (parsedMessage.type === "callback") { + this.handleCallback(parsedMessage) + } else if (parsedMessage.type === "get") { + this.handleGet(parsedMessage) + } else if (parsedMessage.type === "set") { + this.handleSet(parsedMessage) + } else if (parsedMessage.type === "construct") { + this.handleConstruct(parsedMessage) + } else if (parsedMessage.type === "stream-chunk") { + this.handleStreamChunk(parsedMessage) + } else if (parsedMessage.type === "stream-end") { + this.handleStreamEnd(parsedMessage) + } else if (parsedMessage.type === "stream-error") { + this.handleStreamError(parsedMessage) + } else if (parsedMessage.type === "stream-cancel") { + this.handleStreamCancel(parsedMessage) + } else { + console.error("received unknown message type", parsedMessage, typeof parsedMessage) + } + } + + /** + * Calls a method on the remote API + * @param method The name of the method to call + * @param args Arguments to pass to the remote method + * @returns Promise that resolves with the result of the remote call + */ + public callMethod(method: T, args: any[]): Promise { + // Avoid creating a pending request that can never receive a response. + if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) + + return new Promise((resolve, reject) => { + const meta = this.getMetadata?.() + const messageId = generateUUID() + this.pendingRequests[messageId] = { resolve, reject } + this.startTimeout(messageId, method as string) + + const callbackIds: string[] = [] + const argsWithCallbacks = args.map((arg) => { + if (typeof arg === "function") { + let callbackId = this.callbackCache.get(arg) + if (!callbackId) { + callbackId = generateUUID() + this.callbacks[callbackId] = arg + this.callbackCache.set(arg, callbackId) + } + callbackIds.push(callbackId) + return `__callback__${callbackId}` + } + return arg + }) + + let finalArgs = argsWithCallbacks + const transferables: Transferable[] = [] + const transferSlots: TransferSlot[] = [] + const transferredValues: unknown[] = [] + + if (this.supportsTransfer) { + finalArgs = argsWithCallbacks.map((arg) => + processValueForTransfer(arg, transferables, transferSlots, transferredValues) + ) + } + + const message: Message = { + id: messageId, + method: method as string, + args: finalArgs, + type: "request", + callbackIds: callbackIds.length > 0 ? callbackIds : undefined, + transferSlots: transferSlots.length > 0 ? transferSlots : undefined, + ...(meta && Object.keys(meta).length > 0 ? { meta } : {}) + } + + this.sendMessage(message, transferables, transferredValues) + }) + } + + /** + * Gets a property value from the remote API + * @param path The property path (dot notation string or array) + * @returns Promise that resolves with the property value + */ + public getProperty(path: string | string[]): Promise { + // Avoid creating a pending request that can never receive a response. + if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) + + return new Promise((resolve, reject) => { + const messageId = generateUUID() + this.pendingRequests[messageId] = { resolve, reject } + const propertyPath = Array.isArray(path) ? path : path.split(".") + this.startTimeout(messageId, `get:${propertyPath.join(".")}`) + const message: Message = { + id: messageId, + method: "", + args: {}, + type: "get", + path: propertyPath + } + this.sendMessage(message) + }) + } + + /** + * Sets a property value on the remote API + * @param path The property path (dot notation string or array) + * @param value The value to set + * @returns Promise that resolves when the property is set + */ + public setProperty(path: string | string[], value: any): Promise { + // Avoid creating a pending request that can never receive a response. + if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) + + return new Promise((resolve, reject) => { + const messageId = generateUUID() + this.pendingRequests[messageId] = { resolve, reject } + const propertyPath = Array.isArray(path) ? path : path.split(".") + this.startTimeout(messageId, `set:${propertyPath.join(".")}`) + let processedValue = value + const transferables: Transferable[] = [] + const transferSlots: TransferSlot[] = [] + const transferredValues: unknown[] = [] + + if (this.supportsTransfer) { + processedValue = processValueForTransfer( + value, + transferables, + transferSlots, + transferredValues + ) + } + + const message: Message = { + id: messageId, + method: "", + args: {}, + type: "set", + path: propertyPath, + value: processedValue, + transferSlots: transferSlots.length > 0 ? transferSlots : undefined + } + + this.sendMessage(message, transferables, transferredValues) + }) + } + + /** + * Calls a constructor on the remote API + * @param constructor The name of the constructor to call + * @param args Arguments to pass to the remote constructor + * @returns Promise that resolves with the constructed instance + */ + public callConstructor(constructor: T, args: any[]): Promise { + // Avoid creating a pending request that can never receive a response. + if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) + + return new Promise((resolve, reject) => { + const messageId = generateUUID() + this.pendingRequests[messageId] = { resolve, reject } + this.startTimeout(messageId, constructor as string) + + const callbackIds: string[] = [] + const argsWithCallbacks = args.map((arg) => { + if (typeof arg === "function") { + let callbackId = this.callbackCache.get(arg) + if (!callbackId) { + callbackId = generateUUID() + this.callbacks[callbackId] = arg + this.callbackCache.set(arg, callbackId) + } + callbackIds.push(callbackId) + return `__callback__${callbackId}` + } + return arg + }) + + let finalArgs = argsWithCallbacks + const transferables: Transferable[] = [] + const transferSlots: TransferSlot[] = [] + const transferredValues: unknown[] = [] + + if (this.supportsTransfer) { + finalArgs = argsWithCallbacks.map((arg) => + processValueForTransfer(arg, transferables, transferSlots, transferredValues) + ) + } + + const message: Message = { + id: messageId, + method: constructor as string, + args: finalArgs, + type: "construct", + callbackIds: callbackIds.length > 0 ? callbackIds : undefined, + transferSlots: transferSlots.length > 0 ? transferSlots : undefined + } + + this.sendMessage(message, transferables, transferredValues) + }) + } + + /** + * Handles responses received from remote method calls + * @param response The response message to handle + * @private + */ + private handleResponse(response: Message>): void { + const { id } = response + const { result, error } = response.args + if (this.pendingRequests[id]) { + this.clearTimeout(id) + if (error) { + // Handle enhanced error objects + if (typeof error === "object" && error.name && error.message) { + this.pendingRequests[id].reject( + this.serializationRuntime.deserializeError(error as EnhancedError) + ) + } else { + // Fall back to simple string errors for backward compatibility + this.pendingRequests[id].reject(new Error(error as string)) + } + delete this.pendingRequests[id] + } else if (isStreamMarker(result)) { + // Stream response: resolve the pending request with an AsyncIterable + const iterable = this.createStreamIterable(id) + this.pendingRequests[id].resolve(iterable) + delete this.pendingRequests[id] + } else { + let finalResult = result + if (response.transferSlots && response.transferSlots.length > 0) { + const transferredValues = getTransferredValues(response) + finalResult = reconstructValueFromTransfer( + result, + response.transferSlots, + transferredValues + ) + } + this.pendingRequests[id].resolve(finalResult) + delete this.pendingRequests[id] + } + } + } + + /** + * Handles incoming method call requests from the remote endpoint + * @param request The request message to handle + * @private + */ + private async handleRequest(request: Message): Promise { + const { id, method } = request + let incomingArgs = Array.isArray(request.args) ? request.args : [] + + if (request.transferSlots && request.transferSlots.length > 0) { + const transferredValues = getTransferredValues(request) + incomingArgs = incomingArgs.map((arg: any) => + reconstructValueFromTransfer(arg, request.transferSlots!, transferredValues) + ) + } + + const methodPath = method.split(".") + if (!this.apiImplementation) { + // Broadcast transports deliver each request to every peer; peers with no API are not errors. + if (this.shouldIgnoreBroadcastRequestWithoutApi()) return + this.sendError(id, "No API implementation available") + return + } + let target: any = this.apiImplementation + + for (let i = 0; i < methodPath.length - 1; i++) { + target = target[methodPath[i]] + if (!target) { + if (this.shouldIgnoreBroadcastResolutionMiss()) return + this.sendError(id, `Method path ${method} not found at ${methodPath[i]}`) + return + } + } + + const finalMethod = methodPath[methodPath.length - 1] + const targetMethod = target[finalMethod] + + if (typeof targetMethod !== "function") { + // On broadcast transports, another peer may own this method and still respond. + if (this.shouldIgnoreBroadcastResolutionMiss()) return + this.sendError(id, `Method ${method} is not a function`) + return + } + + // Restore callback arguments: the caller serialized callback functions as + // "__callback__" placeholder strings. Here we convert them back into + // real functions that, when called, send a callback message to the remote side. + let processedArgs = incomingArgs.map((arg: any) => { + if (typeof arg === "string" && arg.startsWith("__callback__")) { + const callbackId = arg.slice(12) + return (...callbackArgs: any[]) => { + this.invokeCallback(callbackId, callbackArgs) + } + } + return arg + }) + + // --- Input / Output Validation --- + // Look up the validator for this method path (e.g. "math.divide"). + // If no validators were configured, or this method has none, this is undefined + // and validation is skipped entirely (backward-compatible no-op). + const methodValidator = lookupValidator(this.validators, method) + + if (methodValidator?.input) { + // Filter out callback arguments before validating. After the .map() above, + // callback placeholders have been replaced with real functions, so we filter + // by `typeof a !== "function"`. This matches the FilterCallbacks type + // utility that strips function params from the schema's expected input tuple. + const dataArgs = processedArgs.filter((a: any) => typeof a !== "function") + const inputResult = await runValidation(methodValidator.input, dataArgs) + // Use `=== false` (not `!`) for discriminated union narrowing — without a + // tsconfig, tsc doesn't narrow `!result.success` to the failure branch. + if (inputResult.success === false) { + this.sendError(id, new RPCValidationError("input", method, inputResult.issues)) + return + } + processedArgs = this.mergeValidatedArgs(processedArgs, inputResult.value) + } + + try { + // Wrap the handler invocation with the interceptor chain. + // Interceptors run after input validation, so they see clean args. + const invokeHandler = () => targetMethod.apply(target, processedArgs) + const result = + this.interceptors.length > 0 + ? await runInterceptors( + this.interceptors, + { id, method, args: processedArgs, meta: request.meta, state: {} }, + invokeHandler + ) + : await invokeHandler() + + // If the handler returned an AsyncIterable, stream it instead of + // sending a single response. We notify the consumer with a special + // "stream" flag on the initial response, then send chunks. + if (isAsyncIterable(result)) { + this.sendResponse(id, { __stream: true }) + this.streamResult(id, result, method, methodValidator) + return + } + + // Output validation: if an output schema is defined, validate the handler's + // return value before sending it back. This catches bugs where the handler + // returns an unexpected type (e.g. number instead of string). + if (methodValidator?.output) { + const outputResult = await runValidation(methodValidator.output, result) + if (outputResult.success === false) { + this.sendError(id, new RPCValidationError("output", method, outputResult.issues)) + return + } + // Send the validated value (which may have been transformed/coerced by the schema) + this.sendResponse(id, outputResult.value) + return + } + + this.sendResponse(id, result) + } catch (error: unknown) { + this.sendError(id, error instanceof Error ? error : new Error(String(error))) + } + } + + /** + * Invokes a callback on the remote endpoint + * @param callbackId The ID of the callback to invoke + * @param args Arguments to pass to the callback + * @private + */ + private invokeCallback(callbackId: string, args: any[]): void { + let finalArgs = args + const transferables: Transferable[] = [] + const transferSlots: TransferSlot[] = [] + const transferredValues: unknown[] = [] + + if (this.supportsTransfer) { + finalArgs = args.map((arg) => + processValueForTransfer(arg, transferables, transferSlots, transferredValues) + ) + } + + const message: Message = { + id: generateUUID(), + method: callbackId, + args: finalArgs, + type: "callback", + transferSlots: transferSlots.length > 0 ? transferSlots : undefined + } + this.sendMessage(message, transferables, transferredValues) + } + + /** + * Handles callback invocations received from the remote endpoint + * @param message The callback message to handle + * @private + */ + private handleCallback(message: Message): void { + const { method: callbackId } = message + const callback = this.callbacks[callbackId] + if (callback) { + let callbackArgs = Array.isArray(message.args) ? message.args : [] + if (message.transferSlots && message.transferSlots.length > 0) { + const transferredValues = getTransferredValues(message) + callbackArgs = callbackArgs.map((arg: any) => + reconstructValueFromTransfer(arg, message.transferSlots!, transferredValues) + ) + } + callback(...callbackArgs) + } else { + console.error(`Callback with id ${callbackId} not found`) + } + } + + /** + * Handles property get requests from the remote endpoint + * @param request The get request message to handle + * @private + */ + private handleGet(request: Message): void { + const { id, path } = request + if (!path) { + this.sendError(id, "Invalid get request: missing path") + return + } + if (!this.apiImplementation) { + if (this.shouldIgnoreBroadcastRequestWithoutApi()) return + this.sendError(id, "No API implementation available") + return + } + + try { + // Traverse the object path to get the property value + let target: any = this.apiImplementation + for (const prop of path) { + target = target[prop] + if (target === undefined) { + if (this.shouldIgnoreBroadcastResolutionMiss()) return + this.sendError(id, `Property path ${path.join(".")} not found at ${prop}`) + return + } + } + this.sendResponse(id, target) + } catch (error: unknown) { + this.sendError(id, error instanceof Error ? error : new Error(String(error))) + } + } + + /** + * Handles property set requests from the remote endpoint + * @param request The set request message to handle + * @private + */ + private handleSet(request: Message): void { + const { id, path } = request + if (!path) { + this.sendError(id, "Invalid set request: missing path") + return + } + if (!this.apiImplementation) { + if (this.shouldIgnoreBroadcastRequestWithoutApi()) return + this.sendError(id, "No API implementation available") + return + } + + let incomingValue = request.value + if (request.transferSlots && request.transferSlots.length > 0) { + const transferredValues = getTransferredValues(request) + incomingValue = reconstructValueFromTransfer( + request.value, + request.transferSlots, + transferredValues + ) + } + + try { + // Traverse to the parent object + let target: any = this.apiImplementation + for (let i = 0; i < path.length - 1; i++) { + target = target[path[i]] + if (!target) { + if (this.shouldIgnoreBroadcastResolutionMiss()) return + this.sendError(id, `Property path ${path.join(".")} not found at ${path[i]}`) + return + } + } + + // Set the final property + const finalProp = path[path.length - 1] + if (this.shouldIgnoreBroadcastResolutionMiss() && !(finalProp in Object(target))) return + target[finalProp] = incomingValue + this.sendResponse(id, true) // Return true to indicate success + } catch (error: unknown) { + this.sendError(id, error instanceof Error ? error : new Error(String(error))) + } + } + + /** + * Handles constructor calls from the remote endpoint + * @param request The construct request message to handle + * @private + */ + private handleConstruct(request: Message): void { + const { id, method } = request + let incomingArgs = Array.isArray(request.args) ? request.args : [] + + if (request.transferSlots && request.transferSlots.length > 0) { + const transferredValues = getTransferredValues(request) + incomingArgs = incomingArgs.map((arg: any) => + reconstructValueFromTransfer(arg, request.transferSlots!, transferredValues) + ) + } + + // Split the method path and traverse the API implementation + const methodPath = method.split(".") + if (!this.apiImplementation) { + if (this.shouldIgnoreBroadcastRequestWithoutApi()) return + this.sendError(id, "No API implementation available") + return + } + let target: any = this.apiImplementation + + // Traverse the object path + for (let i = 0; i < methodPath.length - 1; i++) { + target = target[methodPath[i]] + if (!target) { + if (this.shouldIgnoreBroadcastResolutionMiss()) return + this.sendError(id, `Constructor path ${method} not found at ${methodPath[i]}`) + return + } + } + + const finalMethod = methodPath[methodPath.length - 1] + const ConstructorClass = target[finalMethod] + + if (typeof ConstructorClass !== "function") { + if (this.shouldIgnoreBroadcastResolutionMiss()) return + this.sendError(id, `${method} is not a constructor function`) + return + } + + const processedArgs = incomingArgs.map((arg: any) => { + if (typeof arg === "string" && arg.startsWith("__callback__")) { + const callbackId = arg.slice(12) + return (...callbackArgs: any[]) => { + this.invokeCallback(callbackId, callbackArgs) + } + } + return arg + }) + + try { + const instance = new ConstructorClass(...processedArgs) + this.sendResponse(id, instance) + } catch (error: unknown) { + this.sendError(id, error instanceof Error ? error : new Error(String(error))) + } + } + + // --------------------------------------------------------------------------- + // Streaming — producer side + // --------------------------------------------------------------------------- + + /** + * Iterates an AsyncIterable and sends each yielded value as a stream-chunk. + * Sends stream-end when the iterator completes, or stream-error on failure. + * The loop stops if the consumer sends stream-cancel. + */ + private async streamResult( + requestId: string, + iterable: AsyncIterable, + method: string, + methodValidator?: ReturnType + ): Promise { + const abortController = new AbortController() + this.activeStreams.set(requestId, { abortController }) + + try { + for await (const value of iterable) { + // Check cancellation + if (abortController.signal.aborted) break + + let chunk = value + + // Per-chunk output validation + if (methodValidator?.output) { + const outputResult = await runValidation(methodValidator.output, chunk) + if (outputResult.success === false) { + this.sendStreamError( + requestId, + new RPCValidationError("output", method, outputResult.issues) + ) + return + } + chunk = outputResult.value + } + + const transferables: Transferable[] = [] + const transferSlots: TransferSlot[] = [] + const transferredValues: unknown[] = [] + + if (this.supportsTransfer) { + // Stream chunks use the same transfer envelope as normal responses. + chunk = processValueForTransfer(chunk, transferables, transferSlots, transferredValues) + } + + this.sendMessage( + { + id: requestId, + method: "", + args: { value: chunk }, + type: "stream-chunk", + transferSlots: transferSlots.length > 0 ? transferSlots : undefined + }, + transferables, + transferredValues + ) + } + + // Stream completed normally + if (!abortController.signal.aborted) { + this.sendMessage({ + id: requestId, + method: "", + args: {}, + type: "stream-end" + }) + } + } catch (error: unknown) { + if (!abortController.signal.aborted) { + this.sendStreamError(requestId, error instanceof Error ? error : new Error(String(error))) + } + } finally { + this.activeStreams.delete(requestId) + } + } + + /** Send a stream-error message to the consumer. */ + private sendStreamError(requestId: string, error: Error): void { + this.sendMessage({ + id: requestId, + method: "", + args: { error: this.serializationRuntime.serializeError(error) }, + type: "stream-error" + }) + } + + /** Handle a stream-cancel message from the consumer. */ + private handleStreamCancel(message: Message): void { + const state = this.activeStreams.get(message.id) + if (state) { + state.abortController.abort() + this.activeStreams.delete(message.id) + } + } + + // --------------------------------------------------------------------------- + // Streaming — consumer side + // --------------------------------------------------------------------------- + + /** + * Creates an AsyncIterable that yields values from incoming stream-chunk + * messages. The consumer calls `break` or `return()` to send stream-cancel. + */ + private createStreamIterable(requestId: string): AsyncIterable { + const consumer: StreamConsumerState = { + buffer: [], + done: false, + waiters: [] + } + this.streamConsumers.set(requestId, consumer) + + const channel = this + return { + [Symbol.asyncIterator]() { + return { + next(): Promise> { + // If there are buffered results, return immediately + if (consumer.buffer.length > 0) { + return Promise.resolve(consumer.buffer.shift()!) + } + // If stream is done and buffer is empty, return done + if (consumer.done) { + if (consumer.error) { + return Promise.reject(consumer.error) + } + return Promise.resolve({ done: true, value: undefined }) + } + // Wait for next chunk + return new Promise((resolve, reject) => { + consumer.waiters.push({ resolve, reject }) + }) + }, + return(): Promise> { + // Consumer is breaking out of for-await — send cancel + channel.sendMessage({ + id: requestId, + method: "", + args: {}, + type: "stream-cancel" + }) + consumer.done = true + channel.streamConsumers.delete(requestId) + return Promise.resolve({ done: true, value: undefined }) + } + } + } + } + } + + /** Handle a stream-chunk message from the producer. */ + private handleStreamChunk(message: Message): void { + const consumer = this.streamConsumers.get(message.id) + if (!consumer) return + let value = getMessageArgs(message).value + if (message.transferSlots && message.transferSlots.length > 0) { + // Rehydrate zero-copy stream chunks before yielding them to the consumer. + const transferredValues = getTransferredValues(message) + value = reconstructValueFromTransfer(value, message.transferSlots, transferredValues) + } + + const result: IteratorResult = { + done: false, + value + } + + if (consumer.waiters.length > 0) { + consumer.waiters.shift()!.resolve(result) + } else { + consumer.buffer.push(result) + } + } + + /** Handle a stream-end message from the producer. */ + private handleStreamEnd(message: Message): void { + const consumer = this.streamConsumers.get(message.id) + if (!consumer) return + + consumer.done = true + this.streamConsumers.delete(message.id) + + if (consumer.waiters.length > 0) { + // Resolve all waiters with done + for (const waiter of consumer.waiters) { + waiter.resolve({ done: true, value: undefined }) + } + consumer.waiters.length = 0 + } else { + consumer.buffer.push({ done: true, value: undefined }) + } + } + + /** Handle a stream-error message from the producer. */ + private handleStreamError(message: Message): void { + const consumer = this.streamConsumers.get(message.id) + if (!consumer) return + + const rawError = getMessageArgs(message).error + const error = isEnhancedError(rawError) + ? this.serializationRuntime.deserializeError(rawError) + : new Error(typeof rawError === "string" ? rawError : "Stream error") + + consumer.done = true + consumer.error = error + this.streamConsumers.delete(message.id) + + if (consumer.waiters.length > 0) { + for (const waiter of consumer.waiters) { + waiter.reject(error) + } + consumer.waiters.length = 0 + } + } + + /** + * Sends a successful response back to the remote endpoint + * @param id The ID of the request being responded to + * @param result The result to send back + * @private + */ + private sendResponse(id: string, result: T): void { + let responseResult = result + const transferables: Transferable[] = [] + const transferSlots: TransferSlot[] = [] + const transferredValues: unknown[] = [] + + if (this.supportsTransfer) { + responseResult = processValueForTransfer( + result, + transferables, + transferSlots, + transferredValues + ) + } + + const response: Message> = { + id, + method: "", + args: { result: responseResult }, + type: "response", + transferSlots: transferSlots.length > 0 ? transferSlots : undefined + } + + this.sendMessage(response, transferables, transferredValues) + } + + /** + * Sends an error response back to the remote endpoint + * @param id The ID of the request being responded to + * @param error The error message or Error object to send back + * @private + */ + private sendError(id: string, error: string | Error): void { + const errorResponse = error instanceof Error ? this.serializationRuntime.serializeError(error) : error + + const response: Message> = { + id, + method: "", + args: { error: errorResponse }, + type: "response" + } + this.sendMessage(response) + } + + private sendMessage( + message: Message, + transferables: Transferable[] = [], + transferredValues: unknown[] = [] + ): void { + // Centralized send path so write failures can reject the matching pending request. + const encoded = this.serializationRuntime.encodeMessage( + message, + this.serializationOptions, + this.supportsTransfer && transferables.length > 0, + transferredValues + ) + + const outgoing: string | IoMessage = + encoded.mode === "string" + ? encoded.data + : { + data: encoded.data, + transfers: transferables + } + + try { + void this.io.write(outgoing).catch((error: unknown) => { + this.handleWriteFailure(message, error) + }) + } catch (error: unknown) { + this.handleWriteFailure(message, error) + } + } + + /** Rejects the RPC call that initiated this write, instead of leaving it to timeout. */ + private handleWriteFailure(message: Message, error: unknown): void { + const writeError = error instanceof Error ? error : new Error(String(error)) + const pending = this.pendingRequests[message.id] + if (pending) { + this.clearTimeout(message.id) + pending.reject(writeError) + delete this.pendingRequests[message.id] + return + } + + if (!this.isClosed) { + console.error("kkrpc: failed to write message", writeError) + } + } + + /** + * Validation schemas may coerce data arguments, but callback function arguments are not validated. + * This merges transformed data args back into the original argument list without moving callbacks. + */ + private mergeValidatedArgs(originalArgs: any[], validatedValue: unknown): any[] { + const validatedArgs = Array.isArray(validatedValue) ? validatedValue : [validatedValue] + let dataArgIndex = 0 + return originalArgs.map((arg) => { + if (typeof arg === "function") return arg + if (dataArgIndex >= validatedArgs.length) return arg + const validatedArg = validatedArgs[dataArgIndex] + dataArgIndex++ + return validatedArg + }) + } + + /** True when this peer should silently ignore broadcast traffic because it exposes no API. */ + private shouldIgnoreBroadcastRequestWithoutApi(): boolean { + return !this.apiImplementation && this.io.capabilities?.broadcast === true + } + + /** True when another broadcast peer may be the intended owner of the missing path. */ + private shouldIgnoreBroadcastResolutionMiss(): boolean { + return this.io.capabilities?.broadcast === true + } + + /** Rejects all pending RPC calls during transport close or explicit destroy. */ + private rejectPendingRequests(error: Error): void { + for (const [id, pending] of Object.entries(this.pendingRequests)) { + pending.reject(error) + this.clearTimeout(id) + } + this.pendingRequests = {} + this.pendingTimers = {} + } + + /** Stops producer-side async iterators so they do not keep yielding after close. */ + private abortActiveStreams(): void { + for (const [, state] of this.activeStreams) { + state.abortController.abort() + } + this.activeStreams.clear() + } + + /** Rejects consumer-side async iterator waiters that are blocked on the next chunk. */ + private rejectStreamConsumers(error: Error): void { + for (const [, consumer] of this.streamConsumers) { + consumer.done = true + consumer.error = error + for (const waiter of consumer.waiters) { + waiter.reject(error) + } + consumer.waiters.length = 0 + } + this.streamConsumers.clear() + } + + /** Common close path used by EOF, adapter destroy, write-side destroy, and RPCChannel.destroy(). */ + private closeFromTransport(error: Error): void { + if (this.isClosed) return + this.isClosed = true + this.rejectPendingRequests(error) + this.abortActiveStreams() + this.rejectStreamConsumers(error) + this.freeCallbacks() + } + + // --------------------------------------------------------------------------- + // Timeout helpers + // --------------------------------------------------------------------------- + + /** Start a timeout timer for a pending request. No-op if timeout is 0 or Infinity. */ + private startTimeout(messageId: string, method: string): void { + if (!this.timeout || this.timeout === Infinity) return + this.pendingTimers[messageId] = setTimeout(() => { + const pending = this.pendingRequests[messageId] + if (pending) { + pending.reject(new RPCTimeoutError(method, this.timeout)) + delete this.pendingRequests[messageId] + delete this.pendingTimers[messageId] + } + }, this.timeout) + } + + /** Clear the timeout timer for a request that received a response. */ + private clearTimeout(messageId: string): void { + if (this.pendingTimers[messageId]) { + clearTimeout(this.pendingTimers[messageId]) + delete this.pendingTimers[messageId] + } + } + + /** + * Creates a nested proxy object for chaining remote method calls, property access, and constructor calls + * @param chain Array of method names in the chain + * @returns Proxy object that transforms property access into remote method calls + * @private + */ + private createNestedProxy(chain: string[] = []): any { + return new Proxy(() => {}, { + get: (_target, prop: string | symbol) => { + // Handle special properties + if (typeof prop === "string") { + // Handle property access like obj.prop + if (prop !== "then") { + return this.createNestedProxy([...chain, prop]) + } + // Handle thenable for await support + if (prop === "then" && chain.length > 0) { + // Return property value when accessed like: await obj.prop + const promise = this.getProperty(chain) + return promise.then.bind(promise) + } + } + return undefined + }, + set: (_target, prop: string | symbol, value: any) => { + // Handle property setting like obj.prop = value + if (typeof prop === "string") { + const propertyPath = [...chain, prop] + this.setProperty(propertyPath, value) + return true + } + return false + }, + apply: (_target, _thisArg, args: any[]) => { + // Handle method calls like obj.method() + const method = chain.join(".") + return this.callMethod(method as keyof RemoteAPI, args) + }, + construct: (_target, args: any[]) => { + // Handle constructor calls like new obj.Constructor() + return this.callConstructor(chain.join(".") as keyof RemoteAPI, args) + } + }) + } + + /** + * Returns a proxy object that represents the remote API + * Methods called on this proxy will be executed on the remote endpoint + * @returns Proxy object representing the remote API + */ + public getAPI(): RemoteAPI { + return this.createNestedProxy() as RemoteAPI + } + + /** + * Destroys the RPC channel and underlying IO interface if it's destroyable + */ + destroy(): void { + this.closeFromTransport(new Error("RPC channel destroyed")) + + // Clean up IO adapters + if (this.io && this.io.destroy) { + this.io.destroy() + } + } + + /** + * Frees up memory by clearing stored callbacks and callback cache + * Useful when dealing with many anonymous callback functions to prevent memory leaks + */ + freeCallbacks() { + this.callbacks = {} + this.callbackCache.clear() + } +} diff --git a/packages/kkrpc/src/channel-lite.ts b/packages/kkrpc/src/channel-lite.ts new file mode 100644 index 0000000..4e0d968 --- /dev/null +++ b/packages/kkrpc/src/channel-lite.ts @@ -0,0 +1,28 @@ +/** + * Browser-lite RPCChannel facade. + * This wrapper preserves the public `new RPCChannel(...)` API while using the + * JSON-only serialization runtime so SuperJSON stays out of lite bundles. + */ +export { + isRPCTimeoutError, + RPCTimeoutError, + type RPCChannelOptions +} from "./channel-core.ts" +import { RPCChannelCore, type RPCChannelOptions } from "./channel-core.ts" +import type { IoInterface } from "./interface.ts" +import { jsonSerializationRuntime } from "./serialization-json.ts" + +export class RPCChannel< + LocalAPI extends Record, + RemoteAPI extends Record, + Io extends IoInterface = IoInterface +> extends RPCChannelCore { + constructor(io: Io, options?: RPCChannelOptions) { + if (options?.serialization?.version === "superjson") { + throw new Error( + 'SuperJSON serialization is not available in kkrpc/browser-lite. Use kkrpc/browser or configure both endpoints with serialization.version = "json".' + ) + } + super(io, options, jsonSerializationRuntime) + } +} diff --git a/packages/kkrpc/src/channel.ts b/packages/kkrpc/src/channel.ts index faa0c28..1caa4da 100644 --- a/packages/kkrpc/src/channel.ts +++ b/packages/kkrpc/src/channel.ts @@ -1,1350 +1,23 @@ -import type { IoInterface, IoMessage } from "./interface.ts" -import { runInterceptors, type RPCInterceptor } from "./middleware.ts" -import { - decodeMessage, - deserializeError, - encodeMessage, - processValueForTransfer, - reconstructValueFromTransfer, - serializeError, - type EnhancedError, - type Message, - type Response, - type SerializationOptions, - type TransferSlot -} from "./serialization.ts" -import { generateUUID } from "./utils.ts" -import { - lookupValidator, - RPCValidationError, - runValidation, - type RPCValidators -} from "./validation.ts" - -// --------------------------------------------------------------------------- -// RPCTimeoutError -// --------------------------------------------------------------------------- - /** - * Error thrown when an RPC call exceeds the configured timeout. - * - * Custom properties (`method`, `timeoutMs`) survive kkrpc's - * `serializeError()` / `deserializeError()` round-trip (same mechanism - * as `RPCValidationError`). + * Full kkrpc RPCChannel facade. + * This entry keeps existing behavior by using the SuperJSON-enabled + * serialization runtime. */ -export class RPCTimeoutError extends Error { - public readonly method: string - public readonly timeoutMs: number - - constructor(method: string, timeoutMs: number) { - super(`RPC call to "${method}" timed out after ${timeoutMs}ms`) - this.name = "RPCTimeoutError" - this.method = method - this.timeoutMs = timeoutMs - } -} - -/** - * Type guard for RPCTimeoutError. - * - * Uses `.name` check (not `instanceof`) so it works on deserialized - * errors received over the wire. - */ -export function isRPCTimeoutError(error: unknown): error is RPCTimeoutError { - return error instanceof Error && error.name === "RPCTimeoutError" -} - -interface PendingRequest { - resolve: (result: any) => void - reject: (error: any) => void -} - -interface CallbackFunction { - (...args: unknown[]): void -} +export { + isRPCTimeoutError, + RPCTimeoutError, + type RPCChannelOptions +} from "./channel-core.ts" +import { RPCChannelCore, type RPCChannelOptions } from "./channel-core.ts" +import type { IoInterface } from "./interface.ts" +import { fullSerializationRuntime } from "./serialization-full.ts" -type MessageWithTransferredValues = Message & { - __transferredValues?: unknown[] -} - -type StreamMarker = { - __stream: true -} - -function isObjectLike(value: unknown): value is object { - return value !== null && typeof value === "object" -} - -function isRecord(value: unknown): value is Record { - return isObjectLike(value) -} - -function getTransferredValues(message: Message): unknown[] { - const values = (message as MessageWithTransferredValues).__transferredValues - return Array.isArray(values) ? values : [] -} - -function getMessageArgs(message: Message): Record { - return isRecord(message.args) ? message.args : {} -} - -function isStreamMarker(value: unknown): value is StreamMarker { - return isRecord(value) && value.__stream === true -} - -function isEnhancedError(value: unknown): value is EnhancedError { - return isRecord(value) && typeof value.name === "string" && typeof value.message === "string" -} - -// --------------------------------------------------------------------------- -// Streaming internals -// --------------------------------------------------------------------------- - -/** - * Consumer-side state for an active stream. - * - * Chunks arrive asynchronously from the producer. The consumer reads them - * via `for await...of`. We bridge these with a simple queue: chunks that - * arrive before the consumer calls `next()` are buffered; `next()` calls - * that arrive before a chunk are parked as pending resolvers. - */ -interface StreamConsumerState { - /** Buffered iterator results (chunks, end, or error) that arrived before the consumer called next(). */ - buffer: IteratorResult[] - /** Error received from the producer (stream-error). Stored so subsequent next() calls reject. */ - error?: Error - /** Whether the stream has ended (stream-end or stream-error received). */ - done: boolean - /** Pending next() resolve/reject callbacks waiting for the next chunk. */ - waiters: Array<{ - resolve: (result: IteratorResult) => void - reject: (error: Error) => void - }> -} - -/** - * Producer-side state for an active stream. - */ -interface StreamProducerState { - /** AbortController to signal cancellation to the streaming loop. */ - abortController: AbortController -} - -/** Helper to detect AsyncIterable objects. */ -function isAsyncIterable(value: unknown): value is AsyncIterable { - return ( - value !== null && - value !== undefined && - typeof value === "object" && - Symbol.asyncIterator in value - ) -} - -/** - * A bidirectional Stdio IPC channel in RPC style. - * This allows 2 JS/TS processes to call each other's API like using libraries in RPC style, - * without needing to deal with `argv`, `stdin`, `stdout` directly. - */ export class RPCChannel< LocalAPI extends Record, RemoteAPI extends Record, Io extends IoInterface = IoInterface -> { - private pendingRequests: Record = {} - private pendingTimers: Record> = {} - private callbacks: Record = {} - private callbackCache: Map = new Map() - /** Producer-side: active outgoing streams keyed by request id. */ - private activeStreams: Map = new Map() - /** Consumer-side: active incoming streams keyed by request id. */ - private streamConsumers: Map = new Map() - private count: number = 0 - private messageStr = "" - private apiImplementation?: LocalAPI - /** - * Optional validators map that mirrors the LocalAPI shape. - * When set, incoming RPC calls are validated before the handler runs (input) - * and after it returns (output). Validation only applies to the locally - * exposed API — kkrpc is bidirectional, so each side validates its own API. - */ - private validators?: RPCValidators - /** Interceptor chain that wraps handler invocation in handleRequest. */ - private interceptors: ReadonlyArray - private serializationOptions: SerializationOptions - private supportsTransfer = false - private structuredClone = false - /** Timeout in ms for outgoing RPC calls. 0 or Infinity means no timeout. */ - private timeout: number - /** Set after EOF/destroy so new outbound calls reject immediately instead of timing out. */ - private isClosed = false - - constructor( - private io: Io, - options?: { - expose?: LocalAPI - serialization?: SerializationOptions - enableTransfer?: boolean - /** Optional validators for the exposed API. Validates inputs/outputs on the receiving side. */ - validators?: RPCValidators - /** Interceptors that wrap handler invocation on the receiving side. */ - interceptors?: RPCInterceptor[] - /** Timeout in ms for outgoing RPC calls. Default: 0 (no timeout). */ - timeout?: number - } - ) { - this.apiImplementation = options?.expose - this.validators = options?.validators - this.interceptors = options?.interceptors ?? [] - this.timeout = options?.timeout ?? 0 - this.serializationOptions = options?.serialization || {} - this.structuredClone = io.capabilities?.structuredClone === true - if ( - this.structuredClone && - io.capabilities?.transfer === true && - options?.enableTransfer !== false - ) { - this.supportsTransfer = true - } - this.listen() - } - - /** - * Exposes a local API implementation that can be called remotely - * @param api The local API implementation to expose - */ - expose(api: LocalAPI, validators?: RPCValidators) { - this.apiImplementation = api - if (validators) this.validators = validators - } - - /** - * Returns the IO interface used by this channel - * @returns The IO interface instance - */ - getIO(): Io { - return this.io - } - - /** - * Listens for incoming messages on the IO interface and dispatches them. - */ - private async listen(): Promise { - while (!this.isClosed) { - // Check adapters that expose an isDestroyed flag before blocking in read(). - if ("isDestroyed" in this.io && (this.io as Record).isDestroyed) { - this.closeFromTransport(new Error("RPC transport closed")) - break - } - - try { - const incoming = await this.io.read() - // null is the IoInterface EOF signal. Treat it as a hard close and reject waiters. - if (incoming === null) { - this.closeFromTransport(new Error("RPC transport closed")) - break - } - if (incoming === undefined) { - continue - } - await this.handleIncomingMessage(incoming) - } catch (error: unknown) { - // If the error indicates the adapter is destroyed, stop listening - if (error instanceof Error && error.message.includes("destroyed")) { - this.closeFromTransport(error) - break - } - console.error("kkrpc: failed to handle incoming message", error) - } - } - } - - private async handleIncomingMessage(raw: string | IoMessage): Promise { - if (typeof raw === "string") { - if (raw.trim().length === 0) { - return - } - this.bufferString(raw) - return - } - - const payload = raw.data - if (typeof payload === "string") { - await this.handleIncomingMessage(payload) - return - } - - if (!payload || typeof payload !== "object") { - return - } - - if (!payload.__transferredValues && raw.transfers && raw.transfers.length > 0) { - payload.__transferredValues = raw.transfers - } - - const message = await decodeMessage(payload) - await this.processDecodedMessage(message) - } - - private bufferString(chunk: string): void { - this.messageStr += chunk - const lastChar = this.messageStr[this.messageStr.length - 1] - const msgsSplit = this.messageStr.split("\n") - const msgs = lastChar === "\n" ? msgsSplit : msgsSplit.slice(0, -1) - this.messageStr = lastChar === "\n" ? "" : (msgsSplit.at(-1) ?? "") - - for (const msgStr of msgs.map((msg) => msg.trim()).filter(Boolean)) { - if (msgStr.startsWith("{")) { - void this.handleMessageStr(msgStr) - } else { - console.log(`(kkrpc stdout passthrough):`, msgStr) - } - } - } - - private async handleMessageStr(messageStr: string): Promise { - this.count++ - try { - const parsedMessage = await decodeMessage(messageStr) - await this.processDecodedMessage(parsedMessage) - } catch (error) { - console.error("failed to parse message", typeof messageStr, messageStr, error) - } - } - - private async processDecodedMessage(parsedMessage: Message): Promise { - if (parsedMessage.type === "response") { - this.handleResponse(parsedMessage as Message>) - } else if (parsedMessage.type === "request") { - this.handleRequest(parsedMessage) - } else if (parsedMessage.type === "callback") { - this.handleCallback(parsedMessage) - } else if (parsedMessage.type === "get") { - this.handleGet(parsedMessage) - } else if (parsedMessage.type === "set") { - this.handleSet(parsedMessage) - } else if (parsedMessage.type === "construct") { - this.handleConstruct(parsedMessage) - } else if (parsedMessage.type === "stream-chunk") { - this.handleStreamChunk(parsedMessage) - } else if (parsedMessage.type === "stream-end") { - this.handleStreamEnd(parsedMessage) - } else if (parsedMessage.type === "stream-error") { - this.handleStreamError(parsedMessage) - } else if (parsedMessage.type === "stream-cancel") { - this.handleStreamCancel(parsedMessage) - } else { - console.error("received unknown message type", parsedMessage, typeof parsedMessage) - } - } - - /** - * Calls a method on the remote API - * @param method The name of the method to call - * @param args Arguments to pass to the remote method - * @returns Promise that resolves with the result of the remote call - */ - public callMethod(method: T, args: any[]): Promise { - // Avoid creating a pending request that can never receive a response. - if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) - - return new Promise((resolve, reject) => { - const messageId = generateUUID() - this.pendingRequests[messageId] = { resolve, reject } - this.startTimeout(messageId, method as string) - - const callbackIds: string[] = [] - const argsWithCallbacks = args.map((arg) => { - if (typeof arg === "function") { - let callbackId = this.callbackCache.get(arg) - if (!callbackId) { - callbackId = generateUUID() - this.callbacks[callbackId] = arg - this.callbackCache.set(arg, callbackId) - } - callbackIds.push(callbackId) - return `__callback__${callbackId}` - } - return arg - }) - - let finalArgs = argsWithCallbacks - const transferables: Transferable[] = [] - const transferSlots: TransferSlot[] = [] - const transferredValues: unknown[] = [] - - if (this.supportsTransfer) { - finalArgs = argsWithCallbacks.map((arg) => - processValueForTransfer(arg, transferables, transferSlots, transferredValues) - ) - } - - const message: Message = { - id: messageId, - method: method as string, - args: finalArgs, - type: "request", - callbackIds: callbackIds.length > 0 ? callbackIds : undefined, - transferSlots: transferSlots.length > 0 ? transferSlots : undefined - } - - this.sendMessage(message, transferables, transferredValues) - }) - } - - /** - * Gets a property value from the remote API - * @param path The property path (dot notation string or array) - * @returns Promise that resolves with the property value - */ - public getProperty(path: string | string[]): Promise { - // Avoid creating a pending request that can never receive a response. - if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) - - return new Promise((resolve, reject) => { - const messageId = generateUUID() - this.pendingRequests[messageId] = { resolve, reject } - const propertyPath = Array.isArray(path) ? path : path.split(".") - this.startTimeout(messageId, `get:${propertyPath.join(".")}`) - const message: Message = { - id: messageId, - method: "", - args: {}, - type: "get", - path: propertyPath - } - this.sendMessage(message) - }) - } - - /** - * Sets a property value on the remote API - * @param path The property path (dot notation string or array) - * @param value The value to set - * @returns Promise that resolves when the property is set - */ - public setProperty(path: string | string[], value: any): Promise { - // Avoid creating a pending request that can never receive a response. - if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) - - return new Promise((resolve, reject) => { - const messageId = generateUUID() - this.pendingRequests[messageId] = { resolve, reject } - const propertyPath = Array.isArray(path) ? path : path.split(".") - this.startTimeout(messageId, `set:${propertyPath.join(".")}`) - let processedValue = value - const transferables: Transferable[] = [] - const transferSlots: TransferSlot[] = [] - const transferredValues: unknown[] = [] - - if (this.supportsTransfer) { - processedValue = processValueForTransfer( - value, - transferables, - transferSlots, - transferredValues - ) - } - - const message: Message = { - id: messageId, - method: "", - args: {}, - type: "set", - path: propertyPath, - value: processedValue, - transferSlots: transferSlots.length > 0 ? transferSlots : undefined - } - - this.sendMessage(message, transferables, transferredValues) - }) - } - - /** - * Calls a constructor on the remote API - * @param constructor The name of the constructor to call - * @param args Arguments to pass to the remote constructor - * @returns Promise that resolves with the constructed instance - */ - public callConstructor(constructor: T, args: any[]): Promise { - // Avoid creating a pending request that can never receive a response. - if (this.isClosed) return Promise.reject(new Error("RPC channel closed")) - - return new Promise((resolve, reject) => { - const messageId = generateUUID() - this.pendingRequests[messageId] = { resolve, reject } - this.startTimeout(messageId, constructor as string) - - const callbackIds: string[] = [] - const argsWithCallbacks = args.map((arg) => { - if (typeof arg === "function") { - let callbackId = this.callbackCache.get(arg) - if (!callbackId) { - callbackId = generateUUID() - this.callbacks[callbackId] = arg - this.callbackCache.set(arg, callbackId) - } - callbackIds.push(callbackId) - return `__callback__${callbackId}` - } - return arg - }) - - let finalArgs = argsWithCallbacks - const transferables: Transferable[] = [] - const transferSlots: TransferSlot[] = [] - const transferredValues: unknown[] = [] - - if (this.supportsTransfer) { - finalArgs = argsWithCallbacks.map((arg) => - processValueForTransfer(arg, transferables, transferSlots, transferredValues) - ) - } - - const message: Message = { - id: messageId, - method: constructor as string, - args: finalArgs, - type: "construct", - callbackIds: callbackIds.length > 0 ? callbackIds : undefined, - transferSlots: transferSlots.length > 0 ? transferSlots : undefined - } - - this.sendMessage(message, transferables, transferredValues) - }) - } - - /** - * Handles responses received from remote method calls - * @param response The response message to handle - * @private - */ - private handleResponse(response: Message>): void { - const { id } = response - const { result, error } = response.args - if (this.pendingRequests[id]) { - this.clearTimeout(id) - if (error) { - // Handle enhanced error objects - if (typeof error === "object" && error.name && error.message) { - this.pendingRequests[id].reject(deserializeError(error as EnhancedError)) - } else { - // Fall back to simple string errors for backward compatibility - this.pendingRequests[id].reject(new Error(error as string)) - } - delete this.pendingRequests[id] - } else if (isStreamMarker(result)) { - // Stream response: resolve the pending request with an AsyncIterable - const iterable = this.createStreamIterable(id) - this.pendingRequests[id].resolve(iterable) - delete this.pendingRequests[id] - } else { - let finalResult = result - if (response.transferSlots && response.transferSlots.length > 0) { - const transferredValues = getTransferredValues(response) - finalResult = reconstructValueFromTransfer( - result, - response.transferSlots, - transferredValues - ) - } - this.pendingRequests[id].resolve(finalResult) - delete this.pendingRequests[id] - } - } - } - - /** - * Handles incoming method call requests from the remote endpoint - * @param request The request message to handle - * @private - */ - private async handleRequest(request: Message): Promise { - const { id, method } = request - let incomingArgs = Array.isArray(request.args) ? request.args : [] - - if (request.transferSlots && request.transferSlots.length > 0) { - const transferredValues = getTransferredValues(request) - incomingArgs = incomingArgs.map((arg: any) => - reconstructValueFromTransfer(arg, request.transferSlots!, transferredValues) - ) - } - - const methodPath = method.split(".") - if (!this.apiImplementation) { - // Broadcast transports deliver each request to every peer; peers with no API are not errors. - if (this.shouldIgnoreBroadcastRequestWithoutApi()) return - this.sendError(id, "No API implementation available") - return - } - let target: any = this.apiImplementation - - for (let i = 0; i < methodPath.length - 1; i++) { - target = target[methodPath[i]] - if (!target) { - if (this.shouldIgnoreBroadcastResolutionMiss()) return - this.sendError(id, `Method path ${method} not found at ${methodPath[i]}`) - return - } - } - - const finalMethod = methodPath[methodPath.length - 1] - const targetMethod = target[finalMethod] - - if (typeof targetMethod !== "function") { - // On broadcast transports, another peer may own this method and still respond. - if (this.shouldIgnoreBroadcastResolutionMiss()) return - this.sendError(id, `Method ${method} is not a function`) - return - } - - // Restore callback arguments: the caller serialized callback functions as - // "__callback__" placeholder strings. Here we convert them back into - // real functions that, when called, send a callback message to the remote side. - let processedArgs = incomingArgs.map((arg: any) => { - if (typeof arg === "string" && arg.startsWith("__callback__")) { - const callbackId = arg.slice(12) - return (...callbackArgs: any[]) => { - this.invokeCallback(callbackId, callbackArgs) - } - } - return arg - }) - - // --- Input / Output Validation --- - // Look up the validator for this method path (e.g. "math.divide"). - // If no validators were configured, or this method has none, this is undefined - // and validation is skipped entirely (backward-compatible no-op). - const methodValidator = lookupValidator(this.validators, method) - - if (methodValidator?.input) { - // Filter out callback arguments before validating. After the .map() above, - // callback placeholders have been replaced with real functions, so we filter - // by `typeof a !== "function"`. This matches the FilterCallbacks type - // utility that strips function params from the schema's expected input tuple. - const dataArgs = processedArgs.filter((a: any) => typeof a !== "function") - const inputResult = await runValidation(methodValidator.input, dataArgs) - // Use `=== false` (not `!`) for discriminated union narrowing — without a - // tsconfig, tsc doesn't narrow `!result.success` to the failure branch. - if (inputResult.success === false) { - this.sendError(id, new RPCValidationError("input", method, inputResult.issues)) - return - } - processedArgs = this.mergeValidatedArgs(processedArgs, inputResult.value) - } - - try { - // Wrap the handler invocation with the interceptor chain. - // Interceptors run after input validation, so they see clean args. - const invokeHandler = () => targetMethod.apply(target, processedArgs) - const result = - this.interceptors.length > 0 - ? await runInterceptors( - this.interceptors, - { id, method, args: processedArgs, meta: request.meta, state: {} }, - invokeHandler - ) - : await invokeHandler() - - // If the handler returned an AsyncIterable, stream it instead of - // sending a single response. We notify the consumer with a special - // "stream" flag on the initial response, then send chunks. - if (isAsyncIterable(result)) { - this.sendResponse(id, { __stream: true }) - this.streamResult(id, result, method, methodValidator) - return - } - - // Output validation: if an output schema is defined, validate the handler's - // return value before sending it back. This catches bugs where the handler - // returns an unexpected type (e.g. number instead of string). - if (methodValidator?.output) { - const outputResult = await runValidation(methodValidator.output, result) - if (outputResult.success === false) { - this.sendError(id, new RPCValidationError("output", method, outputResult.issues)) - return - } - // Send the validated value (which may have been transformed/coerced by the schema) - this.sendResponse(id, outputResult.value) - return - } - - this.sendResponse(id, result) - } catch (error: unknown) { - this.sendError(id, error instanceof Error ? error : new Error(String(error))) - } - } - - /** - * Invokes a callback on the remote endpoint - * @param callbackId The ID of the callback to invoke - * @param args Arguments to pass to the callback - * @private - */ - private invokeCallback(callbackId: string, args: any[]): void { - let finalArgs = args - const transferables: Transferable[] = [] - const transferSlots: TransferSlot[] = [] - const transferredValues: unknown[] = [] - - if (this.supportsTransfer) { - finalArgs = args.map((arg) => - processValueForTransfer(arg, transferables, transferSlots, transferredValues) - ) - } - - const message: Message = { - id: generateUUID(), - method: callbackId, - args: finalArgs, - type: "callback", - transferSlots: transferSlots.length > 0 ? transferSlots : undefined - } - this.sendMessage(message, transferables, transferredValues) - } - - /** - * Handles callback invocations received from the remote endpoint - * @param message The callback message to handle - * @private - */ - private handleCallback(message: Message): void { - const { method: callbackId } = message - const callback = this.callbacks[callbackId] - if (callback) { - let callbackArgs = Array.isArray(message.args) ? message.args : [] - if (message.transferSlots && message.transferSlots.length > 0) { - const transferredValues = getTransferredValues(message) - callbackArgs = callbackArgs.map((arg: any) => - reconstructValueFromTransfer(arg, message.transferSlots!, transferredValues) - ) - } - callback(...callbackArgs) - } else { - console.error(`Callback with id ${callbackId} not found`) - } - } - - /** - * Handles property get requests from the remote endpoint - * @param request The get request message to handle - * @private - */ - private handleGet(request: Message): void { - const { id, path } = request - if (!path) { - this.sendError(id, "Invalid get request: missing path") - return - } - if (!this.apiImplementation) { - if (this.shouldIgnoreBroadcastRequestWithoutApi()) return - this.sendError(id, "No API implementation available") - return - } - - try { - // Traverse the object path to get the property value - let target: any = this.apiImplementation - for (const prop of path) { - target = target[prop] - if (target === undefined) { - if (this.shouldIgnoreBroadcastResolutionMiss()) return - this.sendError(id, `Property path ${path.join(".")} not found at ${prop}`) - return - } - } - this.sendResponse(id, target) - } catch (error: unknown) { - this.sendError(id, error instanceof Error ? error : new Error(String(error))) - } - } - - /** - * Handles property set requests from the remote endpoint - * @param request The set request message to handle - * @private - */ - private handleSet(request: Message): void { - const { id, path } = request - if (!path) { - this.sendError(id, "Invalid set request: missing path") - return - } - if (!this.apiImplementation) { - if (this.shouldIgnoreBroadcastRequestWithoutApi()) return - this.sendError(id, "No API implementation available") - return - } - - let incomingValue = request.value - if (request.transferSlots && request.transferSlots.length > 0) { - const transferredValues = getTransferredValues(request) - incomingValue = reconstructValueFromTransfer( - request.value, - request.transferSlots, - transferredValues - ) - } - - try { - // Traverse to the parent object - let target: any = this.apiImplementation - for (let i = 0; i < path.length - 1; i++) { - target = target[path[i]] - if (!target) { - if (this.shouldIgnoreBroadcastResolutionMiss()) return - this.sendError(id, `Property path ${path.join(".")} not found at ${path[i]}`) - return - } - } - - // Set the final property - const finalProp = path[path.length - 1] - if (this.shouldIgnoreBroadcastResolutionMiss() && !(finalProp in Object(target))) return - target[finalProp] = incomingValue - this.sendResponse(id, true) // Return true to indicate success - } catch (error: unknown) { - this.sendError(id, error instanceof Error ? error : new Error(String(error))) - } - } - - /** - * Handles constructor calls from the remote endpoint - * @param request The construct request message to handle - * @private - */ - private handleConstruct(request: Message): void { - const { id, method } = request - let incomingArgs = Array.isArray(request.args) ? request.args : [] - - if (request.transferSlots && request.transferSlots.length > 0) { - const transferredValues = getTransferredValues(request) - incomingArgs = incomingArgs.map((arg: any) => - reconstructValueFromTransfer(arg, request.transferSlots!, transferredValues) - ) - } - - // Split the method path and traverse the API implementation - const methodPath = method.split(".") - if (!this.apiImplementation) { - if (this.shouldIgnoreBroadcastRequestWithoutApi()) return - this.sendError(id, "No API implementation available") - return - } - let target: any = this.apiImplementation - - // Traverse the object path - for (let i = 0; i < methodPath.length - 1; i++) { - target = target[methodPath[i]] - if (!target) { - if (this.shouldIgnoreBroadcastResolutionMiss()) return - this.sendError(id, `Constructor path ${method} not found at ${methodPath[i]}`) - return - } - } - - const finalMethod = methodPath[methodPath.length - 1] - const ConstructorClass = target[finalMethod] - - if (typeof ConstructorClass !== "function") { - if (this.shouldIgnoreBroadcastResolutionMiss()) return - this.sendError(id, `${method} is not a constructor function`) - return - } - - const processedArgs = incomingArgs.map((arg: any) => { - if (typeof arg === "string" && arg.startsWith("__callback__")) { - const callbackId = arg.slice(12) - return (...callbackArgs: any[]) => { - this.invokeCallback(callbackId, callbackArgs) - } - } - return arg - }) - - try { - const instance = new ConstructorClass(...processedArgs) - this.sendResponse(id, instance) - } catch (error: unknown) { - this.sendError(id, error instanceof Error ? error : new Error(String(error))) - } - } - - // --------------------------------------------------------------------------- - // Streaming — producer side - // --------------------------------------------------------------------------- - - /** - * Iterates an AsyncIterable and sends each yielded value as a stream-chunk. - * Sends stream-end when the iterator completes, or stream-error on failure. - * The loop stops if the consumer sends stream-cancel. - */ - private async streamResult( - requestId: string, - iterable: AsyncIterable, - method: string, - methodValidator?: ReturnType - ): Promise { - const abortController = new AbortController() - this.activeStreams.set(requestId, { abortController }) - - try { - for await (const value of iterable) { - // Check cancellation - if (abortController.signal.aborted) break - - let chunk = value - - // Per-chunk output validation - if (methodValidator?.output) { - const outputResult = await runValidation(methodValidator.output, chunk) - if (outputResult.success === false) { - this.sendStreamError( - requestId, - new RPCValidationError("output", method, outputResult.issues) - ) - return - } - chunk = outputResult.value - } - - const transferables: Transferable[] = [] - const transferSlots: TransferSlot[] = [] - const transferredValues: unknown[] = [] - - if (this.supportsTransfer) { - // Stream chunks use the same transfer envelope as normal responses. - chunk = processValueForTransfer(chunk, transferables, transferSlots, transferredValues) - } - - this.sendMessage( - { - id: requestId, - method: "", - args: { value: chunk }, - type: "stream-chunk", - transferSlots: transferSlots.length > 0 ? transferSlots : undefined - }, - transferables, - transferredValues - ) - } - - // Stream completed normally - if (!abortController.signal.aborted) { - this.sendMessage({ - id: requestId, - method: "", - args: {}, - type: "stream-end" - }) - } - } catch (error: unknown) { - if (!abortController.signal.aborted) { - this.sendStreamError(requestId, error instanceof Error ? error : new Error(String(error))) - } - } finally { - this.activeStreams.delete(requestId) - } - } - - /** Send a stream-error message to the consumer. */ - private sendStreamError(requestId: string, error: Error): void { - this.sendMessage({ - id: requestId, - method: "", - args: { error: serializeError(error) }, - type: "stream-error" - }) - } - - /** Handle a stream-cancel message from the consumer. */ - private handleStreamCancel(message: Message): void { - const state = this.activeStreams.get(message.id) - if (state) { - state.abortController.abort() - this.activeStreams.delete(message.id) - } - } - - // --------------------------------------------------------------------------- - // Streaming — consumer side - // --------------------------------------------------------------------------- - - /** - * Creates an AsyncIterable that yields values from incoming stream-chunk - * messages. The consumer calls `break` or `return()` to send stream-cancel. - */ - private createStreamIterable(requestId: string): AsyncIterable { - const consumer: StreamConsumerState = { - buffer: [], - done: false, - waiters: [] - } - this.streamConsumers.set(requestId, consumer) - - const channel = this - return { - [Symbol.asyncIterator]() { - return { - next(): Promise> { - // If there are buffered results, return immediately - if (consumer.buffer.length > 0) { - return Promise.resolve(consumer.buffer.shift()!) - } - // If stream is done and buffer is empty, return done - if (consumer.done) { - if (consumer.error) { - return Promise.reject(consumer.error) - } - return Promise.resolve({ done: true, value: undefined }) - } - // Wait for next chunk - return new Promise((resolve, reject) => { - consumer.waiters.push({ resolve, reject }) - }) - }, - return(): Promise> { - // Consumer is breaking out of for-await — send cancel - channel.sendMessage({ - id: requestId, - method: "", - args: {}, - type: "stream-cancel" - }) - consumer.done = true - channel.streamConsumers.delete(requestId) - return Promise.resolve({ done: true, value: undefined }) - } - } - } - } - } - - /** Handle a stream-chunk message from the producer. */ - private handleStreamChunk(message: Message): void { - const consumer = this.streamConsumers.get(message.id) - if (!consumer) return - let value = getMessageArgs(message).value - if (message.transferSlots && message.transferSlots.length > 0) { - // Rehydrate zero-copy stream chunks before yielding them to the consumer. - const transferredValues = getTransferredValues(message) - value = reconstructValueFromTransfer(value, message.transferSlots, transferredValues) - } - - const result: IteratorResult = { - done: false, - value - } - - if (consumer.waiters.length > 0) { - consumer.waiters.shift()!.resolve(result) - } else { - consumer.buffer.push(result) - } - } - - /** Handle a stream-end message from the producer. */ - private handleStreamEnd(message: Message): void { - const consumer = this.streamConsumers.get(message.id) - if (!consumer) return - - consumer.done = true - this.streamConsumers.delete(message.id) - - if (consumer.waiters.length > 0) { - // Resolve all waiters with done - for (const waiter of consumer.waiters) { - waiter.resolve({ done: true, value: undefined }) - } - consumer.waiters.length = 0 - } else { - consumer.buffer.push({ done: true, value: undefined }) - } - } - - /** Handle a stream-error message from the producer. */ - private handleStreamError(message: Message): void { - const consumer = this.streamConsumers.get(message.id) - if (!consumer) return - - const rawError = getMessageArgs(message).error - const error = - isEnhancedError(rawError) - ? deserializeError(rawError) - : new Error(typeof rawError === "string" ? rawError : "Stream error") - - consumer.done = true - consumer.error = error - this.streamConsumers.delete(message.id) - - if (consumer.waiters.length > 0) { - for (const waiter of consumer.waiters) { - waiter.reject(error) - } - consumer.waiters.length = 0 - } - } - - /** - * Sends a successful response back to the remote endpoint - * @param id The ID of the request being responded to - * @param result The result to send back - * @private - */ - private sendResponse(id: string, result: T): void { - let responseResult = result - const transferables: Transferable[] = [] - const transferSlots: TransferSlot[] = [] - const transferredValues: unknown[] = [] - - if (this.supportsTransfer) { - responseResult = processValueForTransfer( - result, - transferables, - transferSlots, - transferredValues - ) - } - - const response: Message> = { - id, - method: "", - args: { result: responseResult }, - type: "response", - transferSlots: transferSlots.length > 0 ? transferSlots : undefined - } - - this.sendMessage(response, transferables, transferredValues) - } - - /** - * Sends an error response back to the remote endpoint - * @param id The ID of the request being responded to - * @param error The error message or Error object to send back - * @private - */ - private sendError(id: string, error: string | Error): void { - const errorResponse = error instanceof Error ? serializeError(error) : error - - const response: Message> = { - id, - method: "", - args: { error: errorResponse }, - type: "response" - } - this.sendMessage(response) - } - - private sendMessage( - message: Message, - transferables: Transferable[] = [], - transferredValues: unknown[] = [] - ): void { - // Centralized send path so write failures can reject the matching pending request. - const encoded = encodeMessage( - message, - this.serializationOptions, - this.supportsTransfer && transferables.length > 0, - transferredValues - ) - - const outgoing: string | IoMessage = - encoded.mode === "string" - ? encoded.data - : { - data: encoded.data, - transfers: transferables - } - - try { - void this.io.write(outgoing).catch((error: unknown) => { - this.handleWriteFailure(message, error) - }) - } catch (error: unknown) { - this.handleWriteFailure(message, error) - } - } - - /** Rejects the RPC call that initiated this write, instead of leaving it to timeout. */ - private handleWriteFailure(message: Message, error: unknown): void { - const writeError = error instanceof Error ? error : new Error(String(error)) - const pending = this.pendingRequests[message.id] - if (pending) { - this.clearTimeout(message.id) - pending.reject(writeError) - delete this.pendingRequests[message.id] - return - } - - if (!this.isClosed) { - console.error("kkrpc: failed to write message", writeError) - } - } - - /** - * Validation schemas may coerce data arguments, but callback function arguments are not validated. - * This merges transformed data args back into the original argument list without moving callbacks. - */ - private mergeValidatedArgs(originalArgs: any[], validatedValue: unknown): any[] { - const validatedArgs = Array.isArray(validatedValue) ? validatedValue : [validatedValue] - let dataArgIndex = 0 - return originalArgs.map((arg) => { - if (typeof arg === "function") return arg - if (dataArgIndex >= validatedArgs.length) return arg - const validatedArg = validatedArgs[dataArgIndex] - dataArgIndex++ - return validatedArg - }) - } - - /** True when this peer should silently ignore broadcast traffic because it exposes no API. */ - private shouldIgnoreBroadcastRequestWithoutApi(): boolean { - return !this.apiImplementation && this.io.capabilities?.broadcast === true - } - - /** True when another broadcast peer may be the intended owner of the missing path. */ - private shouldIgnoreBroadcastResolutionMiss(): boolean { - return this.io.capabilities?.broadcast === true - } - - /** Rejects all pending RPC calls during transport close or explicit destroy. */ - private rejectPendingRequests(error: Error): void { - for (const [id, pending] of Object.entries(this.pendingRequests)) { - pending.reject(error) - this.clearTimeout(id) - } - this.pendingRequests = {} - this.pendingTimers = {} - } - - /** Stops producer-side async iterators so they do not keep yielding after close. */ - private abortActiveStreams(): void { - for (const [, state] of this.activeStreams) { - state.abortController.abort() - } - this.activeStreams.clear() - } - - /** Rejects consumer-side async iterator waiters that are blocked on the next chunk. */ - private rejectStreamConsumers(error: Error): void { - for (const [, consumer] of this.streamConsumers) { - consumer.done = true - consumer.error = error - for (const waiter of consumer.waiters) { - waiter.reject(error) - } - consumer.waiters.length = 0 - } - this.streamConsumers.clear() - } - - /** Common close path used by EOF, adapter destroy, write-side destroy, and RPCChannel.destroy(). */ - private closeFromTransport(error: Error): void { - if (this.isClosed) return - this.isClosed = true - this.rejectPendingRequests(error) - this.abortActiveStreams() - this.rejectStreamConsumers(error) - this.freeCallbacks() - } - - // --------------------------------------------------------------------------- - // Timeout helpers - // --------------------------------------------------------------------------- - - /** Start a timeout timer for a pending request. No-op if timeout is 0 or Infinity. */ - private startTimeout(messageId: string, method: string): void { - if (!this.timeout || this.timeout === Infinity) return - this.pendingTimers[messageId] = setTimeout(() => { - const pending = this.pendingRequests[messageId] - if (pending) { - pending.reject(new RPCTimeoutError(method, this.timeout)) - delete this.pendingRequests[messageId] - delete this.pendingTimers[messageId] - } - }, this.timeout) - } - - /** Clear the timeout timer for a request that received a response. */ - private clearTimeout(messageId: string): void { - if (this.pendingTimers[messageId]) { - clearTimeout(this.pendingTimers[messageId]) - delete this.pendingTimers[messageId] - } - } - - /** - * Creates a nested proxy object for chaining remote method calls, property access, and constructor calls - * @param chain Array of method names in the chain - * @returns Proxy object that transforms property access into remote method calls - * @private - */ - private createNestedProxy(chain: string[] = []): any { - return new Proxy(() => {}, { - get: (_target, prop: string | symbol) => { - // Handle special properties - if (typeof prop === "string") { - // Handle property access like obj.prop - if (prop !== "then") { - return this.createNestedProxy([...chain, prop]) - } - // Handle thenable for await support - if (prop === "then" && chain.length > 0) { - // Return property value when accessed like: await obj.prop - const promise = this.getProperty(chain) - return promise.then.bind(promise) - } - } - return undefined - }, - set: (_target, prop: string | symbol, value: any) => { - // Handle property setting like obj.prop = value - if (typeof prop === "string") { - const propertyPath = [...chain, prop] - this.setProperty(propertyPath, value) - return true - } - return false - }, - apply: (_target, _thisArg, args: any[]) => { - // Handle method calls like obj.method() - const method = chain.join(".") - return this.callMethod(method as keyof RemoteAPI, args) - }, - construct: (_target, args: any[]) => { - // Handle constructor calls like new obj.Constructor() - return this.callConstructor(chain.join(".") as keyof RemoteAPI, args) - } - }) - } - - /** - * Returns a proxy object that represents the remote API - * Methods called on this proxy will be executed on the remote endpoint - * @returns Proxy object representing the remote API - */ - public getAPI(): RemoteAPI { - return this.createNestedProxy() as RemoteAPI - } - - /** - * Destroys the RPC channel and underlying IO interface if it's destroyable - */ - destroy(): void { - this.closeFromTransport(new Error("RPC channel destroyed")) - - // Clean up IO adapters - if (this.io && this.io.destroy) { - this.io.destroy() - } - } - - /** - * Frees up memory by clearing stored callbacks and callback cache - * Useful when dealing with many anonymous callback functions to prevent memory leaks - */ - freeCallbacks() { - this.callbacks = {} - this.callbackCache.clear() +> extends RPCChannelCore { + constructor(io: Io, options?: RPCChannelOptions) { + super(io, options, fullSerializationRuntime) } } diff --git a/packages/kkrpc/src/serialization-full.ts b/packages/kkrpc/src/serialization-full.ts new file mode 100644 index 0000000..8a5cd25 --- /dev/null +++ b/packages/kkrpc/src/serialization-full.ts @@ -0,0 +1,97 @@ +/** + * Full kkrpc serialization runtime with SuperJSON support. + * Existing package entrypoints import this module to preserve the current + * default behavior for Date, Map, Set, BigInt, Uint8Array, and legacy messages. + */ +import superjson from "superjson" +import { + decodeMessage as decodeJsonWireMessage, + decodeStructuredMessage, + deserializeError, + encodeJsonMessage, + encodeStructuredMessage, + processValueForTransfer, + reconstructValueFromTransfer, + serializeError +} from "./serialization-json.ts" +import type { + EncodedMessage, + Message, + RPCSerializationRuntime, + SerializationOptions, + WireFormat +} from "./serialization-types.ts" + +export type { + EncodedMessage, + EnhancedError, + Message, + RPCMessageMetadata, + RPCSerializationRuntime, + Response, + SerializationOptions, + TransferSlot, + WireEnvelope, + WireFormat, + WireV1 +} from "./serialization-types.ts" + +export { deserializeError, processValueForTransfer, reconstructValueFromTransfer, serializeError } + +export function serializeMessage( + message: Message, + options: SerializationOptions = {} +): string { + const version = options.version || "superjson" + const msgWithVersion = { ...message, version } + return version === "json" ? encodeJsonMessage(message) : superjson.stringify(msgWithVersion) + "\n" +} + +export function deserializeMessage(message: string): Promise> { + return new Promise((resolve, reject) => { + try { + if (message.trimStart().startsWith('{"json":')) { + resolve(superjson.parse>(message)) + return + } + decodeJsonWireMessage(message).then(resolve, reject) + } catch (error) { + console.error("failed to parse message", typeof message, message, error) + reject(error) + } + }) +} + +export function encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues: unknown[] = [] +): EncodedMessage { + if (!withTransfers) { + return { + mode: "string", + data: serializeMessage(message, options) + } + } + + return { + mode: "structured", + data: encodeStructuredMessage(message, transferredValues) + } +} + +export async function decodeMessage(raw: WireFormat): Promise> { + if (typeof raw === "string") { + return deserializeMessage(raw) + } + + return decodeStructuredMessage(raw) +} + +export const fullSerializationRuntime: RPCSerializationRuntime = { + encodeMessage, + decodeMessage, + serializeError, + deserializeError +} diff --git a/packages/kkrpc/src/serialization-json.ts b/packages/kkrpc/src/serialization-json.ts new file mode 100644 index 0000000..6ff42a8 --- /dev/null +++ b/packages/kkrpc/src/serialization-json.ts @@ -0,0 +1,384 @@ +/** + * JSON-only kkrpc serialization runtime. + * Browser-lite imports this module to keep SuperJSON out of the dependency graph + * while preserving JSON string messages, structured envelopes, transfer slots, + * and rich error serialization. + */ +import { transferHandlers } from "./transfer-handlers.ts" +import { takeTransferDescriptor } from "./transfer.ts" +import { + TRANSFER_SLOT_PREFIX, + type EncodedMessage, + type EnhancedError, + type Message, + type RPCSerializationRuntime, + type SerializationOptions, + type TransferSlot, + type WireEnvelope, + type WireFormat +} from "./serialization-types.ts" + +export { + TRANSFER_SLOT_PREFIX, + type EncodedMessage, + type EnhancedError, + type Message, + type RPCMessageMetadata, + type RPCSerializationRuntime, + type Response, + type SerializationOptions, + type TransferSlot, + type WireEnvelope, + type WireFormat, + type WireV1 +} from "./serialization-types.ts" + +// Transfer placeholders are tagged objects instead of strings so user strings never collide. +const TRANSFER_SLOT_PLACEHOLDER_KEY = "__kkrpc_transfer_slot__" +const TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY = "__kkrpc_transfer_token__" + +interface TransferPlaceholder { + slotIndex: number + token: string +} + +type ErrorWithProperties = Error & Record + +/** Creates a message-local token to distinguish real placeholders from user objects. */ +function createTransferToken(): string { + return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}` +} + +/** Replaces a transferred value in the JSON payload with a small tagged reference. */ +function createTransferPlaceholder(slotIndex: number, token: string): Record { + return { + [TRANSFER_SLOT_PLACEHOLDER_KEY]: slotIndex, + [TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY]: token + } +} + +/** Only traverse plain objects; Dates, Maps, class instances, etc. should keep identity. */ +function isPlainObject(value: unknown): value is Record { + if (value === null || typeof value !== "object") return false + const prototype = Object.getPrototypeOf(value) + return prototype === Object.prototype || prototype === null +} + +/** Returns a transfer placeholder only when the object exactly matches our tagged shape. */ +function getTransferPlaceholder(value: unknown): TransferPlaceholder | undefined { + if (!isPlainObject(value)) return undefined + const keys = Object.keys(value) + if ( + keys.length !== 2 || + !keys.includes(TRANSFER_SLOT_PLACEHOLDER_KEY) || + !keys.includes(TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY) + ) { + return undefined + } + const slotIndex = value[TRANSFER_SLOT_PLACEHOLDER_KEY] + const token = value[TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY] + return typeof slotIndex === "number" && Number.isInteger(slotIndex) && typeof token === "string" + ? { slotIndex, token } + : undefined +} + +function replacer(_key: string, value: unknown): unknown { + if (value instanceof Uint8Array) { + return { + type: "Uint8Array", + data: Array.from(value) + } + } + return value +} + +function reviver(_key: string, value: unknown): unknown { + if (isPlainObject(value) && value.type === "Uint8Array" && Array.isArray(value.data)) { + return new Uint8Array(value.data as number[]) + } + return value +} + +export function serializeError(error: Error): EnhancedError { + const errorWithProperties = error as ErrorWithProperties + const enhanced: EnhancedError = { + name: error.name, + message: error.message + } + + if (error.stack) { + enhanced.stack = error.stack + } + + if ("cause" in errorWithProperties && errorWithProperties.cause !== undefined) { + enhanced.cause = errorWithProperties.cause + } + + for (const key in errorWithProperties) { + if (key !== "name" && key !== "message" && key !== "stack" && key !== "cause") { + enhanced[key] = errorWithProperties[key] + } + } + + return enhanced +} + +export function deserializeError(enhanced: EnhancedError): Error { + const error = new Error(enhanced.message) + const errorWithProperties = error as ErrorWithProperties + error.name = enhanced.name + + if (enhanced.stack) { + error.stack = enhanced.stack + } + + if (enhanced.cause !== undefined) { + errorWithProperties.cause = enhanced.cause + } + + for (const key in enhanced) { + if (key !== "name" && key !== "message" && key !== "stack" && key !== "cause") { + errorWithProperties[key] = enhanced[key] + } + } + + return error +} + +export function encodeJsonMessage(message: Message): string { + return JSON.stringify({ ...message, version: "json" }, replacer) + "\n" +} + +export function decodeJsonMessage(message: string): Promise> { + return new Promise((resolve, reject) => { + try { + if (message.trimStart().startsWith('{"json":')) { + reject( + new Error( + 'Received a SuperJSON-encoded kkrpc message, but this entrypoint is JSON-only. Use kkrpc/browser or configure both endpoints with serialization.version = "json".' + ) + ) + return + } + resolve(JSON.parse(message, reviver) as Message) + } catch (error) { + console.error("failed to parse message", typeof message, message, error) + reject(error) + } + }) +} + +export function encodeStructuredMessage( + message: Message, + transferredValues: unknown[] = [] +): WireEnvelope { + const envelope: WireEnvelope = { + version: 2, + payload: message as Message, + encoding: "object" + } + + if (message.transferSlots && message.transferSlots.length > 0) { + envelope.transferSlots = message.transferSlots + } + + if (transferredValues.length > 0) { + envelope.__transferredValues = transferredValues + } + + return envelope +} + +export function decodeStructuredMessage(raw: WireEnvelope): Promise> { + const payload = raw.payload as Message + + if (raw.transferSlots?.length) { + payload.transferSlots = raw.transferSlots + } + + if (Array.isArray(raw.__transferredValues)) { + Object.defineProperty(payload, "__transferredValues", { + value: raw.__transferredValues, + enumerable: false, + configurable: true + }) + } + + return Promise.resolve(payload) +} + +export function encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues: unknown[] = [] +): EncodedMessage { + if (options.version === "superjson") { + throw new Error( + 'SuperJSON serialization is not available in kkrpc/browser-lite. Use kkrpc/browser or configure both endpoints with serialization.version = "json".' + ) + } + + if (!withTransfers) { + return { + mode: "string", + data: encodeJsonMessage(message) + } + } + + return { + mode: "structured", + data: encodeStructuredMessage(message, transferredValues) + } +} + +export async function decodeMessage(raw: WireFormat): Promise> { + if (typeof raw === "string") { + return decodeJsonMessage(raw) + } + + return decodeStructuredMessage(raw) +} + +export function processValueForTransfer( + value: T, + transferables?: Transferable[], + transferSlots?: TransferSlot[], + transferredValues?: unknown[], + slotMap?: Map +): T +export function processValueForTransfer( + value: unknown, + transferables: Transferable[] = [], + transferSlots: TransferSlot[] = [], + transferredValues: unknown[] = [], + slotMap: Map = new Map() +): unknown { + if (value === null || typeof value !== "object") { + return value + } + + if (slotMap.has(value)) { + const slotIndex = slotMap.get(value)! + // Preserve shared references by pointing repeat occurrences at the first transfer slot. + return createTransferPlaceholder(slotIndex, transferSlots[slotIndex]?.token ?? "") + } + + const descriptor = takeTransferDescriptor(value) + if (descriptor) { + // Explicit transfer(value, transfers) gets the raw transferred value on structured transports. + const slotIndex = transferSlots.length + const token = createTransferToken() + slotMap.set(value, slotIndex) + transferables.push(...descriptor.transfers) + transferredValues.push(descriptor.value) + transferSlots.push({ + type: "raw", + metadata: { original: true }, + token + }) + return createTransferPlaceholder(slotIndex, token) + } + + for (const [name, handler] of transferHandlers) { + if (handler.canHandle(value)) { + // Registered handlers can serialize custom transferable-like objects into metadata. + const [serialized, handlerTransferables] = handler.serialize(value) + const slotIndex = transferSlots.length + const token = createTransferToken() + slotMap.set(value, slotIndex) + transferables.push(...handlerTransferables) + transferredValues.push(undefined) + transferSlots.push({ + type: "handler", + handlerName: name, + metadata: serialized, + token + }) + return createTransferPlaceholder(slotIndex, token) + } + } + + if (Array.isArray(value)) { + return value.map((item) => + processValueForTransfer(item, transferables, transferSlots, transferredValues, slotMap) + ) + } + + if (isPlainObject(value)) { + // Recurse only through plain payload containers to avoid corrupting runtime objects. + const processed: Record = {} + for (const [key, val] of Object.entries(value)) { + processed[key] = processValueForTransfer( + val, + transferables, + transferSlots, + transferredValues, + slotMap + ) + } + return processed + } + + return value +} + +export function reconstructValueFromTransfer( + value: T, + transferSlots: TransferSlot[], + transferredValues: unknown[] +): T +export function reconstructValueFromTransfer( + value: unknown, + transferSlots: TransferSlot[], + transferredValues: unknown[] +): unknown { + const placeholder = getTransferPlaceholder(value) + if (placeholder !== undefined) { + const { slotIndex } = placeholder + const slot = transferSlots[slotIndex] + if (slot?.token !== placeholder.token) { + // Not a placeholder from this message; preserve user data with the same shape. + } else { + const transferredValue = transferredValues[slotIndex] + + if (slot?.type === "raw") { + if (transferredValue === undefined) { + throw new Error(`Missing transferred value for slot ${slotIndex}`) + } + return transferredValue + } + + if (slot?.type === "handler" && slot.handlerName) { + const handler = transferHandlers.get(slot.handlerName) + if (!handler) { + throw new Error(`Unknown transfer handler: ${slot.handlerName}`) + } + return handler.deserialize(slot.metadata) + } + + throw new Error(`Invalid transfer slot: ${slotIndex}`) + } + } + + if (Array.isArray(value)) { + return value.map((item) => reconstructValueFromTransfer(item, transferSlots, transferredValues)) + } + + if (isPlainObject(value)) { + const reconstructed: Record = {} + for (const [key, val] of Object.entries(value)) { + reconstructed[key] = reconstructValueFromTransfer(val, transferSlots, transferredValues) + } + return reconstructed + } + + return value +} + +export const jsonSerializationRuntime: RPCSerializationRuntime = { + encodeMessage, + decodeMessage, + serializeError, + deserializeError +} diff --git a/packages/kkrpc/src/serialization-types.ts b/packages/kkrpc/src/serialization-types.ts new file mode 100644 index 0000000..dae1468 --- /dev/null +++ b/packages/kkrpc/src/serialization-types.ts @@ -0,0 +1,95 @@ +/** + * Shared wire protocol types for kkrpc serializers and channels. + * This file has no runtime dependencies so lite entrypoints can reuse the + * protocol shape without importing SuperJSON or full serialization helpers. + */ + +export interface Message { + id: string + method: string + args: T + type: + | "request" + | "response" + | "callback" + | "get" + | "set" + | "construct" + | "stream-chunk" + | "stream-end" + | "stream-error" + | "stream-cancel" + callbackIds?: string[] + version?: "json" | "superjson" + meta?: RPCMessageMetadata + path?: string[] + value?: unknown + transferSlots?: TransferSlot[] +} + +/** + * Optional out-of-band metadata carried with an RPC message. + */ +export interface RPCMessageMetadata { + traceparent?: string + tracestate?: string + baggage?: string + requestId?: string + sessionId?: string + runtime?: Record + [key: string]: unknown +} + +export interface Response { + result?: T + error?: string | EnhancedError +} + +export interface EnhancedError { + name: string + message: string + stack?: string + cause?: unknown + [key: string]: unknown +} + +export interface SerializationOptions { + version?: "json" | "superjson" +} + +export const TRANSFER_SLOT_PREFIX = "__kkrpc_transfer_" + +export interface TransferSlot { + type: "raw" | "handler" + handlerName?: string + metadata?: unknown + /** Random per-slot token that proves a placeholder was generated for this message. */ + token?: string +} + +export interface WireEnvelope { + version: 2 + payload: Message + transferSlots?: TransferSlot[] + encoding: "object" + __transferredValues?: unknown[] +} + +export type WireV1 = string +export type WireFormat = WireV1 | WireEnvelope + +export type EncodedMessage = + | { mode: "string"; data: string } + | { mode: "structured"; data: WireEnvelope } + +export interface RPCSerializationRuntime { + encodeMessage( + message: Message, + options: SerializationOptions, + withTransfers: boolean, + transferredValues?: unknown[] + ): EncodedMessage + decodeMessage(raw: WireFormat): Promise> + serializeError(error: Error): EnhancedError + deserializeError(error: EnhancedError): Error +} diff --git a/packages/kkrpc/src/serialization.ts b/packages/kkrpc/src/serialization.ts index caa3763..1e9c534 100644 --- a/packages/kkrpc/src/serialization.ts +++ b/packages/kkrpc/src/serialization.ts @@ -1,418 +1,9 @@ -import superjson from "superjson" -import { transferHandlers } from "./transfer-handlers.ts" -import { takeTransferDescriptor } from "./transfer.ts" - -/** - * This file contains the serialization and deserialization helpers for the RPC protocol. - */ -export interface Message { - id: string - method: string - args: T - type: - | "request" - | "response" - | "callback" - | "get" - | "set" - | "construct" - | "stream-chunk" - | "stream-end" - | "stream-error" - | "stream-cancel" - callbackIds?: string[] - version?: "json" | "superjson" - meta?: RPCMessageMetadata - path?: string[] - value?: unknown - transferSlots?: TransferSlot[] -} - /** - * Optional out-of-band metadata carried with an RPC message. - * - * This is intentionally protocol-level metadata, not another user argument: - * callers can propagate tracing/debug context without changing exposed API - * signatures or leaking implementation details into plugin contracts. + * Compatibility exports for kkrpc's full serialization runtime. + * Existing users import protocol types and helpers from this path, so it remains + * SuperJSON-enabled. Browser-lite must import `serialization-json.ts` or + * `serialization-types.ts` directly instead of this barrel. */ -export interface RPCMessageMetadata { - traceparent?: string - tracestate?: string - baggage?: string - requestId?: string - sessionId?: string - runtime?: Record - [key: string]: unknown -} - -export interface Response { - result?: T - error?: string | EnhancedError -} - -export interface EnhancedError { - name: string - message: string - stack?: string - cause?: unknown - [key: string]: unknown -} - -export interface SerializationOptions { - version?: "json" | "superjson" -} - -export const TRANSFER_SLOT_PREFIX = "__kkrpc_transfer_" -// Transfer placeholders are tagged objects instead of strings so user strings never collide. -const TRANSFER_SLOT_PLACEHOLDER_KEY = "__kkrpc_transfer_slot__" -const TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY = "__kkrpc_transfer_token__" - -export interface TransferSlot { - type: "raw" | "handler" - handlerName?: string - metadata?: unknown - /** Random per-slot token that proves a placeholder was generated for this message. */ - token?: string -} - -export interface WireEnvelope { - version: 2 - payload: Message - transferSlots?: TransferSlot[] - encoding: "object" - __transferredValues?: unknown[] -} - -export type WireV1 = string -export type WireFormat = WireV1 | WireEnvelope - -export type EncodedMessage = - | { mode: "string"; data: string } - | { mode: "structured"; data: WireEnvelope } - -interface TransferPlaceholder { - slotIndex: number - token: string -} - -/** Creates a message-local token to distinguish real placeholders from user objects. */ -function createTransferToken(): string { - return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}` -} - -/** Replaces a transferred value in the JSON payload with a small tagged reference. */ -function createTransferPlaceholder(slotIndex: number, token: string): Record { - return { - [TRANSFER_SLOT_PLACEHOLDER_KEY]: slotIndex, - [TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY]: token - } -} - -/** Only traverse plain objects; Dates, Maps, class instances, etc. should keep identity. */ -function isPlainObject(value: unknown): value is Record { - if (value === null || typeof value !== "object") return false - const prototype = Object.getPrototypeOf(value) - return prototype === Object.prototype || prototype === null -} - -/** Returns a transfer placeholder only when the object exactly matches our tagged shape. */ -function getTransferPlaceholder(value: unknown): TransferPlaceholder | undefined { - if (!isPlainObject(value)) return undefined - const keys = Object.keys(value) - if ( - keys.length !== 2 || - !keys.includes(TRANSFER_SLOT_PLACEHOLDER_KEY) || - !keys.includes(TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY) - ) { - return undefined - } - const slotIndex = value[TRANSFER_SLOT_PLACEHOLDER_KEY] - const token = value[TRANSFER_SLOT_PLACEHOLDER_TOKEN_KEY] - return typeof slotIndex === "number" && Number.isInteger(slotIndex) && typeof token === "string" - ? { slotIndex, token } - : undefined -} - -type ErrorWithProperties = Error & Record - -function replacer(_key: string, value: unknown): unknown { - if (value instanceof Uint8Array) { - return { - type: "Uint8Array", - data: Array.from(value) - } - } - return value -} - -function reviver(_key: string, value: unknown): unknown { - if (isPlainObject(value) && value.type === "Uint8Array" && Array.isArray(value.data)) { - return new Uint8Array(value.data as number[]) - } - return value -} - -export function serializeError(error: Error): EnhancedError { - const errorWithProperties = error as ErrorWithProperties - const enhanced: EnhancedError = { - name: error.name, - message: error.message - } - - if (error.stack) { - enhanced.stack = error.stack - } - - if ("cause" in errorWithProperties && errorWithProperties.cause !== undefined) { - enhanced.cause = errorWithProperties.cause - } - - for (const key in errorWithProperties) { - if (key !== "name" && key !== "message" && key !== "stack" && key !== "cause") { - enhanced[key] = errorWithProperties[key] - } - } - - return enhanced -} - -export function deserializeError(enhanced: EnhancedError): Error { - const error = new Error(enhanced.message) - const errorWithProperties = error as ErrorWithProperties - error.name = enhanced.name - - if (enhanced.stack) { - error.stack = enhanced.stack - } - - if (enhanced.cause !== undefined) { - errorWithProperties.cause = enhanced.cause - } - - for (const key in enhanced) { - if (key !== "name" && key !== "message" && key !== "stack" && key !== "cause") { - errorWithProperties[key] = enhanced[key] - } - } - - return error -} - -export function serializeMessage( - message: Message, - options: SerializationOptions = {} -): string { - const version = options.version || "superjson" - const msgWithVersion = { ...message, version } - return version === "json" - ? JSON.stringify(msgWithVersion, replacer) + "\n" - : superjson.stringify(msgWithVersion) + "\n" -} - -export function deserializeMessage(message: string): Promise> { - return new Promise((resolve, reject) => { - try { - if (message.startsWith('{"json":')) { - const parsed = superjson.parse>(message) - resolve(parsed) - } else { - const parsed = JSON.parse(message, reviver) as Message - resolve(parsed) - } - } catch (error) { - console.error("failed to parse message", typeof message, message, error) - reject(error) - } - }) -} - -export function encodeMessage( - message: Message, - options: SerializationOptions, - withTransfers: boolean, - transferredValues: unknown[] = [] -): EncodedMessage { - if (!withTransfers) { - return { - mode: "string", - data: serializeMessage(message, options) - } - } - - const envelope: WireEnvelope = { - version: 2, - payload: message, - encoding: "object" - } - - if (message.transferSlots && message.transferSlots.length > 0) { - envelope.transferSlots = message.transferSlots - } - - if (transferredValues.length > 0) { - envelope.__transferredValues = transferredValues - } - - return { - mode: "structured", - data: envelope - } -} - -export async function decodeMessage(raw: WireFormat): Promise> { - if (typeof raw === "string") { - return deserializeMessage(raw) - } - - const payload = raw.payload as Message - - if (raw.transferSlots?.length) { - payload.transferSlots = raw.transferSlots - } - - if (Array.isArray(raw.__transferredValues)) { - Object.defineProperty(payload, "__transferredValues", { - value: raw.__transferredValues, - enumerable: false, - configurable: true - }) - } - - return payload -} - -export function processValueForTransfer( - value: T, - transferables?: Transferable[], - transferSlots?: TransferSlot[], - transferredValues?: unknown[], - slotMap?: Map -): T -export function processValueForTransfer( - value: unknown, - transferables: Transferable[] = [], - transferSlots: TransferSlot[] = [], - transferredValues: unknown[] = [], - slotMap: Map = new Map() -): unknown { - if (value === null || typeof value !== "object") { - return value - } - - if (slotMap.has(value)) { - const slotIndex = slotMap.get(value)! - // Preserve shared references by pointing repeat occurrences at the first transfer slot. - return createTransferPlaceholder(slotIndex, transferSlots[slotIndex]?.token ?? "") - } - - const descriptor = takeTransferDescriptor(value) - if (descriptor) { - // Explicit transfer(value, transfers) gets the raw transferred value on structured transports. - const slotIndex = transferSlots.length - const token = createTransferToken() - slotMap.set(value, slotIndex) - transferables.push(...descriptor.transfers) - transferredValues.push(descriptor.value) - transferSlots.push({ - type: "raw", - metadata: { original: true }, - token - }) - return createTransferPlaceholder(slotIndex, token) - } - - for (const [name, handler] of transferHandlers) { - if (handler.canHandle(value)) { - // Registered handlers can serialize custom transferable-like objects into metadata. - const [serialized, handlerTransferables] = handler.serialize(value) - const slotIndex = transferSlots.length - const token = createTransferToken() - slotMap.set(value, slotIndex) - transferables.push(...handlerTransferables) - transferredValues.push(undefined) - transferSlots.push({ - type: "handler", - handlerName: name, - metadata: serialized, - token - }) - return createTransferPlaceholder(slotIndex, token) - } - } - - if (Array.isArray(value)) { - return value.map((item) => - processValueForTransfer(item, transferables, transferSlots, transferredValues, slotMap) - ) - } - - if (isPlainObject(value)) { - // Recurse only through plain payload containers to avoid corrupting runtime objects. - const processed: Record = {} - for (const [key, val] of Object.entries(value)) { - processed[key] = processValueForTransfer( - val, - transferables, - transferSlots, - transferredValues, - slotMap - ) - } - return processed - } - - return value -} - -export function reconstructValueFromTransfer( - value: T, - transferSlots: TransferSlot[], - transferredValues: unknown[] -): T -export function reconstructValueFromTransfer( - value: unknown, - transferSlots: TransferSlot[], - transferredValues: unknown[] -): unknown { - const placeholder = getTransferPlaceholder(value) - if (placeholder !== undefined) { - const { slotIndex } = placeholder - const slot = transferSlots[slotIndex] - if (slot?.token !== placeholder.token) { - // Not a placeholder from this message; preserve user data with the same shape. - } else { - const transferredValue = transferredValues[slotIndex] - - if (slot?.type === "raw") { - if (transferredValue === undefined) { - throw new Error(`Missing transferred value for slot ${slotIndex}`) - } - return transferredValue - } - - if (slot?.type === "handler" && slot.handlerName) { - const handler = transferHandlers.get(slot.handlerName) - if (!handler) { - throw new Error(`Unknown transfer handler: ${slot.handlerName}`) - } - return handler.deserialize(slot.metadata) - } - - throw new Error(`Invalid transfer slot: ${slotIndex}`) - } - } - - if (Array.isArray(value)) { - return value.map((item) => reconstructValueFromTransfer(item, transferSlots, transferredValues)) - } - - if (isPlainObject(value)) { - const reconstructed: Record = {} - for (const [key, val] of Object.entries(value)) { - reconstructed[key] = reconstructValueFromTransfer(val, transferSlots, transferredValues) - } - return reconstructed - } - return value -} +export * from "./serialization-types.ts" +export * from "./serialization-full.ts" diff --git a/packages/kkrpc/tsdown.config.ts b/packages/kkrpc/tsdown.config.ts index 19fda00..4e3f9cd 100644 --- a/packages/kkrpc/tsdown.config.ts +++ b/packages/kkrpc/tsdown.config.ts @@ -4,6 +4,7 @@ export default defineConfig({ entry: [ "./mod.ts", "./browser-mod.ts", + "./browser-lite-mod.ts", "./http.ts", "./deno-mod.ts", "./chrome-extension.ts",