feat(audio-rooms): subscribeCatalog on NestsListener (moq-lite only)

Surface moq-lite's `catalog.json` track on the NestsListener API.
The catalog publishes one JSON object per group describing the
broadcast (codec, sample rate, optional speaker-side hints) and
is the canonical channel a watcher reads first to discover
available tracks.

Wire-up:
  * NestsListener.subscribeCatalog(pubkey) — interface method with
    a default body that throws UnsupportedOperationException, so
    the IETF DefaultNestsListener fails loud rather than returning
    a flow that never delivers data.
  * MoqLiteNestsListener overrides it to wrap
    `session.subscribe(broadcast = pubkey, track = "catalog.json")`
    through the same MoqObject mapping the audio path uses.
  * ReconnectingNestsListener routes both subscribeSpeaker and
    subscribeCatalog through a shared `reissuingSubscribe` helper —
    catalog handles also survive session swaps via the
    MutableSharedFlow re-issuance pump.

Tests:
  * NestsListenerCatalogTest — verifies the interface default
    rejects with UnsupportedOperationException so a caller wired
    against the IETF reference path fails fast.

VM/UI consumers (e.g. "speaker codec" tooltip) are intentionally
out of scope for this commit; this exposes the capability so a
follow-up can plug in a JSON parser + per-pubkey catalog cache.

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt

