From b745977058a61584f4fe6901a0aa2a22a9f9f61e Mon Sep 17 00:00:00 2001 From: Milosz Filimowski Date: Wed, 5 Aug 2026 16:16:51 +0200 Subject: [PATCH] tsunami: TrackDeviceController + TrackPublisher The camera/microphone controller ported from react-client's useTrackManager/useDeviceManager/useTrackMiddleware, preserving the pinned invariants: ideal-vs-exact deviceId constraints, enabled=false reapplied to fresh tracks while muted, selectDevice without a live track only records the selection, stopDevice performs no signalling call, pause/resume via replaceTrack + paused metadata, join auto-publish with TrackTypeError swallowed, and local-to-remote track-id resolution awaiting an in-flight addTrack. TrackPublisher is the narrow signalling adapter that keeps the controller free of Peer generics; LocalDeviceState is the controller's snapshot shape (ClientState wiring lands with the orchestrator). --- .../src/controllers/TrackDeviceController.ts | 393 ++++++++++++++++++ .../tsunami/src/controllers/TrackPublisher.ts | 28 ++ packages/tsunami/src/state/clientState.ts | 16 + packages/tsunami/src/tracks/trackUtils.ts | 51 +++ 4 files changed, 488 insertions(+) create mode 100644 packages/tsunami/src/controllers/TrackDeviceController.ts create mode 100644 packages/tsunami/src/controllers/TrackPublisher.ts create mode 100644 packages/tsunami/src/tracks/trackUtils.ts diff --git a/packages/tsunami/src/controllers/TrackDeviceController.ts b/packages/tsunami/src/controllers/TrackDeviceController.ts new file mode 100644 index 00000000..04e11437 --- /dev/null +++ b/packages/tsunami/src/controllers/TrackDeviceController.ts @@ -0,0 +1,393 @@ +import { type Logger, type TrackMetadata, TrackTypeError, Variant } from "@fishjam-cloud/ts-client"; + +import type { + DeviceItem, + DeviceType, + IDeviceManager, + PlatformMediaStream, + PlatformMediaStreamTrack, +} from "../devices/deviceManager"; +import { DeviceError, UnknownDeviceError } from "../devices/errors"; +import type { BandwidthLimits, StreamConfig, TrackMiddleware } from "../mediaTypes"; +import type { LocalDeviceState, PeerStatus } from "../state/clientState"; +import { getConfigAndBandwidthFromProps, getTrackFromStream, stopStream } from "../tracks/trackUtils"; +import type { TrackPublisher } from "./TrackPublisher"; + +export type TrackDeviceControllerDeps = { + type: DeviceType; + publisher: TrackPublisher; + deviceManager: IDeviceManager; + constraints: MediaTrackConstraints | boolean | undefined; + bandwidthLimits: BandwidthLimits; + streamConfig?: StreamConfig; + logger: Logger; + getPeerStatus: () => PeerStatus; + /** Devices of this controller's kind, from the orchestrator's last enumeration. */ + getAvailableDevices: () => DeviceItem[]; + getInitialStream: () => Promise; + /** Called when this controller replaces its stream, so the shared initial stream stops being reused. */ + invalidateInitialStream: () => void; + onSelectedDeviceChanged: (device: DeviceItem) => void; + onStateChanged: () => void; +}; + +const DEFAULT_SENT_QUALITIES: Variant[] = [Variant.VARIANT_LOW, Variant.VARIANT_MEDIUM, Variant.VARIANT_HIGH]; + +/** + * Owns the full lifecycle of one local media device (camera or microphone): + * acquisition, soft mute, middleware, device switching, and publishing the + * resulting track to the room. + */ +export class TrackDeviceController { + public error: DeviceError | null = null; + + private stream: PlatformMediaStream | null = null; + private processedTrack: PlatformMediaStreamTrack | null = null; + private middleware: TrackMiddleware = null; + private middlewareCleanup: (() => void) | null = null; + private isEnabled = true; + private selectedDevice: DeviceItem | null = null; + + private currentTrackId: string | null = null; + private connectionPromise: Promise | null = null; + + private trackEndCleanup: (() => void) | null = null; + private readonly sessionCleanups: (() => void)[]; + private lastSnapshot: LocalDeviceState | null = null; + + public constructor(private readonly deps: TrackDeviceControllerDeps) { + this.sessionCleanups = [ + deps.publisher.onJoined(() => this.handleJoined()), + deps.publisher.onDisconnected(() => this.handleDisconnected()), + ]; + } + + public get rawTrack(): PlatformMediaStreamTrack | null { + return this.stream && getTrackFromStream(this.stream, this.deps.type); + } + + public get deviceTrack(): PlatformMediaStreamTrack | null { + return this.processedTrack ?? this.rawTrack; + } + + public snapshot(): LocalDeviceState { + const rawDeviceId = this.rawTrack?.getSettings().deviceId; + const next: LocalDeviceState = { + track: this.deviceTrack, + stream: this.stream, + isEnabled: this.isEnabled, + activeDevice: (rawDeviceId && this.deps.getAvailableDevices().find((d) => d.deviceId === rawDeviceId)) || null, + selectedDevice: this.selectedDevice, + middleware: this.middleware, + }; + + const previous = this.lastSnapshot; + const isUnchanged = + previous && + previous.track === next.track && + previous.stream === next.stream && + previous.isEnabled === next.isEnabled && + previous.activeDevice === next.activeDevice && + previous.selectedDevice === next.selectedDevice && + previous.middleware === next.middleware; + if (isUnchanged) return previous; + + this.lastSnapshot = next; + return next; + } + + /** Adopts the stream produced by `initializeDevices` without invalidating it. */ + public adoptInitialStream(stream: PlatformMediaStream | null): void { + if (!stream || this.stream) return; + this.setStream(stream); + this.notify(); + } + + public setSelectedDevice(device: DeviceItem): void { + this.selectedDevice = device; + this.deps.onSelectedDeviceChanged(device); + this.notify(); + } + + public setError(error: DeviceError | null): void { + this.error = error; + this.notify(); + } + + public async startDevice( + deviceId: string | undefined = this.selectedDevice?.deviceId, + ): Promise<[PlatformMediaStreamTrack, null] | [null, DeviceError]> { + const initialStream = await this.deps.getInitialStream(); + + const initialTrack = initialStream && getTrackFromStream(initialStream, this.deps.type); + const isUsingDesiredDevice = !deviceId || deviceId === initialTrack?.getSettings().deviceId; + + if (initialTrack?.enabled && isUsingDesiredDevice) { + if (!this.stream) { + this.setStream(initialStream); + this.notify(); + } + return [initialTrack, null]; + } + + try { + const baseConstraints = typeof this.deps.constraints === "object" ? { ...this.deps.constraints } : {}; + if (deviceId) baseConstraints.deviceId = { exact: deviceId }; + const stream = await this.deps.deviceManager.getUserMedia({ [this.deps.type]: baseConstraints }); + + if (this.stream) stopStream(this.stream, this.deps.type); + this.deps.invalidateInitialStream(); + this.setStream(stream); + + const retrievedTrack = getTrackFromStream(stream, this.deps.type); + if (!retrievedTrack) throw new Error(`getUserMedia returned no ${this.deps.type} track`); + + const retrievedDeviceId = retrievedTrack.getSettings().deviceId; + if (retrievedDeviceId) { + const device = this.deps.getAvailableDevices().find((d) => d.deviceId === retrievedDeviceId); + if (device) this.setSelectedDevice(device); + } + + if (!this.isEnabled) retrievedTrack.enabled = false; + + this.notify(); + return [retrievedTrack, null]; + } catch (err) { + const parsedError = err instanceof DeviceError ? err : new UnknownDeviceError({ cause: err }); + this.error = parsedError; + this.notify(); + return [null, parsedError]; + } + } + + public stopDevice(): void { + if (this.stream) stopStream(this.stream, this.deps.type); + this.deps.invalidateInitialStream(); + this.setStream(null); + this.notify(); + } + + public enableDevice(): void { + const track = this.deviceTrack; + if (!track) return; + track.enabled = true; + this.isEnabled = true; + this.notify(); + } + + public disableDevice(): void { + const track = this.deviceTrack; + if (!track) return; + track.enabled = false; + this.isEnabled = false; + this.notify(); + } + + public async applyMiddleware(newMiddleware: TrackMiddleware): Promise { + this.middlewareCleanup?.(); + this.middlewareCleanup = null; + this.middleware = newMiddleware; + + const rawTrack = this.rawTrack; + if (newMiddleware && rawTrack) { + const { track, onClear } = await newMiddleware(rawTrack); + this.middlewareCleanup = onClear ?? null; + this.processedTrack = track; + this.notify(); + return track; + } + + this.processedTrack = null; + this.notify(); + return rawTrack; + } + + // --- streaming layer (ties the local device to the room) --- + + public async start(deviceId?: string): Promise { + const [track, error] = await this.startDevice(deviceId); + if (error) return error; + + const currentTrackId = await this.getCurrentTrackId(); + if (currentTrackId) { + await this.resumeStreaming(currentTrackId, track); + } else if (this.deps.getPeerStatus() === "connected") { + await this.startStreaming(track); + } + return undefined; + } + + public async stop(): Promise { + const currentTrackId = await this.getCurrentTrackId(); + this.stopDevice(); + if (currentTrackId) await this.pauseStreaming(currentTrackId); + } + + public async toggleDevice(): Promise { + const currentTrackId = await this.getCurrentTrackId(); + if (this.deviceTrack) { + this.stopDevice(); + if (currentTrackId) await this.pauseStreaming(currentTrackId); + return undefined; + } + + const [newTrack, error] = await this.startDevice(); + if (error) return error; + + if (currentTrackId) { + await this.resumeStreaming(currentTrackId, newTrack); + } else if (this.deps.getPeerStatus() === "connected") { + await this.startStreaming(newTrack); + } + return undefined; + } + + public async toggleMute(): Promise { + const currentTrackId = await this.getCurrentTrackId(); + const isTrackCurrentlyEnabled = Boolean(this.deviceTrack?.enabled); + if (!currentTrackId) { + this.deps.logger.warn("Toggling mute is only possible while connected to a room."); + return; + } + + if (isTrackCurrentlyEnabled) { + this.disableDevice(); + await this.pauseStreaming(currentTrackId); + } else if (this.deviceTrack) { + this.enableDevice(); + await this.resumeStreaming(currentTrackId, this.deviceTrack); + } + } + + public async selectDevice(deviceId: string): Promise { + if (!this.deviceTrack) { + const device = this.deps.getAvailableDevices().find((d) => d.deviceId === deviceId); + if (device) this.setSelectedDevice(device); + return undefined; + } + + const [newTrack, error] = await this.startDevice(deviceId); + if (error) return error; + + const currentTrackId = await this.getCurrentTrackId(); + if (!currentTrackId) return undefined; + + await this.deps.publisher.replaceTrack(currentTrackId, newTrack); + return undefined; + } + + public async setTrackMiddleware(middleware: TrackMiddleware): Promise { + const processedTrack = await this.applyMiddleware(middleware); + + const currentTrackId = await this.getCurrentTrackId(); + if (!currentTrackId) return; + + await this.deps.publisher.replaceTrack(currentTrackId, processedTrack); + } + + public dispose(): void { + for (const cleanup of this.sessionCleanups) cleanup(); + this.middlewareCleanup?.(); + this.middlewareCleanup = null; + if (this.stream) stopStream(this.stream, this.deps.type); + this.setStream(null); + } + + private async getCurrentTrackId(): Promise { + if (this.connectionPromise) { + await this.connectionPromise.catch(() => undefined); + } + if (!this.currentTrackId) return null; + return this.deps.publisher.resolveRemoteTrackId(this.currentTrackId); + } + + private async startStreaming(track: PlatformMediaStreamTrack): Promise { + // temporarily setting the local trackId until we have the remoteTrackId + this.currentTrackId = track.id; + + const trackMetadata: TrackMetadata = { type: this.deps.type === "video" ? "camera" : "microphone", paused: false }; + + const displayName = this.deps.publisher.getDisplayName(); + if (displayName) trackMetadata.displayName = displayName; + + const sentQualities = this.deps.streamConfig?.sentQualities ?? DEFAULT_SENT_QUALITIES; + const [maxBandwidth, simulcastConfig] = getConfigAndBandwidthFromProps(sentQualities, this.deps.bandwidthLimits); + + try { + const addTrackJob = this.deps.publisher.addTrack(track, trackMetadata, simulcastConfig, maxBandwidth); + this.connectionPromise = addTrackJob; + this.currentTrackId = await addTrackJob; + } catch (err) { + if (err instanceof TrackTypeError) { + this.deps.logger.warn(err.message); + this.currentTrackId = null; + } + throw err; + } + } + + private async pauseStreaming(trackId: string): Promise { + if (this.deps.getPeerStatus() !== "connected") return; + await this.deps.publisher.replaceTrack(trackId, null); + this.deps.publisher.updateTrackMetadata(trackId, { + type: this.deps.type === "video" ? "camera" : "microphone", + paused: true, + }); + } + + private async resumeStreaming(trackId: string, track: PlatformMediaStreamTrack): Promise { + if (this.deps.getPeerStatus() !== "connected") return; + await this.deps.publisher.replaceTrack(trackId, track); + this.deps.publisher.updateTrackMetadata(trackId, { + type: this.deps.type === "video" ? "camera" : "microphone", + paused: false, + }); + } + + private handleJoined(): void { + const track = this.deviceTrack; + if (!track) return; + // The handler is sync; observe rejections so non-TrackTypeError failures + // from addTrack don't surface as unhandledrejection. + void this.startStreaming(track).catch((err) => { + if (err instanceof TrackTypeError) return; + this.deps.logger.error(err); + }); + } + + private handleDisconnected(): void { + this.currentTrackId = null; + this.connectionPromise = null; + } + + private setStream(stream: PlatformMediaStream | null): void { + this.trackEndCleanup?.(); + this.trackEndCleanup = null; + this.stream = stream; + + if (!stream) { + if (this.processedTrack) { + this.processedTrack.stop(); + this.middlewareCleanup?.(); + this.middlewareCleanup = null; + this.processedTrack = null; + } + return; + } + + const rawTrack = getTrackFromStream(stream, this.deps.type); + if (!rawTrack) return; + + const handleTrackEnded = () => { + if (this.rawTrack !== rawTrack) return; + this.setStream(null); + this.notify(); + }; + rawTrack.addEventListener?.("ended", handleTrackEnded); + this.trackEndCleanup = () => rawTrack.removeEventListener?.("ended", handleTrackEnded); + } + + private notify(): void { + this.deps.onStateChanged(); + } +} diff --git a/packages/tsunami/src/controllers/TrackPublisher.ts b/packages/tsunami/src/controllers/TrackPublisher.ts new file mode 100644 index 00000000..2dee2f7d --- /dev/null +++ b/packages/tsunami/src/controllers/TrackPublisher.ts @@ -0,0 +1,28 @@ +import type { SimulcastConfig, TrackBandwidthLimit, TrackMetadata } from "@fishjam-cloud/ts-client"; + +import type { PlatformMediaStreamTrack } from "../devices/deviceManager"; + +/** + * Narrow signalling surface the device controllers publish through. + * + * Controllers never talk to the signalling client directly — the client hands + * them this adapter, which also makes them trivially testable. + */ +export type TrackPublisher = { + addTrack: ( + track: PlatformMediaStreamTrack, + metadata: TrackMetadata, + simulcastConfig?: SimulcastConfig, + maxBandwidth?: TrackBandwidthLimit, + ) => Promise; + replaceTrack: (trackId: string, newTrack: PlatformMediaStreamTrack | null) => Promise; + removeTrack: (trackId: string) => Promise; + updateTrackMetadata: (trackId: string, metadata: TrackMetadata) => void; + getDisplayName: () => string | undefined; + /** Resolves a local or remote track id to the current remote track id, if the track is still published. */ + resolveRemoteTrackId: (remoteOrLocalTrackId: string) => string | null; + /** `true` while the signalling client exists and can publish tracks. */ + isSignallingActive: () => boolean; + onJoined: (listener: () => void) => () => void; + onDisconnected: (listener: () => void) => () => void; +}; diff --git a/packages/tsunami/src/state/clientState.ts b/packages/tsunami/src/state/clientState.ts index 141af25a..0f04387b 100644 --- a/packages/tsunami/src/state/clientState.ts +++ b/packages/tsunami/src/state/clientState.ts @@ -1,5 +1,8 @@ import type { Component, GenericMetadata, Peer, ReconnectionStatus } from "@fishjam-cloud/ts-client"; +import type { DeviceItem, PlatformMediaStream, PlatformMediaStreamTrack } from "../devices/deviceManager"; +import type { TrackMiddleware } from "../mediaTypes"; + /** * Represents the possible statuses of a peer connection. * @@ -10,6 +13,19 @@ import type { Component, GenericMetadata, Peer, ReconnectionStatus } from "@fish */ export type PeerStatus = "connecting" | "connected" | "error" | "idle"; +export interface LocalDeviceState { + /** Track ready to be rendered or published (post-middleware when one is set). */ + track: PlatformMediaStreamTrack | null; + stream: PlatformMediaStream | null; + /** Soft mute flag — `false` while the track is disabled but the device stays on. */ + isEnabled: boolean; + /** Device backing the current track. */ + activeDevice: DeviceItem | null; + /** Device that will be used on the next start. */ + selectedDevice: DeviceItem | null; + middleware: TrackMiddleware; +} + /** * Flat, synchronously readable snapshot of the client's observable state. * diff --git a/packages/tsunami/src/tracks/trackUtils.ts b/packages/tsunami/src/tracks/trackUtils.ts new file mode 100644 index 00000000..ae8f4b93 --- /dev/null +++ b/packages/tsunami/src/tracks/trackUtils.ts @@ -0,0 +1,51 @@ +import type { SimulcastConfig, TrackMetadata } from "@fishjam-cloud/ts-client"; +import { Variant } from "@fishjam-cloud/ts-client"; + +import type { PlatformMediaStream, PlatformMediaStreamTrack } from "../devices/deviceManager"; +import type { BandwidthLimits } from "../mediaTypes"; + +const ALL_VARIANTS: Variant[] = [Variant.VARIANT_LOW, Variant.VARIANT_MEDIUM, Variant.VARIANT_HIGH]; + +const getDisabledEncodings = (enabledVariants: Variant[]): Variant[] => + ALL_VARIANTS.filter((variant) => !enabledVariants.includes(variant)); + +export const getConfigAndBandwidthFromProps = ( + encodings: Variant[] | false | undefined, + bandwidthLimits: BandwidthLimits, +): readonly [number | Map, SimulcastConfig | undefined] => { + if (!encodings) return [bandwidthLimits.singleStream, undefined] as const; + + const config: SimulcastConfig = { + enabled: true, + enabledVariants: encodings, + disabledVariants: getDisabledEncodings(encodings), + }; + + const variantEntries = Object.entries(bandwidthLimits.simulcast).map( + ([key, value]) => [Number(key), value] as [Variant, number], + ); + + const bandwidth = new Map(variantEntries); + return [bandwidth, config] as const; +}; + +function getCertainTypeTracks(stream: PlatformMediaStream, type: "audio" | "video") { + if (type === "audio") return stream.getAudioTracks(); + return stream.getVideoTracks(); +} + +export function getTrackFromStream( + stream: PlatformMediaStream, + type: "audio" | "video", +): PlatformMediaStreamTrack | null { + return getCertainTypeTracks(stream, type)[0] ?? null; +} + +export function stopStream(stream: PlatformMediaStream, type: "audio" | "video"): void { + getCertainTypeTracks(stream, type).forEach((track) => { + track.enabled = false; + track.stop(); + }); +} + +export type { TrackMetadata };