From 5bdc684f21eeb57bfb6b7d08c6220a785edb8feb Mon Sep 17 00:00:00 2001 From: davotoula Date: Tue, 24 Mar 2026 16:41:41 +0100 Subject: [PATCH] fix: add event-level dedup and handleGameAccepted guard in chess lobby - Dedup incoming events by ID (bounded LRU set of 500) to prevent redundant processing when multiple relays deliver the same event - Filter non-chess kind 30 events early (isStart=false, isMove=false) - Guard handleGameAccepted with recentlyLoadedGames check to prevent N concurrent relay fetches when N move events arrive for same game --- .../datasource/ChessFeedFilterSubAssembler.kt | 34 ++++++++++++++- .../chess/datasource/ChessFilterAssembler.kt | 12 ++++-- .../chess/datasource/ChessSubscription.kt | 13 +++++- .../amethyst/commons/chess/ChessLobbyLogic.kt | 43 +++++++++++++++++++ 4 files changed, 96 insertions(+), 6 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFeedFilterSubAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFeedFilterSubAssembler.kt index 3a1cccd1d..702c7dcbd 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFeedFilterSubAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFeedFilterSubAssembler.kt @@ -22,9 +22,14 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.chess.datasource import com.vitorpamplona.amethyst.service.relayClient.eoseManagers.PerUniqueIdEoseManager import com.vitorpamplona.amethyst.service.relays.SincePerRelayMap +import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener +import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.TimeUtils /** * Query state for chess subscription @@ -39,16 +44,43 @@ data class ChessQueryState( ) /** - * Sub-assembler that creates the actual relay filters + * Sub-assembler that creates the actual relay filters. + * Intercepts incoming chess events from relays and forwards them to the ViewModel. */ class ChessFeedFilterSubAssembler( client: INostrClient, allKeys: () -> Set, ) : PerUniqueIdEoseManager(client, allKeys) { + var onChessEvent: ((Event) -> Unit)? = null + override fun updateFilter( key: ChessQueryState, since: SincePerRelayMap?, ): List = filterChessEvents(key, since) override fun id(key: ChessQueryState): String = key.userPubkey + + override fun newSub(key: ChessQueryState): Subscription = + requestNewSubscription( + object : IRequestListener { + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + newEose(key, relay, TimeUtils.now(), forFilters) + } + + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + if (isLive) { + newEose(key, relay, TimeUtils.now(), forFilters) + } + onChessEvent?.invoke(event) + } + }, + ) } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFilterAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFilterAssembler.kt index 04a4fc5b9..a083cbf0d 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFilterAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessFilterAssembler.kt @@ -21,6 +21,7 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.chess.datasource import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.ComposeSubscriptionManager +import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient /** @@ -29,10 +30,13 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient class ChessFilterAssembler( client: INostrClient, ) : ComposeSubscriptionManager() { - val group = - listOf( - ChessFeedFilterSubAssembler(client, ::allKeys), - ) + private val subAssembler = ChessFeedFilterSubAssembler(client, ::allKeys) + + val group = listOf(subAssembler) + + fun setOnChessEvent(callback: ((Event) -> Unit)?) { + subAssembler.onChessEvent = callback + } override fun invalidateKeys() = invalidateFilters() diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessSubscription.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessSubscription.kt index 11e955ef2..acbe21fc5 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessSubscription.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chess/datasource/ChessSubscription.kt @@ -66,8 +66,19 @@ fun ChessSubscription( ) } + // Wire incoming chess events from relay subscriptions to the ViewModel + val chessAssembler = accountViewModel.dataSources().chess + DisposableEffect(chessAssembler, chessViewModel) { + chessAssembler.setOnChessEvent { event -> + chessViewModel.handleIncomingEvent(event) + } + onDispose { + chessAssembler.setOnChessEvent(null) + } + } + // Register subscription with Amethyst's subscription system - KeyDataSourceSubscription(state, accountViewModel.dataSources().chess) + KeyDataSourceSubscription(state, chessAssembler) // Trigger ViewModel refresh when subscription state changes // This fetches challenges from LocalCache after events arrive diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt index d839285d1..4ed8ed9d4 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/chess/ChessLobbyLogic.kt @@ -128,6 +128,15 @@ class ChessLobbyLogic( ) { val state = ChessLobbyState(userPubkey, scope) + // Track when games were last loaded to prevent duplicate fetches + // (e.g., discoverUserGames loads a game, then polling immediately re-fetches it) + private val recentlyLoadedGames = java.util.concurrent.ConcurrentHashMap() + + // Dedup incoming events (same event delivered by multiple relays) + // Bounded LRU: evict oldest when exceeding capacity + private val seenEventIds = java.util.Collections.synchronizedSet(LinkedHashSet()) + private val seenEventIdsMax = 500 + private val pollingDelegate = ChessPollingDelegate( config = pollingConfig, @@ -188,6 +197,20 @@ class ChessLobbyLogic( * Called by platform subscription callbacks for real-time updates. */ fun handleIncomingEvent(event: JesterEvent) { + // Skip non-chess events (kind 30 but not start or move) + if (!event.isStartEvent() && !event.isMoveEvent()) return + + // Dedup: skip if we already processed this event ID (multiple relays deliver same event) + synchronized(seenEventIds) { + if (!seenEventIds.add(event.id)) return + if (seenEventIds.size > seenEventIdsMax) { + seenEventIds.iterator().let { + it.next() + it.remove() + } + } + } + Log.d("chessdebug", "[Lobby] handleIncomingEvent: id=${event.id.take(8)}, pubkey=${event.pubKey.take(8)}, isStart=${event.isStartEvent()}, isMove=${event.isMoveEvent()}, createdAt=${event.createdAt}") when { event.isStartEvent() -> handleStartEvent(event) @@ -408,6 +431,15 @@ class ChessLobbyLogic( * When we detect our challenge was accepted (opponent made first move), load game from relays. */ fun handleGameAccepted(startEventId: String) { + // Skip if already loaded or in-flight (multiple move events from same game trigger this) + val lastLoaded = recentlyLoadedGames[startEventId] + if (lastLoaded != null && (TimeUtils.now() - lastLoaded) < 10) { + Log.d("chessdebug", "[Lobby] handleGameAccepted: SKIPPED game ${startEventId.take(8)} - loaded ${TimeUtils.now() - lastLoaded}s ago") + return + } + // Mark immediately to prevent concurrent launches + recentlyLoadedGames[startEventId] = TimeUtils.now() + Log.d("chessdebug", "[Lobby] handleGameAccepted: game ${startEventId.take(8)} - fetching from relays") scope.launch(Dispatchers.Default) { state.setBroadcastStatus(ChessBroadcastStatus.Syncing(0f)) @@ -418,6 +450,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { + recentlyLoadedGames[startEventId] = TimeUtils.now() Log.d("chessdebug", "[Lobby] handleGameAccepted SUCCESS: game ${startEventId.take(8)}, role=${result.reconstructedState.viewerRole}") state.addActiveGame(startEventId, result.liveState) pollingDelegate.addGameId(startEventId) @@ -569,6 +602,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { + recentlyLoadedGames[startEventId] = TimeUtils.now() state.addSpectatingGame(startEventId, result.liveState) pollingDelegate.addGameId(startEventId) state.setBroadcastStatus(ChessBroadcastStatus.Idle) @@ -601,6 +635,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { + recentlyLoadedGames[startEventId] = TimeUtils.now() if (result.liveState.isSpectator) { state.addSpectatingGame(startEventId, result.liveState) } else { @@ -636,6 +671,13 @@ class ChessLobbyLogic( } private suspend fun refreshGame(startEventId: String) { + // Skip if this game was just loaded (prevents duplicate fetch after discoverUserGames) + val lastLoaded = recentlyLoadedGames[startEventId] + if (lastLoaded != null && (TimeUtils.now() - lastLoaded) < 10) { + Log.d("chessdebug", "[Lobby] refreshGame: SKIPPED game ${startEventId.take(8)} - loaded ${TimeUtils.now() - lastLoaded}s ago") + return + } + Log.d("chessdebug", "[Lobby] refreshGame: fetching game ${startEventId.take(8)} from relays") val events = fetcher.fetchGameEvents(startEventId) @@ -831,6 +873,7 @@ class ChessLobbyLogic( when (result) { is LoadGameResult.Success -> { + recentlyLoadedGames[startEventId] = TimeUtils.now() // If the discovered game is already finished, send it straight to completed val gameStatus = result.liveState.gameStatus.value if (gameStatus is GameStatus.Finished) {