fix(nip47): parse space-separated encryption/notification tags; drop accumulating notification subscription

Two issues found in an audit of the NWC changes:

1. (correctness, high) NIP-44 negotiation never triggered against real
   wallets. The info event carries schemes in a single space-separated
   tag value (["encryption", "nip44_v2 nip04"]), but encryptionSchemes()
   returned tag.drop(1) = ["nip44_v2 nip04"], so the nip44_v2 membership
   check never matched and every request fell back to NIP-04. Split each
   tag value on whitespace in encryptionSchemes()/notificationTypes() so
   both the spec's space-separated form and a multi-element tag normalize
   to individual tokens. Adds NwcInfoEvent tests for the wire format.

2. (performance) NwcPaymentNotificationWatcher subscribed via
   subscribeAsFlow, which accumulates every event into an ever-growing
   list and re-emits the whole list per event — wrong for a lifetime
   subscription (unbounded retention + O(n) rescan per event). Replace
   with a raw client.subscribe listener (callbackFlow) that emits each
   event once; reconnect re-delivery is still de-duped by the seen set.

Also documents why the watcher keys the account flow on pubkey.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RDAAS4ktFbWtRnEVXQsjfs

amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NwcPaymentNotificationWatcher.kt

@@ -24,10 +24,13 @@ import android.content.Context
 import com.vitorpamplona.amethyst.commons.model.nip47WalletConnect.NwcWalletEntryNorm
 import com.vitorpamplona.amethyst.model.Account
 import com.vitorpamplona.amethyst.service.notifications.renderers.NwcPaymentNotifier
+import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
-import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.subscribeAsFlow
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
+import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
 import com.vitorpamplona.quartz.nip47WalletConnect.events.NwcNotificationEvent
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.PaymentReceivedNotification
@@ -35,8 +38,11 @@ import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.launch
@@ -66,6 +72,9 @@ class NwcPaymentNotificationWatcher(
     fun start() {
         scope.launch(Dispatchers.IO) {
             accountFlow
+                // Key on pubkey to avoid re-subscribing on unrelated Account churn.
+                // Account switches pass through a null (Loading) emission, which is
+                // distinct from any pubkey and so still restarts the watch cleanly.
                 .distinctUntilChanged { a, b -> a?.signer?.pubKey == b?.signer?.pubKey }
                 .collectLatest { account ->
                     account ?: return@collectLatest
@@ -99,18 +108,44 @@ class NwcPaymentNotificationWatcher(
                     )
 
                 val seen = HashSet<HexKey>()
-                client.subscribeAsFlow(wallet.uri.relayUri, filter).collect { events ->
-                    events.forEach { event ->
-                        val notification = event as? NwcNotificationEvent ?: return@forEach
-                        if (seen.add(notification.id)) {
-                            handle(account, notification, signer)
-                        }
+                subscribe(wallet.uri.relayUri, filter).collect { event ->
+                    val notification = event as? NwcNotificationEvent ?: return@collect
+                    if (seen.add(notification.id)) {
+                        handle(account, notification, signer)
                     }
                 }
             }
         }
     }
 
+    /**
+     * A standing subscription that emits each matching event once — unlike the
+     * accumulating `subscribeAsFlow`, which retains and re-emits the full event
+     * list on every event (wrong for a lifetime subscription). Reconnect
+     * re-delivery is de-duplicated by the caller's `seen` set.
+     */
+    private fun subscribe(
+        relay: NormalizedRelayUrl,
+        filter: Filter,
+    ): Flow<Event> =
+        callbackFlow {
+            val subId = newSubId()
+            val listener =
+                object : SubscriptionListener {
+                    override fun onEvent(
+                        event: Event,
+                        isLive: Boolean,
+                        relay: NormalizedRelayUrl,
+                        forFilters: List<Filter>?,
+                    ) {
+                        trySend(event)
+                    }
+                }
+
+            client.subscribe(subId, mapOf(relay to listOf(filter)), listener)
+            awaitClose { client.unsubscribe(subId) }
+        }
+
     private suspend fun handle(
         account: Account,
         event: NwcNotificationEvent,

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip47WalletConnect/events/NwcInfoEvent.kt

@@ -44,9 +44,22 @@ class NwcInfoEvent(
 
     fun supportsNotifications(): Boolean = capabilities().contains("notifications")
 
-    fun encryptionSchemes() = tags.mapNotNull(EncryptionTag::parse).flatten()
+    // NIP-47 carries the schemes/types as a single space-separated string in one
+    // tag value (e.g. ["encryption", "nip44_v2 nip04"]). Split on whitespace so we
+    // return individual tokens, while still tolerating a multi-element tag.
+    fun encryptionSchemes() =
+        tags
+            .mapNotNull(EncryptionTag::parse)
+            .flatten()
+            .flatMap { it.split(" ") }
+            .filter { it.isNotBlank() }
 
-    fun notificationTypes() = tags.mapNotNull(NotificationsTag::parse).flatten()
+    fun notificationTypes() =
+        tags
+            .mapNotNull(NotificationsTag::parse)
+            .flatten()
+            .flatMap { it.split(" ") }
+            .filter { it.isNotBlank() }
 
     companion object {
         const val KIND = 13194

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip47WalletConnect/NwcInfoEventTest.kt

@@ -96,6 +96,37 @@ class NwcInfoEventTest {
         assertTrue(schemes.contains("nip04"))
     }
 
+    @Test
+    fun testEncryptionSchemesSpaceSeparated() {
+        // NIP-47 wire format: all schemes in a single space-separated tag value.
+        val event =
+            NwcInfoEvent("id", "pub", 0L, arrayOf(arrayOf("encryption", "nip44_v2 nip04")), "pay_invoice", "sig")
+
+        val schemes = event.encryptionSchemes()
+        assertEquals(2, schemes.size)
+        assertTrue(schemes.contains("nip44_v2"))
+        assertTrue(schemes.contains("nip04"))
+    }
+
+    @Test
+    fun testNotificationTypesSpaceSeparated() {
+        // NIP-47 wire format: all types in a single space-separated tag value.
+        val event =
+            NwcInfoEvent(
+                "id",
+                "pub",
+                0L,
+                arrayOf(arrayOf("notifications", "payment_received payment_sent")),
+                "pay_invoice notifications",
+                "sig",
+            )
+
+        val types = event.notificationTypes()
+        assertEquals(2, types.size)
+        assertTrue(types.contains("payment_received"))
+        assertTrue(types.contains("payment_sent"))
+    }
+
     @Test
     fun testNotificationTypes() {
         val capabilities = listOf("pay_invoice", "notifications")