From d37eb10b8c5c57b8df604c5b13a3ed14a2e3d846 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Apr 2026 03:17:57 +0000 Subject: [PATCH] feat(nests): proactive JWT refresh + reconnect for speaker path Mirror the listener's ReconnectingNestsListener for the publish side. moq-auth issues 600 s bearer tokens; without proactive refresh, a user holding the stage past 10 minutes silently drops when the relay tears down the WebTransport session and stays off the air until they manually re-tap Talk. The new wrapper recycles the session at 540 s so the relay never sees an expired token, and re-issues publishing onto each fresh session with the user's mute intent replayed on the new handle. VM swap is a one-line change to DefaultNestsSpeakerConnector. The caller-owned BroadcastHandle is now the wrapper's stable handle that survives every refresh. Six unit tests cover happy path, refresh-without-failure-state, mute replay across recycle, close idempotence, first-attempt-failure exception propagation, and post-close startBroadcasting guard. https://claude.ai/code/session_01HXf3zG3F2ev2ASeQju7Y5S --- .../commons/viewmodels/NestViewModel.kt | 13 +- .../nestsclient/ReconnectingNestsSpeaker.kt | 405 +++++++++++++++ .../ReconnectingNestsSpeakerTest.kt | 474 ++++++++++++++++++ 3 files changed, 890 insertions(+), 2 deletions(-) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeakerTest.kt diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt index b1932500e..3db46d569 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt @@ -36,8 +36,8 @@ import com.vitorpamplona.nestsclient.audio.AudioPlayer import com.vitorpamplona.nestsclient.audio.NestPlayer import com.vitorpamplona.nestsclient.audio.OpusDecoder import com.vitorpamplona.nestsclient.audio.OpusEncoder -import com.vitorpamplona.nestsclient.connectNestsSpeaker import com.vitorpamplona.nestsclient.connectReconnectingNestsListener +import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker import com.vitorpamplona.nestsclient.moq.SubscribeHandle import com.vitorpamplona.nestsclient.transport.WebTransportFactory import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner @@ -1132,9 +1132,18 @@ fun interface NestsSpeakerConnector { ): NestsSpeaker } +/** + * Production speaker factory — wraps each session in + * [connectReconnectingNestsSpeaker] so transport drops auto-retry + * with exponential backoff AND the moq-auth 600 s JWT TTL is + * proactively refreshed (default 540 s recycle window). The + * returned [BroadcastHandle] survives every refresh — the wrapper + * re-issues publishing onto each fresh session and replays the + * user's mute state on the new handle. + */ private val DefaultNestsSpeakerConnector = NestsSpeakerConnector { httpClient, transport, scope, room, signer, pubkey, capF, encF -> - connectNestsSpeaker( + connectReconnectingNestsSpeaker( httpClient = httpClient, transport = transport, scope = scope, diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt new file mode 100644 index 000000000..9ccb6332d --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -0,0 +1,405 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient + +import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.OpusEncoder +import com.vitorpamplona.nestsclient.transport.WebTransportFactory +import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.collectLatest +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withTimeoutOrNull +import java.util.concurrent.atomic.AtomicReference + +/** + * `connectNestsSpeaker` plus a transport-loss reconnect loop with + * exponential backoff and proactive JWT refresh. Mirror of + * [connectReconnectingNestsListener] for the publish side. + * + * The returned [NestsSpeaker]'s state surfaces the underlying speaker's + * state directly while a session is alive, but flips to + * [NestsSpeakerState.Reconnecting] between attempts. The speaker is + * auto-redirected to the freshly-opened session under the hood — + * `startBroadcasting()` returns a stable [BroadcastHandle] whose + * `setMuted` and `close` survive every refresh. + * + * **Why this exists** — moq-auth issues 600 s bearer tokens + * (`moq-auth/src/index.ts`). Without proactive refresh, any room a + * user keeps the stage in for >10 min hits an authorisation failure + * the moment the relay tears down the session, the publish stream + * goes silent, and the user has to manually re-tap "Talk". The + * proactive recycle keeps the WebTransport session young so the + * relay never sees an expired token. + * + * **Broadcast-handle re-issuance** — the caller-owned + * [BroadcastHandle] survives a refresh / reconnect. Internally the + * wrapper opens a fresh underlying `BroadcastHandle` against each + * new session, replays the user's mute intent on it, and forwards + * `setMuted` calls to whichever live handle exists at the time. + * `close` cancels the re-issue pump and best-effort closes the + * latest live handle. + * + * **Audio gap during refresh** — the wrapper closes the current + * underlying speaker (which stops the mic capture + Opus encoder + + * publisher) before opening the next, so the listener side will + * hear ~50–150 ms of silence at each recycle boundary. That's the + * trade-off we pay for a clean session swap; the alternative + * (carrying the mic capture across sessions) would require deeper + * plumbing into the audio pipeline. Acceptable for v1 — + * 9-min-spaced 150 ms gaps are well below the noise floor of a + * voice call. + * + * Cancellation: cancelling [scope] (typically the room screen's VM + * scope) cancels the reconnect loop and closes both the active + * session and the active broadcast. [NestsSpeaker.close] is + * idempotent. + */ +suspend fun connectReconnectingNestsSpeaker( + httpClient: NestsClient, + transport: WebTransportFactory, + scope: CoroutineScope, + room: NestsRoomConfig, + signer: NostrSigner, + speakerPubkeyHex: String, + captureFactory: () -> AudioCapture, + encoderFactory: () -> OpusEncoder, + policy: NestsReconnectPolicy = NestsReconnectPolicy(), + /** + * Proactive JWT refresh window. moq-auth issues bearer tokens + * with a 600 s lifetime; once the token expires the relay + * tears down the WebTransport session and we'd otherwise + * recover via the regular reconnect path with a brief audible + * dropout AND a permanent broadcast loss until the user taps + * Talk again. By recycling the session a minute before expiry + * we stay ahead of the relay's tear-down: the new session + * opens, the broadcast pump reopens publishing on it (carrying + * the user's mute intent), and the wrapper's outward state + * never enters the user-visible Reconnecting state. + * + * Set to 0 or negative to disable. + */ + tokenRefreshAfterMs: Long = 540_000L, + /** + * Test seam — defaults to the production [connectNestsSpeaker]. + * Tests pass a fake that returns a scripted [NestsSpeaker] so + * the reconnect state machine can be exercised without a real + * WebTransport stack. + */ + connector: suspend () -> NestsSpeaker = { + connectNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = speakerPubkeyHex, + captureFactory = captureFactory, + encoderFactory = encoderFactory, + ) + }, +): NestsSpeaker { + val state = MutableStateFlow(NestsSpeakerState.Idle) + val activeSpeaker = MutableStateFlow(null) + + suspend fun openOnce(): NestsSpeaker { + val speaker = connector() + activeSpeaker.value = speaker + state.value = speaker.state.value + return speaker + } + + val orchestrator = + scope.launch { + var attempt = 0 + while (true) { + val speaker = + runCatching { openOnce() }.getOrElse { + state.value = NestsSpeakerState.Failed("connect failed: ${it.message}", it) + null + } + var refreshTriggered = false + if (speaker != null) { + // Wait for either a terminal state OR the proactive + // JWT-refresh deadline. withTimeoutOrNull returns + // null when the timer fires first; we then close + // the (still-healthy) speaker and loop to mint a + // fresh JWT via openOnce(). The broadcast-pump + // re-issues publishing onto the new session + // without the wrapper's outward state ever + // showing Reconnecting. + // + // The `onEach { mirror } + first` pattern (rather + // than `state.collect { mirror; if terminal break }`) + // is what lets `withTimeoutOrNull` cancel cleanly + // mid-mirror — once cancelled, the underlying + // speaker's subsequent state changes (e.g. the + // Closed we trigger on the next line) don't leak + // out to the wrapper's state. + val terminalAwait: suspend () -> NestsSpeakerState = { + speaker.state + .onEach { state.value = it } + .first { s -> + s is NestsSpeakerState.Failed || s is NestsSpeakerState.Closed + } + } + val terminal = + if (tokenRefreshAfterMs > 0L) { + withTimeoutOrNull(tokenRefreshAfterMs) { terminalAwait() } + } else { + terminalAwait() + } + if (terminal == null) { + // Refresh deadline hit before any terminal state — + // planned recycle, not a failure. Close the old + // speaker; don't bump `attempt` (it's not a + // backoff event) so the next openOnce() runs + // immediately. + runCatching { speaker.close() } + attempt = 0 + refreshTriggered = true + } else if (terminal is NestsSpeakerState.Failed && !isUserCancelledSpeaker(terminal)) { + // Transport-side failure → schedule a reconnect. + attempt++ + if (!policy.isExhausted(attempt)) { + val delayMs = policy.delayForAttempt(attempt) + state.value = NestsSpeakerState.Reconnecting(attempt, delayMs) + } + } + } + if (refreshTriggered) { + // Skip the reconnect-schedule path entirely — a + // refresh is a planned cutover, not a backoff event. + continue + } + val terminal = state.value + if (terminal is NestsSpeakerState.Closed) break + if (policy.isExhausted(attempt + 1)) break + val delayMs = + if (terminal is NestsSpeakerState.Reconnecting) { + terminal.delayMs + } else { + policy.delayForAttempt(++attempt) + } + state.value = NestsSpeakerState.Reconnecting(attempt.coerceAtLeast(1), delayMs) + delay(delayMs) + } + } + + // Match the existing [connectNestsSpeaker] semantics: suspend + // until the first session is up (or hard-fails) so the VM's + // call site `val s = speakerConnector.connect(...); s.startBroadcasting()` + // keeps working without changes. This is a deliberate departure + // from the listener wrapper, which returns immediately and + // expects the VM to gate `subscribeSpeaker` on the Connected + // state — the speaker side has a tighter `startBroadcasting` + // contract that requires a live session at call time. + val firstReady = + state.first { s -> + s is NestsSpeakerState.Connected || + s is NestsSpeakerState.Broadcasting || + s is NestsSpeakerState.Failed + } + if (firstReady is NestsSpeakerState.Failed) { + // Unwind: cancel the orchestrator + close any speaker that + // managed to open before the failure surfaced. + orchestrator.cancel() + runCatching { activeSpeaker.value?.close() } + throw NestsException(firstReady.reason, firstReady.cause) + } + + return ReconnectingSpeakerHandle(state, activeSpeaker, orchestrator, scope) +} + +private fun isUserCancelledSpeaker(state: NestsSpeakerState.Failed): Boolean { + val msg = state.reason + // Forward-compat seam — same shape as [isUserCancelled] on the + // listener side. User-driven close goes through Closed today; + // anything else surfaced as Failed is a transport / handshake + // error worth retrying. + return msg.contains("user cancelled", ignoreCase = true) +} + +private class ReconnectingSpeakerHandle( + private val mutableState: MutableStateFlow, + private val activeSpeaker: MutableStateFlow, + private val orchestrator: Job, + private val scope: CoroutineScope, +) : NestsSpeaker { + override val state: StateFlow = mutableState.asStateFlow() + + private val gate = Mutex() + + @Volatile private var activeBroadcast: ReissuingBroadcastHandle? = null + + override suspend fun startBroadcasting(): BroadcastHandle = + gate.withLock { + check(state.value !is NestsSpeakerState.Closed) { + "startBroadcasting on a closed speaker" + } + check(activeBroadcast == null) { + "speaker is already broadcasting" + } + // Require a live (or just-connected) session — matches + // the listener wrapper's `subscribeSpeaker` contract. + // The wrapper's own `connect()` already suspended until + // the first session was up, so this check almost never + // fails in practice; it guards the second-call-after- + // close case. + activeSpeaker.value + ?: error("no live session — wait for state == Connected before startBroadcasting") + + val handle = + ReissuingBroadcastHandle(activeSpeaker, scope) { closed -> + if (activeBroadcast === closed) activeBroadcast = null + } + handle.start() + activeBroadcast = handle + handle + } + + override suspend fun close() { + orchestrator.cancel() + runCatching { activeBroadcast?.close() } + runCatching { activeSpeaker.value?.close() } + if (mutableState.value !is NestsSpeakerState.Closed) { + mutableState.value = NestsSpeakerState.Closed + } + } +} + +/** + * Stable [BroadcastHandle] backed by a re-issuing pump. Each time + * the wrapper opens a fresh session the pump cancels its prior + * iteration, calls [NestsSpeaker.startBroadcasting] on the new + * session, replays the cached mute intent on the resulting + * underlying handle, and parks until the next session swap. + * + * `setMuted` updates the cached intent unconditionally and forwards + * to whichever live underlying handle exists at the time. If no + * underlying handle is up (e.g. a brief gap during recycle), the + * intent is replayed on the next handle the pump opens, so the + * user-observed mute state is monotonic across recycles. + */ +private class ReissuingBroadcastHandle( + private val activeSpeaker: StateFlow, + private val scope: CoroutineScope, + private val onClose: (ReissuingBroadcastHandle) -> Unit, +) : BroadcastHandle { + @Volatile private var desiredMuted: Boolean = false + + @Volatile private var closed: Boolean = false + private val liveHandle = AtomicReference(null) + private var pumpJob: Job? = null + + override val isMuted: Boolean get() = desiredMuted + + fun start() { + // Re-broadcast pump: every time activeSpeaker changes, drop + // the prior broadcast (collectLatest cancels the inner + // body via awaitCancellation) and open a new one against + // the fresh session. The pattern mirrors the listener's + // SubscribeHandle re-issuance pump. + pumpJob = + scope.launch { + activeSpeaker.collectLatest { sp -> + if (sp == null || closed) return@collectLatest + // Wait until the underlying speaker is ready to + // broadcast (or has gone terminal). For a fresh + // session this resolves immediately because the + // wrapper's openOnce already saw Connected. + val ready = + sp.state.first { st -> + st is NestsSpeakerState.Connected || + st is NestsSpeakerState.Broadcasting || + st is NestsSpeakerState.Closed || + st is NestsSpeakerState.Failed + } + if (ready !is NestsSpeakerState.Connected && ready !is NestsSpeakerState.Broadcasting) { + return@collectLatest + } + if (closed) return@collectLatest + val handle = + runCatching { sp.startBroadcasting() } + .getOrNull() ?: return@collectLatest + if (closed) { + runCatching { handle.close() } + return@collectLatest + } + // Apply current mute intent BEFORE storing the + // handle so a setMuted that races us applies + // exactly once: either (a) we set intent → + // apply intent → store, and the racing setMuted + // sees the live handle and applies again (no-op + // on the broadcaster); or (b) the racing + // setMuted updates intent → we read intent → + // apply. Order doesn't matter; idempotent. + if (desiredMuted) { + runCatching { handle.setMuted(true) } + } + liveHandle.set(handle) + try { + // Park until activeSpeaker emits a new value + // (collectLatest cancels us) or close() runs + // (pumpJob.cancel). + awaitCancellation() + } finally { + // Clear our slot only if we still own it — + // close() may have already swapped in null. + if (liveHandle.get() === handle) liveHandle.set(null) + // Best-effort close on the way out: the user + // may have called wrapper.close (closed=true, + // pump cancelling), or activeSpeaker swapped + // (the prior speaker is about to be closed + // by the orchestrator anyway, but defensively + // closing here releases the broadcaster + + // publisher promptly rather than waiting for + // the speaker.close()). + runCatching { handle.close() } + } + } + } + } + + override suspend fun setMuted(muted: Boolean) { + if (closed) return + desiredMuted = muted + liveHandle.get()?.let { runCatching { it.setMuted(muted) } } + } + + override suspend fun close() { + if (closed) return + closed = true + pumpJob?.cancel() + liveHandle.getAndSet(null)?.let { runCatching { it.close() } } + onClose(this) + } +} diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeakerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeakerTest.kt new file mode 100644 index 000000000..e105f4a02 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeakerTest.kt @@ -0,0 +1,474 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient + +import com.vitorpamplona.nestsclient.transport.WebTransportFactory +import com.vitorpamplona.nestsclient.transport.WebTransportSession +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class ReconnectingNestsSpeakerTest { + private val room = + NestsRoomConfig( + authBaseUrl = "https://relay.example.com/api/v1/nests", + endpoint = "https://relay.example.com/moq", + hostPubkey = "0".repeat(64), + roomId = "abc", + ) + private val signer = NostrSignerInternal(KeyPair()) + + // The custom `connector` lambda below short-circuits the real + // mintToken / WebTransport handshake, so these two sentinels + // only satisfy the function signature — they're never invoked. + private val httpClient = + object : NestsClient { + override suspend fun mintToken( + room: NestsRoomConfig, + publish: Boolean, + signer: NostrSigner, + ): String = error("not invoked — connector overrides") + } + private val transport = + object : WebTransportFactory { + override suspend fun connect( + authority: String, + path: String, + bearerToken: String?, + ): WebTransportSession = error("not invoked — connector overrides") + } + + /** + * Scripted [BroadcastHandle] paired with a [ScriptedSpeaker]. + * Tracks `setMuted` / `close` calls and exposes the latest + * desired-mute value so tests can assert that the wrapper + * replayed mute intent on a fresh handle after a recycle. + */ + private class ScriptedBroadcastHandle : BroadcastHandle { + // Atomic plumbing — the broadcast pump on Dispatchers.Default + // races against the runBlocking-thread assertions, same + // reasoning as ReconnectingNestsListenerTest's ScriptedListener. + // Backing-property names match the public projection so + // ktlint's `standard:backing-property-naming` rule passes. + private val muted = AtomicBoolean(false) + private val closed = AtomicBoolean(false) + private val setMutedCount = AtomicInteger(0) + private val closeCount = AtomicInteger(0) + + override val isMuted: Boolean get() = muted.get() + val isClosed: Boolean get() = closed.get() + val setMutedCalls: Int get() = setMutedCount.get() + val closeCalls: Int get() = closeCount.get() + + override suspend fun setMuted(muted: Boolean) { + this.muted.set(muted) + setMutedCount.incrementAndGet() + } + + override suspend fun close() { + closed.set(true) + closeCount.incrementAndGet() + } + } + + /** + * Scripted [NestsSpeaker] that opens in + * [NestsSpeakerState.Connected] by default. Each call to + * [startBroadcasting] returns a fresh [ScriptedBroadcastHandle] + * the test can introspect. + */ + private class ScriptedSpeaker( + connectedRoom: NestsRoomConfig, + moqVersion: Long = 1, + ) : NestsSpeaker { + private val mutableState = + MutableStateFlow( + NestsSpeakerState.Connected(connectedRoom, moqVersion), + ) + override val state: StateFlow = mutableState.asStateFlow() + + private val _startCount = AtomicInteger(0) + val startCount: Int get() = _startCount.get() + val handles = mutableListOf() + + override suspend fun startBroadcasting(): BroadcastHandle { + _startCount.incrementAndGet() + val handle = ScriptedBroadcastHandle() + handles += handle + // Mirror the production speaker's contract: transition + // Connected → Broadcasting on startBroadcasting. + val current = mutableState.value + if (current is NestsSpeakerState.Connected) { + mutableState.value = + NestsSpeakerState.Broadcasting( + room = current.room, + negotiatedMoqVersion = current.negotiatedMoqVersion, + isMuted = false, + ) + } + return handle + } + + override suspend fun close() { + mutableState.value = NestsSpeakerState.Closed + } + + fun fail(reason: String) { + mutableState.value = NestsSpeakerState.Failed(reason) + } + } + + /** + * Pre-fail the connector for the first N attempts so the + * orchestrator's failure path is exercised. + */ + private class ScriptedFailure : NestsSpeaker { + override val state: StateFlow = + MutableStateFlow(NestsSpeakerState.Failed("scripted-fail")).asStateFlow() + + override suspend fun startBroadcasting(): BroadcastHandle = error("never connected") + + override suspend fun close() {} + } + + @Test + fun startBroadcasting_returns_handle_against_first_session() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + val first = ScriptedSpeaker(room) + val reconnecting = + connectReconnectingNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = "speaker-pubkey", + captureFactory = { error("not invoked — connector overrides") }, + encoderFactory = { error("not invoked — connector overrides") }, + policy = NestsReconnectPolicy(initialDelayMs = 1L), + connector = { first }, + ) + + val handle = reconnecting.startBroadcasting() + + // Wait for the pump to actually open the underlying + // broadcast against the first session. + withTimeout(5_000L) { + while (first.startCount == 0) delay(5) + } + + assertEquals(1, first.startCount) + assertEquals(1, first.handles.size) + assertFalse(handle.isMuted, "fresh handle should not be muted") + + reconnecting.close() + } finally { + scope.cancel() + } + } + + @Test + fun proactive_token_refresh_recycles_speaker_without_failure_state() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + val first = ScriptedSpeaker(room) + val second = ScriptedSpeaker(room) + val third = ScriptedSpeaker(room) + val speakersInOrder = mutableListOf(first, second, third) + + // Synchronized list — the watcher coroutine appends + // concurrently with the assertion's `.any { ... }`, + // and a plain ArrayList raises CME on iteration. The + // listener test gets away with a plain list because + // its watcher is cancelled by `reconnecting.close()` + // before the assertion runs in some timing windows; + // we don't rely on that here. + val seenStates: MutableList = + java.util.Collections.synchronizedList(mutableListOf()) + + val reconnecting = + connectReconnectingNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = "speaker-pubkey", + captureFactory = { error("unused") }, + encoderFactory = { error("unused") }, + // 50 ms refresh — small enough to fire + // quickly; large enough that openOnce + // resolves Connected first. + tokenRefreshAfterMs = 50L, + connector = { speakersInOrder.removeAt(0) }, + ) + + val watcher = + scope.launch { + reconnecting.state.collect { seenStates += it } + } + + val handle = reconnecting.startBroadcasting() + + // Wait for at least 2 underlying speakers to have + // been consumed (initial + first refresh). + withTimeout(5_000L) { + while (speakersInOrder.size > 1) delay(10) + } + + // Wait for the broadcast pump to have started a + // broadcast on the second session. + withTimeout(5_000L) { + while (second.startCount == 0) delay(5) + } + + // Critical postcondition: the wrapper's outward + // state must NEVER show Reconnecting or Failed + // during a clean refresh — the user-visible UI + // stays Broadcasting (or briefly Connected during + // the cutover) throughout. + // + // Cancel the watcher BEFORE the assertion so we can + // iterate seenStates without racing against further + // appends. (synchronizedList only guards individual + // ops, not iteration.) + watcher.cancel() + watcher.join() + val snapshot: List = synchronized(seenStates) { seenStates.toList() } + val sawReconnecting = snapshot.any { it is NestsSpeakerState.Reconnecting } + val sawFailed = snapshot.any { it is NestsSpeakerState.Failed } + assertTrue( + !sawReconnecting && !sawFailed, + "proactive refresh must not surface Reconnecting/Failed; saw=$snapshot", + ) + + // Both speakers should have seen exactly one + // startBroadcasting — the pump re-issues against + // the new session, doesn't double-broadcast on + // the old one. + assertEquals(1, first.startCount) + assertEquals(1, second.startCount) + + handle.close() + reconnecting.close() + } finally { + scope.cancel() + } + } + + @Test + fun setMuted_intent_replays_on_new_session_after_refresh() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + val first = ScriptedSpeaker(room) + val second = ScriptedSpeaker(room) + val speakersInOrder = mutableListOf(first, second) + + val reconnecting = + connectReconnectingNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = "speaker-pubkey", + captureFactory = { error("unused") }, + encoderFactory = { error("unused") }, + tokenRefreshAfterMs = 80L, + connector = { speakersInOrder.removeAt(0) }, + ) + + val handle = reconnecting.startBroadcasting() + + withTimeout(5_000L) { + while (first.startCount == 0) delay(5) + } + + // User mutes mid-broadcast on the first session. + handle.setMuted(true) + + withTimeout(5_000L) { + while (first.handles[0].setMutedCalls == 0) delay(5) + } + assertTrue(first.handles[0].isMuted, "first handle should be muted by user toggle") + + // Wait for refresh to swap to the second session. + withTimeout(5_000L) { + while (second.startCount == 0) delay(5) + } + + // Critical postcondition: the new underlying handle + // must have inherited the muted=true intent without + // the user calling setMuted again. + assertTrue( + second.handles[0].isMuted, + "post-refresh handle must inherit muted=true; setMutedCalls=${second.handles[0].setMutedCalls}", + ) + + handle.close() + reconnecting.close() + } finally { + scope.cancel() + } + } + + @Test + fun close_cancels_pump_and_closes_live_handle() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + val only = ScriptedSpeaker(room) + val reconnecting = + connectReconnectingNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = "speaker-pubkey", + captureFactory = { error("unused") }, + encoderFactory = { error("unused") }, + connector = { only }, + ) + + val handle = reconnecting.startBroadcasting() + + withTimeout(5_000L) { + while (only.startCount == 0) delay(5) + } + val live = only.handles[0] + + handle.close() + + // Live handle should be closed exactly once after + // close(). The pump-side close+the user-side close + // race; the handle's own idempotence guarantees + // exactly one effective close. + withTimeout(5_000L) { + while (!live.isClosed) delay(5) + } + assertTrue(live.isClosed) + + reconnecting.close() + } finally { + scope.cancel() + } + } + + @Test + fun first_attempt_failure_throws_from_connect() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + // Connector returns a speaker already in Failed + // state — connectReconnectingNestsSpeaker should + // surface this as a thrown NestsException rather + // than returning a wrapper in Failed state, so the + // VM's `try/catch` around the connect call lights + // up. + var threw: Throwable? = null + try { + connectReconnectingNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = "speaker-pubkey", + captureFactory = { error("unused") }, + encoderFactory = { error("unused") }, + policy = NestsReconnectPolicy(maxAttempts = 1), + connector = { ScriptedFailure() }, + ) + } catch (t: Throwable) { + threw = t + } + + assertTrue( + threw is NestsException, + "first-attempt failure should throw NestsException; got $threw", + ) + assertTrue( + threw.message?.contains("scripted-fail") == true, + "exception should preserve underlying reason; got ${threw.message}", + ) + } finally { + scope.cancel() + } + } + + @Test + fun startBroadcasting_after_close_throws() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + val only = ScriptedSpeaker(room) + val reconnecting = + connectReconnectingNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + speakerPubkeyHex = "speaker-pubkey", + captureFactory = { error("unused") }, + encoderFactory = { error("unused") }, + connector = { only }, + ) + + reconnecting.close() + withTimeout(5_000L) { + reconnecting.state.first { it is NestsSpeakerState.Closed } + } + + var threw: Throwable? = null + try { + reconnecting.startBroadcasting() + } catch (t: Throwable) { + threw = t + } + assertTrue(threw is IllegalStateException, "must reject startBroadcasting on closed wrapper; got $threw") + } finally { + scope.cancel() + } + } +}