feat(audio-rooms): connectNestsListener uses moq-lite (phase 5d)

Production wiring switch — `connectNestsListener` now opens a moq-lite (Lite-03) session against the WebTransport peer instead of running the IETF MoQ-transport SETUP handshake. The downstream NestsListener interface is unchanged; consumer code in commons / amethyst keeps working. Listener flow (per the audio-rooms NIP draft + nostrnests JS reference): - subscribeSpeaker(pubkey) → MoqLiteSession.subscribe(broadcast = pubkey, track = "audio/data") - frames map to MoqObject so AudioRoomPlayer / AudioRoomViewModel keep working unchanged. Per-frame `objectId` is synthesised as a monotonic counter (moq-lite has no per-frame ID); `groupId` = moq-lite group sequence; `trackAlias` = subscribe id. NestsListenerState.Connected.negotiatedMoqVersion now reports MOQ_LITE_03_VERSION (a synthetic constant carrying the ALPN suffix in the low bytes) since moq-lite has no in-band SETUP message. NestsConnectTest: - Drop the SETUP-faking peer side — moq-lite has no SETUP, so connectNestsListener returns Connected immediately after the WT handshake. - Assert MOQ_LITE_03_VERSION on Connected. Speaker path (`connectNestsSpeaker`) is untouched — it still uses the IETF DefaultNestsSpeaker. Speaker-side moq-lite needs `acceptBidiStream` on the WebTransportSession interface so the relay can open Announce/Subscribe bidis to the publisher; tracked in plans/2026-04-26-moq-lite-gap.md phase-5c-speaker. Verified `:commons:compileKotlinJvm` + `:amethyst:compilePlayDebugKotlin` both still compile against the swap.
2026-04-26 17:29:51 +00:00
parent 4e136cacf4
commit 41f4dcd9ac
3 changed files with 179 additions and 48 deletions
@@ -0,0 +1,137 @@
+/*
+ * 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.MoqObject
+import com.vitorpamplona.nestsclient.moq.SubscribeHandle
+import com.vitorpamplona.nestsclient.moq.SubscribeOk
+import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession
+import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeException
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.map
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * Moq-lite-backed [NestsListener]. Wraps a connected [MoqLiteSession]
+ * and exposes the same listener API the IETF [DefaultNestsListener]
+ * does, so [connectNestsListener] can swap the framing layer without
+ * changing the public surface that [com.vitorpamplona.amethyst.commons.viewmodels.AudioRoomViewModel]
+ * and downstream UI consume.
+ *
+ * Subscription mapping per the audio-rooms NIP draft + nests JS
+ * reference (`@moq/watch/broadcast-Djx16jPC.js:6237`,
+ * `@moq/publish/screen-B680RFft.js:5641`):
+ *   - subscribeSpeaker(pubkey) → MoqLiteSession.subscribe(
+ *         broadcast = pubkey, track = "audio/data")
+ *   - the catalog (`"catalog.json"`) is not required for audio-only
+ *     listening; we'll add a parallel subscribe in a follow-up if
+ *     speaker-state metadata becomes useful.
+ *
+ * Frame adaptation: each [com.vitorpamplona.nestsclient.moq.lite.MoqLiteFrame]
+ * is wrapped in a [MoqObject] so existing decoder / player code keeps
+ * working unchanged. moq-lite frames carry no per-frame envelope, so
+ * [MoqObject.objectId] is synthesised as a per-subscription monotonic
+ * counter and [MoqObject.publisherPriority] uses the IETF default.
+ */
+class MoqLiteNestsListener internal constructor(
+    private val session: MoqLiteSession,
+    private val mutableState: MutableStateFlow<NestsListenerState>,
+) : NestsListener {
+    override val state: StateFlow<NestsListenerState> = mutableState.asStateFlow()
+
+    override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle {
+        check(state.value is NestsListenerState.Connected) {
+            "NestsListener.subscribeSpeaker requires Connected state, was ${state.value}"
+        }
+        val handle =
+            try {
+                session.subscribe(
+                    broadcast = speakerPubkeyHex,
+                    track = AUDIO_TRACK,
+                )
+            } catch (e: MoqLiteSubscribeException) {
+                // The IETF SubscribeHandle path conventionally surfaces
+                // protocol-level rejections through the same exception
+                // shape MoqProtocolException uses, so wrapping here
+                // avoids leaking moq-lite types into UI consumers.
+                throw com.vitorpamplona.nestsclient.moq.MoqProtocolException(
+                    "moq-lite subscribe rejected: ${e.message}",
+                    e,
+                )
+            }
+
+        val objectIdSeq = AtomicLong(0L)
+        val mapped =
+            handle.frames.map { frame ->
+                MoqObject(
+                    trackAlias = handle.id,
+                    groupId = frame.groupSequence,
+                    objectId = objectIdSeq.getAndIncrement(),
+                    publisherPriority = MoqLiteSession.DEFAULT_PRIORITY,
+                    payload = frame.payload,
+                )
+            }
+
+        return SubscribeHandle(
+            subscribeId = handle.id,
+            trackAlias = handle.id, // moq-lite has no separate alias; reuse id
+            ok = synthesizedOk(handle.ok),
+            objects = mapped,
+            unsubscribeAction = { handle.unsubscribe() },
+        )
+    }
+
+    override suspend fun close() {
+        if (state.value is NestsListenerState.Closed) return
+        runCatching { session.close() }
+        mutableState.value = NestsListenerState.Closed
+    }
+
+    private fun synthesizedOk(ok: com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeOk): SubscribeOk =
+        SubscribeOk(
+            subscribeId = 0L, // not used by downstream consumers
+            expiresMs = ok.maxLatencyMillis,
+            groupOrder = 0x01,
+            // moq-lite has no "content exists" signal — assume no history.
+            contentExists = false,
+            largestGroupId = null,
+            largestObjectId = null,
+        )
+
+    companion object {
+        /**
+         * Track name nests publishers use for the Opus audio stream.
+         * Source: `@moq/publish/screen-B680RFft.js:5641`
+         * (`static TRACK = "audio/data"`).
+         */
+        const val AUDIO_TRACK: String = "audio/data"
+
+        /**
+         * Catalog track name moq-lite watchers subscribe to first to
+         * receive a JSON description of the broadcast. Audio-only
+         * listening doesn't need it; reserved here for the nests
+         * speaker-metadata follow-up.
+         */
+        const val CATALOG_TRACK: String = "catalog.json"
+    }
+}
@@ -39,16 +39,13 @@ import kotlinx.coroutines.flow.MutableStateFlow
 *   1. Mint a JWT — POST `<authBase>/auth` with NIP-98 + namespace body
 *      (see [NestsClient.mintToken]).
 *   2. Open a [com.vitorpamplona.nestsclient.transport.WebTransportSession]
