diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt index b799d3b5ea..f4222d8d6f 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt @@ -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() } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt index 15f9fdedbc..5ce3a777eb 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt @@ -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(extraBufferCapacity = 32, onBufferOverflow = BufferOverflow.DROP_OLDEST) + val incomingNonZapPayments: SharedFlow = _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. */ diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NwcPaymentNotificationWatcher.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NwcPaymentNotificationWatcher.kt index 35fe138cb6..f74bddd9b8 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NwcPaymentNotificationWatcher.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NwcPaymentNotificationWatcher.kt @@ -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, ) { 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, - ) = 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() - 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 = - callbackFlow { - val subId = newSubId() - val listener = - object : SubscriptionListener { - override fun onEvent( - event: Event, - isLive: Boolean, - relay: NormalizedRelayUrl, - forFilters: List?, - ) { - 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) - } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/AccountFilterAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/AccountFilterAssembler.kt index 7220fc739d..f049140194 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/AccountFilterAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/AccountFilterAssembler.kt @@ -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), ) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip47WalletConnect/NwcNotificationsEoseManager.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip47WalletConnect/NwcNotificationsEoseManager.kt new file mode 100644 index 0000000000..fc3a5b8f3a --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/account/nip47WalletConnect/NwcNotificationsEoseManager.kt @@ -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, +) : PerUserEoseManager(client, allKeys) { + private val startSince = TimeUtils.now() + private val seen = ConcurrentHashMap.newKeySet() + private val userJobMap = mutableMapOf>() + + override fun user(key: AccountQueryState) = key.account.userProfile() + + override fun updateFilter( + key: AccountQueryState, + since: SincePerRelayMap?, + ): List { + 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?, + ) { + newEose(key, relay, TimeUtils.now(), forFilters) + } + + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + 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) + } +}