feat: Add experimental rewrite of go2rtc live provider (MSE/WebRTC/MP4/MJPEG) (#2580)

- Closes #2556
- Closes #2450

**Key intended features:**
 - go2rtc compatible
- 100% test coverage to significantly improve ability to test, maintain
and work around browser weirdnesses (e.g. Safari).
 - Written from the ground up in the style of the rest of the project.

**To use:**
 - Change `live_provider` from `go2rtc` to `go2rtc-experimental`.

src/utils/audio.ts

@@ -25,15 +25,17 @@ export const mayHaveAudio = (video: HTMLVideoElement & AudioProperties): boolean
 
 /**
  * Determine if audio is available for a go2rtc stream.
- * @param pc The RTCPeerConnection (for WebRTC streams).
- * @param mseCodecs The negotiated MSE codecs string (for MSE streams).
  * @param video The video element (fallback for browser-based detection).
+ * @param options.pc The RTCPeerConnection (for WebRTC streams).
+ * @param options.mseCodecs The negotiated MSE codecs string (for MSE streams).
  * @returns True if audio is available.
  */
 export const hasAudio = (
   video: HTMLVideoElement & AudioProperties,
-  pc?: RTCPeerConnection | null,
-  mseCodecs?: string,
+  options?: {
+    pc?: RTCPeerConnection | null;
+    mseCodecs?: string | null;
+  },
 ): boolean => {
   // For WebRTC: Check if there's an audio receiver with an active track. We
   // check that the track is not muted because muted means no media data is
@@ -43,15 +45,17 @@ export const hasAudio = (
   // (e.g. WebRTC failed, fell back to MSE) will have receivers with muted
   // tracks that don't reflect actual media availability.
   // See: https://github.com/dermotduffy/advanced-camera-card/issues/2417
-  if (pc && pc.connectionState === 'connected') {
-    const receivers = pc.getReceivers();
+  if (options?.pc && options.pc.connectionState === 'connected') {
+    const receivers = options.pc.getReceivers();
     if (receivers.length > 0) {
       return receivers.some(
         (receiver) => receiver.track?.kind === 'audio' && !receiver.track?.muted,
       );
     }
   }
+
   // For MSE: Check negotiated codecs for audio codecs
+  const mseCodecs = options?.mseCodecs;
   if (mseCodecs) {
     return (
       mseCodecs.includes('mp4a') ||
@@ -59,6 +63,7 @@ export const hasAudio = (
       mseCodecs.includes('flac')
     );
   }
+
   // Fallback to browser-based detection (unreliable in Chrome)
   return mayHaveAudio(video);
 };

src/utils/basic.ts

@@ -352,10 +352,6 @@ export const ignoreFunctionIdentity = (a: unknown, b: unknown): boolean | undefi
     ? typeof a === 'function' && typeof b === 'function'
     : undefined;
 
-export const convertHTTPAdressToWebsocket = (url: string): string => {
-  return url.replace(/^http/i, 'ws');
-};
-
 export const forceReflow = (element: HTMLElement): void => {
   // Force reflow by measuring the height.
   void element.offsetHeight;

src/utils/concurrency/generation.ts

@@ -0,0 +1,29 @@
+// A monotonic counter for guarding against stale asynchronous results. Snapshot
+// the current generation before an `await`; afterwards, `isCurrent()` reports
+// whether anything has invalidated that snapshot since (a reset, a teardown, or
+// a newer operation), so a superseded result can be dropped.
+export class Generation {
+  private _value = 0;
+
+  // Invalidate all outstanding snapshots (e.g. on reset or teardown).
+  public invalidate(): void {
+    this._value++;
+  }
+
+  // Start a new latest-wins operation -- invalidating any outstanding snapshot --
+  // and return its token.
+  public next(): number {
+    return ++this._value;
+  }
+
+  // Snapshot the current generation to compare after an await.
+  public current(): number {
+    return this._value;
+  }
+
+  // Whether the token is still current (nothing has invalidated it since it was
+  // taken).
+  public isCurrent(token: number): boolean {
+    return token === this._value;
+  }
+}

src/utils/concurrency/latest-value-runner.ts

@@ -0,0 +1,71 @@
+/**
+ * Runs an async operation for the latest submitted value, one run at a time.
+ *
+ * While a run is in flight, only the most recently submitted value is kept;
+ * values submitted in between are dropped. The kept value runs when the current
+ * run finishes, so the operation always converges on the newest input without
+ * running more than once at a time.
+ *
+ * Each `submit` returns a promise that resolves after the next run completes, so
+ * a caller can wait for a run to have happened without tracking which value ran
+ * (under load its own value may have been superseded by a newer one).
+ *
+ * The operation owns its own errors: a run that rejects still counts as done
+ * (its waiters resolve and draining continues), so a single bad value cannot
+ * strand later ones.
+ */
+export class LatestValueRunner<T> {
+  private _run: (value: T) => Promise<void>;
+
+  private _running = false;
+
+  private _pending: { value: T } | null = null;
+  private _waiters: Array<() => void> = [];
+
+  constructor(run: (value: T) => Promise<void>) {
+    this._run = run;
+  }
+
+  // Submit a value to run. Returns a promise that resolves once the next run
+  // has completed.
+  public submit(value: T): Promise<void> {
+    this._pending = { value };
+    const ran = new Promise<void>((resolve) => this._waiters.push(resolve));
+    if (!this._running) {
+      this._running = true;
+      // The drain loop cannot reject (the operation's errors are caught within
+      // it); the catch only satisfies the no-floating-promises rule.
+      /* istanbul ignore next -- @preserve */
+      this._drain().catch(() => {});
+    }
+    return ran;
+  }
+
+  // Drop any value still waiting to run.
+  public clear(): void {
+    this._pending = null;
+  }
+
+  private async _drain(): Promise<void> {
+    try {
+      while (this._pending) {
+        const { value } = this._pending;
+        this._pending = null;
+
+        try {
+          await this._run(value);
+        } catch {
+          // The operation owns its errors; the runner only guarantees progress.
+        }
+
+        // Release everyone waiting on a run; values submitted during the run
+        // were queued after this snapshot and wait for the next one.
+        const waiters = this._waiters;
+        this._waiters = [];
+        waiters.forEach((resolve) => resolve());
+      }
+    } finally {
+      this._running = false;
+    }
+  }
+}

src/utils/live-provider.ts

@@ -22,6 +22,13 @@ export const getResolvedLiveProvider = (
   return config?.live_provider ?? 'image';
 };
 
+// Live providers that stream from a go2rtc server and share the `go2rtc`
+// config block and endpoints.
+const GO2RTC_LIVE_PROVIDERS: readonly LiveProvider[] = ['go2rtc', 'go2rtc-experimental'];
+
+export const isGo2RTCLiveProvider = (provider: LiveProvider): boolean =>
+  GO2RTC_LIVE_PROVIDERS.includes(provider);
+
 export const liveProviderSupports2WayAudio = async (
   hass: HomeAssistant,
   config: CameraConfig,
@@ -29,7 +36,7 @@ export const liveProviderSupports2WayAudio = async (
   go2rtcMetadataEndpoint?: Endpoint | null,
   proxyConfig?: EnabledProxyConfig,
 ): Promise<boolean> => {
-  if (getResolvedLiveProvider(config) !== 'go2rtc') {
+  if (!isGo2RTCLiveProvider(getResolvedLiveProvider(config))) {
     return false;
   }
 

src/utils/websocket-url.ts

@@ -0,0 +1,10 @@
+// Convert an HTTP(S) or origin-relative URL to its WS(S) equivalent.
+export const convertToWebSocketURL = (url: string, origin?: string): string => {
+  if (/^http/i.test(url)) {
+    return 'ws' + url.substring(4);
+  }
+  if (url.startsWith('/')) {
+    return 'ws' + (origin ?? location.origin).substring(4) + url;
+  }
+  return url;
+};