From aecbcb68a5f879c23a28554179d4e99653212702 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 15 Mar 2026 14:19:08 +0000 Subject: [PATCH] feat: per-relay recv/new counts and OK-true acceptance tracking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ViewModel: - MAX_ACTIVITY_LOG raised to 5000 - Remove activeRelays from LiveSyncActivity (active relay chips removed) - Add eventsAccepted: Int to CompletedRelayInfo - Add totalEventsAccepted: Int to SyncState.Done - Register IRelayClientListener on account.client for the duration of the sync to intercept OkMessage(success=true) from destination relays - sourceRelayOfEvent map attributes each forwarded event to the relay it was first seen on; ConcurrentHashMap.remove() ensures the first OK true is counted exactly once per event (dedup across dest relays) - acceptedCountPerRelay accumulates per-source-relay accepted counts, read at onRelayComplete time - downloadFromPool.onEvent changed to (Event, NormalizedRelayUrl) to carry the source relay into runSync routing logic Screen: - Remove ActiveRelaysCard and ActiveRelayChip composables + preview - Remove animation imports (no longer needed) - ActivityLogRow now shows two right-hand columns per relay: recv N — events received from that source relay new N — events accepted as new by destination relays (highlighted in primary color when > 0) - DoneCard now shows three lines: Forwarded N events to destination relays. M events accepted as new by destination relays. ← key headline Completed in N seconds. Strings: event_sync_done_body replaced by done_sent / done_accepted / done_duration; event_sync_reading_from and events_found_log removed; log_recv and log_new added. https://claude.ai/code/session_01U8qF9mK4UBvXsXP1zNMXfX --- .../relays/eventsync/EventSyncScreen.kt | 185 +++++------------- .../relays/eventsync/EventSyncViewModel.kt | 78 ++++++-- amethyst/src/main/res/values/strings.xml | 8 +- 3 files changed, 109 insertions(+), 162 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt index 93f1c21c8..4abfd4345 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt @@ -21,11 +21,6 @@ */ package com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync -import androidx.compose.animation.core.RepeatMode -import androidx.compose.animation.core.animateFloat -import androidx.compose.animation.core.infiniteRepeatable -import androidx.compose.animation.core.rememberInfiniteTransition -import androidx.compose.animation.core.tween import androidx.compose.foundation.background import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Box @@ -175,15 +170,6 @@ fun EventSyncScreen( } // ---- Live relay activity (shown during and after sync) ---- - val isRunning = syncState is EventSyncViewModel.SyncState.Running - - if (liveActivity.activeRelays.isNotEmpty()) { - ActiveRelaysCard( - activeRelays = liveActivity.activeRelays, - isRunning = isRunning, - ) - } - if (liveActivity.outboxTargets.isNotEmpty() || liveActivity.inboxTargets.isNotEmpty() || liveActivity.dmTargets.isNotEmpty() @@ -366,17 +352,25 @@ private fun DoneCard(state: EventSyncViewModel.SyncState.Done) { fontWeight = FontWeight.Bold, color = MaterialTheme.colorScheme.onPrimaryContainer, ) - Spacer(Modifier.height(4.dp)) + Spacer(Modifier.height(6.dp)) Text( - text = - stringRes( - R.string.event_sync_done_body, - state.totalEventsSent, - (state.durationMs / 1000).toInt(), - ), + text = stringRes(R.string.event_sync_done_sent, state.totalEventsSent), style = MaterialTheme.typography.bodyMedium, color = MaterialTheme.colorScheme.onPrimaryContainer, ) + Spacer(Modifier.height(2.dp)) + Text( + text = stringRes(R.string.event_sync_done_accepted, state.totalEventsAccepted), + style = MaterialTheme.typography.bodyMedium, + fontWeight = FontWeight.SemiBold, + color = MaterialTheme.colorScheme.onPrimaryContainer, + ) + Spacer(Modifier.height(4.dp)) + Text( + text = stringRes(R.string.event_sync_done_duration, (state.durationMs / 1000).toInt()), + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onPrimaryContainer.copy(alpha = 0.7f), + ) } } } @@ -411,86 +405,6 @@ private fun ErrorCard(message: String) { // Live activity cards // ------------------------------------------------------------------------- -/** - * Shows all relays currently being queried, each with a pulsing dot indicator. - */ -@OptIn(ExperimentalLayoutApi::class) -@Composable -private fun ActiveRelaysCard( - activeRelays: Set, - isRunning: Boolean, -) { - val infiniteTransition = rememberInfiniteTransition(label = "pulse") - val pulseAlpha by infiniteTransition.animateFloat( - initialValue = 1f, - targetValue = 0.2f, - animationSpec = - infiniteRepeatable( - animation = tween(durationMillis = 800), - repeatMode = RepeatMode.Reverse, - ), - label = "pulseAlpha", - ) - - Card( - modifier = Modifier.fillMaxWidth(), - colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surface), - elevation = CardDefaults.cardElevation(defaultElevation = 1.dp), - ) { - Column(modifier = Modifier.padding(16.dp)) { - Text( - text = stringRes(R.string.event_sync_reading_from, activeRelays.size), - style = MaterialTheme.typography.titleSmall, - fontWeight = FontWeight.SemiBold, - ) - Spacer(Modifier.height(10.dp)) - FlowRow( - horizontalArrangement = Arrangement.spacedBy(8.dp), - verticalArrangement = Arrangement.spacedBy(6.dp), - ) { - activeRelays.forEach { relay -> - ActiveRelayChip( - relay = relay, - pulseAlpha = if (isRunning) pulseAlpha else 1f, - ) - } - } - } - } -} - -@Composable -private fun ActiveRelayChip( - relay: NormalizedRelayUrl, - pulseAlpha: Float, -) { - SuggestionChip( - onClick = {}, - label = { - Text( - text = relay.displayHost(), - style = MaterialTheme.typography.labelSmall, - maxLines = 1, - overflow = TextOverflow.Ellipsis, - ) - }, - icon = { - Box( - modifier = - Modifier - .size(8.dp) - .alpha(pulseAlpha) - .clip(CircleShape) - .background(MaterialTheme.colorScheme.primary), - ) - }, - colors = - SuggestionChipDefaults.suggestionChipColors( - containerColor = MaterialTheme.colorScheme.primaryContainer.copy(alpha = 0.4f), - ), - ) -} - /** * Shows where events are being sent: outbox, inbox, and DM relay lists. */ @@ -663,16 +577,31 @@ private fun ActivityLogRow(info: EventSyncViewModel.LiveSyncActivity.CompletedRe overflow = TextOverflow.Ellipsis, modifier = Modifier.weight(1f), ) - Text( - text = - if (hasEvents) { - stringRes(R.string.event_sync_events_found_log, formatCount(info.eventsFound)) - } else { - stringRes(R.string.event_sync_no_events) - }, - style = MaterialTheme.typography.bodySmall, - color = textColor, - ) + if (hasEvents) { + Text( + text = stringRes(R.string.event_sync_log_recv, formatCount(info.eventsFound)), + style = MaterialTheme.typography.bodySmall, + color = textColor, + ) + Spacer(Modifier.width(8.dp)) + Text( + text = stringRes(R.string.event_sync_log_new, formatCount(info.eventsAccepted)), + style = MaterialTheme.typography.bodySmall, + fontWeight = if (info.eventsAccepted > 0) FontWeight.SemiBold else FontWeight.Normal, + color = + if (info.eventsAccepted > 0) { + MaterialTheme.colorScheme.primary + } else { + MaterialTheme.colorScheme.onSurfaceVariant + }, + ) + } else { + Text( + text = stringRes(R.string.event_sync_no_events), + style = MaterialTheme.typography.bodySmall, + color = textColor, + ) + } } } @@ -721,28 +650,18 @@ private fun formatCount(n: Int): String = // Preview data // ------------------------------------------------------------------------- -private val previewRelaySet = - setOf( - NormalizedRelayUrl("wss://relay.damus.io"), - NormalizedRelayUrl("wss://nos.lol"), - NormalizedRelayUrl("wss://relay.nostr.band"), - NormalizedRelayUrl("wss://nostr.bitcoiner.social"), - NormalizedRelayUrl("wss://relay.snort.social"), - ) - private val previewCompletions = listOf( - EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://relay.damus.io"), 1247), - EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://nos.lol"), 892), - EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://relay.nostr.band"), 3500), - EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://slow.relay.example.com"), 0), - EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://nostr.bitcoiner.social"), 15), - EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://unreachable.relay.xyz"), 0), + EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://relay.damus.io"), 1247, 891), + EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://nos.lol"), 892, 45), + EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://relay.nostr.band"), 3500, 3498), + EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://slow.relay.example.com"), 0, 0), + EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://nostr.bitcoiner.social"), 15, 0), + EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://unreachable.relay.xyz"), 0, 0), ) private val previewActivity = EventSyncViewModel.LiveSyncActivity( - activeRelays = previewRelaySet, recentCompletions = previewCompletions, outboxTargets = setOf( @@ -802,6 +721,7 @@ fun DoneCardPreview() { state = EventSyncViewModel.SyncState.Done( totalEventsSent = 18_432, + totalEventsAccepted = 14_891, durationMs = 187_000, ), ) @@ -816,17 +736,6 @@ fun ErrorCardPreview() { } } -@Composable -@Preview -fun ActiveRelaysCardPreview() { - ThemeComparisonColumn { - ActiveRelaysCard( - activeRelays = previewRelaySet, - isRunning = true, - ) - } -} - @Composable @Preview fun DestinationRelaysCardPreview() { diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt index dd6c0449f..be39a1ad0 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt @@ -23,8 +23,13 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync import com.vitorpamplona.amethyst.model.Account import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message +import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import kotlinx.coroutines.CancellationException @@ -56,8 +61,11 @@ import java.util.concurrent.atomic.AtomicLong * Each relay is paginated individually: after EOSE the oldest [Event.createdAt] seen on that * relay becomes the next `until` cursor, repeating until the relay returns no new events. * - * Live activity is emitted via [liveActivity] so the UI can show which relays are currently - * being read, where events are being sent, and a running log of completed relays. + * OK (true) responses from destination relays are tracked via [IRelayClientListener] and + * attributed back to the source relay that contributed each event. + * + * Live activity is emitted via [liveActivity] so the UI can show a per-relay log of events + * received and events accepted by destination relays. * * The sync is pausable: calling [cancel] transitions to [SyncState.Paused] so the user can * [resume] from the last completed relay index rather than starting over. @@ -76,7 +84,7 @@ class EventSyncViewModel( const val RELAY_TIMEOUT_MS = 30_000L /** Maximum number of completed-relay entries kept in the activity log. */ - const val MAX_ACTIVITY_LOG = 100 + const val MAX_ACTIVITY_LOG = 5000 } // ------------------------------------------------------------------------- @@ -101,6 +109,7 @@ class EventSyncViewModel( data class Done( val totalEventsSent: Int, + val totalEventsAccepted: Int, val durationMs: Long, ) : SyncState() @@ -112,23 +121,27 @@ class EventSyncViewModel( /** * Per-relay activity snapshot emitted continuously while the sync runs. * - * @param activeRelays Relays currently being queried. * @param recentCompletions Last [MAX_ACTIVITY_LOG] relays that finished, newest first. * @param outboxTargets Relays receiving events authored by the user. * @param inboxTargets Relays receiving events that mention the user. * @param dmTargets Relays receiving DMs addressed to the user. */ data class LiveSyncActivity( - val activeRelays: Set = emptySet(), val recentCompletions: List = emptyList(), val outboxTargets: Set = emptySet(), val inboxTargets: Set = emptySet(), val dmTargets: Set = emptySet(), ) { + /** + * @param eventsFound Total events received from this relay across all pages. + * @param eventsAccepted Events from this relay that destination relays accepted as new + * (OK true). Reflects the count at relay-completion time; late + * OK responses may not be included. + */ data class CompletedRelayInfo( val relay: NormalizedRelayUrl, - /** Total events received across all pages. 0 = relay was empty or unreachable. */ val eventsFound: Int, + val eventsAccepted: Int, ) } @@ -142,7 +155,6 @@ class EventSyncViewModel( // Live activity tracking (written from worker threads) // ------------------------------------------------------------------------- - private val trackingActiveRelays = ConcurrentHashMap.newKeySet() private val trackingCompletions = ArrayDeque() private val trackingLock = Any() @@ -155,7 +167,6 @@ class EventSyncViewModel( val completions = synchronized(trackingLock) { trackingCompletions.toList() } _liveActivity.value = LiveSyncActivity( - activeRelays = trackingActiveRelays.toSet(), recentCompletions = completions, outboxTargets = liveOutboxTargets, inboxTargets = liveInboxTargets, @@ -171,7 +182,6 @@ class EventSyncViewModel( fun start() { if (_syncState.value is SyncState.Running) return - trackingActiveRelays.clear() synchronized(trackingLock) { trackingCompletions.clear() } _liveActivity.value = LiveSyncActivity() syncJob = @@ -256,12 +266,35 @@ class EventSyncViewModel( return } - // Thread-safe dedup sets and counter — relay workers run concurrently. + // Thread-safe dedup sets and counters — relay workers run concurrently. val outboxSent = ConcurrentHashMap.newKeySet() val inboxSent = ConcurrentHashMap.newKeySet() val dmSent = ConcurrentHashMap.newKeySet() val totalSent = AtomicLong(initialEventsSent.toLong()) + // OK (true) tracking: maps each sent event ID to its source relay. + // The first OK true for an event atomically removes it from this map, + // crediting the acceptance to the source relay and preventing double-counting. + val sourceRelayOfEvent = ConcurrentHashMap() + val acceptedCountPerRelay = ConcurrentHashMap() + val totalAccepted = AtomicLong(0) + + val okListener = + object : IRelayClientListener { + override fun onIncomingMessage( + relay: IRelayClient, + msgStr: String, + msg: Message, + ) { + if (msg is OkMessage && msg.success) { + // remove() is atomic: returns non-null only for the first OK per event. + val sourceRelay = sourceRelayOfEvent.remove(msg.eventId) ?: return + acceptedCountPerRelay.getOrPut(sourceRelay) { AtomicInteger(0) }.incrementAndGet() + totalAccepted.incrementAndGet() + } + } + } + val relaysCompleted = AtomicInteger(startRelayIndex) _syncState.value = @@ -271,13 +304,15 @@ class EventSyncViewModel( eventsSent = totalSent.get().toInt(), ) + account.client.subscribe(okListener) try { downloadFromPool( relays = relaysToProcess, baseFilters = baseFilters, - onEvent = { event -> + onEvent = { event, sourceRelay -> if (event.pubKey == myPubKey && outboxTargets.isNotEmpty()) { if (outboxSent.add(event.id)) { + sourceRelayOfEvent[event.id] = sourceRelay account.client.send(event, outboxTargets) totalSent.incrementAndGet() } @@ -289,24 +324,22 @@ class EventSyncViewModel( if (pTagsMe) { if (event.kind == 4 || event.kind == 1059) { if (dmTargets.isNotEmpty() && dmSent.add(event.id)) { + sourceRelayOfEvent[event.id] = sourceRelay account.client.send(event, dmTargets) totalSent.incrementAndGet() } } else { if (inboxTargets.isNotEmpty() && inboxSent.add(event.id)) { + sourceRelayOfEvent[event.id] = sourceRelay account.client.send(event, inboxTargets) totalSent.incrementAndGet() } } } }, - onRelayStart = { relay -> - trackingActiveRelays.add(relay) - emitLiveSnapshot() - }, onRelayComplete = { relay, eventsFound -> - trackingActiveRelays.remove(relay) - val info = LiveSyncActivity.CompletedRelayInfo(relay, eventsFound) + val eventsAccepted = acceptedCountPerRelay[relay]?.get() ?: 0 + val info = LiveSyncActivity.CompletedRelayInfo(relay, eventsFound, eventsAccepted) synchronized(trackingLock) { trackingCompletions.addFirst(info) while (trackingCompletions.size > MAX_ACTIVITY_LOG) trackingCompletions.removeLast() @@ -325,12 +358,15 @@ class EventSyncViewModel( _syncState.value = SyncState.Done( totalEventsSent = totalSent.get().toInt(), + totalEventsAccepted = totalAccepted.get().toInt(), durationMs = System.currentTimeMillis() - startTime, ) } catch (e: CancellationException) { throw e } catch (e: Exception) { _syncState.value = SyncState.Error(e.message ?: "Unknown error") + } finally { + account.client.unsubscribe(okListener) } } @@ -338,12 +374,13 @@ class EventSyncViewModel( * Maintains a sliding window of up to [MAX_CONCURRENT_RELAYS] active relay workers. * As soon as one relay finishes (all pages exhausted), the next relay from [relays] * starts immediately — no waiting for an entire batch to drain. + * + * [onEvent] receives the event and the URL of the relay it came from. */ private suspend fun downloadFromPool( relays: List, baseFilters: List, - onEvent: (Event) -> Unit, - onRelayStart: (NormalizedRelayUrl) -> Unit, + onEvent: (Event, NormalizedRelayUrl) -> Unit, onRelayComplete: (NormalizedRelayUrl, Int) -> Unit, ) { val semaphore = Semaphore(MAX_CONCURRENT_RELAYS) @@ -353,8 +390,7 @@ class EventSyncViewModel( semaphore.acquire() launch { try { - onRelayStart(relay) - val eventsFound = downloadFromRelay(relay, baseFilters, onEvent) + val eventsFound = downloadFromRelay(relay, baseFilters) { event -> onEvent(event, relay) } onRelayComplete(relay, eventsFound) } finally { semaphore.release() diff --git a/amethyst/src/main/res/values/strings.xml b/amethyst/src/main/res/values/strings.xml index b5eaa0ae2..2cb9e4ad9 100644 --- a/amethyst/src/main/res/values/strings.xml +++ b/amethyst/src/main/res/values/strings.xml @@ -1772,14 +1772,16 @@ Sync Paused Completed %1$d of %2$d relays — %3$d events redistributed so far. Tap Resume to continue. Sync complete - Redistributed %1$d events in %2$d seconds. + Forwarded %1$d events to destination relays. + %1$d events accepted as new by destination relays. + Completed in %1$d seconds. Sync error - Reading (%1$d active) Sending To Outbox Inbox DMs Activity (%1$d relays) - %1$s events + recv %1$s + new %1$s no events