refactor(wallet): move NWC notification subscription into the always-on account subscription layer

The bespoke applicationIOScope watcher that opened its own relay subscription for NIP-47 wallet notifications is replaced by an EOSE manager registered in AccountFilterAssembler.group — the same always-on scheme as the account's zap/notification inbox subscriptions. It is now owned by the single AccountFilterAssemblerSubscription in LoggedInPage, kept warm in the background by NotificationRelayService, and torn down on logout — matching zap-receipt lifecycle exactly (and not gated on OS notification permission). - NwcNotificationsEoseManager (PerUserEoseManager): one filter per connected wallet's own relay (kind 23197/23196, #p = per-wallet client pubkey), re-invalidating when the wallet set changes, `since`-floored at watch start, deduped by a seen set. Because these events are ephemeral, encrypted, and never land in LocalCache, onEvent decrypts them via NwcSignerState.handleIncomingNotification. - NwcSignerState.handleIncomingNotification decrypts with the matching wallet's connection secret, drops zap-carrying payments, and publishes non-zap payments to a new incomingNonZapPayments SharedFlow — a clean seam a future in-app Notifications-tab consumer can also drain. - NwcPaymentNotificationWatcher is now just the Context-bound bridge that drains that flow into an OS tray notification (no relay work, no client dependency). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RDAAS4ktFbWtRnEVXQsjfs
2026-08-08 23:54:39 +00:00 · 2026-07-24 23:17:15 +00:00
parent 9789ff61f4
commit 40820db10f
5 changed files with 214 additions and 117 deletions
@@ -885,10 +885,11 @@ class AppModules(

    // Surfaces non-zap Lightning payments reported by the logged-in account's NWC
    // wallet(s) as tray notifications (zaps are already shown via ZapNotification).
+    // The relay subscription lives in the always-on AccountFilterAssembler; this
+    // only drains the decoded-payment flow into an OS notification.
    val nwcPaymentNotificationWatcher =
        NwcPaymentNotificationWatcher(
            context = appContext,
-            client = client,
            scope = applicationIOScope,
            accountFlow = sessionManager.accountContent.map { (it as? AccountState.LoggedIn)?.account },
        ).also { it.start() }
@@ -37,13 +37,21 @@ import com.vitorpamplona.quartz.nip47WalletConnect.cache.NostrWalletConnectReque
 import com.vitorpamplona.quartz.nip47WalletConnect.cache.NostrWalletConnectResponseCache
 import com.vitorpamplona.quartz.nip47WalletConnect.events.LnZapPaymentRequestEvent
 import com.vitorpamplona.quartz.nip47WalletConnect.events.LnZapPaymentResponseEvent
+import com.vitorpamplona.quartz.nip47WalletConnect.events.NwcNotificationEvent
+import com.vitorpamplona.quartz.nip47WalletConnect.rpc.NwcTransaction
+import com.vitorpamplona.quartz.nip47WalletConnect.rpc.PaymentReceivedNotification
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.Request
 import com.vitorpamplona.quartz.nip47WalletConnect.rpc.Response
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
@@ -154,6 +162,42 @@ class NwcSignerState(
        return zapPaymentResponseDecryptionCache.value.decryptResponse(event)
    }

+    // Non-zap incoming payments reported by connected wallets (NIP-47
+    // payment_received). Buffered + drop-oldest so a burst never blocks the
+    // decrypt coroutine; consumers (e.g. the tray-notification poster) collect it.
+    private val _incomingNonZapPayments =
+        MutableSharedFlow<NwcTransaction>(extraBufferCapacity = 32, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+    val incomingNonZapPayments: SharedFlow<NwcTransaction> = _incomingNonZapPayments.asSharedFlow()
+
+    /**
+     * Decrypts an incoming NWC notification (kind 23197/23196) with the matching
+     * wallet's connection secret and, when it is a non-zap `payment_received`,
+     * publishes its transaction to [incomingNonZapPayments]. Zap-carrying payments
+     * are dropped — those already surface via the kind-9735 ZapNotification path.
+     */
+    suspend fun handleIncomingNotification(event: NwcNotificationEvent) {
+        if (!hasWalletConnectSetup()) return
+
+        // The notification is `p`-tagged to the per-wallet client pubkey; match it
+        // to the wallet whose connection secret derives that key.
+        val clientPubKey = event.clientPubKey() ?: return
+        val wallet = settings.nwcWallets.value.firstOrNull { buildSigner(it.uri)?.pubKey == clientPubKey } ?: return
+        val walletSigner = buildSigner(wallet.uri) ?: return
+
+        val notification =
+            try {
+                event.decryptNotification(walletSigner)
+            } catch (e: Exception) {
+                if (e is CancellationException) throw e
+                return
+            }
+
+        val tx = (notification as? PaymentReceivedNotification)?.notification ?: return
+        if (tx.parsedMetadata()?.nostr != null) return // zap — already shown by ZapNotification
+
+        _incomingNonZapPayments.tryEmit(tx)
+    }
+
    /**
     * Sends a generic NIP-47 request to the default wallet.
     */
@@ -21,149 +21,47 @@
 package com.vitorpamplona.amethyst.service.notifications

 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.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
-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

 /**
- * Always-on watcher that turns NIP-47 wallet notifications (kind 23197/23196) into
- * user-facing tray notifications for the logged-in account.
+ * Posts tray notifications for non-zap Lightning payments reported by the logged-in
+ * account's connected NWC wallets.
 *
- * For each configured NWC wallet it opens a standing subscription on that wallet's
- * own relay, filtered to notifications `p`-tagged to our client pubkey. Each event
- * is decrypted with the per-wallet connection secret and, when it is an incoming
- * `payment_received`, posted via [NwcPaymentNotifier] — **unless** the transaction
- * metadata carries a NIP-57 zap request, in which case it is dropped: zaps already
- * surface through the kind-9735 `ZapNotification` path, so notifying here would
- * double up.
+ * The relay subscription that receives these events is NOT here — it lives in
+ * [com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip47WalletConnect.NwcNotificationsEoseManager],
+ * grouped with the account's always-on zap/notification inbox subscriptions so it
+ * shares their lifecycle (open while logged in, warm in the background). That
+ * manager decrypts each notification and publishes non-zap payments to
+ * `NwcSignerState.incomingNonZapPayments`; this class is only the Context-bound
+ * bridge that drains that flow into an OS notification.
 *
- * Only new notifications (`since` = watch start) are surfaced, so relaunching the
- * app never replays old payments as fresh alerts; a per-subscription id set
- * de-duplicates re-delivery across relay reconnects.
+ * Keeping decode and display decoupled means the flow stays populated even when OS
+ * notifications are denied (a future in-app Notifications-tab consumer can drain
+ * the same flow); [NwcPaymentNotifier] itself is what no-ops when the tray is off.
 */
 class NwcPaymentNotificationWatcher(
    private val context: Context,
-    private val client: INostrClient,
    private val scope: CoroutineScope,
    private val accountFlow: Flow<Account?>,
 ) {
    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
-                    account.settings.nwcWallets.collectLatest { wallets ->
-                        watchWallets(account, wallets)
+                    account.nip47SignerState.incomingNonZapPayments.collect { tx ->
+                        NwcPaymentNotifier.notify(context, account, tx)
                    }
                }
        }
    }
-
-    private suspend fun watchWallets(
-        account: Account,
-        wallets: List<NwcWalletEntryNorm>,
-    ) = coroutineScope {
-        wallets.forEach { wallet ->
-            val signer = account.nip47SignerState.buildSigner(wallet.uri) ?: return@forEach
-
-            launch(Dispatchers.IO) {
-                // Skip wallets that explicitly advertise no notification support, so we
-                // don't hold open a relay connection that will never deliver. Fail open
-                // when the info event is unknown (null) — better to listen than miss.
-                val info = account.nwcInfoCache.getFresh(wallet.uri)
-                if (info != null && !info.supportsNotifications()) return@launch
-
-                val filter =
-                    Filter(
-                        kinds = listOf(NwcNotificationEvent.KIND, NwcNotificationEvent.LEGACY_KIND),
-                        authors = listOf(wallet.uri.pubKeyHex),
-                        tags = mapOf("p" to listOf(signer.pubKey)),
-                        since = TimeUtils.now(),
-                    )
-
-                val seen = HashSet<HexKey>()
-                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,
-        signer: NostrSigner,
-    ) {
-        val notification =
-            try {
-                event.decryptNotification(signer)
-            } catch (e: Exception) {
-                if (e is CancellationException) throw e
-                return
-            }
-
-        val tx = (notification as? PaymentReceivedNotification)?.notification ?: return
-
-        // A payment carrying a NIP-57 zap request is already shown by ZapNotification.
-        if (tx.parsedMetadata()?.nostr != null) return
-
-        NwcPaymentNotifier.notify(context, account, tx)
-    }
 }
@@ -28,6 +28,7 @@ import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.marmot.
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.metadata.AccountMetadataEoseManager
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip01Notifications.AccountNotificationsEoseFromInboxRelaysManager
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip01Notifications.AccountNotificationsHistoryEoseManager
+import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip47WalletConnect.NwcNotificationsEoseManager
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsEoseManager
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.nip59GiftWraps.AccountGiftWrapsHistoryEoseManager
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountFeedContentStates
@@ -71,6 +72,8 @@ class AccountFilterAssembler(
            AccountDraftsEoseManager(client, ::allKeys),
            notifications,
            notificationsHistory,
+            // Live tail: NIP-47 wallet notifications (payment_received) on each connected wallet's own relay.
+            NwcNotificationsEoseManager(client, ::allKeys),
            MarmotGroupEventsEoseManager(client, ::allKeys),
        )

@@ -0,0 +1,151 @@
+/*
+ * 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.amethyst.service.relayClient.reqCommand.account.nip47WalletConnect
+
+import com.vitorpamplona.amethyst.model.User
+import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.PerUserEoseManager
+import com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.AccountQueryState
+import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap
+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.pool.RelayBasedFilter
+import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
+import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip47WalletConnect.events.NwcNotificationEvent
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.sample
+import kotlinx.coroutines.launch
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * Always-on subscription for NIP-47 wallet notifications (kind 23197/23196),
+ * grouped in [com.vitorpamplona.amethyst.service.relayClient.reqCommand.account.AccountFilterAssembler]
+ * so it shares the exact lifecycle of the account's zap/notification inbox
+ * subscription: open while logged in, kept warm in the background by
+ * `NotificationRelayService`, torn down on logout.
+ *
+ * For each configured NWC wallet it queries the wallet's own relay (not the inbox
+ * relays) for notifications `p`-tagged to that wallet's client pubkey. Unlike the
+ * inbox managers — whose events land in `LocalCache` — these are ephemeral and
+ * encrypted per wallet, so decryption + fan-out happens in [onEvent] via
+ * `NwcSignerState.handleIncomingNotification`, which publishes non-zap payments to
+ * `incomingNonZapPayments`. Since `since` is floored at watch start, relaunching
+ * never replays old payments; the `seen` set de-dupes re-delivery on reconnect.
+ */
+class NwcNotificationsEoseManager(
+    client: INostrClient,
+    allKeys: () -> Set<AccountQueryState>,
+) : PerUserEoseManager<AccountQueryState>(client, allKeys) {
+    private val startSince = TimeUtils.now()
+    private val seen = ConcurrentHashMap.newKeySet<HexKey>()
+    private val userJobMap = mutableMapOf<User, List<Job>>()
+
+    override fun user(key: AccountQueryState) = key.account.userProfile()
+
+    override fun updateFilter(
+        key: AccountQueryState,
+        since: SincePerRelayMap?,
+    ): List<RelayBasedFilter> {
+        val account = key.account
+
+        return account.settings.nwcWallets.value.flatMap { wallet ->
+            val signer = account.nip47SignerState.buildSigner(wallet.uri) ?: return@flatMap emptyList()
+
+            // Skip wallets that advertise no notification support; warm the cache
+            // (and re-evaluate on the next invalidation) when the info is unknown.
+            account.nwcInfoCache.refreshIfStale(wallet.uri)
+            if (account.nwcInfoCache.current(wallet.uri)?.supportsNotifications() == false) {
+                return@flatMap emptyList()
+            }
+
+            listOf(
+                RelayBasedFilter(
+                    relay = wallet.uri.relayUri,
+                    filter =
+                        Filter(
+                            kinds = listOf(NwcNotificationEvent.KIND, NwcNotificationEvent.LEGACY_KIND),
+                            authors = listOf(wallet.uri.pubKeyHex),
+                            tags = mapOf("p" to listOf(signer.pubKey)),
+                            since = since?.get(wallet.uri.relayUri)?.time ?: startSince,
+                        ),
+                ),
+            )
+        }
+    }
+
+    @OptIn(FlowPreview::class)
+    override fun newSub(key: AccountQueryState): Subscription {
+        val user = user(key)
+        userJobMap[user]?.forEach { it.cancel() }
+        userJobMap[user] =
+            listOf(
+                // Re-subscribe when the wallet set changes so relays/keys are added or dropped.
+                key.account.scope.launch(Dispatchers.IO) {
+                    key.account.settings.nwcWallets
+                        .sample(1000)
+                        .collectLatest { invalidateFilters() }
+                },
+            )
+
+        return requestNewSubscription(
+            object : SubscriptionListener {
+                override fun onEose(
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) {
+                    newEose(key, relay, TimeUtils.now(), forFilters)
+                }
+
+                override fun onEvent(
+                    event: Event,
+                    isLive: Boolean,
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) {
+                    if (isLive) newEose(key, relay, TimeUtils.now(), forFilters)
+
+                    val notification = event as? NwcNotificationEvent ?: return
+                    if (seen.add(notification.id)) {
+                        key.account.scope.launch(Dispatchers.IO) {
+                            key.account.nip47SignerState.handleIncomingNotification(notification)
+                        }
+                    }
+                }
+            },
+        )
+    }
+
+    override fun endSub(
+        key: User,
+        subId: String,
+    ) {
+        super.endSub(key, subId)
+        userJobMap[key]?.forEach { it.cancel() }
+        userJobMap.remove(key)
+    }
+}