feat(rtc): add RPC interceptors

.changeset/rpc-interceptors.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': minor
+---
+
+Add `RpcInterceptor` support: `LocalParticipant.addRpcInterceptor()` wraps every RPC the participant performs or handles, for logging, tracing, or payload metadata. `RpcInvocationData` now carries the invoked `method`.

packages/livekit-rtc/README.md

@@ -79,7 +79,7 @@ await track.close();
 
 ### RPC
 
-Perform your own predefined method calls from one participant to another. 
+Perform your own predefined method calls from one participant to another.
 
 This feature is especially powerful when used with [Agents](https://docs.livekit.io/agents), for instance to forward LLM function calls to your client application.
 
@@ -89,14 +89,14 @@ The participant who implements the method and will receive its calls must first
 
 ```typescript
 room.localParticipant?.registerRpcMethod(
-   // method name - can be any string that makes sense for your application
+  // method name - can be any string that makes sense for your application
   'greet',
 
   // method handler - will be called when the method is invoked by a RemoteParticipant
   async (data: RpcInvocationData) => {
     console.log(`Received greeting from ${data.callerIdentity}: ${data.payload}`);
     return `Hello, ${data.callerIdentity}!`;
-  }
+  },
 );
 ```
 
@@ -121,9 +121,40 @@ try {
 
 You may find it useful to adjust the `responseTimeout` parameter, which indicates the amount of time you will wait for a response. We recommend keeping this value as low as possible while still satisfying the constraints of your application.
 
+#### Intercepting RPC calls
+
+An `RpcInterceptor` wraps every RPC the local participant performs or handles, which is useful for logging, tracing, or attaching metadata to payloads. Each method receives the call and a `next` continuation; return what `next` returns. Interceptors run in the order they were added, the first being outermost, and errors from the remote side or from your handler flow through them unchanged. Implement only the direction you care about.
+
+```typescript
+const timing: RpcInterceptor = {
+  async interceptOutgoing(call, next) {
+    const start = performance.now();
+    try {
+      return await next(call);
+    } finally {
+      console.log(
+        `call ${call.method} -> ${call.destinationIdentity}: ${performance.now() - start}ms`,
+      );
+    }
+  },
+  async interceptIncoming(invocation, next) {
+    const start = performance.now();
+    try {
+      return await next(invocation);
+    } finally {
+      console.log(
+        `handled ${invocation.method} from ${invocation.callerIdentity}: ${performance.now() - start}ms`,
+      );
+    }
+  },
+};
+
+room.localParticipant!.addRpcInterceptor(timing);
+```
+
 #### Errors
 
-LiveKit is a dynamic realtime environment and calls can fail for various reasons. 
+LiveKit is a dynamic realtime environment and calls can fail for various reasons.
 
 You may throw errors of the type `RpcError` with a string `message` in an RPC method handler and they will be received on the caller's side with the message intact. Other errors will not be transmitted and will instead arrive to the caller as `1500` ("Application Error"). Other built-in errors are detailed in `RpcError`.
 
@@ -132,7 +163,6 @@ You may throw errors of the type `RpcError` with a string `message` in an RPC me
 - [`publish-wav`](https://github.com/livekit/node-sdks/tree/main/examples/publish-wav): connect to a room and publish a .wave file
 - [`rpc`](https://github.com/livekit/node-sdks/tree/main/examples/rpc): simple back-and-forth RPC interaction
 
-
 ## Getting help / Contributing
 
 Please join us on [Slack](https://livekit.io/join-slack) to get help from our devs & community. We welcome your contributions and details can be discussed there.

packages/livekit-rtc/src/index.ts

@@ -44,7 +44,15 @@ export {
   type RoomOptions,
   type RtcConfiguration,
 } from './room.js';
-export { RpcError, type PerformRpcParams, type RpcInvocationData } from './rpc.js';
+export {
+  RpcError,
+  type IncomingRpcNext,
+  type OutgoingRpcNext,
+  type PerformRpcParams,
+  type RpcCallInfo,
+  type RpcInterceptor,
+  type RpcInvocationData,
+} from './rpc.js';
 export {
   LocalAudioTrack,
   LocalVideoTrack,

packages/livekit-rtc/src/participant.ts

@@ -90,7 +90,15 @@ import {
 } from './data_streams/index.js';
 import { FfiClient, FfiHandle } from './ffi_client.js';
 import { log } from './log.js';
-import { type PerformRpcParams, RpcError, type RpcInvocationData } from './rpc.js';
+import {
+  type PerformRpcParams,
+  type RpcCallInfo,
+  RpcError,
+  type RpcInterceptor,
+  type RpcInvocationData,
+  chainIncoming,
+  chainOutgoing,
+} from './rpc.js';
 import type { LocalTrack } from './track.js';
 import type { RemoteTrackPublication, TrackPublication } from './track_publication.js';
 import { LocalTrackPublication } from './track_publication.js';
@@ -167,6 +175,7 @@ export type DataPublishOptions = {
 
 export class LocalParticipant extends Participant {
   private rpcHandlers: Map<string, (data: RpcInvocationData) => Promise<string>> = new Map();
+  private rpcInterceptors: RpcInterceptor[] = [];
 
   private ffiEventLock: Mutex;
 
@@ -833,6 +842,18 @@ export class LocalParticipant extends Participant {
     payload,
     responseTimeout,
   }: PerformRpcParams): Promise<string> {
+    const call: RpcCallInfo = { destinationIdentity, method, payload, responseTimeout };
+    // snapshot the interceptor list so add/remove during a call is well defined
+    const perform = chainOutgoing([...this.rpcInterceptors], (c) => this.performRpcFfi(c));
+    return await perform(call);
+  }
+
+  private async performRpcFfi({
+    destinationIdentity,
+    method,
+    payload,
+    responseTimeout,
+  }: RpcCallInfo): Promise<string> {
     const req = new PerformRpcRequest({
       localParticipantHandle: this.ffi_handle.handle,
       destinationIdentity,
@@ -857,6 +878,29 @@ export class LocalParticipant extends Participant {
     return cb.payload!;
   }
 
+  /**
+   * Add an {@link RpcInterceptor} that wraps every RPC this participant performs or handles.
+   * Interceptors run in the order they were added, the first being outermost. Adding the same
+   * instance twice is a no-op.
+   *
+   * @param interceptor - The interceptor to add
+   */
+  addRpcInterceptor(interceptor: RpcInterceptor) {
+    if (!this.rpcInterceptors.includes(interceptor)) {
+      this.rpcInterceptors.push(interceptor);
+    }
+  }
+
+  /**
+   * Remove a previously added {@link RpcInterceptor}. Calls already in flight keep the chain
+   * they started with.
+   *
+   * @param interceptor - The interceptor to remove
+   */
+  removeRpcInterceptor(interceptor: RpcInterceptor) {
+    this.rpcInterceptors = this.rpcInterceptors.filter((existing) => existing !== interceptor);
+  }
+
   /**
    * Establishes the participant as a receiver for calls of the specified RPC method.
    * Will overwrite any existing callback for the same method.
@@ -928,23 +972,28 @@ export class LocalParticipant extends Participant {
     let responseError: RpcError | null = null;
     let responsePayload: string | null = null;
 
-    const handler = this.rpcHandlers.get(method);
-
-    if (!handler) {
-      responseError = RpcError.builtIn('UNSUPPORTED_METHOD');
-    } else {
-      try {
-        responsePayload = await handler({ requestId, callerIdentity, payload, responseTimeout });
-      } catch (error) {
-        if (error instanceof RpcError) {
-          responseError = error;
-        } else {
-          console.warn(
-            `Uncaught error returned by RPC handler for ${method}. Returning APPLICATION_ERROR instead.`,
-            error,
-          );
-          responseError = RpcError.builtIn('APPLICATION_ERROR');
-        }
+    const invocation: RpcInvocationData = {
+      requestId,
+      callerIdentity,
+      payload,
+      responseTimeout,
+      method,
+    };
+    // the chain sees the handler's outcome unchanged, including UNSUPPORTED_METHOD for a method
+    // nothing is registered for; only after it settles is a non-RpcError turned into
+    // APPLICATION_ERROR for the caller
+    const handle = chainIncoming([...this.rpcInterceptors], (inv) => this.invokeRpcHandler(inv));
+    try {
+      responsePayload = (await handle(invocation)) ?? null;
+    } catch (error) {
+      if (error instanceof RpcError) {
+        responseError = error;
+      } else {
+        console.warn(
+          `Uncaught error returned by RPC handler for ${method}. Returning APPLICATION_ERROR instead.`,
+          error,
+        );
+        responseError = RpcError.builtIn('APPLICATION_ERROR');
       }
     }
 
@@ -963,6 +1012,15 @@ export class LocalParticipant extends Participant {
       console.warn(`error sending rpc method invocation response: ${res.error}`);
     }
   }
+
+  /** The innermost step of the incoming chain: run the registered handler, if any. */
+  private async invokeRpcHandler(invocation: RpcInvocationData): Promise<string> {
+    const handler = this.rpcHandlers.get(invocation.method);
+    if (!handler) {
+      throw RpcError.builtIn('UNSUPPORTED_METHOD');
+    }
+    return await handler(invocation);
+  }
 }
 
 export class RemoteParticipant extends Participant {