{"version":3,"sources":["../../../src/voice/room_io/_input.ts"],"sourcesContent":["// SPDX-FileCopyrightText: 2025 LiveKit, Inc.\n//\n// SPDX-License-Identifier: Apache-2.0\nimport {\n  type AudioFrame,\n  AudioStream,\n  type FrameProcessor,\n  type NoiseCancellationOptions,\n  RemoteParticipant,\n  type RemoteTrack,\n  type RemoteTrackPublication,\n  type Room,\n  RoomEvent,\n  TrackSource,\n  isFrameProcessor,\n} from '@livekit/rtc-node';\nimport { type ReadableStream, TransformStream } from 'node:stream/web';\nimport { log } from '../../log.js';\nimport { resampleStream } from '../../utils.js';\nimport { AudioInput } from '../io.js';\n\nexport class ParticipantAudioInputStream extends AudioInput {\n  private room: Room;\n  private sampleRate: number;\n  private numChannels: number;\n  private noiseCancellation?: NoiseCancellationOptions;\n  private frameProcessor?: FrameProcessor<AudioFrame>;\n  private publication: RemoteTrackPublication | null = null;\n  private track: RemoteTrack | null = null;\n  private participantIdentity: string | null = null;\n  private currentInput: {\n    id: string;\n    stream: ReadableStream<AudioFrame>;\n    pipe: Promise<void>;\n  } | null = null;\n  private streamTransition: Promise<void> | null = null;\n  private attached = true;\n  private closed = false;\n  private logger = log();\n\n  constructor({\n    room,\n    sampleRate,\n    numChannels,\n    noiseCancellation,\n  }: {\n    room: Room;\n    sampleRate: number;\n    numChannels: number;\n    noiseCancellation?: NoiseCancellationOptions | FrameProcessor<AudioFrame>;\n  }) {\n    super();\n    this.room = room;\n    this.sampleRate = sampleRate;\n    this.numChannels = numChannels;\n    if (isFrameProcessor<FrameProcessor<AudioFrame>>(noiseCancellation)) {\n      this.frameProcessor = noiseCancellation;\n    } else {\n      this.noiseCancellation = noiseCancellation;\n    }\n\n    this.room.on(RoomEvent.TrackSubscribed, this.onTrackSubscribed);\n    this.room.on(RoomEvent.TrackUnsubscribed, this.onTrackUnsubscribed);\n    this.room.on(RoomEvent.TrackUnpublished, this.onTrackUnpublished);\n  }\n\n  setParticipant(participant: RemoteParticipant | string | null) {\n    const participantIdentity =\n      participant instanceof RemoteParticipant ? participant.identity : participant;\n    this.logger.debug(\n      { 'lk.pii.participant_identity': participantIdentity },\n      'setting participant audio input',\n    );\n\n    if (this.participantIdentity === participantIdentity) {\n      return;\n    }\n    if (this.participantIdentity) {\n      this.closeStream();\n    }\n    this.participantIdentity = participantIdentity;\n\n    if (!participantIdentity) {\n      return;\n    }\n\n    const participantValue =\n      participant instanceof RemoteParticipant\n        ? participant\n        : this.room.remoteParticipants.get(participantIdentity);\n\n    // Convert Map iterator to array for Pino serialization\n    const trackPublicationsArray = Array.from(participantValue?.trackPublications.values() ?? []);\n\n    this.logger.info(\n      {\n        'lk.pii.participant_identity': participantValue?.identity,\n        trackPublications: trackPublicationsArray,\n        lengthOfTrackPublications: trackPublicationsArray.length,\n      },\n      'participantValue.trackPublications',\n    );\n    // We need to check if the participant has a microphone track and subscribe to it\n    // in case we miss the tracksubscribed event\n    if (participantValue) {\n      for (const publication of participantValue.trackPublications.values()) {\n        if (publication.track && publication.source === TrackSource.SOURCE_MICROPHONE) {\n          this.onTrackSubscribed(publication.track, publication, participantValue);\n          break;\n        }\n      }\n    }\n  }\n\n  override setAttached(attached: boolean): void {\n    this.attached = attached;\n  }\n\n  override onAttached(): void {\n    this.logger.debug(\n      { 'lk.pii.participant_identity': this.participantIdentity },\n      'input stream attached',\n    );\n  }\n\n  override onDetached(): void {\n    this.logger.debug(\n      { 'lk.pii.participant_identity': this.participantIdentity },\n      'input stream detached',\n    );\n  }\n\n  private onTrackUnpublished = (\n    publication: RemoteTrackPublication,\n    participant: RemoteParticipant,\n  ) => {\n    if (\n      this.publication?.sid !== publication.sid ||\n      participant.identity !== this.participantIdentity\n    ) {\n      return;\n    }\n    this.closeStream();\n\n    // subscribe to the first available track\n    for (const publication of participant.trackPublications.values()) {\n      if (\n        publication.track &&\n        this.onTrackSubscribed(publication.track, publication, participant)\n      ) {\n        return;\n      }\n    }\n  };\n\n  private closeStream() {\n    this.updateStream(null, null);\n  }\n\n  private updateStream(track: RemoteTrack | null, publication: RemoteTrackPublication | null) {\n    this.track = track;\n    this.publication = publication;\n\n    if (track && publication && !this.streamTransition && !this.currentInput) {\n      this.openStream(track);\n      return;\n    }\n\n    const previousTransition = this.streamTransition ?? Promise.resolve();\n    const transition = previousTransition.then(async () => {\n      try {\n        await this.closeCurrentInput();\n        if (\n          !track ||\n          !publication ||\n          this.closed ||\n          this.track !== track ||\n          this.publication !== publication\n        ) {\n          return;\n        }\n        this.openStream(track);\n      } catch {\n        this.logger.error('failed to update participant audio input');\n      }\n    });\n    this.streamTransition = transition;\n    void transition.then(() => {\n      if (this.streamTransition === transition) {\n        this.streamTransition = null;\n      }\n    });\n  }\n\n  private async closeCurrentInput() {\n    const input = this.currentInput;\n    this.currentInput = null;\n\n    if (input) {\n      try {\n        await this.multiStream.removeInputStream(input.id);\n      } catch {\n        this.logger.warn('failed to remove participant audio input stream');\n      }\n\n      const [cancelResult] = await Promise.allSettled([input.stream.cancel(), input.pipe]);\n      if (cancelResult.status === 'rejected') {\n        this.logger.warn('failed to cancel participant audio input stream');\n      }\n    }\n  }\n\n  private openStream(track: RemoteTrack) {\n    const output = new TransformStream<AudioFrame, AudioFrame>({\n      transform: (frame, controller) => {\n        if (this.attached) {\n          controller.enqueue(frame);\n        }\n      },\n    });\n    const inputPipe = resampleStream({\n      stream: this.createStream(track),\n      outputRate: this.sampleRate,\n    }).pipeTo(output.writable);\n    const input = {\n      id: this.multiStream.addInputStream(output.readable),\n      stream: output.readable,\n      pipe: inputPipe,\n    };\n    this.currentInput = input;\n    void inputPipe.catch(() => {\n      if (this.currentInput === input) {\n        this.logger.error('participant audio input stream failed');\n      }\n    });\n  }\n\n  private onTrackSubscribed = (\n    track: RemoteTrack,\n    publication: RemoteTrackPublication,\n    participant: RemoteParticipant,\n  ): boolean => {\n    this.logger.debug(\n      { 'lk.pii.participant_identity': participant.identity },\n      'onTrackSubscribed in _input',\n    );\n    if (\n      this.closed ||\n      this.participantIdentity !== participant.identity ||\n      publication.source !== TrackSource.SOURCE_MICROPHONE ||\n      (this.publication?.sid === publication.sid && this.track === track)\n    ) {\n      return false;\n    }\n    this.updateStream(track, publication);\n    return true;\n  };\n\n  private onTrackUnsubscribed = (\n    track: RemoteTrack,\n    publication: RemoteTrackPublication,\n    participant: RemoteParticipant,\n  ) => {\n    if (\n      this.track !== track ||\n      this.publication?.sid !== publication.sid ||\n      participant.identity !== this.participantIdentity\n    ) {\n      return;\n    }\n\n    this.closeStream();\n\n    // Same-publication replacements arrive through TrackSubscribed.\n    for (const candidate of participant.trackPublications.values()) {\n      if (candidate.sid === publication.sid || !candidate.track) {\n        continue;\n      }\n      if (this.onTrackSubscribed(candidate.track, candidate, participant)) {\n        return;\n      }\n    }\n  };\n\n  private createStream(track: RemoteTrack): ReadableStream<AudioFrame> {\n    return new AudioStream(track, {\n      sampleRate: this.sampleRate,\n      numChannels: this.numChannels,\n      noiseCancellation: this.frameProcessor || this.noiseCancellation,\n      // Don't let the AudioStream close the processor when the track switches —\n      // this input stream owns the processor across track changes and closes it\n      // itself in close().\n      autoCloseNoiseCancellation: false,\n      // TODO(AJS-269): resolve compatibility issue with node-sdk to remove the forced type casting\n    }) as unknown as ReadableStream<AudioFrame>;\n  }\n\n  override async close() {\n    if (this.closed) {\n      return;\n    }\n    this.closed = true;\n\n    this.room.off(RoomEvent.TrackSubscribed, this.onTrackSubscribed);\n    this.room.off(RoomEvent.TrackUnsubscribed, this.onTrackUnsubscribed);\n    this.room.off(RoomEvent.TrackUnpublished, this.onTrackUnpublished);\n    this.closeStream();\n    await this.streamTransition;\n    await super.close();\n\n    this.frameProcessor?.close();\n    this.frameProcessor = undefined;\n  }\n}\n"],"mappings":";;;;;;;;;;;;;;;;;;AAAA;AAAA;AAAA;AAAA;AAAA;AAGA,sBAYO;AACP,iBAAqD;AACrD,iBAAoB;AACpB,mBAA+B;AAC/B,gBAA2B;AAEpB,MAAM,oCAAoC,qBAAW;AAAA,EAClD;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA;AAAA,EACA,cAA6C;AAAA,EAC7C,QAA4B;AAAA,EAC5B,sBAAqC;AAAA,EACrC,eAIG;AAAA,EACH,mBAAyC;AAAA,EACzC,WAAW;AAAA,EACX,SAAS;AAAA,EACT,aAAS,gBAAI;AAAA,EAErB,YAAY;AAAA,IACV;AAAA,IACA;AAAA,IACA;AAAA,IACA;AAAA,EACF,GAKG;AACD,UAAM;AACN,SAAK,OAAO;AACZ,SAAK,aAAa;AAClB,SAAK,cAAc;AACnB,YAAI,kCAA6C,iBAAiB,GAAG;AACnE,WAAK,iBAAiB;AAAA,IACxB,OAAO;AACL,WAAK,oBAAoB;AAAA,IAC3B;AAEA,SAAK,KAAK,GAAG,0BAAU,iBAAiB,KAAK,iBAAiB;AAC9D,SAAK,KAAK,GAAG,0BAAU,mBAAmB,KAAK,mBAAmB;AAClE,SAAK,KAAK,GAAG,0BAAU,kBAAkB,KAAK,kBAAkB;AAAA,EAClE;AAAA,EAEA,eAAe,aAAgD;AAC7D,UAAM,sBACJ,uBAAuB,oCAAoB,YAAY,WAAW;AACpE,SAAK,OAAO;AAAA,MACV,EAAE,+BAA+B,oBAAoB;AAAA,MACrD;AAAA,IACF;AAEA,QAAI,KAAK,wBAAwB,qBAAqB;AACpD;AAAA,IACF;AACA,QAAI,KAAK,qBAAqB;AAC5B,WAAK,YAAY;AAAA,IACnB;AACA,SAAK,sBAAsB;AAE3B,QAAI,CAAC,qBAAqB;AACxB;AAAA,IACF;AAEA,UAAM,mBACJ,uBAAuB,oCACnB,cACA,KAAK,KAAK,mBAAmB,IAAI,mBAAmB;AAG1D,UAAM,yBAAyB,MAAM,MAAK,qDAAkB,kBAAkB,aAAY,CAAC,CAAC;AAE5F,SAAK,OAAO;AAAA,MACV;AAAA,QACE,+BAA+B,qDAAkB;AAAA,QACjD,mBAAmB;AAAA,QACnB,2BAA2B,uBAAuB;AAAA,MACpD;AAAA,MACA;AAAA,IACF;AAGA,QAAI,kBAAkB;AACpB,iBAAW,eAAe,iBAAiB,kBAAkB,OAAO,GAAG;AACrE,YAAI,YAAY,SAAS,YAAY,WAAW,4BAAY,mBAAmB;AAC7E,eAAK,kBAAkB,YAAY,OAAO,aAAa,gBAAgB;AACvE;AAAA,QACF;AAAA,MACF;AAAA,IACF;AAAA,EACF;AAAA,EAES,YAAY,UAAyB;AAC5C,SAAK,WAAW;AAAA,EAClB;AAAA,EAES,aAAmB;AAC1B,SAAK,OAAO;AAAA,MACV,EAAE,+BAA+B,KAAK,oBAAoB;AAAA,MAC1D;AAAA,IACF;AAAA,EACF;AAAA,EAES,aAAmB;AAC1B,SAAK,OAAO;AAAA,MACV,EAAE,+BAA+B,KAAK,oBAAoB;AAAA,MAC1D;AAAA,IACF;AAAA,EACF;AAAA,EAEQ,qBAAqB,CAC3B,aACA,gBACG;AAvIP;AAwII,UACE,UAAK,gBAAL,mBAAkB,SAAQ,YAAY,OACtC,YAAY,aAAa,KAAK,qBAC9B;AACA;AAAA,IACF;AACA,SAAK,YAAY;AAGjB,eAAWA,gBAAe,YAAY,kBAAkB,OAAO,GAAG;AAChE,UACEA,aAAY,SACZ,KAAK,kBAAkBA,aAAY,OAAOA,cAAa,WAAW,GAClE;AACA;AAAA,MACF;AAAA,IACF;AAAA,EACF;AAAA,EAEQ,cAAc;AACpB,SAAK,aAAa,MAAM,IAAI;AAAA,EAC9B;AAAA,EAEQ,aAAa,OAA2B,aAA4C;AAC1F,SAAK,QAAQ;AACb,SAAK,cAAc;AAEnB,QAAI,SAAS,eAAe,CAAC,KAAK,oBAAoB,CAAC,KAAK,cAAc;AACxE,WAAK,WAAW,KAAK;AACrB;AAAA,IACF;AAEA,UAAM,qBAAqB,KAAK,oBAAoB,QAAQ,QAAQ;AACpE,UAAM,aAAa,mBAAmB,KAAK,YAAY;AACrD,UAAI;AACF,cAAM,KAAK,kBAAkB;AAC7B,YACE,CAAC,SACD,CAAC,eACD,KAAK,UACL,KAAK,UAAU,SACf,KAAK,gBAAgB,aACrB;AACA;AAAA,QACF;AACA,aAAK,WAAW,KAAK;AAAA,MACvB,QAAQ;AACN,aAAK,OAAO,MAAM,0CAA0C;AAAA,MAC9D;AAAA,IACF,CAAC;AACD,SAAK,mBAAmB;AACxB,SAAK,WAAW,KAAK,MAAM;AACzB,UAAI,KAAK,qBAAqB,YAAY;AACxC,aAAK,mBAAmB;AAAA,MAC1B;AAAA,IACF,CAAC;AAAA,EACH;AAAA,EAEA,MAAc,oBAAoB;AAChC,UAAM,QAAQ,KAAK;AACnB,SAAK,eAAe;AAEpB,QAAI,OAAO;AACT,UAAI;AACF,cAAM,KAAK,YAAY,kBAAkB,MAAM,EAAE;AAAA,MACnD,QAAQ;AACN,aAAK,OAAO,KAAK,iDAAiD;AAAA,MACpE;AAEA,YAAM,CAAC,YAAY,IAAI,MAAM,QAAQ,WAAW,CAAC,MAAM,OAAO,OAAO,GAAG,MAAM,IAAI,CAAC;AACnF,UAAI,aAAa,WAAW,YAAY;AACtC,aAAK,OAAO,KAAK,iDAAiD;AAAA,MACpE;AAAA,IACF;AAAA,EACF;AAAA,EAEQ,WAAW,OAAoB;AACrC,UAAM,SAAS,IAAI,2BAAwC;AAAA,MACzD,WAAW,CAAC,OAAO,eAAe;AAChC,YAAI,KAAK,UAAU;AACjB,qBAAW,QAAQ,KAAK;AAAA,QAC1B;AAAA,MACF;AAAA,IACF,CAAC;AACD,UAAM,gBAAY,6BAAe;AAAA,MAC/B,QAAQ,KAAK,aAAa,KAAK;AAAA,MAC/B,YAAY,KAAK;AAAA,IACnB,CAAC,EAAE,OAAO,OAAO,QAAQ;AACzB,UAAM,QAAQ;AAAA,MACZ,IAAI,KAAK,YAAY,eAAe,OAAO,QAAQ;AAAA,MACnD,QAAQ,OAAO;AAAA,MACf,MAAM;AAAA,IACR;AACA,SAAK,eAAe;AACpB,SAAK,UAAU,MAAM,MAAM;AACzB,UAAI,KAAK,iBAAiB,OAAO;AAC/B,aAAK,OAAO,MAAM,uCAAuC;AAAA,MAC3D;AAAA,IACF,CAAC;AAAA,EACH;AAAA,EAEQ,oBAAoB,CAC1B,OACA,aACA,gBACY;AAjPhB;AAkPI,SAAK,OAAO;AAAA,MACV,EAAE,+BAA+B,YAAY,SAAS;AAAA,MACtD;AAAA,IACF;AACA,QACE,KAAK,UACL,KAAK,wBAAwB,YAAY,YACzC,YAAY,WAAW,4BAAY,uBAClC,UAAK,gBAAL,mBAAkB,SAAQ,YAAY,OAAO,KAAK,UAAU,OAC7D;AACA,aAAO;AAAA,IACT;AACA,SAAK,aAAa,OAAO,WAAW;AACpC,WAAO;AAAA,EACT;AAAA,EAEQ,sBAAsB,CAC5B,OACA,aACA,gBACG;AAtQP;AAuQI,QACE,KAAK,UAAU,WACf,UAAK,gBAAL,mBAAkB,SAAQ,YAAY,OACtC,YAAY,aAAa,KAAK,qBAC9B;AACA;AAAA,IACF;AAEA,SAAK,YAAY;AAGjB,eAAW,aAAa,YAAY,kBAAkB,OAAO,GAAG;AAC9D,UAAI,UAAU,QAAQ,YAAY,OAAO,CAAC,UAAU,OAAO;AACzD;AAAA,MACF;AACA,UAAI,KAAK,kBAAkB,UAAU,OAAO,WAAW,WAAW,GAAG;AACnE;AAAA,MACF;AAAA,IACF;AAAA,EACF;AAAA,EAEQ,aAAa,OAAgD;AACnE,WAAO,IAAI,4BAAY,OAAO;AAAA,MAC5B,YAAY,KAAK;AAAA,MACjB,aAAa,KAAK;AAAA,MAClB,mBAAmB,KAAK,kBAAkB,KAAK;AAAA;AAAA;AAAA;AAAA,MAI/C,4BAA4B;AAAA;AAAA,IAE9B,CAAC;AAAA,EACH;AAAA,EAEA,MAAe,QAAQ;AAzSzB;AA0SI,QAAI,KAAK,QAAQ;AACf;AAAA,IACF;AACA,SAAK,SAAS;AAEd,SAAK,KAAK,IAAI,0BAAU,iBAAiB,KAAK,iBAAiB;AAC/D,SAAK,KAAK,IAAI,0BAAU,mBAAmB,KAAK,mBAAmB;AACnE,SAAK,KAAK,IAAI,0BAAU,kBAAkB,KAAK,kBAAkB;AACjE,SAAK,YAAY;AACjB,UAAM,KAAK;AACX,UAAM,MAAM,MAAM;AAElB,eAAK,mBAAL,mBAAqB;AACrB,SAAK,iBAAiB;AAAA,EACxB;AACF;","names":["publication"]}