nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: per-relay recv/new counts and OK-true acceptance tracking

Browse Source 
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
This commit is contained in:
Claude
2026-03-15 14:19:08 +00:00
parent a49a0ac9ae
commit aecbcb68a5
3 changed files with 109 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt
+47 -138
View File
@@ -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<NormalizedRelayUrl>,
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() {
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt
+57 -21
View File
@@ -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<NormalizedRelayUrl> = emptySet(),
val recentCompletions: List<CompletedRelayInfo> = emptyList(),
val outboxTargets: Set<NormalizedRelayUrl> = emptySet(),
val inboxTargets: Set<NormalizedRelayUrl> = emptySet(),
val dmTargets: Set<NormalizedRelayUrl> = 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<NormalizedRelayUrl>()
private val trackingCompletions = ArrayDeque<LiveSyncActivity.CompletedRelayInfo>()
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<String>()
val inboxSent = ConcurrentHashMap.newKeySet<String>()
val dmSent = ConcurrentHashMap.newKeySet<String>()
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<HexKey, NormalizedRelayUrl>()
val acceptedCountPerRelay = ConcurrentHashMap<NormalizedRelayUrl, AtomicInteger>()
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<NormalizedRelayUrl>,
baseFilters: List<Filter>,
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()
amethyst/src/main/res/values/strings.xml
+5 -3
View File
@@ -1772,14 +1772,16 @@
<string name="event_sync_paused_title">Sync Paused</string>
<string name="event_sync_paused_body">Completed %1$d of %2$d relays — %3$d events redistributed so far. Tap Resume to continue.</string>
<string name="event_sync_done_title">Sync complete</string>
<string name="event_sync_done_body">Redistributed %1$d events in %2$d seconds.</string>
<string name="event_sync_done_sent">Forwarded %1$d events to destination relays.</string>
<string name="event_sync_done_accepted">%1$d events accepted as new by destination relays.</string>
<string name="event_sync_done_duration">Completed in %1$d seconds.</string>
<string name="event_sync_error_title">Sync error</string>
<string name="event_sync_reading_from">Reading (%1$d active)</string>
<string name="event_sync_sending_to">Sending To</string>
<string name="event_sync_outbox_relays">Outbox</string>
<string name="event_sync_inbox_relays">Inbox</string>
<string name="event_sync_dm_relays">DMs</string>
<string name="event_sync_activity_log">Activity (%1$d relays)</string>
<string name="event_sync_events_found_log">%1$s events</string>
<string name="event_sync_log_recv">recv %1$s</string>
<string name="event_sync_log_new">new %1$s</string>
<string name="event_sync_no_events">no events</string>
</resources>