feat(wallet): time-windowed relay sub + coalesce zap bursts

Two improvements to onchain history attribution: 1. Coalesce UI updates. The OnchainZapEvent observation in the ViewModel now passes through .sample(500), so a relay flooding a backlog of historical zaps in quick succession produces at most one map update per 500ms instead of one per event. The downstream combine + StateFlow conflation handles the rest. 2. Bound relay queries to the visible window. New OnchainZapsFilterAssembler (SingleSubEoseManager-based) wakes only while the screen is on top and asks the user's inbox/ outbox relays for kind-8333 events with since = the oldest visible blockTime — incoming (p-tag = user) and outgoing (authors = user) — so we don't drag the whole NIP-BC history when the user only scrolled through last week. As the user pages back, oldestBlockTime drops, the assembler re-subscribes with a wider window. OnchainTransactionsScreen now wires OnchainZapsFilterAssemblerSubscription with the StateFlow-derived window, and the coordinator owns the parent assembler so the lifecycle matches other screen subs.
2026-08-09 16:14:40 +00:00 · 2026-05-19 00:52:12 +00:00
parent 020d5195b5
commit 7dce3a420f
5 changed files with 226 additions and 2 deletions
@@ -60,6 +60,7 @@ import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.datasource.RelayInfo
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.shorts.datasource.ShortsFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.threadview.datasources.ThreadFilterAssembler
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.video.datasource.VideoFilterAssembler
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.wallet.datasource.OnchainZapsFilterAssembler
 import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayOfflineTracker
 import com.vitorpamplona.quartz.nip01Core.relay.client.auth.IAuthStatus
