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<List<TopZapperEntry>>.
  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/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) }

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

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

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

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()
+        }
+    }
 }

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.commons.nip53LiveActivities.TopZapperEntry
-import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.observeNoteZaps
+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.
+ * Horizontally-scrollable leaderboard of the top zappers on a live stream. Matches
- * Aggregates zaps from both the stream's #a subscription and the attached
+ * zap.stream's TopZappers look: pill chips with avatar + lightning + sats.
- * NIP-75 zap goal (if any), de-duplicating by zap-receipt id. Matches
+ *
- * zap.stream's TopZappers UX: 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<TopZapperEntry>`.
  */
 @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 <T : ViewModel> create(modelClass: Class<T>): 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)) }
+        var goalNoteHolder by remember(goalId) {
-        if (goalNote == null) {
+            mutableStateOf(accountViewModel.getNoteIfExists(goalId))
-            LaunchedEffect(goalId) {
+        }
-                goalNote = accountViewModel.checkGetOrCreateNote(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
+    val entries by topZappersVm.topZappers.collectAsStateWithLifecycle()
-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)
-        }
 
     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<TopZapperEntry> {
-    // receiptId -> (zapperBucketKey, sats) — dedupes a zap that appears via both #a and #e.
-    val byReceipt = HashMap<HexKey, Pair<HexKey, BigDecimal>>()
-
-    // 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<HexKey, BigDecimal>(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

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
+ * Kept platform-neutral so Desktop can consume it alongside Android.
- * from regular chat bubbles. The content slot uses a BoxScope so callers can
- * pick their own internal layout (Row / Column / whatever).
  */
 @Composable
 fun StreamSystemCard(

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<ZapContribution>,
+        limit: Int = DEFAULT_LIMIT,
+    ): List<TopZapperEntry> {
+        if (limit <= 0) return emptyList()
+
+        // receiptId -> contribution. Dedupes a single zap that arrives via both #a and #e.
+        val unique = HashMap<HexKey, ZapContribution>()
+        for (c in contributions) {
+            unique[c.receiptId] = c
+        }
+        if (unique.isEmpty()) return emptyList()
+
+        val totals = HashMap<HexKey, Long>(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()
+    }
+}

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<TopZapperEntry>` 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<HexKey, ZapContribution>()
+    private val goalContributions = HashMap<HexKey, ZapContribution>()
+
+    private val _topZappers = MutableStateFlow<List<TopZapperEntry>>(emptyList())
+    val topZappers: StateFlow<List<TopZapperEntry>> = _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<Note>): 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" }

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())
+    }
+}