fix(nwc): trim subscription filter on response to avoid relay replays

Some NWC relays (notably relay-nwc.rizful.com) replay cached kind-23195
events every time a REQ filter mutates. Because we previously left each
reqId in the filter for a full 60s after sending its request, every new
NWC call added a fresh reqId to a filter that still listed several
stale ones, and the relay would re-deliver every matching cached
response. That arrived here as a flurry of NO_MATCH events, wasted work
on each new send, and made the transactions screen feel stuck.

Cleanup is now driven by the response itself:

- NwcSignerState cancels the 60s safety-net job on response and asks
  the assembler to drop the filter through unsubscribeSoon, which is
  debounced 1.5s so a burst of responses produces one relay-side
  filter update instead of one per response.
- subscribeAndFlush drains any pending unsubscribes synchronously
  before adding the new query, so the relay sees a direct {old}->{new}
  transition instead of the {old}->{old,new}->{new} that triggered
  Rizful's replay path.
- The 60s timeout remains as a safety net for wallets that never reply.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

amethyst/src/main/java/com/vitorpamplona/amethyst/model/nip47WalletConnect/NwcSignerState.kt

@@ -157,14 +157,20 @@ class NwcSignerState(
         // be missed.
         assembler.subscribeAndFlush(filter)
 
-        scope.launch(Dispatchers.IO) {
-            delay(60000)
-            assembler.unsubscribe(filter)
-        }
+        // Safety net: drop the filter after 60s if the wallet never replies.
+        // The happy path (response arrives) cancels this job and unsubscribes
+        // through assembler.unsubscribeSoon, which debounces.
+        val timeoutJob =
+            scope.launch(Dispatchers.IO) {
+                delay(60000)
+                assembler.unsubscribe(filter)
+            }
 
         val responseCache = NostrWalletConnectResponseCache(walletSigner)
         cache.consume(event, null, true, walletService.relayUri) {
+            timeoutJob.cancel()
             onResponse(responseCache.decryptResponse(it))
+            assembler.unsubscribeSoon(filter)
         }
 
         return Pair(event, walletService.relayUri)
@@ -195,13 +201,19 @@ class NwcSignerState(
         // See sendNwcRequestToWallet above for the rationale.
         assembler.subscribeAndFlush(filter)
 
-        scope.launch(Dispatchers.IO) {
-            delay(60000) // waits 1 minute to complete payment.
-            assembler.unsubscribe(filter)
-        }
+        // Safety net: drop the filter after 60s if the wallet never replies.
+        // The happy path (response arrives) cancels this job and instead
+        // hands off to assembler.unsubscribeSoon, which debounces.
+        val timeoutJob =
+            scope.launch(Dispatchers.IO) {
+                delay(60000) // waits 1 minute to complete payment.
+                assembler.unsubscribe(filter)
+            }
 
         cache.consume(event, zappedNote, true, walletService.relayUri) {
+            timeoutJob.cancel()
             onResponse(decryptResponse(it))
+            assembler.unsubscribeSoon(filter)
         }
 
         return Pair(event, walletService.relayUri)

amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/reqCommand/nwc/NWCPaymentFilterAssembler.kt

@@ -25,6 +25,14 @@ import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManager
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 
 // This allows multiple screen to be listening to tags, even the same tag.
 // The subscription filter carries `#e: [request id]` plus `#p: [client pubkey]`.
@@ -48,6 +56,20 @@ class NWCPaymentFilterAssembler(
             NWCPaymentWatcherSubAssembler(client, ::allKeys),
         )
 
+    // Trailing-edge debounce: after a response arrives we wait this long
+    // before actually removing the request id from the subscription filter.
+    // If more responses (or a fresh subscribe of any query) arrive within
+    // the window they coalesce into ONE relay-side filter update. This
+    // matters because some NWC relays (e.g. relay-nwc.rizful.com) replay
+    // cached kind-23195 events every time the REQ filter changes — so each
+    // unbatched filter mutation triggers a fresh replay storm.
+    private val unsubDelayMs = 1500L
+    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
+    private val pendingLock = Any()
+    private val pendingUnsubs = mutableSetOf<NWCPaymentQueryState>()
+
+    @Volatile private var pendingJob: Job? = null
+
     override fun invalidateFilters() = group.forEach { it.invalidateFilters() }
 
     override fun invalidateKeys() = invalidateFilters()
@@ -58,11 +80,69 @@ class NWCPaymentFilterAssembler(
      * ephemeral event (kind 23195) and the subscription must be active on the
      * relay before we publish the request event — otherwise the relay drops
      * the response with no replay.
+     *
+     * Also drains any pending unsubscribes synchronously so the relay sees
+     * a direct {old} → {new} transition instead of {old} → {old, new} → {new}.
+     * The intermediate `{old, new}` state is the trigger that makes Rizful's
+     * relay replay cached responses for `old`.
      */
     fun subscribeAndFlush(query: NWCPaymentQueryState) {
+        drainPendingUnsubs()
         subscribe(query)
         group.forEach { it.forceInvalidate() }
     }
 
-    override fun destroy() = group.forEach { it.destroy() }
+    /**
+     * Removes [query] from the active subscription after a short debounce.
+     * Repeated calls within the window coalesce into a single batch
+     * unsubscribe. Idempotent: calling with a query that's already been
+     * unsubscribed is a no-op.
+     */
+    fun unsubscribeSoon(query: NWCPaymentQueryState) {
+        val oldJob: Job?
+        synchronized(pendingLock) {
+            pendingUnsubs.add(query)
+            oldJob = pendingJob
+            pendingJob =
+                scope.launch {
+                    try {
+                        delay(unsubDelayMs)
+                        val batch =
+                            synchronized(pendingLock) {
+                                val copy = pendingUnsubs.toList()
+                                pendingUnsubs.clear()
+                                pendingJob = null
+                                copy
+                            }
+                        if (batch.isNotEmpty()) {
+                            unsubscribe(batch)
+                        }
+                    } catch (_: CancellationException) {
+                        // Superseded by a newer schedule; the replacement
+                        // job owns the pending set from here.
+                    }
+                }
+        }
+        oldJob?.cancel()
+    }
+
+    private fun drainPendingUnsubs() {
+        val batch: List<NWCPaymentQueryState>
+        val cancelled: Job?
+        synchronized(pendingLock) {
+            batch = pendingUnsubs.toList()
+            pendingUnsubs.clear()
+            cancelled = pendingJob
+            pendingJob = null
+        }
+        cancelled?.cancel()
+        if (batch.isNotEmpty()) {
+            unsubscribe(batch)
+        }
+    }
+
+    override fun destroy() {
+        scope.cancel()
+        group.forEach { it.destroy() }
+    }
 }