Add typeguards for frame processors in order to avoid dual package hazard

.changeset/dry-buttons-help.md

@@ -0,0 +1,5 @@
+---
+"@livekit/rtc-node": patch
+---
+
+ensure userdata is passed through when resampling

.changeset/eleven-timers-kick.md

@@ -0,0 +1,5 @@
+---
+"@livekit/rtc-node": minor
+---
+
+Add typeguards for frame processors in order to avoid dual package hazard

packages/livekit-rtc/src/audio_frame.ts

@@ -32,9 +32,14 @@ export class AudioFrame {
     this._userdata = userdata;
   }
 
-  static create(sampleRate: number, channels: number, samplesPerChannel: number): AudioFrame {
+  static create(
+    sampleRate: number,
+    channels: number,
+    samplesPerChannel: number,
+    userdata?: Record<string, unknown>,
+  ): AudioFrame {
     const data = new Int16Array(channels * samplesPerChannel);
-    return new AudioFrame(data, sampleRate, channels, samplesPerChannel);
+    return new AudioFrame(data, sampleRate, channels, samplesPerChannel, userdata);
   }
 
   /** @internal */
@@ -103,5 +108,12 @@ export const combineAudioFrames = (buffer: AudioFrame | AudioFrame[]): AudioFram
   }
 
   const data = new Int16Array(buffer.map((x) => [...x.data]).flat());
-  return new AudioFrame(data, sampleRate, channels, totalSamplesPerChannel);
+
+  // Merge userdata from all frames
+  const mergedUserdata: Record<string, unknown> = {};
+  for (const frame of buffer) {
+    Object.assign(mergedUserdata, frame.userdata);
+  }
+
+  return new AudioFrame(data, sampleRate, channels, totalSamplesPerChannel, mergedUserdata);
 };

packages/livekit-rtc/src/audio_resampler.ts

@@ -131,12 +131,14 @@ export class AudioResampler {
     }
 
     const outputData = FfiClient.instance.copyBuffer(res.outputPtr, res.size!);
+
     return [
       new AudioFrame(
         new Int16Array(outputData.buffer),
         this.#outputRate,
         this.#channels,
         Math.trunc(outputData.length / this.#channels / 2),
+        data.userdata,
       ),
     ];
   }

packages/livekit-rtc/src/audio_stream.ts

@@ -5,7 +5,7 @@ import type { UnderlyingSource } from 'node:stream/web';
 import { AudioFrame } from './audio_frame.js';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
-import { FrameProcessor } from './frame_processor.js';
+import { type FrameProcessor, isAudioFrameProcessor } from './frame_processor.js';
 import { log } from './log.js';
 import type { NewAudioStreamResponse } from './proto/audio_frame_pb.js';
 import { AudioStreamType, NewAudioStreamRequest } from './proto/audio_frame_pb.js';
@@ -41,7 +41,7 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
     if (sampleRateOrOptions !== undefined && typeof sampleRateOrOptions !== 'number') {
       this.sampleRate = sampleRateOrOptions.sampleRate ?? 48000;
       this.numChannels = sampleRateOrOptions.numChannels ?? 1;
-      if (sampleRateOrOptions.noiseCancellation instanceof FrameProcessor) {
+      if (isAudioFrameProcessor(sampleRateOrOptions.noiseCancellation)) {
         this.frameProcessor = sampleRateOrOptions.noiseCancellation;
       } else {
         this.legacyNcOptions = sampleRateOrOptions.noiseCancellation;

packages/livekit-rtc/src/frame_processor.ts

@@ -15,7 +15,42 @@ export type FrameProcessorCredentials = {
   url: string;
 };
 
+export const FrameProcessorSymbol = Symbol.for('lk.frame-processor');
+
+export function isFrameProcessor<Type extends 'audio' | 'video'>(
+  maybeProcessor: unknown,
+  type?: Type,
+): maybeProcessor is FrameProcessor<
+  Type extends 'audio' ? AudioFrame : Type extends 'video' ? VideoFrame : AudioFrame | VideoFrame
+> {
+  return (
+    maybeProcessor !== null &&
+    typeof maybeProcessor === 'object' &&
+    'symbol' in maybeProcessor &&
+    maybeProcessor.symbol === FrameProcessorSymbol &&
+    (!type || ('type' in maybeProcessor && maybeProcessor.type === type))
+  );
+}
+
+export function isAudioFrameProcessor(
+  maybeProcessor: unknown,
+): maybeProcessor is FrameProcessor<AudioFrame> {
+  return isFrameProcessor(maybeProcessor, 'audio');
+}
+
+export function isVideoFrameProcessor(
+  maybeProcessor: unknown,
+): maybeProcessor is FrameProcessor<VideoFrame> {
+  return isFrameProcessor(maybeProcessor, 'video');
+}
+
 export abstract class FrameProcessor<Frame extends VideoFrame | AudioFrame> {
+  readonly symbol = FrameProcessorSymbol;
+  abstract readonly type: Frame extends VideoFrame
+    ? 'video'
+    : Frame extends AudioFrame
+      ? 'audio'
+      : never;
   abstract isEnabled(): boolean;
   abstract setEnabled(enabled: boolean): void;