diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt index 587a637ae..6046daf7b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt @@ -38,7 +38,6 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.tags.people.isTaggedUser -import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job @@ -50,8 +49,7 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.supervisorScope import kotlinx.coroutines.sync.Semaphore import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.atomic.AtomicInteger -import java.util.concurrent.atomic.AtomicLong +import kotlin.coroutines.cancellation.CancellationException /** * Syncs the user's events across all known relays: @@ -103,12 +101,24 @@ class EventSync( object Idle : SyncState() data class Running( - val relaysCompleted: Int, - val totalRelays: Int, - val eventsSent: Int, - ) : SyncState() + val relaysCompleted: MutableStateFlow, + val totalRelays: MutableStateFlow, + val eventsSent: MutableStateFlow, + val eventsReceived: MutableStateFlow, + val eventsAccepted: MutableStateFlow, + ) : SyncState() { + constructor(relaysCompleted: Int, totalRelays: Int, eventsSent: Int, eventsReceived: Int, eventsAccepted: Int) : + this( + relaysCompleted = MutableStateFlow(relaysCompleted), + totalRelays = MutableStateFlow(totalRelays), + eventsSent = MutableStateFlow(eventsSent), + eventsReceived = MutableStateFlow(eventsReceived), + eventsAccepted = MutableStateFlow(eventsAccepted), + ) + } data class Done( + val totalEventsReceived: Int, val totalEventsSent: Int, val totalEventsAccepted: Int, val durationMs: Long, @@ -180,9 +190,16 @@ class EventSync( val status: MutableStateFlow, val eventsFound: MutableStateFlow, val eventsAccepted: MutableStateFlow, + val pageUntil: MutableStateFlow = MutableStateFlow(null), ) { - constructor(relay: NormalizedRelayUrl, status: ConnectionStatus, eventsFound: Int, eventsAccepted: Int) : - this(relay, MutableStateFlow(status), MutableStateFlow(eventsFound), MutableStateFlow(eventsAccepted)) + constructor(relay: NormalizedRelayUrl, status: ConnectionStatus, eventsFound: Int, eventsAccepted: Int, untilPage: Long? = null) : + this( + relay = relay, + status = MutableStateFlow(status), + eventsFound = MutableStateFlow(eventsFound), + eventsAccepted = MutableStateFlow(eventsAccepted), + pageUntil = MutableStateFlow(untilPage), + ) } /** @@ -346,14 +363,17 @@ class EventSync( val outboxDedup = ConcurrentHashMap.newKeySet() val inboxDedup = ConcurrentHashMap.newKeySet() val dmDedup = ConcurrentHashMap.newKeySet() - val totalSent = AtomicLong(0) - // 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 acceptedCountPerSourceRelay = ConcurrentHashMap() - val totalAccepted = AtomicLong(0) + + val runningState = + SyncState.Running( + relaysCompleted = 0, + totalRelays = totalRelays, + eventsSent = 0, + eventsReceived = 0, + eventsAccepted = 0, + ) val okListener = object : IRelayClientListener { @@ -375,22 +395,30 @@ class EventSync( success: Boolean, ) { super.onSent(relay, cmdStr, cmd, success) - if (cmd is EventCmd) { + var hasSent = false + if (outboxDedup.contains(cmd.event.id)) { liveActivity.value.outboxTargets[relay.url] ?.eventsSent ?.update { it + 1 } + hasSent = true } if (inboxDedup.contains(cmd.event.id)) { liveActivity.value.inboxTargets[relay.url] ?.eventsSent ?.update { it + 1 } + hasSent = true } if (dmDedup.contains(cmd.event.id)) { liveActivity.value.dmTargets[relay.url] ?.eventsSent ?.update { it + 1 } + hasSent = true + } + + if (hasSent) { + runningState.eventsSent.update { it + 1 } } } else if (cmd is ReqCmd) { val currentStatus = liveActivity.value.runningRelays[relay.url]?.status @@ -407,37 +435,39 @@ class EventSync( ) { if (msg is OkMessage && msg.success && msg.message.isBlank()) { // remove() is atomic: returns non-null only for the first OK per event. - val sourceRelay = sourceRelayOfEvent.remove(msg.eventId) ?: return - acceptedCountPerSourceRelay.getOrPut(sourceRelay) { AtomicInteger(0) }.incrementAndGet() - totalAccepted.incrementAndGet() + val sourceRelay = sourceRelayOfEvent.remove(msg.eventId) + if (sourceRelay != null) { + liveActivity.value.runningRelays[sourceRelay] + ?.eventsAccepted + ?.update { it + 1 } + } if (outboxDedup.contains(msg.eventId)) { - liveActivity.value.outboxTargets[relay.url] - ?.eventsAccepted - ?.update { it + 1 } + val relayTarget = liveActivity.value.outboxTargets[relay.url] + if (relayTarget != null) { + relayTarget.eventsAccepted.update { it + 1 } + runningState.eventsAccepted.update { it + 1 } + } } if (dmDedup.contains(msg.eventId)) { - liveActivity.value.dmTargets[relay.url] - ?.eventsAccepted - ?.update { it + 1 } + val relayTarget = liveActivity.value.dmTargets[relay.url] + if (relayTarget != null) { + relayTarget.eventsAccepted.update { it + 1 } + runningState.eventsAccepted.update { it + 1 } + } } if (inboxDedup.contains(msg.eventId)) { - liveActivity.value.inboxTargets[relay.url] - ?.eventsAccepted - ?.update { it + 1 } + val relayTarget = liveActivity.value.inboxTargets[relay.url] + if (relayTarget != null) { + relayTarget.eventsAccepted.update { it + 1 } + runningState.eventsAccepted.update { it + 1 } + } } } } } - val relaysCompleted = AtomicInteger(0) - - _syncState.value = - SyncState.Running( - relaysCompleted = relaysCompleted.get(), - totalRelays = totalRelays, - eventsSent = totalSent.get().toInt(), - ) + _syncState.emit(runningState) clientBuilder().use { client -> client.subscribe(okListener) @@ -445,6 +475,11 @@ class EventSync( client.downloadFromPool( relays = relaysToProcess, filters = perRelayFilters, + onNewPage = { until, sourceRelay -> + _liveActivity.value.runningRelays[sourceRelay] + ?.pageUntil + ?.tryEmit(until) + }, onEvent = { event, sourceRelay -> val isMyEvent = event.pubKey == myPubKey val mentionsMe = event.tags.isTaggedUser(myPubKey) @@ -452,36 +487,41 @@ class EventSync( val live = liveActivity.value + var newEvent = false + var matchesAtLeastOneFilter = false + // Each routing rule is independent: an event can match more than one. if (isMyEvent && outboxTargets.isNotEmpty()) { if (outboxDedup.add(event.id)) { - sourceRelayOfEvent[event.id] = sourceRelay client.send(event, outboxTargets) - totalSent.incrementAndGet() - - live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 } - live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 } + newEvent = true } + matchesAtLeastOneFilter = true } if (mentionsMe && isDmKind && dmTargets.isNotEmpty()) { if (dmDedup.add(event.id)) { - sourceRelayOfEvent[event.id] = sourceRelay client.send(event, dmTargets) - totalSent.incrementAndGet() - - live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 } - live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 } + newEvent = true } + matchesAtLeastOneFilter = true } if (mentionsMe && !isDmKind && inboxTargets.isNotEmpty()) { if (inboxDedup.add(event.id)) { - sourceRelayOfEvent[event.id] = sourceRelay client.send(event, inboxTargets) - totalSent.incrementAndGet() - - live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 } - live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 } + newEvent = true } + matchesAtLeastOneFilter = true + } + + if (newEvent) { + sourceRelayOfEvent[event.id] = sourceRelay + } + + if (matchesAtLeastOneFilter) { + runningState.eventsReceived.update { it + 1 } + + live.runningRelays[sourceRelay]?.eventsFound?.update { it + 1 } + live.completedRelays[sourceRelay]?.eventsFound?.update { it + 1 } } }, onRelayStart = { relay -> @@ -513,30 +553,27 @@ class EventSync( status?.tryEmit(LiveSyncActivity.ConnectionStatus.Completed) } - _syncState.value = - SyncState.Running( - relaysCompleted = relaysCompleted.incrementAndGet(), - totalRelays = totalRelays, - eventsSent = totalSent.get().toInt(), - ) + runningState.relaysCompleted.update { it + 1 } }, ) _syncState.value = SyncState.Done( - totalEventsSent = totalSent.get().toInt(), - totalEventsAccepted = totalAccepted.get().toInt(), + totalEventsReceived = runningState.eventsReceived.value, + totalEventsSent = runningState.eventsSent.value, + totalEventsAccepted = runningState.eventsAccepted.value, durationMs = System.currentTimeMillis() - startTime, ) - } catch (e: CancellationException) { + } catch (e: Exception) { _syncState.value = SyncState.Done( - totalEventsSent = totalSent.get().toInt(), - totalEventsAccepted = totalAccepted.get().toInt(), + totalEventsReceived = runningState.eventsReceived.value, + totalEventsSent = runningState.eventsSent.value, + totalEventsAccepted = runningState.eventsAccepted.value, durationMs = System.currentTimeMillis() - startTime, ) - throw e - } catch (e: Exception) { + + if (e is CancellationException) throw e _syncState.value = SyncState.Error(e.message ?: "Unknown error") } finally { client.unsubscribe(okListener) @@ -554,6 +591,7 @@ class EventSync( private suspend fun INostrClient.downloadFromPool( relays: List, filters: Map>, + onNewPage: (Long, NormalizedRelayUrl) -> Unit, onEvent: (Event, NormalizedRelayUrl) -> Unit, onRelayStart: (NormalizedRelayUrl) -> Unit, onRelayComplete: (NormalizedRelayUrl) -> Unit, @@ -567,7 +605,12 @@ class EventSync( try { onRelayStart(relay) filters[relay]?.let { filtersForRelay -> - downloadFromRelay(relay, filtersForRelay) { event -> onEvent(event, relay) } + downloadFromRelay( + relay = relay, + filters = filtersForRelay, + onNewPage = { onNewPage(it, relay) }, + onEvent = { onEvent(it, relay) }, + ) } ?: 0 onRelayComplete(relay) } finally { @@ -587,6 +630,7 @@ class EventSync( private suspend fun INostrClient.downloadFromRelay( relay: NormalizedRelayUrl, filters: List, + onNewPage: (Long) -> Unit, onEvent: (Event) -> Unit, - ): Int = reqBypassingRelayLimits(relay, filters, RELAY_TIMEOUT_MS, onEvent) + ): Int = reqBypassingRelayLimits(relay, filters, RELAY_TIMEOUT_MS, onNewPage, onEvent) } 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 100e8d575..ce0020da2 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 @@ -54,6 +54,7 @@ import androidx.compose.runtime.setValue import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.draw.clip +import androidx.compose.ui.platform.LocalContext import androidx.compose.ui.text.font.FontWeight import androidx.compose.ui.text.style.TextAlign import androidx.compose.ui.text.style.TextOverflow @@ -63,6 +64,7 @@ import androidx.lifecycle.compose.collectAsStateWithLifecycle import com.vitorpamplona.amethyst.R import com.vitorpamplona.amethyst.ui.navigation.navs.INav import com.vitorpamplona.amethyst.ui.navigation.topbars.TopBarWithBackButton +import com.vitorpamplona.amethyst.ui.note.timeAgoNoDot import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel import com.vitorpamplona.amethyst.ui.stringRes import com.vitorpamplona.amethyst.ui.theme.ThemeComparisonColumn @@ -292,28 +294,9 @@ private fun SyncProgressCard( elevation = CardDefaults.cardElevation(defaultElevation = 1.dp), ) { Column(modifier = Modifier.padding(16.dp)) { - Text( - text = stringRes(R.string.event_sync_relays_progress, state.relaysCompleted, state.totalRelays), - style = MaterialTheme.typography.titleSmall, - fontWeight = FontWeight.SemiBold, - ) + RelayStatement(state) Spacer(Modifier.height(8.dp)) - LinearProgressIndicator( - progress = { - if (state.totalRelays > 0) { - state.relaysCompleted.toFloat() / state.totalRelays - } else { - 0f - } - }, - modifier = Modifier.fillMaxWidth(), - ) - Spacer(Modifier.height(8.dp)) - Text( - text = stringRes(R.string.event_sync_events_sent, state.eventsSent), - style = MaterialTheme.typography.bodySmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - ) + EventsReceivedStatement(state) Spacer(Modifier.height(10.dp)) OutlinedButton( onClick = onCancel, @@ -329,6 +312,42 @@ private fun SyncProgressCard( } } +@Composable +private fun EventsReceivedStatement(state: EventSync.SyncState.Running) { + val eventsReceived by state.eventsReceived.collectAsStateWithLifecycle() + val eventsSent by state.eventsSent.collectAsStateWithLifecycle() + val eventsAccepted by state.eventsAccepted.collectAsStateWithLifecycle() + + Text( + text = stringRes(R.string.event_sync_events_sent, eventsAccepted, eventsSent, eventsReceived), + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) +} + +@Composable +private fun RelayStatement(state: EventSync.SyncState.Running) { + val relaysCompleted by state.relaysCompleted.collectAsStateWithLifecycle() + val totalRelays by state.totalRelays.collectAsStateWithLifecycle() + + Text( + text = stringRes(R.string.event_sync_relays_progress, relaysCompleted, totalRelays), + style = MaterialTheme.typography.titleSmall, + fontWeight = FontWeight.SemiBold, + ) + Spacer(Modifier.height(8.dp)) + LinearProgressIndicator( + progress = { + if (totalRelays > 0) { + relaysCompleted / totalRelays.toFloat() + } else { + 0f + } + }, + modifier = Modifier.fillMaxWidth(), + ) +} + @Composable private fun DoneCard( state: EventSync.SyncState.Done, @@ -351,7 +370,7 @@ private fun DoneCard( ) Spacer(Modifier.height(6.dp)) Text( - text = stringRes(R.string.event_sync_done_sent, state.totalEventsSent), + text = stringRes(R.string.event_sync_done_sent, state.totalEventsSent, state.totalEventsReceived), style = MaterialTheme.typography.bodyMedium, color = MaterialTheme.colorScheme.onPrimaryContainer, ) @@ -544,7 +563,6 @@ private fun DestinationRelayRow( @Composable private fun ActivityLogRow(info: EventSync.LiveSyncActivity.SourceRelayInfo) { val eventsFound by info.eventsFound.collectAsStateWithLifecycle() - val status by info.status.collectAsStateWithLifecycle() val hasEvents = eventsFound > 0 val dotColor = if (hasEvents) { @@ -583,12 +601,25 @@ private fun ActivityLogRow(info: EventSync.LiveSyncActivity.SourceRelayInfo) { modifier = Modifier.weight(0.6f), ) if (hasEvents) { + val context = LocalContext.current + val untilPage by info.pageUntil.collectAsStateWithLifecycle() val eventsAccepted by info.eventsAccepted.collectAsStateWithLifecycle() + untilPage?.let { + Text( + text = stringRes(R.string.event_sync_less_than_until, timeAgoNoDot(it, context)), + style = MaterialTheme.typography.bodySmall, + color = textColor, + modifier = Modifier.weight(0.3f), + maxLines = 1, + textAlign = TextAlign.End, + overflow = TextOverflow.StartEllipsis, + ) + } Text( text = stringRes(R.string.event_sync_log_recv, formatCount(eventsFound)), style = MaterialTheme.typography.bodySmall, color = textColor, - modifier = Modifier.weight(0.2f), + modifier = Modifier.weight(0.3f), maxLines = 1, textAlign = TextAlign.End, overflow = TextOverflow.StartEllipsis, @@ -606,7 +637,7 @@ private fun ActivityLogRow(info: EventSync.LiveSyncActivity.SourceRelayInfo) { maxLines = 1, textAlign = TextAlign.End, overflow = TextOverflow.StartEllipsis, - modifier = Modifier.weight(0.2f), + modifier = Modifier.weight(0.3f), ) } else { val status by info.status.collectAsStateWithLifecycle() @@ -737,7 +768,9 @@ fun SyncProgressCardPreview() { EventSync.SyncState.Running( relaysCompleted = 312, totalRelays = 1024, + eventsAccepted = 12, eventsSent = 4821, + eventsReceived = 10000, ), onCancel = {}, ) @@ -751,6 +784,7 @@ fun DoneCardPreview() { DoneCard( state = EventSync.SyncState.Done( + totalEventsReceived = 20_000, totalEventsSent = 18_432, totalEventsAccepted = 14_891, durationMs = 187_000, @@ -791,7 +825,7 @@ fun EventScreenBodyPreview() { fun EventScreenBody2Preview() { ThemeComparisonRow { EventScreenBody( - EventSync.SyncState.Running(1247, 1024, 4821), + EventSync.SyncState.Running(1047, 1224, 100, 4821, 10000), previewActivity, ) } diff --git a/amethyst/src/main/res/values/strings.xml b/amethyst/src/main/res/values/strings.xml index 91a96bcc7..5df2fa4d1 100644 --- a/amethyst/src/main/res/values/strings.xml +++ b/amethyst/src/main/res/values/strings.xml @@ -1829,11 +1829,11 @@ Start Over Cancel Relays: %1$d / %2$d - Events redistributed: %1$d + Events redistributed: %1$d new out of %2$d sent and %3$d received Sync Paused Completed %1$d of %2$d relays — %3$d events redistributed so far. Tap Resume to continue. Sync complete - Forwarded %1$d events to destination relays. + Forwarded %1$d events to destination relays out of %2$d received. %1$d events accepted as new by destination relays. Completed in %1$d seconds. Sync error @@ -1854,6 +1854,7 @@ relay settings Last seen %1$s ago + <%1$s Connecting Downloading Error diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientReqBypassingRelayLimitsExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientReqBypassingRelayLimitsExt.kt index e2438dff5..22ea0fd24 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientReqBypassingRelayLimitsExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientReqBypassingRelayLimitsExt.kt @@ -21,6 +21,8 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.accessories import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient.close +import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient.openReqSubscription import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId @@ -54,6 +56,7 @@ suspend fun INostrClient.reqBypassingRelayLimits( relay: NormalizedRelayUrl, filters: List, timeoutMs: Long = 30_000L, + onNewPage: ((Long) -> Unit)? = null, onEvent: (Event) -> Unit, ): Int { var until: Long? = null @@ -82,6 +85,7 @@ suspend fun INostrClient.reqBypassingRelayLimits( if (until == null) { remainingFilters } else { + onNewPage?.invoke(until) remainingFilters.map { it.copy(until = until) } } @@ -162,11 +166,13 @@ suspend fun INostrClient.reqBypassingRelayLimits( relay: String, filters: List, timeoutMs: Long = 30_000L, + onNewPage: ((Long) -> Unit)? = null, onEvent: (Event) -> Unit, ): Int = reqBypassingRelayLimits( relay = RelayUrlNormalizer.normalize(relay), filters = filters, timeoutMs = timeoutMs, + onNewPage = onNewPage, onEvent = onEvent, )