From 3acd84bf7651e5a1c248474ab40f11adf47bbccc Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 1 May 2026 19:24:48 +0000 Subject: [PATCH] fix(nests): reactions dedup, drift-fade animation, 10s window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit-driven fixes for the audio-room reactions overlay: 1. RoomReactionsAggregator now dedups by event id. The previous code appended every event to a flat list, but LocalCache.observeEvents re-emits the full matching list on every cache mutation — so an N-reaction window grew quadratically and the overlay rendered the same emoji once per replay. Keyed by event id collapses re-emits. 2. RoomPresenceAggregator gains applyOrNull that returns null when an incoming heartbeat is older-or-equal to the cached presence; the VM skips the StateFlow write in that case. In a 200-peer room, every replay used to copy a 200-entry map plus run an O(N) StateFlow equality check 200 times per emission. Now it's one-and-skip. 3. ReactionsEvictionTicker only ticks while the aggregator is non-empty. Once the last reaction expires the loop self-cancels until the next reaction lands — a quiet room costs no scheduled work. 4. REACTION_WINDOW_SEC dropped 30s -> 10s. Reactions are about what the speaker is saying right now; a 10s window keeps the overlay timely instead of bleeding into the next paragraph. 5. SpeakerReactionOverlay drives a per-chip lifecycle animation: each chip drifts upward ~16dp and fades over the 10s window. A fresh reaction (same emoji) restarts the chip's animation. The AnimatedVisibility outer entry/exit still smooths first-arrival and final disappearance. Tests: added a re-emit dedup case to RoomReactionsStateTest plus an end-to-end replay assertion in NestViewModelTest. Existing tests updated to use unique event ids. https://claude.ai/code/session_01DMeCvWyBYVVVPez2hwqCs4 --- .../room/lifecycle/NestRoomEventCollectors.kt | 14 ++- .../room/stage/SpeakerReactionOverlay.kt | 90 +++++++++++++++---- .../commons/viewmodels/NestViewModel.kt | 32 ++++--- .../commons/viewmodels/RoomPresenceState.kt | 26 +++++- .../commons/viewmodels/RoomReactionsState.kt | 36 ++++++-- .../commons/viewmodels/NestViewModelTest.kt | 23 ++++- .../viewmodels/RoomReactionsStateTest.kt | 21 ++++- 7 files changed, 196 insertions(+), 46 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/lifecycle/NestRoomEventCollectors.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/lifecycle/NestRoomEventCollectors.kt index d4d243912..5e67573ea 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/lifecycle/NestRoomEventCollectors.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/lifecycle/NestRoomEventCollectors.kt @@ -22,6 +22,8 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.nests.room.lifecycle import androidx.compose.runtime.Composable import androidx.compose.runtime.LaunchedEffect +import androidx.compose.runtime.collectAsState +import androidx.compose.runtime.getValue import com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel import com.vitorpamplona.amethyst.model.LocalCache import com.vitorpamplona.quartz.experimental.nests.admin.AdminCommandEvent @@ -144,7 +146,15 @@ private fun ReactionsCollector( @Composable private fun ReactionsEvictionTicker(viewModel: NestViewModel) { - LaunchedEffect(viewModel) { + // Only tick while there are reactions to evict. The aggregator's + // last eviction empties [NestViewModel.recentReactions], which + // flips [hasReactions] to false and cancels the loop until the + // next reaction arrives. A perpetually-quiet room costs no + // scheduled work. + val reactions by viewModel.recentReactions.collectAsState() + val hasReactions = reactions.values.any { it.isNotEmpty() } + LaunchedEffect(viewModel, hasReactions) { + if (!hasReactions) return@LaunchedEffect while (isActive) { delay(REACTIONS_TICK_MS) viewModel.evictReactions(System.currentTimeMillis() / 1000 - REACTION_WINDOW_SEC_LOCAL) @@ -215,4 +225,4 @@ private const val REACTIONS_TICK_MS = 1_000L // — the platform layer doesn't import from commons here, and a // duplicate constant is cheaper than another import. If these // drift, the 1-s tick still self-heals within one window length. -private const val REACTION_WINDOW_SEC_LOCAL = 30L +private const val REACTION_WINDOW_SEC_LOCAL = 10L diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/SpeakerReactionOverlay.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/SpeakerReactionOverlay.kt index 3b883abaf..32ac342f8 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/SpeakerReactionOverlay.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/nests/room/stage/SpeakerReactionOverlay.kt @@ -21,39 +21,49 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.nests.room.stage import androidx.compose.animation.AnimatedVisibility +import androidx.compose.animation.core.LinearEasing +import androidx.compose.animation.core.animateFloatAsState +import androidx.compose.animation.core.tween import androidx.compose.animation.fadeIn import androidx.compose.animation.fadeOut import androidx.compose.animation.scaleIn import androidx.compose.animation.scaleOut import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Row +import androidx.compose.foundation.layout.offset import androidx.compose.foundation.layout.padding import androidx.compose.material3.MaterialTheme import androidx.compose.material3.Surface import androidx.compose.material3.Text import androidx.compose.runtime.Composable +import androidx.compose.runtime.LaunchedEffect +import androidx.compose.runtime.getValue +import androidx.compose.runtime.mutableStateOf +import androidx.compose.runtime.remember +import androidx.compose.runtime.setValue import androidx.compose.ui.Modifier +import androidx.compose.ui.draw.alpha import androidx.compose.ui.unit.dp +import com.vitorpamplona.amethyst.commons.viewmodels.REACTION_WINDOW_SEC import com.vitorpamplona.amethyst.commons.viewmodels.RoomReaction +import kotlinx.coroutines.delay /** - * Floating-emoji overlay drawn under a speaker's avatar. Aggregates - * the same emoji into one chip with a count badge so a burst of - * "🔥🔥🔥" reads as `🔥 ×3` rather than three stacked chips. + * Floating-emoji overlay drawn under a speaker's avatar. Each chip + * is keyed by emoji content; bursts of the same emoji collapse into + * a single `🔥 ×3` chip whose count tracks live as new reactions land. * - * Hides itself when there are no reactions in the window — the - * 30-s sliding-window aggregator in - * [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel.recentReactions] - * drops stale entries on the 1-s tick. + * Reactions are about what the speaker is saying RIGHT NOW, so each + * chip lives for [REACTION_WINDOW_SEC] seconds: the moment its + * youngest reaction arrives the chip fades+scales in, and over that + * window it slowly drifts upward and fades out before the eviction + * tick removes it from the underlying list. */ @Composable internal fun SpeakerReactionOverlay( reactions: List, modifier: Modifier = Modifier, ) { - // Wrap the row in AnimatedVisibility so the burst scales-and-fades - // in instead of snapping; a fading-out tail also smooths the - // 30 s eviction sweep instead of having chips just disappear. AnimatedVisibility( visible = reactions.isNotEmpty(), enter = fadeIn() + scaleIn(initialScale = 0.6f), @@ -64,10 +74,20 @@ internal fun SpeakerReactionOverlay( // "🔥 ×N" chip. Order by most recent so a fresh reaction lands // at the front of the row. val byContent = reactions.groupBy { it.content } - val ordered = byContent.toList().sortedByDescending { (_, list) -> list.maxOf { it.createdAtSec } } + val ordered = + byContent.toList().sortedByDescending { (_, list) -> + list.maxOf { it.createdAtSec } + } Row(horizontalArrangement = Arrangement.spacedBy(4.dp)) { ordered.forEach { (content, list) -> - ReactionChip(content = content, count = list.size) + // Newest createdAt drives the chip's lifecycle — + // a fresh reaction restarts the upward-drift+fade. + val youngestSec = list.maxOf { it.createdAtSec } + ReactionChip( + content = content, + count = list.size, + youngestSec = youngestSec, + ) } } } @@ -77,11 +97,49 @@ internal fun SpeakerReactionOverlay( private fun ReactionChip( content: String, count: Int, + youngestSec: Long, ) { - // Tonal Surface picks up the right elevation tint in both light - // and dark themes — softer than the flat secondaryContainer fill - // and keeps a tiny shadow so the chip reads as floating. + // Tick a [progress] state from 0f → 1f over the eviction window + // so the chip can drift + fade in lockstep with how close it is + // to falling out of the aggregator. Reset whenever a fresher + // reaction lands by re-keying on [youngestSec]. + var progress by remember(youngestSec) { mutableStateOf(0f) } + LaunchedEffect(youngestSec) { + // Re-sync against wall-clock so a chip whose window started + // before this composable mounted (e.g. user rotated the + // device mid-burst) still ages correctly. + val ageMs = (System.currentTimeMillis() / 1000L - youngestSec).coerceAtLeast(0L) * 1000L + val remaining = (REACTION_WINDOW_MS - ageMs).coerceAtLeast(0L) + progress = (ageMs.toFloat() / REACTION_WINDOW_MS).coerceIn(0f, 1f) + if (remaining <= 0L) return@LaunchedEffect + // 100 ms ticks keep the drift smooth without burning a frame + // budget — the avatar is small and the chip's motion is + // sub-pixel between ticks anyway. + val steps = (remaining / 100L).coerceAtLeast(1L) + repeat(steps.toInt()) { + delay(100L) + progress = + ((System.currentTimeMillis() / 1000L - youngestSec).toFloat() * 1000f / REACTION_WINDOW_MS) + .coerceIn(0f, 1f) + } + progress = 1f + } + + // Drift up by ~16 dp over the window so the chip reads as + // "rising and dissipating", and fade out over the second half + // so the first half stays solidly readable. + val driftDp = (-16f * progress).dp + val animatedAlpha by animateFloatAsState( + targetValue = (1f - ((progress - 0.5f).coerceAtLeast(0f) * 2f)).coerceIn(0f, 1f), + animationSpec = tween(durationMillis = 100, easing = LinearEasing), + label = "reaction-chip-alpha", + ) + Surface( + modifier = + Modifier + .offset(y = driftDp) + .alpha(animatedAlpha), shape = MaterialTheme.shapes.small, tonalElevation = 2.dp, shadowElevation = 1.dp, @@ -95,3 +153,5 @@ private fun ReactionChip( ) } } + +private const val REACTION_WINDOW_MS = REACTION_WINDOW_SEC * 1000L diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt index da52af697..e9c64fddf 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt @@ -139,10 +139,11 @@ class NestViewModel( /** * Chat ledger for the live-activities chat panel (#1) — every - * kind-1311 ([LiveActivitiesChatMessageEvent]) - * tagged with this room's `a`-pointer, ordered by `created_at` - * ascending so the newest message is at the end (the panel - * auto-scrolls to it). + * kind-1311 ([LiveActivitiesChatMessageEvent]) tagged with this + * room's `a`-pointer, ordered by [DefaultFeedOrder] (descending + * `created_at`, then by id for stability). The chat panel renders + * with `LazyColumn(reverseLayout = true)`, which puts data index 0 + * (the newest message) at the bottom of the viewport. * * Dedupes by event id — a relay re-emit on reconnect can't * produce a duplicate row. @@ -155,8 +156,8 @@ class NestViewModel( /** * Recent kind-7 reactions for the floating speaker-avatar overlay * (#3). Keyed by target pubkey; room-wide reactions land under the - * empty-string key. Sliding 30 s window driven by the platform - * layer's tick (typically every 1 s). + * empty-string key. Sliding [REACTION_WINDOW_SEC] window driven + * by the platform layer's tick (typically every 1 s). */ private val reactionsAgg = RoomReactionsAggregator() private val _recentReactions = MutableStateFlow>>(emptyMap()) @@ -371,7 +372,11 @@ class NestViewModel( */ fun onPresenceEvent(event: com.vitorpamplona.quartz.nip53LiveActivities.presence.MeetingRoomPresenceEvent) { if (closed) return - _presences.value = presenceAgg.apply(event) + // Skip the StateFlow write when the aggregator says nothing + // changed — saves a snapshot copy + an O(N) map equality check + // per replay in big rooms (LocalCache.observeEvents re-emits + // the full list on every cache mutation). + presenceAgg.applyOrNull(event)?.let { _presences.value = it } } /** @@ -401,8 +406,8 @@ class NestViewModel( /** * Apply one kind-7 reaction event to the sliding-window aggregator. * Caller passes [nowSec] (so tests can be deterministic) and the - * fixed 30-s window. Mirror of [evictReactions] for the per-tick - * cleanup. + * fixed [REACTION_WINDOW_SEC] window. Mirror of [evictReactions] + * for the per-tick cleanup. */ fun onReactionEvent( event: com.vitorpamplona.quartz.nip25Reactions.ReactionEvent, @@ -1222,10 +1227,13 @@ const val LEVEL_TICK_MS: Long = 100L /** * How long a kind-7 reaction stays in * [NestViewModel.recentReactions] before the eviction sweep - * drops it. Matches the duration of the floating-up animation in the - * SpeakerReactionOverlay. + * drops it. Reactions are about what the speaker is talking about + * RIGHT NOW — a 10 s window keeps the floating-emoji overlay + * timely (a reaction to one sentence shouldn't bleed into the + * next paragraph) and matches the duration of the fade-up + * animation in the SpeakerReactionOverlay. */ -const val REACTION_WINDOW_SEC: Long = 30L +const val REACTION_WINDOW_SEC: Long = 10L /** * Indirection over the top-level `connectNestsListener` so tests can drive diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomPresenceState.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomPresenceState.kt index 95c20a576..a2b97b589 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomPresenceState.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomPresenceState.kt @@ -79,13 +79,31 @@ data class RoomPresence( class RoomPresenceAggregator { private val byPubkey = mutableMapOf() - /** Apply one presence event. Returns the updated snapshot. */ - fun apply(event: MeetingRoomPresenceEvent): Map { + /** + * Apply one presence event. Returns a fresh snapshot when state + * actually changed, or `null` when the event was a no-op (older + * or equal to the cached presence for the same pubkey). + * + * `LocalCache.observeEvents` re-emits the full matching list on + * every cache mutation, so in a 200-peer room the VM forwards + * 200 events for a single new heartbeat. Returning `null` for + * the 199 no-op replays lets the VM skip the + * `_presences.value = ...` write — saves both the snapshot copy + * and the StateFlow equality check (both O(N) per call). + */ + fun applyOrNull(event: MeetingRoomPresenceEvent): Map? { val incoming = RoomPresence.from(event) val current = byPubkey[incoming.pubkey] - if (current == null || current.updatedAtSec < incoming.updatedAtSec) { - byPubkey[incoming.pubkey] = incoming + if (current != null && current.updatedAtSec >= incoming.updatedAtSec) { + return null } + byPubkey[incoming.pubkey] = incoming + return snapshot() + } + + /** Apply-or-no-op variant kept for tests + callers that always want a snapshot. */ + fun apply(event: MeetingRoomPresenceEvent): Map { + applyOrNull(event) return snapshot() } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsState.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsState.kt index 192095bb8..a538d1c59 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsState.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsState.kt @@ -35,6 +35,8 @@ import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent */ @Immutable data class RoomReaction( + /** Source event id — used for dedup across re-emits. */ + val eventId: String, /** Who reacted. */ val sourcePubkey: String, /** Speaker the reaction is aimed at, or `null` for room-wide. */ @@ -52,6 +54,7 @@ data class RoomReaction( */ fun from(event: ReactionEvent): RoomReaction = RoomReaction( + eventId = event.id, sourcePubkey = event.pubKey, targetPubkey = event.originalAuthor().firstOrNull(), content = event.content, @@ -61,16 +64,19 @@ data class RoomReaction( } /** - * Sliding-window aggregator. Holds reactions keyed by target pubkey - * (with `null` lumped under the empty-string key so the map's - * value-type is uniform); the room screen reads - * `byTarget()` per render and the UI naturally fades as the window - * slides. + * Sliding-window aggregator. Holds reactions keyed by event id (so + * re-emits from `LocalCache.observeEvents` — which republishes the + * full matching list on every cache mutation — collapse into a + * single overlay entry instead of stacking duplicates), and surfaces + * them grouped by target pubkey for the UI (with `null` lumped under + * the empty-string key so the map's value-type is uniform). * * Not thread-safe — call from the VM's single coroutine. */ class RoomReactionsAggregator { - private val all = mutableListOf() + // Insertion-ordered so groupBy preserves arrival order; keyed by + // event id so the same kind-7 from two relays only counts once. + private val byEventId = LinkedHashMap() /** Stable key for room-wide reactions in the returned map. */ private val roomWideKey = "" @@ -81,7 +87,13 @@ class RoomReactionsAggregator { nowSec: Long, windowSec: Long, ): Map> { - all += RoomReaction.from(event) + val incoming = RoomReaction.from(event) + // Dedup: a relay re-delivery (or LocalCache.observeEvents's + // full-list re-emit) of the same kind-7 must not stack. + if (byEventId.put(incoming.eventId, incoming) != null) { + // Re-emit of an existing reaction: just return the post- + // evict snapshot without growing the map. + } return evictAndSnapshot(nowSec - windowSec) } @@ -92,7 +104,13 @@ class RoomReactionsAggregator { * per-Composable timer). */ fun evictAndSnapshot(olderThanSec: Long): Map> { - all.removeAll { it.createdAtSec < olderThanSec } - return all.groupBy { it.targetPubkey ?: roomWideKey } + val it = byEventId.entries.iterator() + while (it.hasNext()) { + if (it.next().value.createdAtSec < olderThanSec) it.remove() + } + return byEventId.values.groupBy { it.targetPubkey ?: roomWideKey } } + + /** Whether the aggregator currently holds any unevicted reactions. */ + fun isEmpty(): Boolean = byEventId.isEmpty() } diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt index 8783edcfd..2cddb7a0d 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModelTest.kt @@ -286,13 +286,19 @@ class NestViewModelTest { val alice = "a".repeat(64) val bob = "b".repeat(64) + // Vary id per reaction so the dedup gate doesn't collapse + // distinct kind-7s; the relay-replay dedup case has its own + // assertion below. + var nextId = 1 + fun rxn( from: String, to: String, content: String, createdAt: Long, + id: String = "%064x".format(nextId++), ) = com.vitorpamplona.quartz.nip25Reactions.ReactionEvent( - id = "0".repeat(64), + id = id, pubKey = from, createdAt = createdAt, tags = arrayOf(arrayOf("a", "30312:host:room"), arrayOf("p", to)), @@ -305,7 +311,15 @@ class NestViewModelTest { vm.onReactionEvent(rxn(alice, bob, "👏", 105L), nowSec = 105L) assertEquals(2, vm.recentReactions.value[bob]!!.size) - // Tick advances past the window — both reactions evicted. + // LocalCache.observeEvents re-emits the full matching list + // on every cache mutation; the same kind-7 must collapse + // into one overlay entry instead of stacking on each replay. + val replayed = rxn(alice, bob, "🔥", 100L, id = "f".repeat(64)) + vm.onReactionEvent(replayed, nowSec = 105L) + vm.onReactionEvent(replayed, nowSec = 105L) + assertEquals(3, vm.recentReactions.value[bob]!!.size) + + // Tick advances past the window — all reactions evicted. vm.evictReactions(olderThanSec = 200L) assertEquals(emptyMap(), vm.recentReactions.value) } @@ -464,7 +478,10 @@ class NestViewModelTest { mutable.value = s } - override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = error("subscribeSpeaker not exercised in these tests — see NestPlayerTest in :nestsClient") + override suspend fun subscribeSpeaker( + speakerPubkeyHex: String, + maxLatencyMs: Long, + ): SubscribeHandle = error("subscribeSpeaker not exercised in these tests — see NestPlayerTest in :nestsClient") override suspend fun close() { closeCallCount++ diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsStateTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsStateTest.kt index 820091ff0..38df29857 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsStateTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/RoomReactionsStateTest.kt @@ -31,11 +31,14 @@ class RoomReactionsStateTest { private val bob = "b".repeat(64) private val charlie = "c".repeat(64) + private var nextEventId = 1 + private fun reaction( from: String, to: String?, content: String, createdAt: Long, + id: String = "%064x".format(nextEventId++), ): ReactionEvent { val tags = buildList> { @@ -43,7 +46,7 @@ class RoomReactionsStateTest { if (to != null) add(arrayOf("p", to)) }.toTypedArray() return ReactionEvent( - id = "0".repeat(64), + id = id, pubKey = from, createdAt = createdAt, tags = tags, @@ -112,4 +115,20 @@ class RoomReactionsStateTest { // RoomReaction so the StateFlow doesn't re-emit on no-op ticks). assertEquals(a, b) } + + @Test + fun aggregatorDedupsRepeatedEventIds() { + val agg = RoomReactionsAggregator() + // LocalCache.observeEvents re-emits the full matching list on + // every cache mutation; the same kind-7 must collapse into one + // overlay entry instead of stacking on each replay. + val sharedId = "f".repeat(64) + val first = reaction(alice, bob, "🔥", 100L, id = sharedId) + val replay = reaction(alice, bob, "🔥", 100L, id = sharedId) + agg.apply(first, nowSec = 100L, windowSec = 30L) + agg.apply(replay, nowSec = 100L, windowSec = 30L) + val snap = agg.apply(replay, nowSec = 100L, windowSec = 30L) + + assertEquals(1, snap[bob]!!.size) + } }