feat(napplet): live subscription tail, multi-filters, resource.cancel

Implements the remaining tractable conformance follow-ups: - Live subscription tail: relay.subscribe now opens a real client.subscribe whose SubscriptionListener streams relay.event (stored + live), relay.eose, and relay.closed pushes keyed by subId, instead of a one-shot snapshot. relay.close unsubscribes (tracked in liveSubs, torn down in onDestroy). The broker only authorizes the subscription (RELAY consent) and returns Subscribed; the host owns the live stream. Shim dispatches relay.closed too. - Multi-filters: relay.query/subscribe honor every filter in filters[], not just the first (decodeFilterList; gateway query(List<Filter>); queryEvents unions across filters; max limit applied). - resource.cancel: accepted at the host edge as a no-op Done. Conformance tests extended (multi-filter decode, relay.closed push); commons:jvmTest and the amethyst napplet suite pass. Still open: identity getList/getZaps/getBadges + onChanged (shapes underspecified), the keys.action push (needs a host trigger UI), the resource nostr: scheme, inc + the niche domains, and on-device verification. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ncMHuBBVHEf7spAoSssde
2026-08-09 08:04:45 +00:00 · 2026-06-21 21:43:36 +00:00
parent 0f23051d54
commit cd2bae081e
11 changed files with 182 additions and 50 deletions
@@ -26,9 +26,18 @@ All four conformance breakers below are fixed (codec pinned by `NappletSdkConfor
   request `Blob` as base64 so it survives the bridge; the gateway uploads via the app's
   `BlossomUploader` to the user's kind:10063 server with a signed auth event.

-Still open (◐): multi-`filters` queries, identity `getList`/`getZaps`/`getBadges` + `onChanged`,
-`resource` `nostr:` + `cancel`, `relay.closed` + live subscription tail, and `inc`/`intent`/the
-niche domains. Plus **on-device verification** of the host/shell behavior.
+Also now implemented (the ◐ follow-ups):
+- **Live subscription tail** — `relay.subscribe` opens a real `client.subscribe` whose listener
+  streams `relay.event` (stored + live), `relay.eose`, and `relay.closed` pushes by `subId`;
+  `relay.close` unsubscribes (tracked in `liveSubs`, torn down in `onDestroy`).
+- **Multi-`filters`** — `relay.query`/`subscribe` honor every filter in the `filters[]` array,
+  not just the first (`decodeFilterList`, gateway `query(List<Filter>)`).
+- **`resource.cancel`** — accepted at the host edge as a no-op `Done`.
+
+Still open: identity `getList`/`getZaps`/`getBadges` + `onChanged` (object shapes / list-type
+semantics underspecified), the `keys.action` push (needs a host command-palette UI to *trigger*
+actions — registration already conforms), the `resource` `nostr:` scheme (unspecified bytes),
+`inc`/`intent`/the niche domains, and **on-device verification** of the host/shell behavior.

 ## Base envelope & error convention (verified)

@@ -56,7 +56,9 @@ import com.vitorpamplona.quartz.lightning.LnInvoiceUtil
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.toHexKey
 import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.NwcErrorResponse
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.PayInvoiceErrorResponse
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.PayInvoiceSuccessResponse
@@ -83,6 +85,8 @@ import java.io.ByteArrayInputStream
 import java.net.InetSocketAddress
 import java.net.Proxy
 import java.net.URLDecoder
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger

 /**
 * The trust boundary's main-process endpoint. The untrusted `:napplet` process binds this
@@ -106,6 +110,10 @@ class NappletBrokerService : Service() {

    private val incoming by lazy { Messenger(Handler(Looper.getMainLooper(), ::handleMessage)) }

+    // Live relay subscriptions: applet subId -> client subId, so relay.close (and teardown) can stop them.
+    private val liveSubs = ConcurrentHashMap<String, String>()
+    private val liveSeq = AtomicInteger(0)
+
    override fun onBind(intent: Intent?): IBinder? {
        // Defense in depth on top of exported=false: only our own UID may bind.
        if (Binder.getCallingUid() != Process.myUid()) return null
@@ -113,6 +121,12 @@ class NappletBrokerService : Service() {
    }

    override fun onDestroy() {
+        val client =
+            Amethyst.instance.sessionManager
+                .loggedInAccount()
+                ?.client
+        liveSubs.values.forEach { runCatching { client?.unsubscribe(it) } }
+        liveSubs.clear()
        scope.cancel()
        super.onDestroy()
    }
@@ -135,23 +149,31 @@ class NappletBrokerService : Service() {

        val requestType = runCatching { NappletProtocolJson.readType(payload) }.getOrNull() ?: "napplet"
        scope.launch {
-            // Unsubscribe is fire-and-forget: snapshot subscriptions have no live tail to cancel.
-            if (requestType == "relay.close") {
-                reply(replyTo, requestId, NappletProtocolJson.encodeResponse(requestType, NappletResponse.Done))
-                return@launch
+            // Fire-and-forget edge ops that don't go through the broker.
+            when (requestType) {
+                "relay.close" -> {
+                    runCatching { NappletProtocolJson.readSubId(payload) }.getOrNull()?.let { closeLiveSubscription(it) }
+                    reply(replyTo, requestId, NappletProtocolJson.encodeResponse(requestType, NappletResponse.Done))
+                    return@launch
+                }
+                "resource.cancel" -> {
+                    reply(replyTo, requestId, NappletProtocolJson.encodeResponse(requestType, NappletResponse.Done))
+                    return@launch
+                }
            }

            val response = process(identity, declared, payload)

-            // A subscription is answered with relay.event/relay.eose pushes (keyed by subId), not a
-            // .result — matching @napplet/shim. Today it delivers the initial matches then EOSE; a
-            // live tail is a follow-up.
+            // A subscription is answered with relay.event/relay.eose/relay.closed pushes keyed by
+            // subId (matching @napplet/shim) — the host opens a live relay subscription once the
+            // broker authorizes it. A non-authorized subscription closes immediately with an EOSE.
            val subId = if (requestType == "relay.subscribe") runCatching { NappletProtocolJson.readSubId(payload) }.getOrNull() else null
            if (subId != null) {
-                if (response is NappletResponse.Events) {
-                    response.events.forEach { push(replyTo, NappletProtocolJson.encodeRelayEvent(subId, it)) }
+                if (response is NappletResponse.Subscribed) {
+                    openLiveSubscription(subId, payload, replyTo)
+                } else {
+                    push(replyTo, NappletProtocolJson.encodeRelayEose(subId))
                }
-                push(replyTo, NappletProtocolJson.encodeRelayEose(subId))
            } else {
                reply(replyTo, requestId, NappletProtocolJson.encodeResponse(requestType, response))
            }
@@ -191,7 +213,7 @@ class NappletBrokerService : Service() {
                    return relays.map { it.url }
                }

-                override suspend fun query(filter: Filter): List<Event> = queryEvents(account, filter)
+                override suspend fun query(filters: List<Filter>): List<Event> = queryEvents(account, filters)
            }

        val consent =
@@ -416,27 +438,86 @@ class NappletBrokerService : Service() {
        return NappletResource(bytes, contentType)
    }

-    /** Bounded live relay fetch (EOSE/timeout) merged with the local cache, newest-first. */
+    /** Bounded live relay fetch (EOSE/timeout) for all [filters], merged with the local cache, newest-first. */
    private suspend fun queryEvents(
        account: Account,
-        filter: Filter,
+        filters: List<Filter>,
    ): List<Event> {
+        if (filters.isEmpty()) return emptyList()
        val relays = account.homeRelays.flow.value
        val fromRelays =
            if (relays.isEmpty()) {
                emptyList()
            } else {
                runCatching {
-                    account.client.fetchAll(filters = relays.associateWith { listOf(filter) }, timeoutMs = QUERY_TIMEOUT_MS)
+                    account.client.fetchAll(filters = relays.associateWith { filters }, timeoutMs = QUERY_TIMEOUT_MS)
                }.getOrDefault(emptyList())
            }
-        val fromCache = account.cache.filter(filter).mapNotNull { it.event }
+        val fromCache = filters.flatMap { filter -> account.cache.filter(filter).mapNotNull { it.event } }

        val merged =
            (fromRelays + fromCache)
                .distinctBy { it.id }
                .sortedByDescending { it.createdAt }
-        return filter.limit?.let { merged.take(it) } ?: merged
+        val limit = filters.mapNotNull { it.limit }.maxOrNull()
+        return limit?.let { merged.take(it) } ?: merged
+    }
+
+    /**
+     * Opens a live relay subscription for [nappletSubId], streaming `relay.event`/`relay.eose`/
+     * `relay.closed` pushes to the applet as events arrive. Replaces any existing subscription for
+     * the same id. Reached only after the broker authorized the subscription (RELAY consent).
+     */
+    private fun openLiveSubscription(
+        nappletSubId: String,
+        payload: String,
+        replyTo: Messenger,
+    ) {
+        val account = Amethyst.instance.sessionManager.loggedInAccount()
+        val filters = runCatching { NappletProtocolJson.decodeFilterList(payload) }.getOrDefault(emptyList())
+        val relays = account?.homeRelays?.flow?.value ?: emptySet()
+        if (account == null || filters.isEmpty() || relays.isEmpty()) {
+            push(replyTo, NappletProtocolJson.encodeRelayEose(nappletSubId))
+            return
+        }
+
+        closeLiveSubscription(nappletSubId)
+        val clientSubId = "napplet-$nappletSubId-${liveSeq.incrementAndGet()}"
+        liveSubs[nappletSubId] = clientSubId
+
+        val listener =
+            object : SubscriptionListener {
+                override fun onEvent(
+                    event: Event,
+                    isLive: Boolean,
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) = push(replyTo, NappletProtocolJson.encodeRelayEvent(nappletSubId, event))
+
+                override fun onEose(
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) = push(replyTo, NappletProtocolJson.encodeRelayEose(nappletSubId))
+
+                override fun onClosed(
+                    message: String,
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) = push(replyTo, NappletProtocolJson.encodeRelayClosed(nappletSubId, message))
+            }
+
+        runCatching { account.client.subscribe(clientSubId, relays.associateWith { filters }, listener) }
+    }
+
+    /** Stops the live subscription for [nappletSubId], if any. */
+    private fun closeLiveSubscription(nappletSubId: String) {
+        val clientSubId = liveSubs.remove(nappletSubId) ?: return
+        runCatching {
+            Amethyst.instance.sessionManager
+                .loggedInAccount()
+                ?.client
+                ?.unsubscribe(clientSubId)
+        }
    }

    /**
@@ -511,10 +511,11 @@ class NappletHostActivity : ComponentActivity() {
    var msg; if (typeof e.data === 'string') { try { msg = JSON.parse(e.data); } catch (_) { return; } } else { msg = e.data; }
    if (!msg) return;
    // Subscription pushes are keyed by subId, not a request id.
-    if (msg.type === 'relay.event' || msg.type === 'relay.eose') {
+    if (msg.type === 'relay.event' || msg.type === 'relay.eose' || msg.type === 'relay.closed') {
      var sub = subs[msg.subId]; if (!sub) return;
      if (msg.type === 'relay.event') { if (sub.onEvent) sub.onEvent(msg.event); }
-      else { if (sub.onEose) sub.onEose(); }
+      else if (msg.type === 'relay.eose') { if (sub.onEose) sub.onEose(); }
+      else { delete subs[msg.subId]; if (sub.onClosed) sub.onClosed(msg.reason); }
      return;
    }
    // keys.action push: the shell triggers a registered keyboard/command action.
@@ -77,6 +77,17 @@ object NappletProtocolJson {
            put("subId", subId)
        }.toString()

+    /** A `relay.closed` push: a relay (or the shell) ended the subscription [subId]. */
+    fun encodeRelayClosed(
+        subId: String,
+        reason: String,
+    ): String =
+        buildJsonObject {
+            put("type", "relay.closed")
+            put("subId", subId)
+            put("reason", reason)
+        }.toString()
+
    /**
     * The `shell.init` handshake reply (`@napplet/core`): the capability environment the napplet
     * caches and answers `shell.supports()` from. [domains] is the set of NAP domains this shell
@@ -115,8 +126,8 @@ object NappletProtocolJson {
                    encryption = o.str("encryption") ?: "nip44",
                )
            }
-            "relay.query" -> NappletRequest.QueryEvents(decodeFilter(o))
-            "relay.subscribe" -> NappletRequest.Subscribe(decodeFilter(o))
+            "relay.query" -> NappletRequest.QueryEvents(decodeFilterList(o))
+            "relay.subscribe" -> NappletRequest.Subscribe(decodeFilterList(o))
            "storage.get" -> NappletRequest.StorageGet(o.req("key"))
            "storage.set" -> NappletRequest.StorageSet(o.req("key"), o.req("value"))
            "storage.remove" -> NappletRequest.StorageRemove(o.req("key"))
@@ -220,6 +231,10 @@ object NappletProtocolJson {
                is NappletResponse.Done -> {
                    put("ok", true)
                }
+                // Subscribe is acknowledged here, but the host streams pushes instead of sending this.
+                is NappletResponse.Subscribed -> {
+                    put("ok", true)
+                }
                is NappletResponse.Denied -> {
                    put("ok", false)
                    put("error", "denied")
@@ -239,10 +254,16 @@ object NappletProtocolJson {
            }
        }.toString()

-    /** Parses a Nostr filter from `filter` (object) or the first of `filters` (array). */
-    private fun decodeFilter(o: JsonObject): Filter {
-        val f = o["filter"]?.jsonObject ?: o["filters"]?.jsonArray?.firstOrNull()?.jsonObject ?: JsonObject(emptyMap())
+    /** Parses every Nostr filter from a `filters` array (each entry) or a single `filter` object. */
+    fun decodeFilterList(envelopeJson: String): List<Filter> = decodeFilterList(json.parseToJsonElement(envelopeJson).jsonObject)

+    private fun decodeFilterList(o: JsonObject): List<Filter> {
+        o["filters"]?.jsonArray?.let { arr -> return arr.map { decodeFilterObject(it.jsonObject) } }
+        o["filter"]?.jsonObject?.let { return listOf(decodeFilterObject(it)) }
+        return emptyList()
+    }
+
+    private fun decodeFilterObject(f: JsonObject): Filter {
        val tags = mutableMapOf<String, List<String>>()
        for ((key, value) in f) {
            if (key.startsWith("#") && key.length == 2) {
@@ -96,15 +96,24 @@ class NappletProtocolJsonTest {
    @Test
    fun decodesQueryAndSubscribeFromFilterObjectOrFiltersArray() {
        val single = NappletProtocolJson.decodeRequest("""{"type":"relay.query","filter":{"kinds":[1],"#t":["nostr"],"limit":5}}""") as NappletRequest.QueryEvents
-        assertEquals(listOf(1), single.filter.kinds)
-        assertEquals(listOf("nostr"), single.filter.tags?.get("t"))
-        assertEquals(5, single.filter.limit)
+        assertEquals(listOf(1), single.filters.first().kinds)
+        assertEquals(
+            listOf("nostr"),
+            single.filters
+                .first()
+                .tags
+                ?.get("t"),
+        )
+        assertEquals(5, single.filters.first().limit)

-        val array = NappletProtocolJson.decodeRequest("""{"type":"relay.query","filters":[{"authors":["aa"]}]}""") as NappletRequest.QueryEvents
-        assertEquals(listOf("aa"), array.filter.authors)
+        // Multiple filters are all honored, not just the first.
+        val array = NappletProtocolJson.decodeRequest("""{"type":"relay.query","filters":[{"authors":["aa"]},{"kinds":[7]}]}""") as NappletRequest.QueryEvents
+        assertEquals(2, array.filters.size)
+        assertEquals(listOf("aa"), array.filters[0].authors)
+        assertEquals(listOf(7), array.filters[1].kinds)

        val sub = NappletProtocolJson.decodeRequest("""{"type":"relay.subscribe","filter":{"kinds":[1]}}""") as NappletRequest.Subscribe
-        assertEquals(listOf(1), sub.filter.kinds)
+        assertEquals(listOf(1), sub.filters.first().kinds)
    }

    @Test
@@ -79,14 +79,15 @@ class NappletSdkConformanceTest {

    @Test
    fun relayQueryAndSubscribeReadTheFiltersArray() {
-        // RelayQueryMessage: { type:'relay.query', id, filters: NostrFilter[] }
-        val q = NappletProtocolJson.decodeRequest("""{"type":"relay.query","id":"1","filters":[{"kinds":[1],"authors":["aa"]}]}""") as NappletRequest.QueryEvents
-        assertEquals(listOf(1), q.filter.kinds)
-        assertEquals(listOf("aa"), q.filter.authors)
+        // RelayQueryMessage: { type:'relay.query', id, filters: NostrFilter[] } — all filters honored.
+        val q = NappletProtocolJson.decodeRequest("""{"type":"relay.query","id":"1","filters":[{"kinds":[1]},{"authors":["aa"]}]}""") as NappletRequest.QueryEvents
+        assertEquals(2, q.filters.size)
+        assertEquals(listOf(1), q.filters[0].kinds)
+        assertEquals(listOf("aa"), q.filters[1].authors)

        // RelaySubscribeMessage: { type:'relay.subscribe', id, subId, filters, relay? }
        val s = NappletProtocolJson.decodeRequest("""{"type":"relay.subscribe","id":"1","subId":"s1","filters":[{"kinds":[7]}]}""") as NappletRequest.Subscribe
-        assertEquals(listOf(7), s.filter.kinds)
+        assertEquals(listOf(7), s.filters.first().kinds)
        // The subId the pushes are keyed by is read from the raw envelope by the service.
        assertEquals("s1", NappletProtocolJson.readSubId("""{"type":"relay.subscribe","id":"1","subId":"s1","filters":[{}]}"""))
    }
@@ -133,6 +134,12 @@ class NappletSdkConformanceTest {
        val eose = json.parseToJsonElement(NappletProtocolJson.encodeRelayEose("s1")).jsonObject
        assertEquals("relay.eose", eose["type"]?.jsonPrimitive?.content)
        assertEquals("s1", eose["subId"]?.jsonPrimitive?.content)
+
+        // RelayClosedMessage (PUSH): { type:'relay.closed', subId, reason }
+        val closed = json.parseToJsonElement(NappletProtocolJson.encodeRelayClosed("s1", "done")).jsonObject
+        assertEquals("relay.closed", closed["type"]?.jsonPrimitive?.content)
+        assertEquals("s1", closed["subId"]?.jsonPrimitive?.content)
+        assertEquals("done", closed["reason"]?.jsonPrimitive?.content)
    }

    // ---------- identity (Identity*Message) ----------
@@ -168,13 +168,14 @@ class NappletBroker(

            is NappletRequest.QueryEvents -> {
                val gateway = relay ?: return NappletResponse.Unsupported("relay.query")
-                NappletResponse.Events(gateway.query(request.filter))
+                NappletResponse.Events(gateway.query(request.filters))
            }

-            // Live tailing is a follow-up; for now subscribe returns the initial matches.
+            // The broker only authorizes the subscription (consent + declaration); the host opens the
+            // live relay subscription and streams relay.event/relay.eose/relay.closed by subId.
            is NappletRequest.Subscribe -> {
-                val gateway = relay ?: return NappletResponse.Unsupported("relay.subscribe")
-                NappletResponse.Events(gateway.query(request.filter))
+                relay ?: return NappletResponse.Unsupported("relay.subscribe")
+                NappletResponse.Subscribed
            }

            is NappletRequest.StorageGet -> {
@@ -50,8 +50,8 @@ interface NappletRelayGateway {
    /** Publishes [event] and returns the relay URLs that accepted it (empty = nowhere reached). */
    suspend fun publish(event: Event): List<String>

-    /** Returns events matching [filter] (e.g. from the local cache and/or a bounded relay fetch). */
-    suspend fun query(filter: Filter): List<Event>
+    /** Returns events matching any of [filters] (e.g. from the local cache and/or a bounded relay fetch). */
+    suspend fun query(filters: List<Filter>): List<Event>
 }

 /**
@@ -140,19 +140,19 @@ sealed interface NappletRequest {
        }
    }

-    /** Read events matching [filter] (from the cache and/or a bounded relay fetch). */
+    /** Read events matching [filters] (from the cache and/or a bounded relay fetch). */
    data class QueryEvents(
-        val filter: Filter,
+        val filters: List<Filter>,
    ) : NappletRequest {
        override val capability get() = NappletCapability.RELAY
    }

    /**
-     * Subscribe to events matching [filter]. The shell currently answers with the initial matches
-     * (like a query); a live tail over the existing reply channel is a follow-up.
+     * Subscribe to events matching [filters]. The shell opens a live relay subscription and pushes
+     * `relay.event`/`relay.eose`/`relay.closed` keyed by the applet's `subId`.
     */
    data class Subscribe(
-        val filter: Filter,
+        val filters: List<Filter>,
    ) : NappletRequest {
        override val capability get() = NappletCapability.RELAY
    }
@@ -60,6 +60,9 @@ sealed interface NappletResponse {
        val actionId: String,
    ) : NappletResponse

+    /** `relay.subscribe` was authorized; the host now streams `relay.event`/`relay.eose` pushes. */
+    data object Subscribed : NappletResponse
+
    /** Result of a storage read; [value] is null when the key is absent. */
    data class StorageValue(
        val value: String?,
@@ -77,7 +77,7 @@ class NappletBrokerTest {
            return listOf("wss://relay.example")
        }

-        override suspend fun query(filter: Filter): List<Event> = queryResult
+        override suspend fun query(filters: List<Filter>): List<Event> = queryResult
    }

    private class RecordingWallet(
@@ -276,7 +276,7 @@ class NappletBrokerTest {

            val response =
                broker(ScriptedPrompt(GrantState.ALLOW_ONCE), relay = relay)
-                    .handle(applet, NappletRequest.QueryEvents(Filter(kinds = listOf(1))), allDeclared)
+                    .handle(applet, NappletRequest.QueryEvents(listOf(Filter(kinds = listOf(1)))), allDeclared)

            assertEquals(NappletResponse.Events(listOf(cached)), response)
        }