diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalog.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalog.kt index 3694843e8..9bbf04840 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalog.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalog.kt @@ -22,35 +22,67 @@ package com.vitorpamplona.amethyst.commons.viewmodels import androidx.compose.runtime.Immutable import com.vitorpamplona.quartz.nip01Core.core.JsonMapper -import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable /** * Decoded shape of a moq-lite `catalog.json` track for a single - * audio-room speaker. Each broadcast advertises one or more tracks - * (one per codec / quality tier) — for nests audio rooms there's - * exactly one Opus track today, but the model carries a list for - * forward compatibility. + * audio-room speaker. Mirrors the kixelated/moq `hang` reference + * catalog format — the canonical shape that `@kixelated/hang`'s + * browser watcher and the `moq-rs` Rust hang crate both produce and + * consume — so Amethyst publishers are visible to the moq-lite browser + * reference, and Amethyst listeners parse standards-aligned catalogs + * from non-Amethyst publishers. * - * Best-effort schema: nostrnests' upstream moq-lite catalog spec - * (kixelated/moq) has evolved across revisions, and not every - * publisher fills in every field. The parser tolerates missing - * keys via `ignoreUnknownKeys = true` on [JsonMapper] and the - * `?`-marked properties. + * Shape (verbatim from `kixelated/moq/rs/hang/src/catalog/`): + * + * { + * "audio": { + * "renditions": { + * "": { + * "codec": "opus", + * "container": { "kind": "legacy" }, + * "sampleRate": 48000, + * "numberOfChannels": 1, + * "bitrate": 32000 // optional + * } + * } + * } + * } + * + * Keys are camelCase per the upstream serde `rename_all = "camelCase"`. + * Every field except the rendition-key string is optional in the parser + * — older / partial publishers are tolerated. Field semantics: + * + * - rendition key: the moq-lite `track` name a subscriber should + * subscribe to for that audio rendition's frames (commonly the same + * string for single-rendition broadcasts). Subscribers pick a + * rendition (e.g. by codec / bitrate) and use this key as the + * Subscribe.track string. + * - `codec`: codec mimetype string (`"opus"`, `"mp4a.40.2"` for AAC). + * - `container.kind`: frame wrapper. `"legacy"` = each frame is + * `varint(timestamp_us) + raw_codec_payload` inside the moq-lite + * group. `"cmaf"` = MOOF/MDAT-fragmented MP4. We only emit/parse + * `legacy` today; unknown kinds are ignored at parse time. + * - `sampleRate` / `numberOfChannels`: PCM source params. */ @Immutable @Serializable data class RoomSpeakerCatalog( - val version: Int? = null, - val audio: List = emptyList(), + val audio: Audio? = null, ) { @Immutable @Serializable - data class AudioTrack( - val track: String? = null, + data class Audio( + val renditions: Map = emptyMap(), + ) + + @Immutable + @Serializable + data class AudioConfig( val codec: String? = null, - @SerialName("sample_rate") val sampleRate: Int? = null, - @SerialName("channel_count") val channelCount: Int? = null, + val container: Container? = null, + val sampleRate: Int? = null, + val numberOfChannels: Int? = null, val bitrate: Int? = null, ) { /** @@ -64,14 +96,25 @@ data class RoomSpeakerCatalog( buildList { codec?.takeIf { it.isNotBlank() }?.let { add(it.uppercase()) } sampleRate?.let { add("${it / 1000}kHz") } - channelCount?.let { add(if (it == 1) "mono" else "${it}ch") } + numberOfChannels?.let { add(if (it == 1) "mono" else "${it}ch") } } return parts.takeIf { it.isNotEmpty() }?.joinToString(" · ") } } - /** First audio track, if any. The current single-Opus reality. */ - fun primaryAudio(): AudioTrack? = audio.firstOrNull() + @Immutable + @Serializable + data class Container( + val kind: String? = null, + ) + + /** + * First audio rendition, if any. The current single-Opus reality. + * Map iteration order is the JSON insertion order (kotlinx.serialization + * uses LinkedHashMap), so this picks the first rendition the + * publisher declared rather than an arbitrary one. + */ + fun primaryAudio(): AudioConfig? = audio?.renditions?.values?.firstOrNull() companion object { /** diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalogTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalogTest.kt index 4f4ffab80..f8a45bcff 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalogTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomSpeakerCatalogTest.kt @@ -27,69 +27,72 @@ import kotlin.test.assertNull class RoomSpeakerCatalogTest { @Test - fun parsesNostrnestsShape() { + fun parsesKixelatedHangShape() { + // Verbatim shape produced by the canonical kixelated/moq `hang` + // crate (`rs/hang/src/catalog/`). Verifies that an Amethyst + // listener picks up a standards-aligned publisher's catalog. val json = """ { - "version": 1, - "audio": [ - { - "track": "audio/data", - "codec": "opus", - "sample_rate": 48000, - "channel_count": 2, - "bitrate": 64000 + "audio": { + "renditions": { + "audio/data": { + "codec": "opus", + "container": { "kind": "legacy" }, + "sampleRate": 48000, + "numberOfChannels": 2, + "bitrate": 64000 + } } - ] + } } """.trimIndent() val catalog = RoomSpeakerCatalog.parseOrNull(json.encodeToByteArray()) assertNotNull(catalog) - assertEquals(1, catalog.version) - assertEquals(1, catalog.audio.size) - val track = catalog.primaryAudio() - assertNotNull(track) - assertEquals("audio/data", track.track) - assertEquals("opus", track.codec) - assertEquals(48_000, track.sampleRate) - assertEquals(2, track.channelCount) - assertEquals(64_000, track.bitrate) + assertEquals(1, catalog.audio?.renditions?.size) + val rendition = catalog.primaryAudio() + assertNotNull(rendition) + assertEquals("opus", rendition.codec) + assertEquals("legacy", rendition.container?.kind) + assertEquals(48_000, rendition.sampleRate) + assertEquals(2, rendition.numberOfChannels) + assertEquals(64_000, rendition.bitrate) } @Test fun describeFormatsHumanReadable() { - val track = - RoomSpeakerCatalog.AudioTrack( + val rendition = + RoomSpeakerCatalog.AudioConfig( codec = "opus", sampleRate = 48_000, - channelCount = 2, + numberOfChannels = 2, ) // Codec uppercased, kHz / channel count short-formed — // intended for a single-line tooltip in the participant // sheet, not a verbose codec dump. - assertEquals("OPUS · 48kHz · 2ch", track.describe()) + assertEquals("OPUS · 48kHz · 2ch", rendition.describe()) } @Test fun describeMonoIsLabelled() { - val track = RoomSpeakerCatalog.AudioTrack(codec = "opus", channelCount = 1) - assertEquals("OPUS · mono", track.describe()) + val rendition = RoomSpeakerCatalog.AudioConfig(codec = "opus", numberOfChannels = 1) + assertEquals("OPUS · mono", rendition.describe()) } @Test fun describeAllNullReturnsNull() { - val track = RoomSpeakerCatalog.AudioTrack() + val rendition = RoomSpeakerCatalog.AudioConfig() // Empty catalog → caller doesn't render a "unknown · unknown" // line; describe() returns null so the UI can omit cleanly. - assertNull(track.describe()) + assertNull(rendition.describe()) } @Test fun toleratesUnknownKeys() { - // Forward-compat: future moq-lite catalog revisions can add + // Forward-compat: future hang catalog revisions can add // fields without breaking older clients. val json = - """{"version":2,"audio":[{"track":"audio/data","codec":"opus","extra":"future-only"}],"new_top_level":true}""" + """{"audio":{"renditions":{"audio/data":{"codec":"opus","extra":"future-only"}}},"video":{},"newTopLevel":true}""" val catalog = RoomSpeakerCatalog.parseOrNull(json.encodeToByteArray()) assertNotNull(catalog) assertEquals("opus", catalog.primaryAudio()?.codec) @@ -104,12 +107,43 @@ class RoomSpeakerCatalogTest { } @Test - fun emptyAudioListIsAllowed() { + fun emptyAudioRenditionsIsAllowed() { // A publisher might emit an "advertising" catalog with no - // tracks yet (e.g. between codec switches). Accept the shape - // and let primaryAudio() return null. - val catalog = RoomSpeakerCatalog.parseOrNull("""{"version":1,"audio":[]}""".encodeToByteArray()) + // renditions yet (e.g. between codec switches). Accept the + // shape and let primaryAudio() return null. + val catalog = RoomSpeakerCatalog.parseOrNull("""{"audio":{"renditions":{}}}""".encodeToByteArray()) assertNotNull(catalog) assertNull(catalog.primaryAudio()) } + + @Test + fun missingAudioReturnsNullPrimary() { + // hang catalogs declaring only video MUST still parse without + // throwing — primaryAudio() returns null. + val catalog = RoomSpeakerCatalog.parseOrNull("""{}""".encodeToByteArray()) + assertNotNull(catalog) + assertNull(catalog.primaryAudio()) + } + + @Test + fun stripPrefixRoundTripsCanonicalCatalog() { + // The catalog payload [com.vitorpamplona.nestsclient.MoqLiteNestsSpeaker] + // emits MUST round-trip through the parser — guards against + // either side drifting from the kixelated/hang shape. + // SPEAKER_CATALOG_JSON lives in nestsClient and isn't + // accessible from commons; keep an inline literal that mirrors + // it verbatim so a desync triggers this assertion. + val emitted = + "{\"audio\":{\"renditions\":{\"audio/data\":{" + + "\"codec\":\"opus\",\"container\":{\"kind\":\"legacy\"}," + + "\"sampleRate\":48000,\"numberOfChannels\":1}}}}" + val catalog = RoomSpeakerCatalog.parseOrNull(emitted.encodeToByteArray()) + assertNotNull(catalog) + val rendition = catalog.primaryAudio() + assertNotNull(rendition) + assertEquals("opus", rendition.codec) + assertEquals("legacy", rendition.container?.kind) + assertEquals(48_000, rendition.sampleRate) + assertEquals(1, rendition.numberOfChannels) + } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 6b6c308bb..424ad38d9 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -82,14 +82,36 @@ class MoqLiteNestsListener internal constructor( override suspend fun subscribeSpeaker( speakerPubkeyHex: String, maxLatencyMs: Long, - ): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK, maxLatencyMs = maxLatencyMs) + ): SubscribeHandle = + wrapSubscription( + broadcast = speakerPubkeyHex, + track = AUDIO_TRACK, + maxLatencyMs = maxLatencyMs, + // Audio frames arrive in kixelated/moq `hang` "legacy" + // container layout: each moq-lite frame is + // `varint(timestamp_us) + raw_opus_packet`. Strip the + // leading varint so downstream decoders see a pristine + // Opus packet exactly as if it came from + // [com.vitorpamplona.nestsclient.audio.OpusEncoder]. + stripLegacyTimestamp = true, + ) - override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK, maxLatencyMs = 0L) + override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = + wrapSubscription( + broadcast = speakerPubkeyHex, + track = CATALOG_TRACK, + maxLatencyMs = 0L, + // Catalog frames carry raw JSON bytes — no container + // wrapping (the catalog itself is what tells you which + // container the audio track uses). + stripLegacyTimestamp = false, + ) private suspend fun wrapSubscription( broadcast: String, track: String, maxLatencyMs: Long, + stripLegacyTimestamp: Boolean, ): SubscribeHandle { check(state.value is NestsListenerState.Connected) { "NestsListener.subscribe requires Connected state, was ${state.value}" @@ -112,12 +134,13 @@ class MoqLiteNestsListener internal constructor( val objectIdSeq = AtomicLong(0L) val mapped = handle.frames.map { frame -> + val payload = if (stripLegacyTimestamp) stripLegacyTimestampPrefix(frame.payload) else frame.payload MoqObject( trackAlias = handle.id, groupId = frame.groupSequence, objectId = objectIdSeq.getAndIncrement(), publisherPriority = MoqLiteSession.DEFAULT_PRIORITY, - payload = frame.payload, + payload = payload, ) } @@ -242,5 +265,23 @@ class MoqLiteNestsListener internal constructor( * [subscribeCatalog]. */ const val CATALOG_TRACK: String = "catalog.json" + + /** + * Strip the leading `varint(timestamp_us)` prefix from a + * kixelated/moq `hang` "legacy" container frame, returning + * the bare codec payload (e.g. an Opus packet) for downstream + * decoders. The varint length is encoded in the top 2 bits of + * the first byte per RFC 9000 §16: `00`→1, `01`→2, `10`→4, + * `11`→8 bytes. Returns the payload unchanged when the + * varint header would overrun the payload (malformed frame — + * surface upstream rather than silently mask). + */ + internal fun stripLegacyTimestampPrefix(payload: ByteArray): ByteArray { + if (payload.isEmpty()) return payload + val tag = (payload[0].toInt() ushr 6) and 0x3 + val varintLen = 1 shl tag + if (payload.size < varintLen) return payload + return payload.copyOfRange(varintLen, payload.size) + } } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index 00a8622af..67cbf3217 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -203,17 +203,29 @@ class MoqLiteNestsSpeaker internal constructor( companion object { /** * moq-lite catalog manifest broadcast on the - * [MoqLiteNestsListener.CATALOG_TRACK] track. The shape mirrors - * what the kixelated/moq browser watcher parses - * ([com.vitorpamplona.amethyst.commons.viewmodels.RoomSpeakerCatalog]): - * a versioned envelope with an `audio[]` of track descriptors. + * [MoqLiteNestsListener.CATALOG_TRACK] track. Verbatim + * canonical kixelated/moq `hang` shape (camelCase, nested + * `audio.renditions[]`, `container.kind = "legacy"`). + * The rendition key MUST equal the moq-lite track we publish + * audio frames on ([MoqLiteNestsListener.AUDIO_TRACK]) so the + * watcher's "subscribe to this rendition" path resolves to our + * actual audio stream. + * + * `container.kind = "legacy"` declares the wire layout the + * watcher must expect inside each moq-lite frame: a single + * `varint(timestamp_us)` prefix followed by the raw codec + * payload (no MOOF/MDAT wrapping). The publisher side honours + * this contract in + * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster]. + * * Hard-coded because the speaker's encoder is fixed at * 48 kHz mono Opus today; if [OpusEncoder] becomes parameterised * this should be derived from the encoder config instead. */ const val SPEAKER_CATALOG_JSON: String = - "{\"version\":1,\"audio\":[{\"track\":\"audio/data\"," + - "\"codec\":\"opus\",\"sample_rate\":48000,\"channel_count\":1}]}" + "{\"audio\":{\"renditions\":{\"" + MoqLiteNestsListener.AUDIO_TRACK + "\":{" + + "\"codec\":\"opus\",\"container\":{\"kind\":\"legacy\"}," + + "\"sampleRate\":48000,\"numberOfChannels\":1}}}}" /** * How often the catalog group is re-emitted. The relay drops diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index 525f72c45..cf10c7a57 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -21,11 +21,14 @@ package com.vitorpamplona.nestsclient.audio import com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle +import com.vitorpamplona.quic.Varint import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.launch +import kotlin.time.TimeMark +import kotlin.time.TimeSource /** * Mirror of [NestBroadcaster] but driving a moq-lite @@ -195,6 +198,21 @@ class NestMoqLiteBroadcaster( // and the relay would see two unrelated uni streams under // the same logical group. var lastPublisher: MoqLitePublisherHandle = publisher + // kixelated/moq `hang` "legacy" container wire format: + // every frame inside a moq-lite group is + // varint(timestamp_us) + raw_codec_payload + // (`rs/hang/src/container/frame.rs`, + // Timescale<1_000_000>). Watchers that read our + // catalog's `container.kind = "legacy"` declaration + // will skip the leading varint as a microsecond + // timestamp; they MUST receive a real timestamp or + // their decoder picks up garbage bytes ahead of the + // Opus packet. We use a monotonic mark captured at + // capture-loop start so timestamps are robust to mute + // gaps and survive publisher hot-swap (the wire is + // single-broadcast, single-encoder; timestamps are + // codec-payload metadata, not stream-position). + val frameStartMark: TimeMark = TimeSource.Monotonic.markNow() try { while (true) { val pcm = capture.readFrame() ?: break @@ -233,7 +251,11 @@ class NestMoqLiteBroadcaster( // for the production cliff this works around. val sendOutcome = runCatching { - val accepted = current.send(opus) + val tsBytes = Varint.encode(frameStartMark.elapsedNow().inWholeMicroseconds) + val payload = ByteArray(tsBytes.size + opus.size) + tsBytes.copyInto(payload, 0) + opus.copyInto(payload, tsBytes.size) + val accepted = current.send(payload) framesInCurrentGroup += 1 if (framesInCurrentGroup >= framesPerGroup) { current.endGroup() diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt index 3d5b6445a..239a41d55 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt @@ -251,6 +251,20 @@ object MoqLiteCodec { return MoqLiteProbe(bitrate = bitrate) } + /** + * Encode a single Lite-03 Probe message body + * (`lite/probe.rs` — `bitrate: u62` only; `rtt` is Lite-04+). + * The publisher writes these size-prefixed onto a Probe bidi the + * subscriber opened, advertising the publisher's expected + * bandwidth. Wrapping (size prefix) is the caller's responsibility, + * matching [encodeAnnouncePlease] / [encodeAnnounce]. + */ + fun encodeProbe(probe: MoqLiteProbe): ByteArray { + val body = MoqWriter() + body.writeVarint(probe.bitrate) + return wrapSizePrefixed(body) + } + // ---------------- internals ---------------- /** diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 55f27fa12..ff31ab993 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -883,9 +883,50 @@ class MoqLiteSession internal constructor( dispatched = true } - else -> { - // Lite-03 treats Session/Fetch/Probe as - // separate flows; we don't implement them. + MoqLiteControlType.Probe -> { + // Subscriber-opened bidi asking us (the + // publisher) for a bitrate hint. Per + // `lite/probe.rs:Lite03+`, the publisher + // writes one or more size-prefixed + // `Probe { bitrate: u62 }` messages on + // this bidi. We're a fixed-rate Opus + // publisher, so emit a single hint and + // FIN our write side — the peer treats + // FIN as "no further updates" rather than + // an error. Better than the old behaviour + // of FINing without writing anything, + // which left the subscriber's ABR + // estimator with no signal. + runCatching { + bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = NESTS_AUDIO_BITRATE_HINT_BPS))) + bidi.finish() + } + dispatched = true + } + + MoqLiteControlType.Fetch -> { + // Subscriber-opened bidi requesting a + // historical group (`lite/fetch.rs`). + // Audio rooms are live-only — we have no + // group history to serve. FINing the + // write side without any reply is the + // spec-clean way to signal "no groups + // available"; the subscriber's wait on + // its receive side resolves to + // end-of-stream and it falls back to a + // live Subscribe. Ignoring inbound bytes + // (the request body) is fine: we don't + // need to know which group was requested + // because we couldn't serve any of them. + runCatching { bidi.finish() } + dispatched = true + } + + MoqLiteControlType.Session -> { + // ControlType=0 was the Lite-01/02 setup + // exchange. Lite-03 doesn't use it; if a + // legacy peer sends one, FIN cleanly so + // their bidi resolves rather than hanging. runCatching { bidi.finish() } dispatched = true } @@ -1194,6 +1235,15 @@ class MoqLiteSession internal constructor( /** moq-lite priority byte midpoint — neutral default. */ const val DEFAULT_PRIORITY: Int = 0x80 + /** + * Bitrate hint (bits/sec) we report on inbound moq-lite Probe + * bidis as a publisher. Mirrors the upper-bound of an Opus + * voice profile at 48 kHz mono, ≈32 kbps. Subscriber-side ABR + * estimators use this to size their forward queue; we emit the + * single hint and FIN since our encoder runs at a fixed bitrate. + */ + const val NESTS_AUDIO_BITRATE_HINT_BPS: Long = 32_000L + // Diagnostic: log "send returned false" once every N invocations. // At 50 fps and N=50 → ≤ 1 log/sec for a sustained no-sub window. private const val SEND_LOG_THROTTLE: Long = 50L