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

@@ -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;
       }),
     ),

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<ConnectionState>
-    ).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>(
-        ConnectionState.LivekitConnected,
-      );
+      const connectionState$ = new BehaviorSubject<ConnectionState>({
+        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({

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<LocalMemberMediaState> = 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]) => {

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);
   });
 

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<ConnectionState>({
+    state: "initialized",
+  });
 
   /**
    * The current state of the connection to the media transport.
    */
-  public readonly state$: Behavior<ConnectionState | Error> = this._state$;
+  public readonly state$: Behavior<ConnectionState> = 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(
+      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}`,
     );