- *      against the [room.endpoint] via [transport], passing the JWT as the
- *      bearer token.
- *   3. Run the MoQ SETUP handshake.
- *
- * Note: step 3 runs the IETF `draft-ietf-moq-transport-17` SETUP, which is
- * NOT what nostrnests's relay (moq-lite) expects. Steps 1 and 2 are
- * already on the moq-rs wire shape (path = `/<namespace>`, JWT in
- * `?jwt=`). The MoQ framing layer needs a moq-lite codec swap before
- * end-to-end interop works — see
- * `nestsClient/plans/2026-04-26-moq-lite-gap.md`.
+ *      against the [room.endpoint] via [transport]. The path is
+ *      `/<moqNamespace>?jwt=<token>` — moq-rs treats the URL path as
+ *      `claims.root` and reads the JWT from `?jwt=`.
+ *   3. Wrap the WT session in a [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession].
+ *      moq-lite Lite-03 has NO in-band SETUP message — the WT handshake
+ *      itself is the handshake, version is selected by the ALPN
+ *      `moq-lite-03`.
 *
 * The returned [NestsListener] is in state [NestsListenerState.Connected];
 * if any step fails, the listener is returned in
@@ -58,9 +55,12 @@ import kotlinx.coroutines.flow.MutableStateFlow
 * @param room per-room config built from the NIP-53 kind 30312 event by
 *   the caller (UI / VM).
 * @param signer NIP-98 signer for the mintToken HTTP call.
