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
This commit is contained in:
davotoula
2026-03-24 16:41:41 +01:00
parent 944fbcd000
commit 5bdc684f21
4 changed files with 96 additions and 6 deletions
@@ -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<ChessQueryState>,
) : PerUniqueIdEoseManager<ChessQueryState, String>(client, allKeys) {
var onChessEvent: ((Event) -> Unit)? = null
override fun updateFilter(
key: ChessQueryState,
since: SincePerRelayMap?,
): List<RelayBasedFilter> = 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<Filter>?,
) {
newEose(key, relay, TimeUtils.now(), forFilters)
}
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
if (isLive) {
newEose(key, relay, TimeUtils.now(), forFilters)
}
onChessEvent?.invoke(event)
}
},
)
}
@@ -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<ChessQueryState>() {
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()
@@ -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
@@ -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<String, Long>()
// Dedup incoming events (same event delivered by multiple relays)
// Bounded LRU: evict oldest when exceeding capacity
private val seenEventIds = java.util.Collections.synchronizedSet(LinkedHashSet<String>())
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) {