@@ -59,16 +59,20 @@ class MoqLiteNestsListener internal constructor(
 ) : NestsListener {
     override val state: StateFlow<NestsListenerState> = mutableState.asStateFlow()
 
-    override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle {
+    override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK)
+
+    override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK)
+
+    private suspend fun wrapSubscription(
+        broadcast: String,
+        track: String,
+    ): SubscribeHandle {
         check(state.value is NestsListenerState.Connected) {
-            "NestsListener.subscribeSpeaker requires Connected state, was ${state.value}"
+            "NestsListener.subscribe requires Connected state, was ${state.value}"
         }
         val handle =
             try {
-                session.subscribe(
-                    broadcast = speakerPubkeyHex,
-                    track = AUDIO_TRACK,
-                )
+                session.subscribe(broadcast = broadcast, track = track)
             } catch (e: MoqLiteSubscribeException) {
                 // The IETF SubscribeHandle path conventionally surfaces
                 // protocol-level rejections through the same exception

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsListener.kt

@@ -52,6 +52,26 @@ interface NestsListener {
      */
     suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle
 
+    /**
+     * Subscribe to a speaker's `catalog.json` track — moq-lite's
+     * canonical channel for "what is this broadcast publishing" JSON
+     * metadata (codec, sample rate, optional speaker-side hints).
+     * The publisher emits one JSON object per group; consumers
+     * typically read the latest frame and parse it.
+     *
+     * Throws on the IETF reference path because moq-transport-17
+     * doesn't define a catalog convention. Callers that want
+     * cross-protocol behavior should `runCatching` this.
+     *
+     * @throws UnsupportedOperationException on the IETF listener.
+     * @throws MoqProtocolException if the publisher rejects the subscription.
+     * @throws IllegalStateException if the listener is not in [NestsListenerState.Connected].
+     */
+    suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle =
+        throw UnsupportedOperationException(
+            "subscribeCatalog is moq-lite-only; IETF listener has no catalog channel.",
+        )
+
     /** Tear down the MoQ session + underlying transport. Idempotent. */
     suspend fun close()
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt

@@ -156,7 +156,20 @@ private class ReconnectingHandle(
 ) : NestsListener {
     override val state: StateFlow<NestsListenerState> = mutableState.asStateFlow()
 
-    override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle {
+    override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = reissuingSubscribe { listener -> listener.subscribeSpeaker(speakerPubkeyHex) }
+
+    override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = reissuingSubscribe { listener -> listener.subscribeCatalog(speakerPubkeyHex) }
+
+    /**
+     * Open a SubscribeHandle whose `objects` flow survives session
+     * swaps. Each time the wrapper opens a fresh inner listener, the
+     * pump cancels the prior subscription and calls [opener] against
+     * the new session, forwarding its frames into the consumer-facing
+     * [MutableSharedFlow]. Same shape for both `subscribeSpeaker`
+     * and `subscribeCatalog`; the only difference is which inner
+     * subscribe call to make.
+     */
+    private fun reissuingSubscribe(opener: suspend (NestsListener) -> SubscribeHandle): SubscribeHandle {
         // Require a live (or just-connected) session at call time —
         // matches the existing IETF / moq-lite listener contract so
         // a caller can't accidentally subscribe in Idle and stall
@@ -170,10 +183,6 @@ private class ReconnectingHandle(
         // Re-subscribe pump: every time activeListener changes, drop
         // the prior subscription (collectLatest cancels the inner
         // body) and open a new one against the fresh session.
-        // Wait for the inner listener to reach Connected before
-        // calling subscribeSpeaker — IETF + moq-lite listeners both
-        // throw IllegalStateException if subscribed before Connected,
-        // and a fresh session may go through Connecting first.
         val pumpJob =
             scope.launch {
                 activeListener.collectLatest { listener ->
@@ -186,36 +195,23 @@ private class ReconnectingHandle(
                         }
                     if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest
                     val handle =
-                        runCatching { listener.subscribeSpeaker(speakerPubkeyHex) }
+                        runCatching { opener(listener) }
                             .getOrNull() ?: return@collectLatest
                     liveHandleRef.set(handle)
                     try {
                         handle.objects.collect { frames.emit(it) }
                     } finally {
-                        // Either the inner session ended, or we're
-                        // being replaced because activeListener
-                        // changed. Drop the dead reference; the
-                        // inner unsubscribe is handled by the
-                        // session's own teardown.
                         if (liveHandleRef.get() === handle) liveHandleRef.set(null)
                     }
                 }
             }
 
         return SubscribeHandle(
-            // Synthetic ids — the consumer-facing handle is logical,
-            // not bound to any one session's wire ids. -1 is a
-            // reserved sentinel callers shouldn't compare against.
             subscribeId = -1L,
             trackAlias = -1L,
             ok = SYNTH_OK,
             objects = frames.asSharedFlow(),
             unsubscribeAction = {
-                // Capture the live underlying handle BEFORE cancelling
-                // the pump — the pump's finally clears `liveHandleRef`
-                // on cancellation, and we need a stable reference to
-                // forward the user-initiated unsubscribe to the live
-                // session's wire SUBSCRIBE_DONE.
                 val live = liveHandleRef.getAndSet(null)
                 pumpJob.cancel()
                 live?.let { runCatching { it.unsubscribe() } }

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsListenerCatalogTest.kt

@@ -0,0 +1,56 @@
+/*
+ * 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.moq.SubscribeHandle
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertFailsWith
+
+class NestsListenerCatalogTest {
+    private class IetfStyleListener : NestsListener {
+        override val state: StateFlow<NestsListenerState> =
+            MutableStateFlow<NestsListenerState>(NestsListenerState.Idle).asStateFlow()
+
+        override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = throw UnsupportedOperationException("not under test")
+
+        // Intentionally does NOT override subscribeCatalog so the
+        // interface's default body fires.
+
+        override suspend fun close() = Unit
+    }
+
+    @Test
+    fun ietfDefaultListenerRejectsCatalogSubscribe() =
+        runTest {
+            // The catalog channel is moq-lite only; the IETF listener
+            // path doesn't define one, so the interface default throws
+            // UnsupportedOperationException rather than silently
+            // returning a subscription that will never deliver data.
+            val listener = IetfStyleListener()
+            assertFailsWith<UnsupportedOperationException> {
+                listener.subscribeCatalog("speakerPubkey")
+            }
+        }
+}