- * @param scope where the [MoqSession] pumps live (typically the caller's
+ * @param scope where the moq-lite pumps live (typically the caller's
 *   ViewModel scope so they cancel when the screen leaves).
- * @param supportedMoqVersions in preference order; defaults to draft-17.
+ * @param supportedMoqVersions retained for source-compatible callers but
+ *   currently a no-op — moq-lite negotiates via ALPN, not an in-band
+ *   SETUP message. Will be repurposed to drive ALPN preference once a
+ *   second moq-lite version ships.
 */
 suspend fun connectNestsListener(
    httpClient: NestsClient,
@@ -113,35 +113,37 @@ suspend fun connectNestsListener(

    state.value = NestsListenerState.Connecting(NestsListenerState.Connecting.ConnectStep.MoqHandshake)

+    // moq-lite Lite-03 has NO setup message — the WebTransport handshake
+    // itself is the handshake, version is selected by ALPN. The
+    // `supportedMoqVersions` parameter is retained for backward compat
+    // with callers that may want to gate on a version mismatch in
+    // future, but it is currently a no-op.
    val moq =
        try {
-            MoqSession.client(webTransport, scope).also { it.setup(supportedMoqVersions) }
+            com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession
+                .client(webTransport, scope)
        } catch (t: Throwable) {
-            runCatching { webTransport.close(0, "moq setup failed") }
-            state.value = NestsListenerState.Failed("MoQ handshake failed: ${t.message}", t)
+            runCatching { webTransport.close(0, "moq-lite session init failed") }
+            state.value = NestsListenerState.Failed("moq-lite session init failed: ${t.message}", t)
            return failedListener(state)
        }

-    val negotiatedVersion =
-        moq.selectedVersion ?: run {
-            runCatching { moq.close() }
-            state.value = NestsListenerState.Failed("MoQ session reported no negotiated version")
-            return failedListener(state)
-        }
-
-    state.value = NestsListenerState.Connected(room, negotiatedVersion)
-    return DefaultNestsListener(
+    state.value = NestsListenerState.Connected(room, MOQ_LITE_03_VERSION)
+    return MoqLiteNestsListener(
        session = moq,
-        // moq-auth's JWT claim is `root: "<moqNamespace>"` — the simplest
-        // wire-level mapping is a 1-segment TrackNamespace whose only
-        // element is that exact string. Phase-3 interop test will
-        // confirm; if the real relay expects a multi-segment tuple
-        // (e.g. split on `/` or `:`), adjust here.
-        roomNamespace = TrackNamespace.of(room.moqNamespace()),
        mutableState = state,
    )
 }

+/**
+ * Synthetic version code reported on [NestsListenerState.Connected.negotiatedMoqVersion]
+ * for moq-lite Lite-03 sessions. moq-lite negotiates the wire version
+ * via ALPN rather than an in-band SETUP exchange, so there is no real
+ * "version" the peer reports back. The constant carries the ALPN suffix
+ * `-03` in the low bytes for diagnostic display.
+ */
+const val MOQ_LITE_03_VERSION: Long = 0x6D71_6C03L
+
 /**
 * Build a no-op [NestsListener] in a Failed state for callers that want a
 * uniform return type even on early-failure paths.
@@ -20,10 +20,6 @@
 */
 package com.vitorpamplona.nestsclient

-import com.vitorpamplona.nestsclient.moq.ClientSetup
-import com.vitorpamplona.nestsclient.moq.MoqCodec
-import com.vitorpamplona.nestsclient.moq.MoqVersion
-import com.vitorpamplona.nestsclient.moq.ServerSetup
 import com.vitorpamplona.nestsclient.transport.FakeWebTransport
 import com.vitorpamplona.nestsclient.transport.WebTransportException
 import com.vitorpamplona.nestsclient.transport.WebTransportFactory
@@ -31,8 +27,6 @@ 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.async
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -50,20 +44,15 @@ class NestsConnectTest {
        )

    @Test
-    fun connect_walks_mintToken_then_transport_then_moq_handshake() =
+    fun connect_walks_mintToken_then_transport_then_moq_lite_session() =
        runTest {
-            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val (clientSide, _) = FakeWebTransport.pair()
            val httpClient = FakeNestsClient(token = "tok-abc")
            val transport = ConstantWebTransportFactory(clientSide)

-            // Server-side raw peer answers SETUP.
-            val server =
-                async {
-                    val control = serverSide.peerOpenedBidiStreams().first()
-                    val cs = MoqCodec.decode(control.incoming().first())!!.message as ClientSetup
-                    control.write(MoqCodec.encode(ServerSetup(cs.supportedVersions.first())))
-                }
-
+            // moq-lite Lite-03 has NO setup message — the WT handshake
+            // itself is the handshake. As soon as connect returns, the
+            // listener is in Connected and ready to subscribe.
            val listener =
                connectNestsListener(
                    httpClient = httpClient,
@@ -72,11 +61,14 @@ class NestsConnectTest {
                    room = room,
                    signer = NostrSignerInternal(KeyPair()),
                )
-            server.await()

            val connected = assertIs<NestsListenerState.Connected>(listener.state.value)
            assertEquals(room, connected.room)
-            assertEquals(MoqVersion.DRAFT_17, connected.negotiatedMoqVersion)
+            assertEquals(
+                MOQ_LITE_03_VERSION,
+                connected.negotiatedMoqVersion,
+                "synthesised version code identifies the moq-lite-03 ALPN",
+            )

            assertEquals("relay.example.com", transport.lastConnectedAuthority)
            assertEquals(