Data track schema definition storage

.changeset/schema-storage-api.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': minor
+---
+
+Added `defineSchema` and `getSchema` to `LocalParticipant` for storing and retrieving data track schema definitions

src/api/SignalClient.test.ts

@@ -1,12 +1,16 @@
 import {
   ClientInfo_Capability,
+  DataBlob,
+  DataBlobKey,
   DisconnectReason,
+  GetDataBlobResponse,
   JoinRequest,
   JoinResponse,
   LeaveRequest,
   ReconnectResponse,
   SignalRequest,
   SignalResponse,
+  StoreDataBlobResponse,
   WrappedJoinRequest,
   WrappedJoinRequest_Compression,
 } from '@livekit/protocol';
@@ -33,7 +37,8 @@ function createJoinResponse() {
 }
 
 function createSignalResponse(
-  messageCase: 'join' | 'reconnect' | 'leave' | 'update',
+  messageCase:
+    'join' | 'reconnect' | 'leave' | 'update' | 'storeDataBlobResponse' | 'getDataBlobResponse',
   value: any,
 ): SignalResponse {
   return new SignalResponse({
@@ -573,6 +578,88 @@ describe('SignalClient.connect', () => {
   });
 });
 
+describe('SignalClient data blobs', () => {
+  let signalClient: SignalClient;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    signalClient = new SignalClient(false);
+  });
+
+  function clientInternals() {
+    return signalClient as unknown as {
+      state: SignalConnectionState;
+      streamWriter: { write: (chunk: ArrayBuffer) => Promise<void> } | undefined;
+      handleSignalResponse: (res: SignalResponse) => void;
+    };
+  }
+
+  it('dispatches store data blob responses', () => {
+    const received: StoreDataBlobResponse[] = [];
+    signalClient.onStoreDataBlobResponse = (res) => received.push(res);
+
+    clientInternals().handleSignalResponse(
+      createSignalResponse('storeDataBlobResponse', new StoreDataBlobResponse({ requestId: 42 })),
+    );
+
+    expect(received).toHaveLength(1);
+    expect(received[0].requestId).toBe(42);
+  });
+
+  it('dispatches get data blob responses', () => {
+    const received: GetDataBlobResponse[] = [];
+    signalClient.onGetDataBlobResponse = (res) => received.push(res);
+
+    clientInternals().handleSignalResponse(
+      createSignalResponse(
+        'getDataBlobResponse',
+        new GetDataBlobResponse({
+          requestId: 43,
+          blob: new DataBlob({ contents: new TextEncoder().encode('definition') }),
+        }),
+      ),
+    );
+
+    expect(received).toHaveLength(1);
+    expect(received[0].requestId).toBe(43);
+    expect(new TextDecoder().decode(received[0].blob?.contents)).toBe('definition');
+  });
+
+  it('sends blob requests with increasing request ids', async () => {
+    const written: SignalRequest[] = [];
+    const internals = clientInternals();
+    internals.state = SignalConnectionState.CONNECTED;
+    internals.streamWriter = {
+      write: async (chunk) => {
+        written.push(SignalRequest.fromBinary(new Uint8Array(chunk)));
+      },
+    };
+
+    const key = new DataBlobKey({ key: { case: 'generic', value: 'my-key' } });
+    const storeRequestId = await signalClient.sendStoreDataBlobRequest(
+      new DataBlob({ key, contents: new TextEncoder().encode('contents') }),
+    );
+    const getRequestId = await signalClient.sendGetDataBlobRequest(key, 'publisher-identity');
+
+    expect(getRequestId).toBe(storeRequestId + 1);
+    expect(written).toHaveLength(2);
+
+    const storeMessage = written[0].message;
+    expect(storeMessage.case).toStrictEqual('storeDataBlobRequest');
+    if (storeMessage.case !== 'storeDataBlobRequest') throw new Error('unreachable');
+    expect(storeMessage.value.requestId).toBe(storeRequestId);
+    expect(storeMessage.value.blob?.key?.key).toStrictEqual({ case: 'generic', value: 'my-key' });
+    expect(new TextDecoder().decode(storeMessage.value.blob?.contents)).toBe('contents');
+
+    const getMessage = written[1].message;
+    expect(getMessage.case).toStrictEqual('getDataBlobRequest');
+    if (getMessage.case !== 'getDataBlobRequest') throw new Error('unreachable');
+    expect(getMessage.value.requestId).toBe(getRequestId);
+    expect(getMessage.value.participantIdentity).toStrictEqual('publisher-identity');
+    expect(getMessage.value.key?.key).toStrictEqual({ case: 'generic', value: 'my-key' });
+  });
+});
+
 describe('SignalClient utility functions', () => {
   describe('toProtoSessionDescription', () => {
     it('should convert RTCSessionDescriptionInit to proto SessionDescription', async () => {

src/api/SignalClient.ts

@@ -6,9 +6,13 @@ import {
   ClientInfo_Capability,
   ConnectionQualityUpdate,
   ConnectionSettings,
+  DataBlob,
+  DataBlobKey,
   DataTrackSubscriberHandles,
   DisconnectReason,
   Encryption_Type,
+  GetDataBlobRequest,
+  GetDataBlobResponse,
   JoinRequest,
   JoinResponse,
   LeaveRequest,
@@ -30,6 +34,8 @@ import {
   SignalTarget,
   SimulateScenario,
   SpeakerInfo,
+  StoreDataBlobRequest,
+  StoreDataBlobResponse,
   StreamStateUpdate,
   SubscribedQualityUpdate,
   SubscriptionPermission,
@@ -203,6 +209,10 @@ export class SignalClient {
 
   onDataTrackSubscriberHandles?: (event: DataTrackSubscriberHandles) => void;
 
+  onStoreDataBlobResponse?: (res: StoreDataBlobResponse) => void;
+
+  onGetDataBlobResponse?: (res: GetDataBlobResponse) => void;
+
   onJoined?: (event: JoinResponse) => void;
 
   connectOptions?: ConnectOpts;
@@ -774,6 +784,24 @@ export class SignalClient {
     });
   }
 
+  async sendStoreDataBlobRequest(blob: DataBlob) {
+    const requestId = this.getNextRequestId();
+    await this.sendRequest({
+      case: 'storeDataBlobRequest',
+      value: new StoreDataBlobRequest({ requestId, blob }),
+    });
+    return requestId;
+  }
+
+  async sendGetDataBlobRequest(key: DataBlobKey, participantIdentity: string) {
+    const requestId = this.getNextRequestId();
+    await this.sendRequest({
+      case: 'getDataBlobRequest',
+      value: new GetDataBlobRequest({ requestId, participantIdentity, key }),
+    });
+    return requestId;
+  }
+
   sendUpdateDataSubscription(sid: DataTrackSid, subscribe: boolean) {
     return this.sendRequest({
       case: 'updateDataSubscription',
@@ -940,6 +968,14 @@ export class SignalClient {
       if (this.onDataTrackSubscriberHandles) {
         this.onDataTrackSubscriberHandles(msg.value);
       }
+    } else if (msg.case === 'storeDataBlobResponse') {
+      if (this.onStoreDataBlobResponse) {
+        this.onStoreDataBlobResponse(msg.value);
+      }
+    } else if (msg.case === 'getDataBlobResponse') {
+      if (this.onGetDataBlobResponse) {
+        this.onGetDataBlobResponse(msg.value);
+      }
     } else {
       this.log.debug('unsupported message', { msgCase: msg.case });
     }

src/room/RTCEngine.ts

@@ -13,6 +13,7 @@ import {
   EncryptedPacket,
   EncryptedPacketPayload,
   Encryption_Type,
+  type GetDataBlobResponse,
   type JoinResponse,
   type LeaveRequest,
   LeaveRequest_Action,
@@ -31,6 +32,7 @@ import {
   SessionDescription,
   SignalTarget,
   SpeakerInfo,
+  type StoreDataBlobResponse,
   type StreamStateUpdate,
   SubscribedQualityUpdate,
   type SubscriptionPermissionUpdate,
@@ -735,6 +737,14 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       this.emit(EngineEvent.DataTrackSubscriberHandles, event);
     };
 
+    this.client.onStoreDataBlobResponse = (res: StoreDataBlobResponse) => {
+      this.emit(EngineEvent.StoreDataBlobResponse, res);
+    };
+
+    this.client.onGetDataBlobResponse = (res: GetDataBlobResponse) => {
+      this.emit(EngineEvent.GetDataBlobResponse, res);
+    };
+
     this.client.onClose = () => {
       this.handleDisconnect('signal', ReconnectReason.RR_SIGNAL_DISCONNECTED);
     };
@@ -2054,6 +2064,8 @@ export type EngineEventCallbacks = {
   unPublishDataTrackResponse: (event: UnpublishDataTrackResponse) => void;
   dataTrackSubscriberHandles: (event: DataTrackSubscriberHandles) => void;
   dataTrackPacketReceived: (packet: Uint8Array) => void;
+  storeDataBlobResponse: (res: StoreDataBlobResponse) => void;
+  getDataBlobResponse: (res: GetDataBlobResponse) => void;
   joined: (joinResponse: JoinResponse) => void;
   tokenRefreshed: (token: string) => void;
   serverRegionsReported: (regions: RegionSettings) => void;

src/room/data-track/schema-storage.ts

@@ -0,0 +1,86 @@
+import { RequestResponse_Reason } from '@livekit/protocol';
+import { LivekitReasonedError } from '../errors';
+
+export enum DataTrackSchemaStorageErrorReason {
+  /** Request to store or retrieve a schema definition timed-out */
+  Timeout = 0,
+  /** The participant has not defined a schema with this ID */
+  NotFound = 1,
+  /** The server rejected the request */
+  RequestFailed = 2,
+  /** The server response is malformed */
+  MalformedResponse = 3,
+  /** The retrieved schema definition is not valid UTF-8 */
+  InvalidDefinition = 4,
+  /** Request cancelled by caller */
+  Cancelled = 5,
+  /** Cannot store or retrieve a schema definition when disconnected */
+  Disconnected = 6,
+}
+
+export class DataTrackSchemaStorageError<
+  Reason extends DataTrackSchemaStorageErrorReason = DataTrackSchemaStorageErrorReason,
+> extends LivekitReasonedError<Reason> {
+  readonly name = 'DataTrackSchemaStorageError';
+
+  reason: Reason;
+
+  reasonName: string;
+
+  constructor(message: string, reason: Reason, options?: { cause?: unknown }) {
+    super(24, message, options);
+    this.reason = reason;
+    this.reasonName = DataTrackSchemaStorageErrorReason[reason];
+  }
+
+  static timeout() {
+    return new DataTrackSchemaStorageError(
+      'Schema storage request timed out',
+      DataTrackSchemaStorageErrorReason.Timeout,
+    );
+  }
+
+  static notFound(message: string) {
+    return new DataTrackSchemaStorageError(
+      message || 'The participant has not defined a schema with this ID',
+      DataTrackSchemaStorageErrorReason.NotFound,
+    );
+  }
+
+  static requestFailed(reason: RequestResponse_Reason, message: string) {
+    return new DataTrackSchemaStorageError(
+      `Schema storage request failed (${RequestResponse_Reason[reason]}): ${message}`,
+      DataTrackSchemaStorageErrorReason.RequestFailed,
+    );
+  }
+
+  static malformedResponse() {
+    return new DataTrackSchemaStorageError(
+      'Schema storage response is malformed',
+      DataTrackSchemaStorageErrorReason.MalformedResponse,
+    );
+  }
+
+  static invalidDefinition(options?: { cause?: unknown }) {
+    return new DataTrackSchemaStorageError(
+      'Schema definition is not valid UTF-8',
+      DataTrackSchemaStorageErrorReason.InvalidDefinition,
+      options,
+    );
+  }
+
+  // NOTE: this was introduced by web / there isn't a corresponding case in the rust version.
+  static cancelled() {
+    return new DataTrackSchemaStorageError(
+      'Schema storage request cancelled by caller',
+      DataTrackSchemaStorageErrorReason.Cancelled,
+    );
+  }
+
+  static disconnected() {
+    return new DataTrackSchemaStorageError(
+      'Cannot store or retrieve a schema definition when disconnected',
+      DataTrackSchemaStorageErrorReason.Disconnected,
+    );
+  }
+}

src/room/data-track/schema.test.ts

@@ -1,4 +1,5 @@
 import {
+  DataBlobKey,
   DataTrackFrameEncoding as ProtocolDataTrackFrameEncoding,
   DataTrackSchemaEncoding as ProtocolDataTrackSchemaEncoding,
   DataTrackSchemaId as ProtocolDataTrackSchemaId,
@@ -112,6 +113,19 @@ describe('DataTrackSchemaId', () => {
     const protobuf = new ProtocolDataTrackSchemaId({ name: 'rgb' });
     expect(DataTrackSchemaId.from(protobuf)).toStrictEqual({ name: 'rgb', encoding: 'other' });
   });
+
+  it.each([
+    { title: 'well-known encoding', encoding: 'jsonSchema' as DataTrackSchemaEncoding },
+    { title: 'custom encoding', encoding: { custom: 'my_encoding' } },
+  ])('converts to a data blob key ($title)', ({ encoding }) => {
+    const schemaId: DataTrackSchemaId = { name: 'rgb', encoding };
+    const key = DataTrackSchemaId.toDataBlobKey(schemaId);
+    expect(key).toBeInstanceOf(DataBlobKey);
+    expect(key.key.case).toStrictEqual('schemaId');
+    expect(DataTrackSchemaId.from(key.key.value as ProtocolDataTrackSchemaId)).toStrictEqual(
+      schemaId,
+    );
+  });
 });
 
 describe('validateSchemaMetadata', () => {

src/room/data-track/schema.ts

@@ -1,4 +1,5 @@
 import {
+  DataBlobKey,
   DataTrackFrameEncoding as ProtocolDataTrackFrameEncoding,
   DataTrackSchemaEncoding as ProtocolDataTrackSchemaEncoding,
   DataTrackSchemaId as ProtocolDataTrackSchemaId,
@@ -232,6 +233,12 @@ export const DataTrackSchemaId = {
       encoding: DataTrackSchemaEncoding.toProtobuf(schemaId.encoding),
     });
   },
+  /** Key under which the schema's definition is stored as a data blob. */
+  toDataBlobKey(schemaId: DataTrackSchemaId): DataBlobKey {
+    return new DataBlobKey({
+      key: { case: 'schemaId', value: DataTrackSchemaId.toProtobuf(schemaId) },
+    });
+  },
 };
 
 export enum DataTrackSchemaErrorReason {

src/room/events.ts

@@ -630,6 +630,8 @@ export enum EngineEvent {
   UnPublishDataTrackResponse = 'unPublishDataTrackResponse',
   DataTrackSubscriberHandles = 'dataTrackSubscriberHandles',
   DataTrackPacketReceived = 'dataTrackPacketReceived',
+  StoreDataBlobResponse = 'storeDataBlobResponse',
+  GetDataBlobResponse = 'getDataBlobResponse',
   Joined = 'joined',
   TokenRefreshed = 'tokenRefreshed',
   ServerRegionsReported = 'serverRegionsReported',