@@ -127,6 +128,9 @@ class RelaySubscriptionsCoordinator(
    // active when sending zaps via NWC
    val nwc = NWCPaymentFilterAssembler(client)

+    // active when the wallet's on-chain transactions screen is on top.
+    val onchainZaps = OnchainZapsFilterAssembler(client)
+
    val all =
        listOf(
            account,
@@ -167,6 +171,7 @@ class RelaySubscriptionsCoordinator(
            relayInfoNip66,
            chess,
            nwc,
+            onchainZaps,
        )

    fun destroy() = all.forEach { it.destroy() }
@@ -67,6 +67,7 @@ import com.vitorpamplona.amethyst.ui.note.UserPicture
 import com.vitorpamplona.amethyst.ui.note.UsernameDisplay
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
 import com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.rooms.LoadUser
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.wallet.datasource.OnchainZapsFilterAssemblerSubscription
 import com.vitorpamplona.amethyst.ui.stringRes
 import com.vitorpamplona.amethyst.ui.theme.bitcoinColor
 import java.text.NumberFormat
@@ -88,6 +89,13 @@ fun OnchainTransactionsScreen(
        viewModel.fetchTransactions()
    }

+    val windowSince by viewModel.oldestBlockTime.collectAsState()
+    OnchainZapsFilterAssemblerSubscription(
+        user = accountViewModel.account.userProfile(),
+        windowSinceSeconds = windowSince,
+        accountViewModel = accountViewModel,
+    )
+
    val transactions by viewModel.filteredTransactions.collectAsState()
    val isLoading by viewModel.isLoading.collectAsState()
    val isLoadingMore by viewModel.isLoadingMore.collectAsState()
@@ -31,10 +31,14 @@ import com.vitorpamplona.quartz.nipBCOnchainZaps.chain.OnchainBackend
 import com.vitorpamplona.quartz.nipBCOnchainZaps.taproot.TaprootAddress
 import com.vitorpamplona.quartz.nipBCOnchainZaps.zap.OnchainZapEvent
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.sample
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
@@ -65,6 +69,8 @@ data class OnchainTxView(
        }
 }

+private const val SAMPLE_MILLIS = 500L
+
 class OnchainTransactionsViewModel : ViewModel() {
    private var address: String? = null
    private var backend: OnchainBackend? = null
@@ -113,6 +119,21 @@ class OnchainTransactionsViewModel : ViewModel() {
    private val _displayAddress = MutableStateFlow<String?>(null)
    val displayAddress = _displayAddress.asStateFlow()

+    /**
+     * Oldest `blockTime` (or null = unconfirmed/unknown) across the currently
+     * loaded chain rows. The screen feeds this into the relay subscription so
+     * we only ask relays for zap events from that point onwards instead of
+     * pulling the user's entire NIP-BC history.
+     *
+     * Returns null while the page is empty (no constraint — fall back to the
+     * relay's full history for the first call) or when the oldest row is in
+     * the mempool (we don't have a timestamp to bound on).
+     */
+    val oldestBlockTime: StateFlow<Long?> =
+        chainTxs
+            .map { txs -> txs.minOfOrNull { it.blockTime ?: Long.MAX_VALUE }?.takeIf { it != Long.MAX_VALUE } }
+            .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), null)
+
    fun init(accountViewModel: AccountViewModel) {
        if (address != null) return
        val pubKey = accountViewModel.account.signer.pubKey
@@ -122,10 +143,13 @@ class OnchainTransactionsViewModel : ViewModel() {

        // Stand up the reactive zap cache. The two filters cover both
        // directions; LocalCache's filter index narrows the fanout, so we
-        // only get woken for kind-8333 events that mention us.
+        // only get woken for kind-8333 events that mention us. .sample(500)
+        // coalesces bursts so a relay flooding 50 historical zaps at once
+        // produces at most one UI update.
        viewModelScope.launch(Dispatchers.IO) {
            val incoming = Filter(kinds = listOf(OnchainZapEvent.KIND), tags = mapOf("p" to listOf(pubKey)))
            val outgoing = Filter(kinds = listOf(OnchainZapEvent.KIND), authors = listOf(pubKey))
+            @OptIn(FlowPreview::class)
            combine(
                LocalCache.observeEvents<OnchainZapEvent>(incoming),
                LocalCache.observeEvents<OnchainZapEvent>(outgoing),
@@ -136,7 +160,7 @@ class OnchainTransactionsViewModel : ViewModel() {
                    merged[txid] = z
                }
                merged
-            }.collect { zapsByTxid.value = it }
+            }.sample(SAMPLE_MILLIS).collect { zapsByTxid.value = it }
        }
    }

@@ -0,0 +1,137 @@
+/*
+ * 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.ui.screen.loggedIn.wallet.datasource
+
+import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.ComposeSubscriptionManager
+import com.vitorpamplona.amethyst.commons.relayClient.eoseManagers.SingleSubEoseManager
+import com.vitorpamplona.amethyst.model.LocalCache
+import com.vitorpamplona.amethyst.model.User
+import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
+import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nipBCOnchainZaps.zap.OnchainZapEvent
+
+/**
+ * Per-screen subscription key for the on-chain wallet history.
+ *
+ * @property user the logged-in user whose taproot address is being viewed.
+ * @property windowSinceSeconds lower bound (`since`) for the relay query, in
+ *           seconds. Derived from the oldest visible chain transaction's
+ *           `blockTime` so we only fetch zap events relevant to what the
+ *           screen will actually display. `null` disables the lower bound —
+ *           the relay returns the full history capped by `limit`.
+ */
+data class OnchainZapsQueryState(
+    val user: User,
+    val windowSinceSeconds: Long?,
+)
+
+private const val PER_FILTER_LIMIT = 500
+
+/**
+ * Build per-relay filters for kind-8333 zaps that involve [pubkey] — either
+ * as the recipient (`p`-tag, incoming) or the author (outgoing). The
+ * effective `since` is the larger of the visible-window lower bound (so we
+ * don't drag the whole history every time) and any EOSE-tracked timestamp
+ * from a previous round (so we only ask the relay for events newer than
+ * what we already have).
+ */
+private fun filterOnchainZaps(
+    pubkey: String,
+    relays: Collection<NormalizedRelayUrl>,
+    windowSinceSeconds: Long?,
+    eoseSince: SincePerRelayMap?,
+): List<RelayBasedFilter> =
+    relays.flatMap { relay ->
+        val eoseTime = eoseSince?.get(relay)?.time
+        val since =
+            listOfNotNull(windowSinceSeconds, eoseTime)
+                .maxOrNull()
+                ?.takeIf { it > 0L }
+
+        listOf(
+            RelayBasedFilter(
+                relay = relay,
+                filter =
+                    Filter(
+                        kinds = listOf(OnchainZapEvent.KIND),
+                        tags = mapOf("p" to listOf(pubkey)),
+                        limit = PER_FILTER_LIMIT,
+                        since = since,
+                    ),
+            ),
+            RelayBasedFilter(
+                relay = relay,
+                filter =
+                    Filter(
+                        kinds = listOf(OnchainZapEvent.KIND),
+                        authors = listOf(pubkey),
+                        limit = PER_FILTER_LIMIT,
+                        since = since,
+                    ),
+            ),
+        )
+    }
+
+class OnchainZapsFilterSubAssembler(
+    client: INostrClient,
+    allKeys: () -> Set<OnchainZapsQueryState>,
+) : SingleSubEoseManager<OnchainZapsQueryState>(client, allKeys) {
+    override fun updateFilter(
+        keys: List<OnchainZapsQueryState>,
+        since: SincePerRelayMap?,
+    ): List<RelayBasedFilter> {
+        val key = keys.firstOrNull() ?: return emptyList()
+        val user = key.user
+        val relays: Collection<NormalizedRelayUrl> =
+            user.inboxRelays()?.ifEmpty { null }
+                ?: user.outboxRelays()?.ifEmpty { null }
+                ?: user.allUsedRelaysOrNull()
+                ?: LocalCache.relayHints.hintsForKey(user.pubkeyHex)
+        if (relays.isEmpty()) return emptyList()
+
+        // Take the tightest window across all live keys — if any subscriber
+        // is showing older transactions, widen the query to cover them too.
+        val windowSince =
+            keys
+                .mapNotNull { it.windowSinceSeconds }
+                .minOrNull()
+                ?.takeIf { keys.all { k -> k.windowSinceSeconds != null } }
+
+        return filterOnchainZaps(user.pubkeyHex, relays, windowSince, since)
+    }
+
+    override fun distinct(key: OnchainZapsQueryState) = key.user.pubkeyHex
+}
+
+class OnchainZapsFilterAssembler(
+    client: INostrClient,
+) : ComposeSubscriptionManager<OnchainZapsQueryState>() {
+    private val sub = OnchainZapsFilterSubAssembler(client, ::allKeys)
+
+    override fun invalidateFilters() = sub.invalidateFilters()
+
+    override fun invalidateKeys() = invalidateFilters()
+
+    override fun destroy() = sub.destroy()
+}
@@ -0,0 +1,50 @@
+/*
+ * 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.ui.screen.loggedIn.wallet.datasource
+
+import androidx.compose.runtime.Composable
+import androidx.compose.runtime.remember
+import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.LifecycleAwareKeyDataSourceSubscription
+import com.vitorpamplona.amethyst.model.User
+import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
+
+/**
+ * Lifecycle-aware relay subscription for kind-8333 zaps involving [user],
+ * bounded by [windowSinceSeconds] (the oldest visible chain transaction's
+ * blockTime) so the relay only ships zaps that could plausibly attribute to
+ * what we're showing.
+ *
+ * Re-subscribes whenever either input changes — scrolling back into older
+ * transactions widens the window and re-queries; switching accounts swaps
+ * the key entirely.
+ */
+@Composable
+fun OnchainZapsFilterAssemblerSubscription(
+    user: User,
+    windowSinceSeconds: Long?,
+    accountViewModel: AccountViewModel,
+) {
+    val state =
+        remember(user, windowSinceSeconds) {
+            OnchainZapsQueryState(user, windowSinceSeconds)
+        }
+    LifecycleAwareKeyDataSourceSubscription(state, accountViewModel.dataSources().onchainZaps)
+}