From 4e4ab23cdad7ba1e25ddc4f2cd699626ecb20682 Mon Sep 17 00:00:00 2001 From: Robin Date: Mon, 12 Oct 2026 10:08:22 +0200 Subject: [PATCH] Model ConnectionState as a discriminated union The enum mixed our own lifecycle states with a copy of livekit-client's connection states, and errors were smuggled through the same observable as a separate union member. A discriminated union keeps the LiveKit state nested under "authenticated", carries the error as data, and lets the local member represent the time before a connection exists as an "initializing" state instead of null. --- src/state/CallViewModel/CallViewModel.ts | 4 +- .../localMember/LocalMember.test.ts | 63 ++++++++++++---- .../CallViewModel/localMember/LocalMember.ts | 20 +++-- .../remoteMembers/Connection.test.ts | 70 ++++++++++-------- .../CallViewModel/remoteMembers/Connection.ts | 74 ++++++++----------- 5 files changed, 134 insertions(+), 97 deletions(-) diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index a58b0837f..3b457f719 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -1662,9 +1662,9 @@ export function createCallViewModel$( returnObject.publishError = value.media; else if ( typeof value.media === "object" && - value.media.connection instanceof ElementCallError + value.media.connection.state === "error" ) - returnObject.connectionError = value.media.connection; + returnObject.connectionError = value.media.connection.error; return returnObject; }), ), diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index 282759fdd..da4da45aa 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -23,7 +23,11 @@ import { } from "vitest"; import { BehaviorSubject, map, of, Subject } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; -import { type LocalParticipant, type LocalTrack } from "livekit-client"; +import { + ConnectionState as LivekitConnectionState, + type LocalParticipant, + type LocalTrack, +} from "livekit-client"; import fetchMock from "fetch-mock"; import { PosthogAnalytics } from "../../../analytics/PosthogAnalytics"; @@ -54,7 +58,10 @@ import { MatrixRTCTransportMissingError } from "../../../utils/errors"; import { Epoch, ObservableScope } from "../../ObservableScope"; import { constant } from "../../Behavior"; import { ConnectionManagerData } from "../remoteMembers/ConnectionManager"; -import { ConnectionState, type Connection } from "../remoteMembers/Connection"; +import { + type ConnectionState, + type Connection, +} from "../remoteMembers/Connection"; import { type Publisher } from "./Publisher"; import { initializeWidget } from "../../../widget"; import { nullHostBridge } from "../../../HostBridge"; @@ -336,12 +343,18 @@ describe("LocalMembership", () => { trackPublications: [], } as unknown as LocalParticipant, }), - state$: constant(ConnectionState.LivekitConnected), + state$: constant({ + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }), transport: mockTransportConfig, } as Connection; const connectionTransportAConnecting = { ...connectionTransportAConnected, - state$: constant(ConnectionState.LivekitConnecting), + state$: constant({ + state: "authenticated", + livekitState: LivekitConnectionState.Connecting, + }), livekitRoom: mockLivekitRoom({}), } as unknown as Connection; @@ -532,7 +545,10 @@ describe("LocalMembership", () => { await flushPromises(); expect(localMembership.localMemberState$.value).toStrictEqual({ matrix: RTCMemberStatus.Connected, - media: { connection: null, tracks: TrackState.WaitingForUser }, + media: { + connection: { state: "initializing" }, + tracks: TrackState.WaitingForUser, + }, }); const connectionManagerData2 = new ConnectionManagerData(); @@ -547,7 +563,10 @@ describe("LocalMembership", () => { expect(localMembership.localMemberState$.value).toStrictEqual({ matrix: RTCMemberStatus.Connected, media: { - connection: ConnectionState.LivekitConnecting, + connection: { + state: "authenticated", + livekitState: LivekitConnectionState.Connecting, + }, tracks: TrackState.WaitingForUser, }, }); @@ -555,11 +574,17 @@ describe("LocalMembership", () => { ( connectionManagerData2.getConnectionForTransport(mockTransportConfig)! .state$ as BehaviorSubject - ).next(ConnectionState.LivekitConnected); + ).next({ + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }); expect(localMembership.localMemberState$.value).toStrictEqual({ matrix: RTCMemberStatus.Connected, media: { - connection: ConnectionState.LivekitConnected, + connection: { + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }, tracks: TrackState.WaitingForUser, }, }); @@ -717,9 +742,10 @@ describe("LocalMembership", () => { "track", ); - const connectionState$ = new BehaviorSubject( - ConnectionState.LivekitConnected, - ); + const connectionState$ = new BehaviorSubject({ + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }); const mutableConnection = { ...connectionTransportAConnected, state$: connectionState$, @@ -746,8 +772,14 @@ describe("LocalMembership", () => { await flushPromises(); - connectionState$.next(ConnectionState.LivekitDisconnected); - connectionState$.next(ConnectionState.LivekitConnected); + connectionState$.next({ + state: "authenticated", + livekitState: LivekitConnectionState.Disconnected, + }); + connectionState$.next({ + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }); expect(trackSpy).toHaveBeenCalledWith( defaultCreateLocalMemberValues.roomId, @@ -857,7 +889,10 @@ describe("LocalMembership", () => { const error = new Error("NotReadableError"); const setScreenShareEnabled = vi.fn().mockRejectedValue(error); const connection = { - state$: constant(ConnectionState.LivekitConnected), + state$: constant({ + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }), transport: mockTransportConfig, livekitRoom: mockLivekitRoom({ localParticipant: mockLocalParticipant({ diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index feb49554c..c06190d3f 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -6,6 +6,7 @@ Please see LICENSE in the repository root for full details. */ import { + ConnectionState as LivekitConnectionState, type Participant, ParticipantEvent, type LocalParticipant, @@ -74,9 +75,8 @@ import { } from "../../../config/ConfigOptions.ts"; import { Config } from "../../../config/Config.ts"; import { - ConnectionState, + type ConnectionState, type Connection, - type FailedToStartError, } from "../remoteMembers/Connection.ts"; import { type HomeserverConnected } from "./HomeserverConnected.ts"; import { type LocalTransport } from "./LocalTransport.ts"; @@ -113,7 +113,7 @@ export enum TrackState { export type LocalMemberMediaState = | { tracks: TrackState; - connection: ConnectionState | FailedToStartError; + connection: ConnectionState; } | PublishState | ElementCallError; @@ -475,7 +475,9 @@ export const createLocalMembership$ = ({ }; const localConnectionState$ = localConnection$.pipe( - switchMap((connection) => (connection ? connection.state$ : of(null))), + switchMap((connection) => + connection ? connection.state$ : of({ state: "initializing" as const }), + ), ); const mediaState$: Behavior = scope.behavior( @@ -493,7 +495,9 @@ export const createLocalMembership$ = ({ : TrackState.WaitingForUser; if ( - localConnectionState !== ConnectionState.LivekitConnected || + localConnectionState.state !== "authenticated" || + localConnectionState.livekitState !== + LivekitConnectionState.Connected || trackState !== TrackState.Ready ) return { @@ -561,7 +565,11 @@ export const createLocalMembership$ = ({ combineLatest([ homeserverConnected.combined$, localConnectionState$.pipe( - map((state) => state === ConnectionState.LivekitConnected), + map( + (state) => + state.state === "authenticated" && + state.livekitState === LivekitConnectionState.Connected, + ), ), ]).pipe( map(([[hsConnected, hsReason], livekitConnected]) => { diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index 62dc17a31..aa2429f88 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.test.ts @@ -34,15 +34,12 @@ import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; import { Connection, - ConnectionState, + type ConnectionState, type ConnectionOpts, } from "./Connection.ts"; import { ObservableScope } from "../../ObservableScope.ts"; import { type OpenIDClientParts } from "../../../livekit/openIDSFU.ts"; -import { - ElementCallError, - FailToGetOpenIdToken, -} from "../../../utils/errors.ts"; +import { FailToGetOpenIdToken } from "../../../utils/errors.ts"; import { testJWTToken } from "../../../utils/test-fixtures.ts"; import { mockRemoteParticipant, ownMemberMock } from "../../../utils/test.ts"; @@ -181,7 +178,7 @@ describe("Start connection states", () => { }; const connection = new Connection(opts, logger); - expect(connection.state$.getValue()).toEqual("Initialized"); + expect(connection.state$.getValue()).toEqual({ state: "initialized" }); }); it("fail to getOpenId token then error state", async () => { @@ -199,7 +196,7 @@ describe("Start connection states", () => { const connection = new Connection(opts, logger); - const capturedStates: (ConnectionState | Error)[] = []; + const capturedStates: ConnectionState[] = []; const s = connection.state$.subscribe((value) => { capturedStates.push(value); }); @@ -219,18 +216,20 @@ describe("Start connection states", () => { let capturedState = capturedStates.pop(); expect(capturedState).toBeDefined(); - expect(capturedState!).toEqual("FetchingConfig"); + expect(capturedState).toEqual({ state: "authenticating" }); deferred.reject(new FailToGetOpenIdToken(new Error("Failed to get token"))); await vi.runAllTimersAsync(); capturedState = capturedStates.pop(); - if (capturedState instanceof Error) { - expect(capturedState.message).toEqual("Something went wrong"); + if (capturedState?.state === "error") { + expect(capturedState.error.message).toEqual("Something went wrong"); expect(connection.transport).toEqual(livekitFocus); } else { - expect.fail("Expected FailedToStart state but got " + capturedState); + expect.fail( + "Expected FailedToStart state but got " + JSON.stringify(capturedState), + ); } }); @@ -249,7 +248,7 @@ describe("Start connection states", () => { const connection = new Connection(opts, logger); - const capturedStates: (ConnectionState | Error)[] = []; + const capturedStates: ConnectionState[] = []; const s = connection.state$.subscribe((value) => { capturedStates.push(value); }); @@ -274,7 +273,7 @@ describe("Start connection states", () => { let capturedState = capturedStates.pop(); expect(capturedState).toBeDefined(); - expect(capturedState).toEqual(ConnectionState.FetchingConfig); + expect(capturedState).toEqual({ state: "authenticating" }); deferredSFU.resolve(); await vi.runAllTimersAsync(); @@ -282,15 +281,17 @@ describe("Start connection states", () => { capturedState = capturedStates.pop(); if ( - capturedState instanceof ElementCallError && - capturedState.cause instanceof Error + capturedState?.state === "error" && + capturedState.error.cause instanceof Error ) { - expect(capturedState.cause.message).toContain( + expect(capturedState.error.cause.message).toContain( "Failed to look up user info from homeserver", ); expect(connection.transport).toEqual(livekitFocus); } else { - expect.fail("Expected FailedToStart state but got " + capturedState); + expect.fail( + "Expected FailedToStart state but got " + JSON.stringify(capturedState), + ); } }); @@ -309,7 +310,7 @@ describe("Start connection states", () => { const connection = new Connection(opts, logger); - const capturedStates: (ConnectionState | Error)[] = []; + const capturedStates: ConnectionState[] = []; const s = connection.state$.subscribe((value) => { capturedStates.push(value); }); @@ -339,7 +340,7 @@ describe("Start connection states", () => { let capturedState = capturedStates.pop(); expect(capturedState).toBeDefined(); - expect(capturedState).toEqual(ConnectionState.FetchingConfig); + expect(capturedState).toEqual({ state: "authenticating" }); deferredSFU.resolve(); await vi.runAllTimersAsync(); @@ -347,10 +348,10 @@ describe("Start connection states", () => { capturedState = capturedStates.pop(); if ( - capturedState instanceof ElementCallError && - capturedState.cause instanceof Error + capturedState?.state === "error" && + capturedState.error.cause instanceof Error ) { - expect(capturedState.cause.message).toContain( + expect(capturedState.error.cause.message).toContain( "Failed to connect to livekit", ); expect(connection.transport).toEqual(livekitFocus); @@ -367,7 +368,7 @@ describe("Start connection states", () => { const connection = setupRemoteConnection(); - const capturedStates: (ConnectionState | Error)[] = []; + const capturedStates: ConnectionState[] = []; const s = connection.state$.subscribe((value) => { capturedStates.push(value); }); @@ -377,22 +378,31 @@ describe("Start connection states", () => { await vi.runAllTimersAsync(); const initialState = capturedStates.shift(); - expect(initialState).toEqual(ConnectionState.Initialized); + expect(initialState).toEqual({ state: "initialized" }); const fetchingState = capturedStates.shift(); - expect(fetchingState).toEqual(ConnectionState.FetchingConfig); + expect(fetchingState).toEqual({ state: "authenticating" }); const disconnectedState = capturedStates.shift(); - expect(disconnectedState).toEqual(ConnectionState.LivekitDisconnected); + expect(disconnectedState).toEqual({ + state: "authenticated", + livekitState: LivekitConnectionState.Disconnected, + }); const connectingState = capturedStates.shift(); - expect(connectingState).toEqual(ConnectionState.LivekitConnecting); + expect(connectingState).toEqual({ + state: "authenticated", + livekitState: LivekitConnectionState.Connecting, + }); const connectedState = capturedStates.shift(); - expect(connectedState).toEqual(ConnectionState.LivekitConnected); + expect(connectedState).toEqual({ + state: "authenticated", + livekitState: LivekitConnectionState.Connected, + }); }); it("stopping while connecting does not report an error", async () => { setupTest(); const connection = setupRemoteConnection(); - const capturedStates: (ConnectionState | Error)[] = []; + const capturedStates: ConnectionState[] = []; const s = connection.state$.subscribe((value) => { capturedStates.push(value); }); @@ -412,7 +422,7 @@ describe("Start connection states", () => { // start() resolves rather than rejecting (it is not awaited by the // ConnectionManager, so a rejection would be unhandled). await expect(started).resolves.toBeUndefined(); - expect(capturedStates.at(-1)).toEqual(ConnectionState.Stopped); + expect(capturedStates.at(-1)).toEqual({ state: "stopped" }); expect(capturedStates.some((st) => st instanceof Error)).toBe(false); }); diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index cc84142d9..7face646a 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.ts @@ -13,6 +13,7 @@ import { import { ConnectionError, ConnectionErrorReason, + type ConnectionState as LivekitConnectionState, type Participant, type RemoteParticipant, type RemoteTrackPublication, @@ -23,7 +24,7 @@ import { type TrackPublication, } from "livekit-client"; import { type LivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; -import { BehaviorSubject, map } from "rxjs"; +import { BehaviorSubject } from "rxjs"; import { type Logger } from "matrix-js-sdk/lib/logger"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; @@ -63,30 +64,15 @@ export interface ConnectionOpts { /** Optional factory to create the LiveKit room, mainly for testing purposes. */ livekitRoomFactory: () => LivekitRoom; } -export class FailedToStartError extends Error { - public constructor(message: string) { - super(message); - this.name = "FailedToStartError"; - } -} -export enum ConnectionState { - /** The start state of a connection. It has been created but nothing has loaded yet. */ - Initialized = "Initialized", - /** `start` has been called on the connection. It aquires the jwt info to conenct to the LK Room */ - FetchingConfig = "FetchingConfig", - Stopped = "Stopped", - /** The same as ConnectionState.Disconnected from `livekit-client` */ - LivekitDisconnected = "disconnected", - /** The same as ConnectionState.Connecting from `livekit-client` */ - LivekitConnecting = "connecting", - /** The same as ConnectionState.Connected from `livekit-client` */ - LivekitConnected = "connected", - /** The same as ConnectionState.Reconnecting from `livekit-client` */ - LivekitReconnecting = "reconnecting", - /** The same as ConnectionState.SignalReconnecting from `livekit-client` */ - LivekitSignalReconnecting = "signalReconnecting", -} +export type ConnectionState = + // The start state of a connection. It has been created but nothing has loaded yet. + | { state: "initialized" } + // `start` has been called on the connection. It acquires the JWT token to connect to the LK room. + | { state: "authenticating" } + | { state: "authenticated"; livekitState: LivekitConnectionState } + | { state: "error"; error: ElementCallError } + | { state: "stopped" }; /** * A connection to a Matrix RTC LiveKit backend. @@ -95,14 +81,14 @@ export enum ConnectionState { */ export class Connection { // Private Behavior - private readonly _state$ = new BehaviorSubject< - ConnectionState | ElementCallError - >(ConnectionState.Initialized); + private readonly _state$ = new BehaviorSubject({ + state: "initialized", + }); /** * The current state of the connection to the media transport. */ - public readonly state$: Behavior = this._state$; + public readonly state$: Behavior = this._state$; /** * The media transport to connect to. @@ -304,7 +290,7 @@ export class Connection { this.logger.debug("Starting Connection"); this.stopped = false; try { - this._state$.next(ConnectionState.FetchingConfig); + this._state$.next({ state: "authenticating" }); // We should already have this information after creating the localTransport. // only call getSFUConfigWithOpenID for connections where we do not have a token yet. (existingJwtTokenData === undefined) const { url, jwt, livekitAlias } = @@ -326,14 +312,10 @@ export class Connection { // Setup observer once we are done with getSFUConfigWithOpenID connectionStateObserver(this.livekitRoom) - .pipe( - this.scope.bind(), - map((s) => s as unknown as ConnectionState), - ) - .subscribe((lkState) => { - // It is save to cast lkState to ConnectionState as they are fully overlapping. - this._state$.next(lkState); - }); + .pipe(this.scope.bind()) + .subscribe((livekitState) => + this._state$.next({ state: "authenticated", livekitState }), + ); try { this.logger.info(`livekitRoom.connect ${url}`); @@ -383,13 +365,15 @@ export class Connection { return; } this.logger.debug(`Failed to connect to LiveKit room: ${error}`); - this._state$.next( - error instanceof ElementCallError - ? error - : error instanceof Error - ? new UnknownCallError(error) - : new UnknownCallError(new Error(`${error}`)), - ); + this._state$.next({ + state: "error", + error: + error instanceof ElementCallError + ? error + : error instanceof Error + ? new UnknownCallError(error) + : new UnknownCallError(new Error(`${error}`)), + }); // Its okay to ignore the throw. The error is part of the state. throw error; } @@ -424,7 +408,7 @@ export class Connection { // disconnect sees the flag and does not report the abort as an error. this.stopped = true; await this.livekitRoom.disconnect(); - this._state$.next(ConnectionState.Stopped); + this._state$.next({ state: "stopped" }); this.logger.debug( `stop: DONE disconnecing from lk room ${this.transport.livekit_service_url}`, );