feat(ts-sdk): add schemas

docs/WASM-CLIENT.md

@@ -104,6 +104,9 @@ if (!MTPClient.isSupported()) {
 | `requestTimeoutMs` | 30 seconds | Default `request()` timeout. |
 | `pings` | `false` | Protocol pings, or an object with `intervalMs`. |
 | `logger` | No-op | Receives SDK state and error events. |
+| `schemas` | None | Client-wide request and response schema registry. |
+| `throwProtocolErrors` | `false` | Reject requests whose correlated response is an `Error*` frame. |
+| `onValidationError` | No-op | Receives subscription validation failures. |
 | `sessionStorage` | In-memory | E2EE session state storage. |
 | `encryptedSecretProvider` | In-memory | Independent caller-managed encrypted secret storage. |
 | `defaultSignatureVerificationPolicy` | `"ed25519"` | Receiver policy for protected signatures. |
@@ -471,6 +474,60 @@ const unsubscribe = client.subscribe("SomeType", (message) => {
 unsubscribe();
 ```
 
+### Zod request and response schemas
+
+Applications can provide their request and response schemas once when creating
+the client. MTP uses `parseAsync`, so synchronous schemas, async refinements,
+defaults, coercions, and transforms all work. MTP has no runtime dependency on
+Zod; the application supplies its preferred Zod version.
+
+```typescript
+import { z } from "zod";
+import { MTPClient, MTPValidationError } from "mtp";
+
+const schemas = {
+  GetUser: {
+    request: z.object({ UserId: z.number().int().positive() }),
+    response: z.object({
+      UserId: z.number().int().positive(),
+      Display: z.string(),
+    }),
+  },
+};
+
+const client = await MTPClient.create({
+  url,
+  schemas,
+  throwProtocolErrors: true,
+  onValidationError(error) {
+    console.error(error.messageType, error.cause);
+  },
+});
+
+const response = await client.request("GetUser", { UserId: 42 });
+console.log(response.data.Display);
+```
+
+Request schemas run before frame encoding and transmission. Their transformed
+output is sent. Response schemas run after request correlation, and their
+transformed output replaces `frame.data`; `frame.raw`, when present, remains the
+original wire frame. Invalid requests and responses reject with
+`MTPValidationError`. Invalid subscription messages do not reach the handler
+and are reported through `onValidationError`.
+
+`throwProtocolErrors: true` converts correlated `Error*` frames into
+`MTPProtocolError`. It defaults to `false` for compatibility.
+
+`MTPProxyConnection` applies the same schema registry to another TypeScript
+request/subscription transport, such as a Tauri command and event proxy:
+
+```typescript
+const connection = new MTPProxyConnection(adapter, {
+  schemas,
+  throwProtocolErrors: true,
+});
+```
+
 Protocol ping behavior is defined in [Protocol Reference](PROTOCOL-REFERENCE.md#protocol-keepalive). The SDK configuration is:
 
 ```typescript

src/sdk/client.ts

@@ -10,6 +10,15 @@ import * as bindings from "mtp/raw";
 import { unixTimeMillis, utf8Encode } from "./utils.js";
 import type * as RawBindings from "../raw/index";
 import type { MTPCommunicationType } from "../type-map/index";
+import { MTPProtocol } from "./schema.js";
+import type {
+  MTPMessageType,
+  MTPFrame,
+  MTPNoSchemas,
+  MTPRequestData,
+  MTPResponseFrame,
+  MTPSchemaRegistry,
+} from "./schema.js";
 import type { MTPSessionStorage, MTPSessionState } from "./session";
 import { MTPSessionManager } from "./session.js";
 import {
@@ -249,7 +258,9 @@ export interface MTPPublicKeyBundleKeys {
   sigClPublicKey: Uint8Array;
 }
 
-export interface MTPClientOptions {
+export interface MTPClientOptions<
+  Registry extends MTPSchemaRegistry = MTPNoSchemas,
+> {
   url: string;
   descriptor?: string;
   hostPublicKey?: MTPKeyMaterialInput;
@@ -282,6 +293,12 @@ export interface MTPClientOptions {
   securityProfile?: MTPSecurityProfile;
   /** One receive resource policy shared by frame and protected-value opening. */
   receiveLimits?: MTPReceiveLimits;
+  /** Application request and response schemas, keyed by communication type. */
+  schemas?: Registry;
+  /** Reject `request()` when the correlated response is an `Error*` frame. */
+  throwProtocolErrors?: boolean;
+  /** Receives subscription validation failures. Request failures reject normally. */
+  onValidationError?: (error: import("./schema.js").MTPValidationError) => void;
 }
 
 export interface MTPSecurityProfile {
@@ -562,8 +579,8 @@ export interface MTPAcceptEncryptedPipeOptions {
   signaturePolicy?: MTPSignatureVerificationPolicy;
 }
 
-type NormalizedMTPClientOptions = Omit<
+type NormalizedMTPClientOptions<Registry extends MTPSchemaRegistry> = Omit<
-  MTPClientOptions,
+  MTPClientOptions<Registry>,
   "hostPublicKey" | "receiveLimits"
 > & {
   hostPublicKey?: Uint8Array;
@@ -914,14 +931,34 @@ function validateOptions(options) {
   ) {
     throw new TypeError("requestTimeoutMs must be a positive safe integer");
   }
+  if (options.schemas != null) {
+    if (typeof options.schemas !== "object" || Array.isArray(options.schemas)) {
+      throw new TypeError("schemas must be an object");
+    }
+    for (const [type, pair] of Object.entries(options.schemas)) {
+      if (
+        !pair ||
+        typeof pair !== "object" ||
+        typeof (pair as { request?: { parseAsync?: unknown } }).request
+          ?.parseAsync !== "function" ||
+        typeof (pair as { response?: { parseAsync?: unknown } }).response
+          ?.parseAsync !== "function"
+      ) {
+        throw new TypeError(
+          `schemas.${type} must contain request and response schemas with parseAsync()`,
+        );
+      }
+    }
+  }
 }
 
-export class MTPClient {
+export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
   static readonly crypto = crypto;
   static readonly codec = codec;
 
   #credentials: InternalCredentials | null;
-  #options: NormalizedMTPClientOptions;
+  #options: NormalizedMTPClientOptions<Registry>;
+  readonly #protocol: MTPProtocol<Registry> | undefined;
   readonly #protectedReplayGuard = new InMemoryReplayGuard();
   readonly #relayReplayGuard = new InMemoryReplayGuard();
   readonly raw: MTPRaw;
@@ -934,10 +971,17 @@ export class MTPClient {
   readonly encryptedSecretProvider: MTPEncryptedSecretProvider;
 
   private constructor(
-    options: NormalizedMTPClientOptions,
+    options: NormalizedMTPClientOptions<Registry>,
     client: RawBindings.WasmClient,
   ) {
     this.#options = options;
+    this.#protocol = options.schemas
+      ? new MTPProtocol({
+          schemas: options.schemas,
+          throwProtocolErrors: options.throwProtocolErrors,
+          onValidationError: options.onValidationError,
+        })
+      : undefined;
     this.#credentials = deserializeCredentials(options.credentials);
     this.raw = { client, bindings };
     this.encryptedSecretProvider =
@@ -947,7 +991,11 @@ export class MTPClient {
     );
   }
 
-  static async create(options: MTPClientOptions): Promise<MTPClient> {
+  static async create<
+    const Registry extends MTPSchemaRegistry = MTPNoSchemas,
+  >(
+    options: MTPClientOptions<Registry>,
+  ): Promise<MTPClient<Registry>> {
     validateOptions(options);
     await MTPClient.init(options.wasm);
 
@@ -968,7 +1016,7 @@ export class MTPClient {
       securityProfile: resolveSecurityProfile(options),
     };
 
-    let sdk: MTPClient | undefined;
+    let sdk: MTPClient<Registry> | undefined;
     const client = new WasmClient(
       (state) =>
         emit(normalizedOptions.logger, {
@@ -1004,7 +1052,7 @@ export class MTPClient {
       setReceiveLimits.call(rawClient, normalizedOptions.receiveLimits);
     }
 
-    sdk = new MTPClient(normalizedOptions, client);
+    sdk = new MTPClient<Registry>(normalizedOptions, client);
     await sdk.#loadStoredCredentials();
     if (!sdk.#credentials) {
       sdk.#credentials = {
@@ -1238,6 +1286,35 @@ export class MTPClient {
     };
   }
 
+  async #parseRequestData(
+    type: MTPCommunicationType,
+    data: unknown,
+  ): Promise<Record<string, unknown>> {
+    if (!this.#protocol || !this.#protocol.schemas[type]) {
+      return (data ?? {}) as Record<string, unknown>;
+    }
+    const parsed = await this.#protocol.parseRequest(
+      type as MTPMessageType<Registry>,
+      data as never,
+    );
+    return (parsed ?? {}) as Record<string, unknown>;
+  }
+
+  async #parseResponseData(
+    requestedType: MTPCommunicationType,
+    frame: ParsedFrame,
+    phase: "response" | "subscription" = "response",
+  ): Promise<MTPFrame<unknown>> {
+    if (!this.#protocol || !this.#protocol.schemas[requestedType]) {
+      return frame;
+    }
+    return await this.#protocol.parseResponse(
+      requestedType as MTPMessageType<Registry>,
+      frame,
+      phase,
+    );
+  }
+
   #buildFrame(typeOrFrame, data, options) {
     if (typeOrFrame instanceof Uint8Array) {
       if (
@@ -1279,6 +1356,11 @@ export class MTPClient {
   }
 
   async send(message: Uint8Array): Promise<void>;
+  async send<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data?: MTPRequestData<Registry, Type>,
+    options?: MTPSendOptions,
+  ): Promise<void>;
   async send(
     type: MTPCommunicationType,
     data: Record<string, unknown>,
@@ -1286,10 +1368,14 @@ export class MTPClient {
   ): Promise<void>;
   async send(
     typeOrFrame: Uint8Array | MTPCommunicationType,
-    data?: Record<string, unknown>,
+    data?: unknown,
     options?: MTPSendOptions,
   ): Promise<void> {
-    const message = this.#buildFrame(typeOrFrame, data, options);
+    const parsedData =
+      typeof typeOrFrame === "string"
+        ? await this.#parseRequestData(typeOrFrame, data)
+        : data;
+    const message = this.#buildFrame(typeOrFrame, parsedData, options);
 
     try {
       const frame = this.raw.bindings.parse_frame(message);
@@ -1345,6 +1431,11 @@ export class MTPClient {
     data?: never,
     options?: MTPRequestOptions,
   ): Promise<ParsedFrame>;
+  async request<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data?: MTPRequestData<Registry, Type>,
+    options?: MTPRequestOptions,
+  ): Promise<MTPResponseFrame<Registry, Type>>;
   async request(
     type: MTPCommunicationType,
     data: Record<string, unknown>,
@@ -1352,15 +1443,19 @@ export class MTPClient {
   ): Promise<ParsedFrame>;
   async request(
     typeOrFrame: Uint8Array | MTPCommunicationType,
-    data?: Record<string, unknown>,
+    data?: unknown,
     options: MTPRequestOptions = {},
-  ): Promise<ParsedFrame> {
+  ): Promise<ParsedFrame | MTPFrame<unknown>> {
     const timeoutMs =
       options.timeoutMs ?? this.#options.requestTimeoutMs ?? 30_000;
     if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) {
       throw new TypeError("request timeoutMs must be a positive safe integer");
     }
-    const frame = this.#buildFrame(typeOrFrame, data, options);
+    const parsedData =
+      typeof typeOrFrame === "string"
+        ? await this.#parseRequestData(typeOrFrame, data)
+        : data;
+    const frame = this.#buildFrame(typeOrFrame, parsedData, options);
     try {
       const parsed = this.raw.bindings.parse_frame(frame);
       emit(
@@ -1391,16 +1486,25 @@ export class MTPClient {
     // The WASM client owns request expiry and its late-response tombstones.
     // Keeping a second Promise timer here can reject the SDK call while the
     // protocol request is still allowed to complete successfully.
-    return await this.raw.client.request(
+    const response = await this.raw.client.request(
       frame,
       options.responseType ?? null,
       timeoutMs,
     );
+    return typeof typeOrFrame === "string"
+      ? await this.#parseResponseData(typeOrFrame, response)
+      : response;
   }
 
+  subscribe<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    handler: (
+      message: MTPResponseFrame<Registry, Type>,
+    ) => void | Promise<void>,
+  ): Unsubscribe;
   subscribe(
     type: MTPCommunicationType,
-    handler: (message: ParsedFrame) => void,
+    handler: (message: MTPFrame<unknown>) => void | Promise<void>,
   ): Unsubscribe {
     if (typeof type !== "string" || !type) {
       throw new TypeError("subscription type must be a non-empty string");
@@ -1408,8 +1512,25 @@ export class MTPClient {
     if (typeof handler !== "function") {
       throw new TypeError("subscription handler must be a function");
     }
-    const id = this.raw.client.subscribe(type, handler);
+    let active = true;
-    return () => this.raw.client.unsubscribe(id);
+    const id = this.raw.client.subscribe(type, (message) => {
+      if (!this.#protocol || !this.#protocol.schemas[type]) {
+        void handler(message);
+        return;
+      }
+      void this.#parseResponseData(type, message, "subscription").then(
+        (parsed) => {
+          if (active) void handler(parsed);
+        },
+        (error) => {
+          this.#protocol?.reportValidationError(error);
+        },
+      );
+    });
+    return () => {
+      active = false;
+      this.raw.client.unsubscribe(id);
+    };
   }
 
   #handleFrame(frame) {

src/sdk/index.ts

@@ -5,3 +5,4 @@
  * keeps the package's historical exports stable.
  */
 export * from "./client.js";
+export * from "./schema.js";

src/sdk/schema.ts

@@ -0,0 +1,234 @@
+import type { MTPRequestOptions, ParsedFrame, Unsubscribe } from "./client.js";
+import type { MTPCommunicationType } from "../type-map/index.js";
+
+export interface MTPSchema<Input = unknown, Output = Input> {
+  readonly _input: Input;
+  readonly _output: Output;
+  parseAsync(value: unknown): Promise<Output>;
+}
+
+export interface MTPSchemaPair<
+  Request extends MTPSchema = MTPSchema,
+  Response extends MTPSchema = MTPSchema,
+> {
+  request: Request;
+  response: Response;
+}
+
+export type MTPSchemaRegistry = Record<string, MTPSchemaPair>;
+export type MTPNoSchemas = Record<never, never>;
+
+export type MTPSchemaInput<Schema extends MTPSchema> = Schema["_input"];
+export type MTPSchemaOutput<Schema extends MTPSchema> = Schema["_output"];
+export type MTPMessageType<Registry extends MTPSchemaRegistry> =
+  keyof Registry & string;
+
+export type MTPFrame<Data = ParsedFrame["data"]> = {
+  id?: number;
+  type: string;
+  data: Data;
+  sender?: ParsedFrame["sender"];
+  receiver?: ParsedFrame["receiver"];
+  raw?: ParsedFrame["raw"];
+};
+
+export type MTPTypedFrame<Data = ParsedFrame["data"]> = MTPFrame<Data>;
+
+export type MTPResponseFrame<
+  Registry extends MTPSchemaRegistry,
+  Type extends MTPMessageType<Registry>,
+> = MTPTypedFrame<MTPSchemaOutput<Registry[Type]["response"]>>;
+
+export type MTPRequestData<
+  Registry extends MTPSchemaRegistry,
+  Type extends MTPMessageType<Registry>,
+> = MTPSchemaInput<Registry[Type]["request"]>;
+
+export type MTPRequestFunction<Registry extends MTPSchemaRegistry> = <
+  Type extends MTPMessageType<Registry>,
+>(
+  type: Type,
+  data?: MTPRequestData<Registry, Type>,
+  options?: MTPRequestOptions,
+) => Promise<MTPResponseFrame<Registry, Type>>;
+
+export type MTPSubscriptionFunction<Registry extends MTPSchemaRegistry> = <
+  Type extends MTPMessageType<Registry>,
+>(
+  type: Type,
+  handler: (message: MTPResponseFrame<Registry, Type>) => void | Promise<void>,
+) => Unsubscribe;
+
+export class MTPValidationError extends Error {
+  readonly phase: "request" | "response" | "subscription";
+  readonly messageType: string;
+  readonly frame?: MTPFrame;
+
+  constructor(
+    phase: MTPValidationError["phase"],
+    messageType: string,
+    cause: unknown,
+    frame?: MTPFrame,
+  ) {
+    super(`${phase} validation failed for ${messageType}`, { cause });
+    this.name = "MTPValidationError";
+    this.phase = phase;
+    this.messageType = messageType;
+    this.frame = frame;
+  }
+}
+
+export class MTPProtocolError extends Error {
+  readonly type: string;
+  readonly id: number | undefined;
+  readonly communicationType: string;
+  readonly requestId: number | undefined;
+  readonly errorType: string | undefined;
+  readonly frame: MTPFrame;
+
+  constructor(frame: MTPFrame) {
+    const errorType =
+      frame.data &&
+      typeof frame.data === "object" &&
+      !Array.isArray(frame.data) &&
+      typeof (frame.data as Record<string, unknown>).ErrorType === "string"
+        ? ((frame.data as Record<string, unknown>).ErrorType as string)
+        : undefined;
+    super(errorType ? `${frame.type}: ${errorType}` : frame.type);
+    this.name = "MTPProtocolError";
+    this.type = frame.type;
+    this.id = frame.id;
+    this.communicationType = frame.type;
+    this.requestId = frame.id;
+    this.errorType = errorType;
+    this.frame = frame;
+  }
+}
+
+export interface MTPProtocolOptions<Registry extends MTPSchemaRegistry> {
+  schemas: Registry;
+  throwProtocolErrors?: boolean;
+  onValidationError?: (error: MTPValidationError) => void;
+}
+
+function isErrorFrame(frame: MTPFrame): boolean {
+  return frame.type.startsWith("Error");
+}
+
+export class MTPProtocol<Registry extends MTPSchemaRegistry> {
+  readonly schemas: Registry;
+  readonly #throwProtocolErrors: boolean;
+  readonly #onValidationError:
+    | ((error: MTPValidationError) => void)
+    | undefined;
+
+  constructor(options: MTPProtocolOptions<Registry>) {
+    this.schemas = options.schemas;
+    this.#throwProtocolErrors = options.throwProtocolErrors ?? false;
+    this.#onValidationError = options.onValidationError;
+  }
+
+  async parseRequest<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data: MTPRequestData<Registry, Type> | undefined,
+  ): Promise<MTPSchemaOutput<Registry[Type]["request"]>> {
+    try {
+      return await this.schemas[type].request.parseAsync(data);
+    } catch (error) {
+      throw new MTPValidationError("request", type, error);
+    }
+  }
+
+  async parseResponse<Type extends MTPMessageType<Registry>>(
+    requestedType: Type,
+    frame: MTPFrame,
+    phase: "response" | "subscription" = "response",
+  ): Promise<MTPResponseFrame<Registry, Type>> {
+    if (isErrorFrame(frame)) {
+      if (phase === "response" && this.#throwProtocolErrors) {
+        throw new MTPProtocolError(frame);
+      }
+      return frame as MTPResponseFrame<Registry, Type>;
+    }
+
+    const schema =
+      this.schemas[frame.type]?.response ??
+      this.schemas[requestedType].response;
+    try {
+      const data = await schema.parseAsync(frame.data);
+      return { ...frame, data } as MTPResponseFrame<Registry, Type>;
+    } catch (error) {
+      throw new MTPValidationError(
+        phase,
+        frame.type || requestedType,
+        error,
+        frame,
+      );
+    }
+  }
+
+  reportValidationError(error: unknown): void {
+    if (error instanceof MTPValidationError) {
+      this.#onValidationError?.(error);
+    }
+  }
+}
+
+export interface MTPProxyAdapter {
+  request(
+    type: MTPCommunicationType,
+    data: Record<string, unknown>,
+    options?: MTPRequestOptions,
+  ): Promise<MTPFrame>;
+  subscribe(
+    type: MTPCommunicationType,
+    handler: (message: MTPFrame) => void,
+  ): Unsubscribe;
+}
+
+export class MTPProxyConnection<Registry extends MTPSchemaRegistry> {
+  readonly #adapter: MTPProxyAdapter;
+  readonly #protocol: MTPProtocol<Registry>;
+
+  constructor(adapter: MTPProxyAdapter, options: MTPProtocolOptions<Registry>) {
+    this.#adapter = adapter;
+    this.#protocol = new MTPProtocol(options);
+  }
+
+  async request<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    data?: MTPRequestData<Registry, Type>,
+    options?: MTPRequestOptions,
+  ): Promise<MTPResponseFrame<Registry, Type>> {
+    const parsed = await this.#protocol.parseRequest(type, data);
+    const response = await this.#adapter.request(
+      type,
+      (parsed ?? {}) as Record<string, unknown>,
+      options,
+    );
+    return await this.#protocol.parseResponse(type, response);
+  }
+
+  subscribe<Type extends MTPMessageType<Registry>>(
+    type: Type,
+    handler: (
+      message: MTPResponseFrame<Registry, Type>,
+    ) => void | Promise<void>,
+  ): Unsubscribe {
+    let active = true;
+    const unsubscribe = this.#adapter.subscribe(type, (message) => {
+      void this.#protocol.parseResponse(type, message, "subscription").then(
+        (parsed) => {
+          if (active) void handler(parsed);
+        },
+        (error) => {
+          this.#protocol.reportValidationError(error);
+        },
+      );
+    });
+    return () => {
+      active = false;
+      unsubscribe();
+    };
+  }
+}