From 4888d30f6cf08a7ed7f95941154fee017742ebce Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Apr 2026 00:20:01 +0000 Subject: [PATCH] refactor(live-chat): lift NIP-53 live-stream state into commons Address the architectural critique from self-review. The leaderboard math and the shared system card are now platform-neutral primitives in commons, the on-UI-thread aggregation is gone, and two latent correctness bugs around subscription lifecycle and zap routing are fixed. - Introduce a pure LiveActivityTopZappersAggregator in commons that takes plain ZapContribution values and returns a sorted, deduped, anon-bucketed top-N list. Covered by a unit-test suite that is Compose/Android-independent. - Introduce LiveStreamTopZappersViewModel in commons/commonMain. It owns two partitioned contribution maps (stream-scoped and goal-scoped), mutates them under a Mutex on the IO dispatcher in response to channel.changesFlow and Note.zaps.stateFlow emissions, and publishes the aggregator result via a StateFlow>. UI is now a dumb consumer with no aggregation logic left in the composable. - Move StreamSystemCard to commons/commonMain/compose/ so Desktop can consume it alongside Android. - Fix attachZapToLiveActivityChannel to run even when a zap receipt was already consumed by another subscription. Previously a zap first seen by notifications/profile would never reach the stream's channel cache; addNote is already idempotent so this is a safe retro-route. - Fix ChannelFilterAssemblerSubscription to re-invalidate filters when a live-activity channel's metadata arrives. The 30311 stream event usually lands after the initial assembler run, so the goal-tag-driven subscription never fired. Now it re-runs as metadata resolves and the goal id is discovered. https://claude.ai/code/session_01WJA5PvDABegBYUu6h42YZi --- .../amethyst/model/LocalCache.kt | 10 +- .../chats/feed/types/RenderChatClip.kt | 1 + .../chats/feed/types/RenderChatRaid.kt | 1 + .../chats/feed/types/RenderChatZap.kt | 1 + .../ChannelFilterAssemblerSubscription.kt | 18 ++ .../header/LiveStreamTopZappers.kt | 131 ++++--------- .../nip53LiveActivities}/StreamSystemCard.kt | 6 +- .../LiveActivityTopZappersAggregator.kt | 106 +++++++++++ .../LiveStreamTopZappersViewModel.kt | 173 ++++++++++++++++++ .../LiveActivityTopZappersAggregatorTest.kt | 153 ++++++++++++++++ 10 files changed, 497 insertions(+), 103 deletions(-) rename {amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types => commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/compose/nip53LiveActivities}/StreamSystemCard.kt (89%) create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregator.kt create mode 100644 commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/LiveStreamTopZappersViewModel.kt create mode 100644 commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregatorTest.kt diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt index c58e725ef..ac8d998f3 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt @@ -1555,8 +1555,14 @@ object LocalCache : ILocalCache, ICacheProvider { wasVerified: Boolean, ): Boolean { val note = getOrCreateNote(event.id) - // Already processed this event. - if (note.event != null) return false + + // Already processed this event — still ensure it's routed into any live-activity + // channel(s) it references. A zap that was first consumed by, e.g., the notifications + // subscription must still appear in the stream's chat when the user opens the stream. + if (note.event != null) { + attachZapToLiveActivityChannel(event, note, relay) + return false + } if (wasVerified || justVerify(event)) { val existingZapRequest = event.zapRequest?.id?.let { getNoteIfExists(it) } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatClip.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatClip.kt index 4561d7025..bb5a8f39e 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatClip.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatClip.kt @@ -36,6 +36,7 @@ import androidx.compose.ui.layout.ContentScale import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.unit.dp import com.vitorpamplona.amethyst.R +import com.vitorpamplona.amethyst.commons.compose.nip53LiveActivities.StreamSystemCard import com.vitorpamplona.amethyst.model.Note import com.vitorpamplona.amethyst.service.playback.composable.VideoView import com.vitorpamplona.amethyst.ui.navigation.navs.INav diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatRaid.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatRaid.kt index 3bc3d1bf3..5838ff092 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatRaid.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatRaid.kt @@ -39,6 +39,7 @@ import androidx.compose.ui.Modifier import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.unit.dp import com.vitorpamplona.amethyst.R +import com.vitorpamplona.amethyst.commons.compose.nip53LiveActivities.StreamSystemCard import com.vitorpamplona.amethyst.model.Note import com.vitorpamplona.amethyst.ui.navigation.navs.INav import com.vitorpamplona.amethyst.ui.navigation.routes.Route diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatZap.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatZap.kt index b066890b8..967ca1ec5 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatZap.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/RenderChatZap.kt @@ -36,6 +36,7 @@ import androidx.compose.ui.Modifier import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.unit.dp import com.vitorpamplona.amethyst.R +import com.vitorpamplona.amethyst.commons.compose.nip53LiveActivities.StreamSystemCard import com.vitorpamplona.amethyst.model.Note import com.vitorpamplona.amethyst.ui.navigation.navs.INav import com.vitorpamplona.amethyst.ui.note.CrossfadeToDisplayComment diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/datasource/ChannelFilterAssemblerSubscription.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/datasource/ChannelFilterAssemblerSubscription.kt index ea531c1bd..e8231e71b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/datasource/ChannelFilterAssemblerSubscription.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/datasource/ChannelFilterAssemblerSubscription.kt @@ -21,8 +21,12 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.datasource import androidx.compose.runtime.Composable +import androidx.compose.runtime.LaunchedEffect +import androidx.compose.runtime.getValue import androidx.compose.runtime.remember +import androidx.lifecycle.compose.collectAsStateWithLifecycle import com.vitorpamplona.amethyst.commons.model.Channel +import com.vitorpamplona.amethyst.commons.model.nip53LiveActivities.LiveActivitiesChannel import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.KeyDataSourceSubscription import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel @@ -40,4 +44,18 @@ fun ChannelFilterAssemblerSubscription( } KeyDataSourceSubscription(state, dataSource) + + // Live streams need the 30311 event to populate `channel.info` before we can read its + // `goal` tag and add the goal+zap subscriptions. Re-invalidate when metadata changes so + // the goal filter fires as soon as the stream event arrives. + if (channel is LiveActivitiesChannel) { + val metadataState by channel + .flow() + .metadata.stateFlow + .collectAsStateWithLifecycle() + val goalId = channel.info?.goalEventId() + LaunchedEffect(goalId, metadataState) { + dataSource.invalidateFilters() + } + } } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/nip53LiveActivities/header/LiveStreamTopZappers.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/nip53LiveActivities/header/LiveStreamTopZappers.kt index 15fdec655..1624e3f51 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/nip53LiveActivities/header/LiveStreamTopZappers.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/nip53LiveActivities/header/LiveStreamTopZappers.kt @@ -43,11 +43,14 @@ import androidx.compose.runtime.setValue import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.unit.dp +import androidx.lifecycle.ViewModel +import androidx.lifecycle.ViewModelProvider import androidx.lifecycle.compose.collectAsStateWithLifecycle +import androidx.lifecycle.viewmodel.compose.viewModel import com.vitorpamplona.amethyst.R import com.vitorpamplona.amethyst.commons.model.nip53LiveActivities.LiveActivitiesChannel -import com.vitorpamplona.amethyst.model.Note -import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.observeNoteZaps +import com.vitorpamplona.amethyst.commons.nip53LiveActivities.TopZapperEntry +import com.vitorpamplona.amethyst.commons.viewmodels.LiveStreamTopZappersViewModel import com.vitorpamplona.amethyst.ui.navigation.navs.INav import com.vitorpamplona.amethyst.ui.navigation.routes.Route import com.vitorpamplona.amethyst.ui.note.UserPicture @@ -58,27 +61,14 @@ import com.vitorpamplona.amethyst.ui.stringRes import com.vitorpamplona.amethyst.ui.theme.BitcoinOrange import com.vitorpamplona.amethyst.ui.theme.Size16Modifier import com.vitorpamplona.amethyst.ui.theme.Size24dp -import com.vitorpamplona.quartz.nip01Core.core.HexKey -import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent -import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent import java.math.BigDecimal -private const val TOP_ZAPPERS_LIMIT = 10 - -/** Sentinel key used to bucket every anonymous/private zap into a single leaderboard entry. */ -private const val ANON_KEY = "anon" - -private data class TopZapperEntry( - val zapperPubKey: HexKey, - val totalSats: BigDecimal, - val isAnonymous: Boolean, -) - /** - * Horizontally-scrollable leaderboard of the top zappers on a live stream. - * Aggregates zaps from both the stream's #a subscription and the attached - * NIP-75 zap goal (if any), de-duplicating by zap-receipt id. Matches - * zap.stream's TopZappers UX: pill chips with avatar + lightning + sats. + * Horizontally-scrollable leaderboard of the top zappers on a live stream. Matches + * zap.stream's TopZappers look: pill chips with avatar + lightning + sats. + * + * State is owned by [LiveStreamTopZappersViewModel], which maintains the aggregation + * incrementally off the UI thread and publishes a stable `List`. */ @Composable fun LiveStreamTopZappers( @@ -86,43 +76,33 @@ fun LiveStreamTopZappers( accountViewModel: AccountViewModel, nav: INav, ) { + val topZappersVm: LiveStreamTopZappersViewModel = + viewModel( + key = "TopZappers-${channel.address.toValue()}", + factory = + object : ViewModelProvider.Factory { + @Suppress("UNCHECKED_CAST") + override fun create(modelClass: Class): T = LiveStreamTopZappersViewModel(channel) as T + }, + ) + + // Track the goal note lifecycle — pass it to the VM as it resolves, clear on channel change. val goalId = channel.info?.goalEventId() - if (goalId != null) { - var goalNote by remember(goalId) { mutableStateOf(accountViewModel.getNoteIfExists(goalId)) } - if (goalNote == null) { - LaunchedEffect(goalId) { - goalNote = accountViewModel.checkGetOrCreateNote(goalId) + var goalNoteHolder by remember(goalId) { + mutableStateOf(accountViewModel.getNoteIfExists(goalId)) + } + LaunchedEffect(goalId) { + if (goalNoteHolder == null) { + goalNoteHolder = accountViewModel.checkGetOrCreateNote(goalId) } + topZappersVm.setGoalNote(goalNoteHolder) } - TopZappersStrip(channel, goalNote, accountViewModel, nav) } else { - TopZappersStrip(channel, null, accountViewModel, nav) + LaunchedEffect(channel) { topZappersVm.setGoalNote(null) } } -} -@Composable -private fun TopZappersStrip( - channel: LiveActivitiesChannel, - goalNote: Note?, - accountViewModel: AccountViewModel, - nav: INav, -) { - // Trigger recomposition whenever a new zap lands in the channel cache. - val channelTick by channel.changesFlow().collectAsStateWithLifecycle(initialValue = null) - - // Trigger recomposition when zaps arrive for the goal note. - val goalZapsState = - if (goalNote != null) { - observeNoteZaps(goalNote, accountViewModel).value - } else { - null - } - - val entries = - remember(channel, channelTick, goalNote, goalZapsState) { - aggregateTopZappers(channel, goalNote) - } + val entries by topZappersVm.topZappers.collectAsStateWithLifecycle() if (entries.isEmpty()) return @@ -131,7 +111,7 @@ private fun TopZappersStrip( horizontalArrangement = Arrangement.spacedBy(6.dp), contentPadding = PaddingValues(horizontal = 4.dp), ) { - items(entries, key = { it.zapperPubKey }) { entry -> + items(entries, key = { it.bucketKey }) { entry -> ZapperPill(entry, accountViewModel, nav) } } @@ -147,7 +127,7 @@ private fun ZapperPill( if (entry.isAnonymous) { Modifier } else { - Modifier.clickable { nav.nav(Route.Profile(entry.zapperPubKey)) } + Modifier.clickable { nav.nav(Route.Profile(entry.bucketKey)) } } Surface( @@ -168,7 +148,7 @@ private fun ZapperPill( ) } else { UserPicture( - userHex = entry.zapperPubKey, + userHex = entry.bucketKey, size = Size24dp, accountViewModel = accountViewModel, nav = nav, @@ -176,53 +156,10 @@ private fun ZapperPill( } ZapIcon(Size16Modifier, BitcoinOrange) Text( - text = showAmountInteger(entry.totalSats), + text = showAmountInteger(BigDecimal.valueOf(entry.totalSats)), style = MaterialTheme.typography.labelMedium, ) Spacer(Modifier.padding(start = 2.dp)) } } } - -private fun aggregateTopZappers( - channel: LiveActivitiesChannel, - goalNote: Note?, -): List { - // receiptId -> (zapperBucketKey, sats) — dedupes a zap that appears via both #a and #e. - val byReceipt = HashMap>() - - // Stream-level zaps routed to the channel cache. - channel.notes.forEach { _, note -> - val ev = note.event - if (ev is LnZapEvent) { - val request = ev.zapRequest ?: return@forEach - val sats = ev.amount() ?: return@forEach - byReceipt[note.idHex] = bucketKeyFor(request) to sats - } - } - - // Goal-scoped zaps attached to the goal note via #e. - goalNote?.zaps?.forEach { (zapRequestNote, receiptNote) -> - val receiptEv = receiptNote?.event as? LnZapEvent ?: return@forEach - val request = zapRequestNote.event as? LnZapRequestEvent ?: return@forEach - val sats = receiptEv.amount() ?: return@forEach - byReceipt[receiptNote.idHex] = bucketKeyFor(request) to sats - } - - if (byReceipt.isEmpty()) return emptyList() - - val totals = HashMap(byReceipt.size) - byReceipt.values.forEach { (pk, sats) -> - totals[pk] = (totals[pk] ?: BigDecimal.ZERO) + sats - } - - return totals.entries - .asSequence() - .sortedByDescending { it.value } - .take(TOP_ZAPPERS_LIMIT) - .map { TopZapperEntry(it.key, it.value, isAnonymous = it.key == ANON_KEY) } - .toList() -} - -/** Returns the aggregation key for a zap request: real pubkey, or the anon sentinel for any `anon`-tagged zap. */ -private fun bucketKeyFor(request: LnZapRequestEvent): HexKey = if (request.tags.any { it.isNotEmpty() && it[0] == "anon" }) ANON_KEY else request.pubKey diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/StreamSystemCard.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/compose/nip53LiveActivities/StreamSystemCard.kt similarity index 89% rename from amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/StreamSystemCard.kt rename to commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/compose/nip53LiveActivities/StreamSystemCard.kt index ae6d80efb..52355902d 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/feed/types/StreamSystemCard.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/compose/nip53LiveActivities/StreamSystemCard.kt @@ -18,7 +18,7 @@ * 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.chats.feed.types +package com.vitorpamplona.amethyst.commons.compose.nip53LiveActivities import androidx.compose.foundation.background import androidx.compose.foundation.clickable @@ -38,9 +38,7 @@ import androidx.compose.ui.unit.dp * Shared rounded, accent-tinted container used by the centered "system-style" * cards rendered inside the live stream chat feed (zaps, raids, clips). * - * Keeps the three renderers visually consistent so they clearly stand apart - * from regular chat bubbles. The content slot uses a BoxScope so callers can - * pick their own internal layout (Row / Column / whatever). + * Kept platform-neutral so Desktop can consume it alongside Android. */ @Composable fun StreamSystemCard( diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregator.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregator.kt new file mode 100644 index 000000000..2d0e88564 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregator.kt @@ -0,0 +1,106 @@ +/* + * 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.commons.nip53LiveActivities + +import androidx.compose.runtime.Immutable +import com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** + * A single entry on a stream's top-zappers leaderboard. + * + * @param bucketKey either a real zapper pubkey or [LiveActivityTopZappersAggregator.ANON_KEY] + * for zaps whose zap-request carried an `anon` tag. + * @param totalSats sum of amounts across every zap that mapped to this bucket. + * @param isAnonymous true when the bucket is the shared anonymous pool. + */ +@Immutable +data class TopZapperEntry( + val bucketKey: HexKey, + val totalSats: Long, + val isAnonymous: Boolean, +) + +/** + * Represents a single zap-receipt contribution before aggregation. All inputs are + * plain values so this stays testable in pure Kotlin without Compose or Android. + * + * @param receiptId the kind-9735 receipt event id — used to de-duplicate receipts that + * appear in both the stream's #a and the goal's #e feed. + * @param zapperPubKey pubkey of the zap-request signer (random one-time key for anon/private). + * @param isAnonymous true when the zap-request carried an `anon` tag (empty or encrypted). + * @param sats amount in satoshis. + */ +@Immutable +data class ZapContribution( + val receiptId: HexKey, + val zapperPubKey: HexKey, + val isAnonymous: Boolean, + val sats: Long, +) + +/** + * Computes a top-N zap leaderboard for a live stream. + * + * Inputs are raw [ZapContribution]s from any number of sources; the aggregator handles: + * - de-duplication by receipt id (a zap that tags both `#a` and `#e` counts once), + * - anonymous bucketing (all `anon`-tagged zaps collapse into one entry), + * - sum-by-contributor and sort-desc-by-total, + * - top-N truncation. + * + * This is a pure function — no Compose, no Android, no LocalCache — so it can be unit + * tested in isolation and reused from any platform target. + */ +object LiveActivityTopZappersAggregator { + /** Sentinel key that collapses every anonymous / private zap into a single leaderboard entry. */ + const val ANON_KEY: HexKey = "anon" + + /** Default cap — matches zap.stream's TopZappers limit. */ + const val DEFAULT_LIMIT = 10 + + fun aggregate( + contributions: Iterable, + limit: Int = DEFAULT_LIMIT, + ): List { + if (limit <= 0) return emptyList() + + // receiptId -> contribution. Dedupes a single zap that arrives via both #a and #e. + val unique = HashMap() + for (c in contributions) { + unique[c.receiptId] = c + } + if (unique.isEmpty()) return emptyList() + + val totals = HashMap(unique.size) + var hasAnon = false + for (c in unique.values) { + val key = if (c.isAnonymous) ANON_KEY else c.zapperPubKey + if (c.isAnonymous) hasAnon = true + totals[key] = (totals[key] ?: 0L) + c.sats + } + + return totals.entries + .asSequence() + .sortedByDescending { it.value } + .take(limit) + .map { TopZapperEntry(it.key, it.value, isAnonymous = it.key == ANON_KEY && hasAnon) } + .toList() + } +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/LiveStreamTopZappersViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/LiveStreamTopZappersViewModel.kt new file mode 100644 index 000000000..032c53a75 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/LiveStreamTopZappersViewModel.kt @@ -0,0 +1,173 @@ +/* + * 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.commons.viewmodels + +import androidx.compose.runtime.Stable +import androidx.lifecycle.ViewModel +import androidx.lifecycle.viewModelScope +import com.vitorpamplona.amethyst.commons.model.ListChange +import com.vitorpamplona.amethyst.commons.model.Note +import com.vitorpamplona.amethyst.commons.model.nip53LiveActivities.LiveActivitiesChannel +import com.vitorpamplona.amethyst.commons.nip53LiveActivities.LiveActivityTopZappersAggregator +import com.vitorpamplona.amethyst.commons.nip53LiveActivities.TopZapperEntry +import com.vitorpamplona.amethyst.commons.nip53LiveActivities.ZapContribution +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip57Zaps.LnZapEvent +import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + +/** + * Live-stream top-zappers leaderboard state, computed off the UI thread and maintained + * incrementally. + * + * The leaderboard draws from two independent contribution sets keyed by zap-receipt id: + * - [streamContributions] — zaps routed into the [LiveActivitiesChannel]'s cache via the + * stream's `#a` subscription; + * - [goalContributions] — zaps attached to a NIP-75 goal note via `#e`. + * + * The union is flattened on each change and fed into [LiveActivityTopZappersAggregator] for + * de-dup, anonymous bucketing, sum-by-contributor, and top-N sort, publishing a stable + * `List` via [topZappers] for dumb UI consumption. + */ +@Stable +class LiveStreamTopZappersViewModel( + private val channel: LiveActivitiesChannel, + private val limit: Int = LiveActivityTopZappersAggregator.DEFAULT_LIMIT, +) : ViewModel() { + private val mutex = Mutex() + + // receiptId -> ZapContribution, partitioned by source. Aggregator dedupes at merge. + private val streamContributions = HashMap() + private val goalContributions = HashMap() + + private val _topZappers = MutableStateFlow>(emptyList()) + val topZappers: StateFlow> = _topZappers.asStateFlow() + + private var goalNote: Note? = null + private var goalObserverJob: Job? = null + + init { + viewModelScope.launch(Dispatchers.IO) { + // Seed from the channel cache. + mutex.withLock { + channel.notes.forEach { _, note -> + contributionFromStreamZap(note)?.let { streamContributions[it.receiptId] = it } + } + } + publish() + + // Keep in sync with subsequent arrivals. + channel.changesFlow().collect { change -> + val mutated = applyChannelChange(change) + if (mutated) publish() + } + } + } + + /** + * Attach (or detach) a NIP-75 goal note whose zaps should flow into the leaderboard. + * Passing the same instance twice is a no-op; passing null clears the goal source. + */ + fun setGoalNote(newGoalNote: Note?) { + if (newGoalNote === goalNote) return + goalNote = newGoalNote + + goalObserverJob?.cancel() + goalObserverJob = + viewModelScope.launch(Dispatchers.IO) { + // Reset goal bucket and re-seed from the new note (if any) before subscribing. + rebuildGoalContributions(newGoalNote) + publish() + + if (newGoalNote == null) return@launch + + // Observe future goal-zap arrivals. Each emission rebuilds the goal bucket; + // the aggregator already handles de-dup at merge-time. + newGoalNote.flow().zaps.stateFlow.collect { + rebuildGoalContributions(newGoalNote) + publish() + } + } + } + + private suspend fun applyChannelChange(change: ListChange): Boolean = + mutex.withLock { + when (change) { + is ListChange.Addition -> upsertStream(change.item) + is ListChange.Deletion -> removeStream(change.item.idHex) + is ListChange.SetAddition -> change.item.fold(false) { acc, n -> upsertStream(n) || acc } + is ListChange.SetDeletion -> change.item.fold(false) { acc, n -> removeStream(n.idHex) || acc } + } + } + + private fun upsertStream(note: Note): Boolean { + val c = contributionFromStreamZap(note) ?: return false + val prev = streamContributions[c.receiptId] + if (prev == c) return false + streamContributions[c.receiptId] = c + return true + } + + private fun removeStream(receiptId: HexKey): Boolean = streamContributions.remove(receiptId) != null + + private suspend fun rebuildGoalContributions(goal: Note?) { + mutex.withLock { + goalContributions.clear() + goal?.zaps?.forEach { (zapRequestNote, receiptNote) -> + contributionFromGoalZap(zapRequestNote, receiptNote)?.let { + goalContributions[it.receiptId] = it + } + } + } + } + + private fun publish() { + val merged = streamContributions.values + goalContributions.values + _topZappers.value = LiveActivityTopZappersAggregator.aggregate(merged, limit) + } + + private fun contributionFromStreamZap(note: Note): ZapContribution? { + val ev = note.event as? LnZapEvent ?: return null + val request = ev.zapRequest ?: return null + val sats = ev.amount()?.toLong() ?: return null + return ZapContribution(note.idHex, request.pubKey, request.isAnonTagged(), sats) + } + + private fun contributionFromGoalZap( + zapRequestNote: Note, + receiptNote: Note?, + ): ZapContribution? { + val receiptEv = receiptNote?.event as? LnZapEvent ?: return null + val request = zapRequestNote.event as? LnZapRequestEvent ?: return null + val sats = receiptEv.amount()?.toLong() ?: return null + return ZapContribution(receiptNote.idHex, request.pubKey, request.isAnonTagged(), sats) + } +} + +/** True for both public anonymous and NIP-57 private zaps (any `anon` tag, empty or encrypted). */ +private fun LnZapRequestEvent.isAnonTagged(): Boolean = tags.any { it.isNotEmpty() && it[0] == "anon" } diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregatorTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregatorTest.kt new file mode 100644 index 000000000..f1d64c7e1 --- /dev/null +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/nip53LiveActivities/LiveActivityTopZappersAggregatorTest.kt @@ -0,0 +1,153 @@ +/* + * 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.commons.nip53LiveActivities + +import com.vitorpamplona.amethyst.commons.nip53LiveActivities.LiveActivityTopZappersAggregator.ANON_KEY +import com.vitorpamplona.amethyst.commons.nip53LiveActivities.LiveActivityTopZappersAggregator.aggregate +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class LiveActivityTopZappersAggregatorTest { + private val alice = "a".repeat(64) + private val bob = "b".repeat(64) + private val carol = "c".repeat(64) + + private fun zap( + receipt: String, + zapper: String = alice, + sats: Long = 100, + anon: Boolean = false, + ) = ZapContribution( + receiptId = receipt, + zapperPubKey = zapper, + isAnonymous = anon, + sats = sats, + ) + + @Test + fun sortsByTotalDescending() { + val result = + aggregate( + listOf( + zap("r1", alice, 100), + zap("r2", bob, 500), + zap("r3", carol, 250), + ), + ) + + assertEquals(listOf(bob, carol, alice), result.map { it.bucketKey }) + assertEquals(listOf(500L, 250L, 100L), result.map { it.totalSats }) + } + + @Test + fun sumsMultipleZapsFromSameContributor() { + val result = + aggregate( + listOf( + zap("r1", alice, 100), + zap("r2", alice, 200), + zap("r3", bob, 250), + ), + ) + + assertEquals(listOf(alice, bob), result.map { it.bucketKey }) + assertEquals(300L, result[0].totalSats) + assertEquals(250L, result[1].totalSats) + } + + @Test + fun dedupesByReceiptId() { + // Same zap arriving via both #a (stream) and #e (goal) should count once. + val result = + aggregate( + listOf( + zap("r1", alice, 500), + zap("r1", alice, 500), + ), + ) + + assertEquals(1, result.size) + assertEquals(500L, result[0].totalSats) + } + + @Test + fun bucketsAnonymousZapsUnderSingleSentinel() { + // Each anon zap has a random one-time pubkey; they must collapse into one entry. + val anon1 = "1".repeat(64) + val anon2 = "2".repeat(64) + val result = + aggregate( + listOf( + zap("r1", alice, 100), + zap("r2", anon1, 200, anon = true), + zap("r3", anon2, 300, anon = true), + ), + ) + + assertEquals(2, result.size) + val anon = result.first { it.isAnonymous } + assertEquals(ANON_KEY, anon.bucketKey) + assertEquals(500L, anon.totalSats) + val realUser = result.first { !it.isAnonymous } + assertEquals(alice, realUser.bucketKey) + assertEquals(100L, realUser.totalSats) + } + + @Test + fun doesNotFalselyMarkRealPubkeyAnonymousWhenItCollidesWithSentinel() { + // Defensive: if some stream has a zapper whose pubkey literally equals "anon", + // we don't collide because the sentinel is 4 chars and real pubkeys are 64. + val result = + aggregate( + listOf( + zap("r1", alice, 100), + ), + ) + + assertTrue(result.none { it.isAnonymous }) + } + + @Test + fun respectsTopNLimit() { + val contributions = + (1..15).map { + zap(receipt = "r$it", zapper = "$it".repeat(64), sats = it.toLong() * 10) + } + + val result = aggregate(contributions, limit = 5) + + assertEquals(5, result.size) + // Highest total first. + assertEquals(15 * 10L, result[0].totalSats) + assertEquals(11 * 10L, result[4].totalSats) + } + + @Test + fun handlesEmptyInput() { + assertTrue(aggregate(emptyList()).isEmpty()) + } + + @Test + fun handlesZeroLimit() { + assertTrue(aggregate(listOf(zap("r1")), limit = 0).isEmpty()) + } +}