From 3e3adaff100df34499a9f4c3541af30d54c08117 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 18 Mar 2026 17:35:05 -0400 Subject: [PATCH] Finalizes the RelaySync utility --- .../relays/eventsync/EventSyncTest.kt | 73 +++ .../ui/screen/loggedIn/AccountViewModel.kt | 66 +- .../loggedIn/relays/eventsync/EventSync.kt | 541 ++++++++++------ .../relays/eventsync/EventSyncScreen.kt | 598 +++++++++--------- amethyst/src/main/res/values/strings.xml | 11 +- .../nip01Core/relay/client/INostrClient.kt | 4 +- .../nip01Core/relay/client/NostrClient.kt | 7 +- .../NostrClientReqBypassingRelayLimitsExt.kt | 62 +- .../NostrClientSingleDownloadExt.kt | 25 +- .../relay/client/pool/PoolEventOutbox.kt | 70 +- .../relay/client/pool/PoolEventOutboxState.kt | 68 +- .../relay/client/stats/RelayStats.kt | 2 + .../signer/NostrSignerRemoteIsolationTest.kt | 2 + 13 files changed, 959 insertions(+), 570 deletions(-) create mode 100644 amethyst/src/androidTest/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncTest.kt diff --git a/amethyst/src/androidTest/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncTest.kt b/amethyst/src/androidTest/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncTest.kt new file mode 100644 index 000000000..387d40572 --- /dev/null +++ b/amethyst/src/androidTest/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncTest.kt @@ -0,0 +1,73 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync + +import androidx.test.ext.junit.runners.AndroidJUnit4 +import com.vitorpamplona.amethyst.model.Constants +import com.vitorpamplona.amethyst.service.okhttp.DefaultContentTypeInterceptor +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.normalizeRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.sockets.okhttp.BasicOkHttpWebSocket +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.runBlocking +import okhttp3.OkHttpClient +import org.junit.Test +import org.junit.runner.RunWith + +@RunWith(AndroidJUnit4::class) +class EventSyncTest { + companion object { + val vitor = "wss://vitor.nostr1.com".normalizeRelayUrl() + val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + + val rootClient = + OkHttpClient + .Builder() + .followRedirects(true) + .followSslRedirects(true) + .addInterceptor(DefaultContentTypeInterceptor("Amethyst/v1.05")) + .build() + val socketBuilder = BasicOkHttpWebSocket.Builder { url -> rootClient } + } + + @Test + fun testSync() = + runBlocking { + val sync = + EventSync( + accountPubKey = "460c25e682fda7832b52d1f22d3d22b3176d972f60dcdc3212ed8c92ef85065c", + relayDb = { + listOf(Constants.mom, Constants.nos) + }, + outboxTargets = { setOf(vitor) }, + inboxTargets = { setOf(vitor) }, + dmTargets = { setOf(vitor) }, + clientBuilder = { + NostrClient(socketBuilder, appScope) + }, + scope = appScope, + ) + + sync.runSync() + } +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt index a1a22994b..b3bd79893 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt @@ -35,6 +35,7 @@ import coil3.asDrawable import coil3.imageLoader import coil3.request.ImageRequest import com.vitorpamplona.amethyst.AccountInfo +import com.vitorpamplona.amethyst.Amethyst import com.vitorpamplona.amethyst.LocalPreferences import com.vitorpamplona.amethyst.R import com.vitorpamplona.amethyst.commons.compose.GenericBaseCache @@ -94,8 +95,10 @@ import com.vitorpamplona.quartz.nip01Core.core.toHexKey import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.hints.EventHintBundle import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.RelayOfflineTracker import com.vitorpamplona.quartz.nip01Core.relay.client.auth.EmptyIAuthStatus +import com.vitorpamplona.quartz.nip01Core.relay.client.auth.RelayAuthenticator import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal import com.vitorpamplona.quartz.nip01Core.signers.SignerExceptions @@ -145,8 +148,10 @@ import kotlinx.collections.immutable.persistentSetOf import kotlinx.collections.immutable.toImmutableList import kotlinx.collections.immutable.toImmutableSet import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.StateFlow @@ -175,7 +180,66 @@ class AccountViewModel( val toastManager = ToastManager() val broadcastTracker = BroadcastTracker() val feedStates = AccountFeedContentStates(account, viewModelScope) - val eventSync = EventSync(account, viewModelScope) + + val eventSync = + EventSync( + accountPubKey = account.signer.pubKey, + relayDb = { + val stats = Amethyst.instance.relayStats.snapshot() + + val relays = + account.cache.relayHints.relayDB + .keys() + .filter { url -> + val relayStat = stats[url] + // has connected at least once OR never tried. + if (relayStat != null) { + relayStat.connectionCompleted > 0 || relayStat.connectionTentatives == 0 + } else { + true + } + } + + val sortMap = relays.associateWith { stats.get(it)?.receivedBytes } + + relays.sortedByDescending { sortMap[it] } + }, + outboxTargets = { account.nip65RelayList.outboxFlow.value }, + inboxTargets = { account.nip65RelayList.inboxFlow.value }, + dmTargets = { account.dmRelayList.flow.value }, + clientBuilder = { + // creates a new client to make sure these events don't end up polluting the local cache. + + // Create a new scope that inherits the ViewModel's lifecycle + // but uses a SupervisorJob so child failures are independent. + val customScope = CoroutineScope(viewModelScope.coroutineContext + SupervisorJob()) + + // Provides a relay pool + val newClient = NostrClient(Amethyst.instance.websocketBuilder, customScope) + + // Authenticates with relays. + val auth = + RelayAuthenticator( + newClient, + customScope, + signWithAllLoggedInUsers = { authTemplate -> + if (account.signer.isWriteable()) { + try { + listOf(account.signer.sign(authTemplate)) + } catch (e: Exception) { + Log.e("AuthCoordinator", "Failed trying to authenticate a writeable account", e) + emptyList() + } + } else { + emptyList() + } + }, + ) + + newClient + }, + scope = viewModelScope, + ) val tempManualPaymentCache = LruCache>(5) 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 204757ec2..587a637ae 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 @@ -20,17 +20,23 @@ */ package com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync -import com.vitorpamplona.amethyst.model.Account +import androidx.compose.runtime.Stable +import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync.EventSync.Companion.MAX_ACTIVITY_LOG +import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync.EventSync.Companion.MAX_CONCURRENT_RELAYS +import com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync.EventSync.LiveSyncActivity.SourceRelayInfo import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.reqBypassingRelayLimits import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd +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.relay.normalizer.normalizeRelayUrl import com.vitorpamplona.quartz.nip01Core.tags.people.isTaggedUser import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope @@ -38,6 +44,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.update import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.supervisorScope @@ -65,18 +72,21 @@ import java.util.concurrent.atomic.AtomicLong * 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. - * * Scoped to the AccountViewModel so the sync survives navigation within the same session. */ +@Stable class EventSync( - val account: Account, + private val accountPubKey: HexKey, + private val relayDb: () -> List, + private val outboxTargets: () -> Set, + private val inboxTargets: () -> Set, + private val dmTargets: () -> Set, + private val clientBuilder: () -> INostrClient, private val scope: CoroutineScope, ) { companion object { /** Maximum number of relays queried at the same time. */ - const val MAX_CONCURRENT_RELAYS = 50 + const val MAX_CONCURRENT_RELAYS = 10 /** How long (ms) to wait for a single relay to reply per page before giving up. */ const val RELAY_TIMEOUT_MS = 30_000L @@ -98,13 +108,6 @@ class EventSync( val eventsSent: Int, ) : SyncState() - /** Cancelled mid-run; can be resumed from [nextRelayIndex]. */ - data class Paused( - val nextRelayIndex: Int, - val totalRelays: Int, - val eventsSent: Int, - ) : SyncState() - data class Done( val totalEventsSent: Int, val totalEventsAccepted: Int, @@ -119,39 +122,83 @@ class EventSync( /** * Per-relay activity snapshot emitted continuously while the sync runs. * - * @param recentCompletions Last [MAX_ACTIVITY_LOG] relays that finished, newest first. + * @param completedRelays Last [MAX_ACTIVITY_LOG] relays that finished, sorted by most events + * found (descending) so the most productive sources appear 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. */ + @Stable data class LiveSyncActivity( - val recentCompletions: List = emptyList(), - val outboxTargets: List = emptyList(), - val inboxTargets: List = emptyList(), - val dmTargets: List = emptyList(), + val runningRelays: Map = emptyMap(), + val completedRelays: Map = emptyMap(), + val outboxTargets: Map = emptyMap(), + val inboxTargets: Map = emptyMap(), + val dmTargets: Map = emptyMap(), ) { + companion object { + val DefaultOrder = compareByDescending { it.eventsFound.value }.thenByDescending { it.status.value == ConnectionStatus.Completed } + } + + val sortedCompletedRelays = completedRelays.values.sortedWith(DefaultOrder) + + constructor( + runningRelays: List, + completedRelays: List, + outboxTargets: List, + inboxTargets: List, + dmTargets: List, + ) : this( + runningRelays.associateBy { it.relay }, + completedRelays.associateBy { it.relay }, + outboxTargets.associateBy { it.relay }, + inboxTargets.associateBy { it.relay }, + dmTargets.associateBy { it.relay }, + ) + + sealed interface ConnectionStatus { + object Connecting : ConnectionStatus + + object Querying : ConnectionStatus + + class Error( + val msg: String, + ) : ConnectionStatus + + object Completed : ConnectionStatus + } + /** * @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( + @Stable + data class SourceRelayInfo( val relay: NormalizedRelayUrl, - val eventsFound: Int, - val eventsAccepted: Int, - ) + val status: MutableStateFlow, + val eventsFound: MutableStateFlow, + val eventsAccepted: MutableStateFlow, + ) { + constructor(relay: NormalizedRelayUrl, status: ConnectionStatus, eventsFound: Int, eventsAccepted: Int) : + this(relay, MutableStateFlow(status), MutableStateFlow(eventsFound), MutableStateFlow(eventsAccepted)) + } /** * @param relay The destination relay URL. * @param eventsSent Number of events sent to this relay. * @param eventsAccepted Number of OK=true responses received from this relay. */ + @Stable data class DestinationRelayInfo( val relay: NormalizedRelayUrl, - val eventsSent: Int, - val eventsAccepted: Int, - ) + val eventsSent: MutableStateFlow, + val eventsAccepted: MutableStateFlow, + ) { + constructor(relay: NormalizedRelayUrl, eventsSent: Int, eventsAccepted: Int) : + this(relay, MutableStateFlow(eventsSent), MutableStateFlow(eventsAccepted)) + } } private val _syncState = MutableStateFlow(SyncState.Idle) @@ -160,42 +207,57 @@ class EventSync( private val _liveActivity = MutableStateFlow(LiveSyncActivity()) val liveActivity: StateFlow = _liveActivity - // ------------------------------------------------------------------------- - // Live activity tracking (written from worker threads) - // ------------------------------------------------------------------------- - - private val trackingCompletions = ArrayDeque() - private val trackingLock = Any() - - /** Destination relay sets, set at the start of each sync run. */ - @Volatile private var liveOutboxTargets: Set = emptySet() - - @Volatile private var liveInboxTargets: Set = emptySet() - - @Volatile private var liveDmTargets: Set = emptySet() - - /** Per-destination-relay counters, updated atomically during a sync run. */ - private val liveSentCountPerDestRelay = ConcurrentHashMap() - private val liveAcceptedCountPerDestRelay = ConcurrentHashMap() - - private fun emitLiveSnapshot() { - val completions = synchronized(trackingLock) { trackingCompletions.toList() } - - fun buildDestInfo(relays: Set) = - relays.map { relay -> - LiveSyncActivity.DestinationRelayInfo( - relay = relay, - eventsSent = liveSentCountPerDestRelay[relay]?.get() ?: 0, - eventsAccepted = liveAcceptedCountPerDestRelay[relay]?.get() ?: 0, - ) - } - + private fun emitLiveSnapshot( + runningRelays: Set = emptySet(), + completedRelays: Set = emptySet(), + liveOutboxTargets: Set = emptySet(), + liveInboxTargets: Set = emptySet(), + liveDmTargets: Set = emptySet(), + ) { _liveActivity.value = LiveSyncActivity( - recentCompletions = completions, - outboxTargets = buildDestInfo(liveOutboxTargets), - inboxTargets = buildDestInfo(liveInboxTargets), - dmTargets = buildDestInfo(liveDmTargets), + runningRelays = + runningRelays.associateWith { + SourceRelayInfo( + relay = it, + status = MutableStateFlow(LiveSyncActivity.ConnectionStatus.Connecting), + eventsFound = MutableStateFlow(0), + eventsAccepted = MutableStateFlow(0), + ) + }, + completedRelays = + completedRelays.associateWith { + SourceRelayInfo( + relay = it, + status = MutableStateFlow(LiveSyncActivity.ConnectionStatus.Connecting), + eventsFound = MutableStateFlow(0), + eventsAccepted = MutableStateFlow(0), + ) + }, + outboxTargets = + liveOutboxTargets.associateWith { relay -> + LiveSyncActivity.DestinationRelayInfo( + relay = relay, + eventsSent = MutableStateFlow(0), + eventsAccepted = MutableStateFlow(0), + ) + }, + inboxTargets = + liveInboxTargets.associateWith { relay -> + LiveSyncActivity.DestinationRelayInfo( + relay = relay, + eventsSent = MutableStateFlow(0), + eventsAccepted = MutableStateFlow(0), + ) + }, + dmTargets = + liveDmTargets.associateWith { relay -> + LiveSyncActivity.DestinationRelayInfo( + relay = relay, + eventsSent = MutableStateFlow(0), + eventsAccepted = MutableStateFlow(0), + ) + }, ) } @@ -207,82 +269,42 @@ class EventSync( fun start() { if (_syncState.value is SyncState.Running) return - synchronized(trackingLock) { trackingCompletions.clear() } - liveSentCountPerDestRelay.clear() - liveAcceptedCountPerDestRelay.clear() _liveActivity.value = LiveSyncActivity() syncJob = scope.launch(Dispatchers.IO) { - runSync(startRelayIndex = 0, initialEventsSent = 0) - } - } - - fun resume() { - val paused = _syncState.value as? SyncState.Paused ?: return - if (_syncState.value is SyncState.Running) return - syncJob = - scope.launch(Dispatchers.IO) { - runSync( - startRelayIndex = paused.nextRelayIndex, - initialEventsSent = paused.eventsSent, - ) + runSync() } } fun cancel() { syncJob?.cancel() - val current = _syncState.value - _syncState.value = - if (current is SyncState.Running) { - SyncState.Paused( - nextRelayIndex = maxOf(0, current.relaysCompleted - MAX_CONCURRENT_RELAYS), - totalRelays = current.totalRelays, - eventsSent = current.eventsSent, - ) - } else { - SyncState.Idle - } } // ------------------------------------------------------------------------- // Sync logic // ------------------------------------------------------------------------- - - private suspend fun runSync( - startRelayIndex: Int, - initialEventsSent: Int, - ) { + suspend fun runSync() { val startTime = System.currentTimeMillis() - val myPubKey = account.signer.pubKey - val allRelays = listOf("wss://relay.damus.io".normalizeRelayUrl()) + _liveActivity.value = LiveSyncActivity() - /* - account.cache.relayHints.relayDB - .keys() - .toList() - */ + val myPubKey = accountPubKey - if (allRelays.isEmpty()) { + val relaysToProcess = relayDb() + + if (relaysToProcess.isEmpty()) { _syncState.value = SyncState.Error("No known relays found. Browse some content first to discover relays.") return } - val relaysToProcess = if (startRelayIndex > 0) allRelays.drop(startRelayIndex) else allRelays - val totalRelays = allRelays.size + val totalRelays = relaysToProcess.size - val outboxTargets = account.outboxRelays.flow.value - val inboxTargets = account.nip65RelayList.inboxFlow.value - val dmTargets = account.dmRelays.flow.value + val outboxTargets = outboxTargets() + val inboxTargets = inboxTargets() + val dmTargets = dmTargets() - // Publish destination relays so the UI can show them before events arrive. - liveOutboxTargets = outboxTargets - liveInboxTargets = inboxTargets - liveDmTargets = dmTargets - emitLiveSnapshot() - - val baseFilters = + val defaultFilters = buildList { if (outboxTargets.isNotEmpty()) add(Filter(authors = listOf(myPubKey))) if (inboxTargets.isNotEmpty() || dmTargets.isNotEmpty()) { @@ -290,43 +312,125 @@ class EventSync( } } - if (baseFilters.isEmpty()) { - _syncState.value = - SyncState.Error("No outbox, inbox, or DM relays configured.") + if (defaultFilters.isEmpty()) { + _syncState.value = SyncState.Error("No outbox, inbox, or DM relays configured.") return } - // 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()) + val usersRelays = outboxTargets + inboxTargets + dmTargets + + val perRelayFilters = + relaysToProcess.associateWith { + if (it !in usersRelays) { + defaultFilters + } else { + buildList { + if (it !in outboxTargets) add(Filter(authors = listOf(myPubKey))) + if (it !in inboxTargets && it !in dmTargets) { + add(Filter(tags = mapOf("p" to listOf(myPubKey)))) + } + } + } + } + + emitLiveSnapshot( + emptySet(), + emptySet(), + outboxTargets, + inboxTargets, + dmTargets, + ) + + // Thread-safe dedup sets — prevent the same event from being forwarded twice when + // multiple source relays return the same event concurrently. + 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 acceptedCountPerRelay = ConcurrentHashMap() + val acceptedCountPerSourceRelay = ConcurrentHashMap() val totalAccepted = AtomicLong(0) val okListener = object : IRelayClientListener { + override fun onCannotConnect( + relay: IRelayClient, + errorMessage: String, + ) { + super.onCannotConnect(relay, errorMessage) + val currentStatus = liveActivity.value.runningRelays[relay.url]?.status + if (currentStatus?.value !is LiveSyncActivity.ConnectionStatus.Error) { + currentStatus?.tryEmit(LiveSyncActivity.ConnectionStatus.Error(errorMessage)) + } + } + + override fun onSent( + relay: IRelayClient, + cmdStr: String, + cmd: Command, + success: Boolean, + ) { + super.onSent(relay, cmdStr, cmd, success) + + if (cmd is EventCmd) { + if (outboxDedup.contains(cmd.event.id)) { + liveActivity.value.outboxTargets[relay.url] + ?.eventsSent + ?.update { it + 1 } + } + if (inboxDedup.contains(cmd.event.id)) { + liveActivity.value.inboxTargets[relay.url] + ?.eventsSent + ?.update { it + 1 } + } + if (dmDedup.contains(cmd.event.id)) { + liveActivity.value.dmTargets[relay.url] + ?.eventsSent + ?.update { it + 1 } + } + } else if (cmd is ReqCmd) { + val currentStatus = liveActivity.value.runningRelays[relay.url]?.status + if (currentStatus?.value != LiveSyncActivity.ConnectionStatus.Querying) { + currentStatus?.tryEmit(LiveSyncActivity.ConnectionStatus.Querying) + } + } + } + override fun onIncomingMessage( relay: IRelayClient, msgStr: String, msg: Message, ) { - if (msg is OkMessage && msg.success) { + 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 - acceptedCountPerRelay.getOrPut(sourceRelay) { AtomicInteger(0) }.incrementAndGet() + acceptedCountPerSourceRelay.getOrPut(sourceRelay) { AtomicInteger(0) }.incrementAndGet() totalAccepted.incrementAndGet() - liveAcceptedCountPerDestRelay.getOrPut(relay.url) { AtomicInteger(0) }.incrementAndGet() + + if (outboxDedup.contains(msg.eventId)) { + liveActivity.value.outboxTargets[relay.url] + ?.eventsAccepted + ?.update { it + 1 } + } + if (dmDedup.contains(msg.eventId)) { + liveActivity.value.dmTargets[relay.url] + ?.eventsAccepted + ?.update { it + 1 } + } + if (inboxDedup.contains(msg.eventId)) { + liveActivity.value.inboxTargets[relay.url] + ?.eventsAccepted + ?.update { it + 1 } + } } } } - val relaysCompleted = AtomicInteger(startRelayIndex) + val relaysCompleted = AtomicInteger(0) _syncState.value = SyncState.Running( @@ -335,75 +439,108 @@ class EventSync( eventsSent = totalSent.get().toInt(), ) - account.client.subscribe(okListener) - try { - downloadFromPool( - relays = relaysToProcess, - baseFilters = baseFilters, - 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() - outboxTargets.forEach { dest -> - liveSentCountPerDestRelay.getOrPut(dest) { AtomicInteger(0) }.incrementAndGet() - } - } - } - val pTagsMe = event.tags.isTaggedUser(myPubKey) - 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() - dmTargets.forEach { dest -> - liveSentCountPerDestRelay.getOrPut(dest) { AtomicInteger(0) }.incrementAndGet() - } - } - } else { - if (inboxTargets.isNotEmpty() && inboxSent.add(event.id)) { - sourceRelayOfEvent[event.id] = sourceRelay - account.client.send(event, inboxTargets) - totalSent.incrementAndGet() - inboxTargets.forEach { dest -> - liveSentCountPerDestRelay.getOrPut(dest) { AtomicInteger(0) }.incrementAndGet() - } - } - } - } - }, - onRelayComplete = { 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() - } - val completed = relaysCompleted.incrementAndGet() - emitLiveSnapshot() - _syncState.value = - SyncState.Running( - relaysCompleted = completed, - totalRelays = totalRelays, - eventsSent = totalSent.get().toInt(), - ) - }, - ) + clientBuilder().use { client -> + client.subscribe(okListener) + try { + client.downloadFromPool( + relays = relaysToProcess, + filters = perRelayFilters, + onEvent = { event, sourceRelay -> + val isMyEvent = event.pubKey == myPubKey + val mentionsMe = event.tags.isTaggedUser(myPubKey) + val isDmKind = event.kind == 4 || event.kind == 1059 - _syncState.value = - SyncState.Done( - totalEventsSent = totalSent.get().toInt(), - totalEventsAccepted = totalAccepted.get().toInt(), - durationMs = System.currentTimeMillis() - startTime, + val live = liveActivity.value + + // 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 } + } + } + 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 } + } + } + 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 } + } + } + }, + onRelayStart = { relay -> + _liveActivity.update { + it.copy( + runningRelays = + it.runningRelays + + Pair(relay, SourceRelayInfo(relay, LiveSyncActivity.ConnectionStatus.Connecting, 0, 0)), + ) + } + }, + onRelayComplete = { relay -> + _liveActivity.update { + val newCompleted = it.runningRelays[relay] + it.copy( + runningRelays = it.runningRelays.minus(relay), + completedRelays = + if (newCompleted != null) { + it.completedRelays.plus(relay to newCompleted) + } else { + it.completedRelays + }, + ) + } + + val status = _liveActivity.value.completedRelays[relay]?.status + + if (status?.value !is LiveSyncActivity.ConnectionStatus.Error) { + status?.tryEmit(LiveSyncActivity.ConnectionStatus.Completed) + } + + _syncState.value = + SyncState.Running( + relaysCompleted = relaysCompleted.incrementAndGet(), + totalRelays = totalRelays, + eventsSent = totalSent.get().toInt(), + ) + }, ) - } catch (e: CancellationException) { - throw e - } catch (e: Exception) { - _syncState.value = SyncState.Error(e.message ?: "Unknown error") - } finally { - account.client.unsubscribe(okListener) + + _syncState.value = + SyncState.Done( + totalEventsSent = totalSent.get().toInt(), + totalEventsAccepted = totalAccepted.get().toInt(), + durationMs = System.currentTimeMillis() - startTime, + ) + } catch (e: CancellationException) { + _syncState.value = + SyncState.Done( + totalEventsSent = totalSent.get().toInt(), + totalEventsAccepted = totalAccepted.get().toInt(), + durationMs = System.currentTimeMillis() - startTime, + ) + throw e + } catch (e: Exception) { + _syncState.value = SyncState.Error(e.message ?: "Unknown error") + } finally { + client.unsubscribe(okListener) + } } } @@ -414,11 +551,12 @@ class EventSync( * * [onEvent] receives the event and the URL of the relay it came from. */ - private suspend fun downloadFromPool( + private suspend fun INostrClient.downloadFromPool( relays: List, - baseFilters: List, + filters: Map>, onEvent: (Event, NormalizedRelayUrl) -> Unit, - onRelayComplete: (NormalizedRelayUrl, Int) -> Unit, + onRelayStart: (NormalizedRelayUrl) -> Unit, + onRelayComplete: (NormalizedRelayUrl) -> Unit, ) { val semaphore = Semaphore(MAX_CONCURRENT_RELAYS) supervisorScope { @@ -427,8 +565,11 @@ class EventSync( semaphore.acquire() launch { try { - val eventsFound = downloadFromRelay(relay, baseFilters) { event -> onEvent(event, relay) } - onRelayComplete(relay, eventsFound) + onRelayStart(relay) + filters[relay]?.let { filtersForRelay -> + downloadFromRelay(relay, filtersForRelay) { event -> onEvent(event, relay) } + } ?: 0 + onRelayComplete(relay) } finally { semaphore.release() } @@ -443,9 +584,9 @@ class EventSync( * * @return total number of events received across all pages. */ - private suspend fun downloadFromRelay( + private suspend fun INostrClient.downloadFromRelay( relay: NormalizedRelayUrl, - baseFilters: List, + filters: List, onEvent: (Event) -> Unit, - ): Int = account.client.reqBypassingRelayLimits(relay, baseFilters, RELAY_TIMEOUT_MS, onEvent) + ): Int = reqBypassingRelayLimits(relay, filters, RELAY_TIMEOUT_MS, 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 b6aa38343..100e8d575 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 @@ -31,10 +31,9 @@ import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.height import androidx.compose.foundation.layout.padding import androidx.compose.foundation.layout.size -import androidx.compose.foundation.layout.width -import androidx.compose.foundation.rememberScrollState +import androidx.compose.foundation.lazy.LazyColumn +import androidx.compose.foundation.lazy.itemsIndexed import androidx.compose.foundation.shape.CircleShape -import androidx.compose.foundation.verticalScroll import androidx.compose.material3.AlertDialog import androidx.compose.material3.Button import androidx.compose.material3.ButtonDefaults @@ -56,6 +55,7 @@ import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.draw.clip import androidx.compose.ui.text.font.FontWeight +import androidx.compose.ui.text.style.TextAlign import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.tooling.preview.Preview import androidx.compose.ui.unit.dp @@ -66,6 +66,7 @@ import com.vitorpamplona.amethyst.ui.navigation.topbars.TopBarWithBackButton import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel import com.vitorpamplona.amethyst.ui.stringRes import com.vitorpamplona.amethyst.ui.theme.ThemeComparisonColumn +import com.vitorpamplona.amethyst.ui.theme.ThemeComparisonRow import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl @Composable @@ -74,11 +75,9 @@ fun EventSyncScreen( nav: INav, ) { val syncViewModel = accountViewModel.eventSync - + val isMobileOrMetered by accountViewModel.settings.isMobileOrMeteredConnection.collectAsStateWithLifecycle() val syncState by syncViewModel.syncState.collectAsStateWithLifecycle() val liveActivity by syncViewModel.liveActivity.collectAsStateWithLifecycle() - val isMobileOrMetered by accountViewModel.settings.isMobileOrMeteredConnection.collectAsStateWithLifecycle() - var showMobileDataDialog by remember { mutableStateOf(false) } Scaffold( topBar = { @@ -88,172 +87,192 @@ fun EventSyncScreen( ) }, ) { padding -> - Column( - modifier = - Modifier - .fillMaxSize() - .padding(padding) - .padding(horizontal = 16.dp) - .verticalScroll(rememberScrollState()), - verticalArrangement = Arrangement.spacedBy(16.dp), - ) { + Column(Modifier.fillMaxSize().padding(padding)) { + EventScreenBody( + syncState = syncState, + liveActivity = liveActivity, + isMobileOrMetered = isMobileOrMetered, + onStart = syncViewModel::start, + onCancel = syncViewModel::cancel, + ) + } + } +} + +@Composable +fun EventScreenBody( + syncState: EventSync.SyncState, + liveActivity: EventSync.LiveSyncActivity, + isMobileOrMetered: Boolean = false, + onStart: () -> Unit = {}, + onCancel: () -> Unit = {}, +) { + LazyColumn( + modifier = + Modifier + .fillMaxSize() + .padding(horizontal = 16.dp), + verticalArrangement = Arrangement.spacedBy(2.dp), + ) { + item { // ---- Progress / Status area ---- - when (val state = syncState) { - is EventSync.SyncState.Idle -> { - // ---- Explanation card ---- - Card( - modifier = Modifier.fillMaxWidth(), - colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surfaceVariant), - ) { - Column(modifier = Modifier.padding(16.dp)) { - Text( - text = stringRes(R.string.event_sync_what_happens_title), - style = MaterialTheme.typography.titleMedium, - fontWeight = FontWeight.Bold, - ) - Spacer(Modifier.height(8.dp)) - Text( - text = stringRes(R.string.event_sync_what_happens_body), - style = MaterialTheme.typography.bodyMedium, - ) - Spacer(Modifier.height(12.dp)) - StepRow(number = "1", text = stringRes(R.string.event_sync_step1)) - Spacer(Modifier.height(4.dp)) - StepRow(number = "2", text = stringRes(R.string.event_sync_step2)) - Spacer(Modifier.height(4.dp)) - StepRow(number = "3", text = stringRes(R.string.event_sync_step3)) - } - } - - // ---- WiFi warning ---- - if (isMobileOrMetered) { - Card( - modifier = Modifier.fillMaxWidth(), - colors = - CardDefaults.cardColors( - containerColor = MaterialTheme.colorScheme.errorContainer, - ), - ) { - Text( - text = stringRes(R.string.event_sync_wifi_warning), - style = MaterialTheme.typography.bodyMedium, - color = MaterialTheme.colorScheme.onErrorContainer, - modifier = Modifier.padding(16.dp), - ) - } - } - } - - is EventSync.SyncState.Running -> { - SyncProgressCard(state = state) - } - - is EventSync.SyncState.Paused -> { - PausedCard(state = state) - } - - is EventSync.SyncState.Done -> { - DoneCard(state = state) - } - - is EventSync.SyncState.Error -> { - ErrorCard(message = state.message) - } + when (syncState) { + is EventSync.SyncState.Idle -> ExplanationCard(isMobileOrMetered, onStart) + is EventSync.SyncState.Running -> SyncProgressCard(state = syncState, onCancel) + is EventSync.SyncState.Done -> DoneCard(state = syncState, isMobileOrMetered, onStart) + is EventSync.SyncState.Error -> ErrorCard(syncState.message, isMobileOrMetered, onStart) } + } - // ---- Live relay activity (shown during and after sync) ---- - if (liveActivity.outboxTargets.isNotEmpty() || - liveActivity.inboxTargets.isNotEmpty() || - liveActivity.dmTargets.isNotEmpty() - ) { + // ---- Live relay activity (shown during and after sync) ---- + if (liveActivity.outboxTargets.isNotEmpty() || + liveActivity.inboxTargets.isNotEmpty() || + liveActivity.dmTargets.isNotEmpty() + ) { + item { + Spacer(Modifier.height(16.dp)) DestinationRelaysCard(activity = liveActivity) } + } - if (liveActivity.recentCompletions.isNotEmpty()) { - ActivityLogCard(completions = liveActivity.recentCompletions) - } - - // ---- Action buttons ---- - when (val state = syncState) { - is EventSync.SyncState.Idle, - is EventSync.SyncState.Done, - is EventSync.SyncState.Error, - -> { - Button( - onClick = { - if (isMobileOrMetered) { - showMobileDataDialog = true - } else { - syncViewModel.start() - } - }, - modifier = Modifier.fillMaxWidth(), - ) { - Text(stringRes(R.string.event_sync_start)) - } - } - - is EventSync.SyncState.Paused -> { - Button( - onClick = { - if (isMobileOrMetered) { - showMobileDataDialog = true - } else { - syncViewModel.resume() - } - }, - modifier = Modifier.fillMaxWidth(), - ) { - Text(stringRes(R.string.event_sync_resume)) - } - OutlinedButton( - onClick = { syncViewModel.start() }, - modifier = Modifier.fillMaxWidth(), - ) { - Text(stringRes(R.string.event_sync_start_over)) - } - } - - is EventSync.SyncState.Running -> { - OutlinedButton( - onClick = { syncViewModel.cancel() }, - modifier = Modifier.fillMaxWidth(), - colors = - ButtonDefaults.outlinedButtonColors( - contentColor = MaterialTheme.colorScheme.error, - ), - ) { - Text(stringRes(R.string.event_sync_pause)) - } - } - } - - // ---- Mobile-data confirmation dialog ---- - if (showMobileDataDialog) { - val isPaused = syncState is EventSync.SyncState.Paused - AlertDialog( - onDismissRequest = { showMobileDataDialog = false }, - title = { Text(stringRes(R.string.event_sync_mobile_data_dialog_title)) }, - text = { Text(stringRes(R.string.event_sync_wifi_warning)) }, - confirmButton = { - Button( - onClick = { - showMobileDataDialog = false - if (isPaused) syncViewModel.resume() else syncViewModel.start() - }, - ) { - Text(stringRes(R.string.event_sync_start_anyway)) - } - }, - dismissButton = { - TextButton(onClick = { showMobileDataDialog = false }) { - Text(stringRes(R.string.event_sync_cancel)) - } - }, + val runningSize = liveActivity.runningRelays.size + if (runningSize > 0) { + item { + Spacer(Modifier.height(16.dp)) + Text( + text = stringRes(R.string.event_sync_activity_log, runningSize), + style = MaterialTheme.typography.titleSmall, + fontWeight = FontWeight.SemiBold, ) + Spacer(Modifier.height(5.dp)) } - Spacer(Modifier.height(16.dp)) + itemsIndexed(liveActivity.runningRelays.values.toList(), key = { _, item -> item.relay.url }) { index, info -> + if (index > 0) { + HorizontalDivider( + color = MaterialTheme.colorScheme.outlineVariant.copy(alpha = 0.5f), + ) + } + ActivityLogRow(info = info) + } + } + + val completedSize = liveActivity.completedRelays.size + if (completedSize > 0) { + item { + Spacer(Modifier.height(16.dp)) + Text( + text = stringRes(R.string.event_sync_activity_log_finished, completedSize), + style = MaterialTheme.typography.titleSmall, + fontWeight = FontWeight.SemiBold, + ) + Spacer(Modifier.height(5.dp)) + } + + itemsIndexed(liveActivity.sortedCompletedRelays, key = { _, item -> item.relay.url }) { index, info -> + if (index > 0) { + HorizontalDivider( + color = MaterialTheme.colorScheme.outlineVariant.copy(alpha = 0.5f), + ) + } + ActivityLogRow(info = info) + } + } + } +} + +@Composable +private fun StartSyncButton( + isMobileOrMetered: Boolean, + onClick: () -> Unit, +) { + var showMobileDataDialog by remember { mutableStateOf(false) } + + Button( + onClick = { + if (isMobileOrMetered) { + showMobileDataDialog = true + } else { + onClick() + } + }, + modifier = Modifier.fillMaxWidth(), + ) { + Text(stringRes(R.string.event_sync_start)) + } + + // ---- Mobile-data confirmation dialog ---- + if (showMobileDataDialog) { + AlertDialog( + onDismissRequest = { showMobileDataDialog = false }, + title = { Text(stringRes(R.string.event_sync_mobile_data_dialog_title)) }, + text = { Text(stringRes(R.string.event_sync_wifi_warning)) }, + confirmButton = { + Button( + onClick = { + showMobileDataDialog = false + onClick() + }, + ) { + Text(stringRes(R.string.event_sync_start_anyway)) + } + }, + dismissButton = { + TextButton(onClick = { showMobileDataDialog = false }) { + Text(stringRes(R.string.event_sync_cancel)) + } + }, + ) + } +} + +@Composable +private fun ExplanationCard( + isMobileOrMetered: Boolean, + onStart: () -> Unit, +) { + // ---- Explanation card ---- + Card( + modifier = Modifier.fillMaxWidth(), + colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surfaceVariant), + ) { + Column(modifier = Modifier.padding(16.dp)) { + Text( + text = stringRes(R.string.event_sync_what_happens_title), + style = MaterialTheme.typography.titleMedium, + fontWeight = FontWeight.Bold, + ) + Spacer(Modifier.height(8.dp)) + Text( + text = stringRes(R.string.event_sync_what_happens_body), + style = MaterialTheme.typography.bodyMedium, + ) + Spacer(Modifier.height(12.dp)) + StepRow(number = "1", text = stringRes(R.string.event_sync_step1)) + Spacer(Modifier.height(4.dp)) + StepRow(number = "2", text = stringRes(R.string.event_sync_step2)) + Spacer(Modifier.height(4.dp)) + StepRow(number = "3", text = stringRes(R.string.event_sync_step3)) + Spacer(Modifier.height(10.dp)) + // ---- WiFi warning ---- + if (isMobileOrMetered) { + Card( + modifier = Modifier.fillMaxWidth(), + colors = + CardDefaults.cardColors( + containerColor = MaterialTheme.colorScheme.errorContainer, + ), + ) { + Text( + text = stringRes(R.string.event_sync_wifi_warning), + style = MaterialTheme.typography.bodyMedium, + color = MaterialTheme.colorScheme.onErrorContainer, + modifier = Modifier.padding(16.dp), + ) + } + Spacer(Modifier.height(10.dp)) + } + StartSyncButton(isMobileOrMetered = isMobileOrMetered, onStart) } } } @@ -263,7 +282,10 @@ fun EventSyncScreen( // ------------------------------------------------------------------------- @Composable -private fun SyncProgressCard(state: EventSync.SyncState.Running) { +private fun SyncProgressCard( + state: EventSync.SyncState.Running, + onCancel: () -> Unit, +) { Card( modifier = Modifier.fillMaxWidth(), colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surface), @@ -292,44 +314,27 @@ private fun SyncProgressCard(state: EventSync.SyncState.Running) { style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onSurfaceVariant, ) - } - } -} - -@Composable -private fun PausedCard(state: EventSync.SyncState.Paused) { - Card( - modifier = Modifier.fillMaxWidth(), - colors = - CardDefaults.cardColors( - containerColor = MaterialTheme.colorScheme.secondaryContainer, - ), - ) { - Column(modifier = Modifier.padding(16.dp)) { - Text( - text = stringRes(R.string.event_sync_paused_title), - style = MaterialTheme.typography.titleSmall, - fontWeight = FontWeight.Bold, - color = MaterialTheme.colorScheme.onSecondaryContainer, - ) - Spacer(Modifier.height(4.dp)) - Text( - text = - stringRes( - R.string.event_sync_paused_body, - state.nextRelayIndex, - state.totalRelays, - state.eventsSent, + Spacer(Modifier.height(10.dp)) + OutlinedButton( + onClick = onCancel, + modifier = Modifier.fillMaxWidth(), + colors = + ButtonDefaults.outlinedButtonColors( + contentColor = MaterialTheme.colorScheme.error, ), - style = MaterialTheme.typography.bodyMedium, - color = MaterialTheme.colorScheme.onSecondaryContainer, - ) + ) { + Text(stringRes(R.string.event_sync_cancel)) + } } } } @Composable -private fun DoneCard(state: EventSync.SyncState.Done) { +private fun DoneCard( + state: EventSync.SyncState.Done, + isMobileOrMetered: Boolean = false, + onStart: () -> Unit = { }, +) { Card( modifier = Modifier.fillMaxWidth(), colors = @@ -363,12 +368,18 @@ private fun DoneCard(state: EventSync.SyncState.Done) { style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onPrimaryContainer.copy(alpha = 0.7f), ) + Spacer(Modifier.height(10.dp)) + StartSyncButton(isMobileOrMetered, onStart) } } } @Composable -private fun ErrorCard(message: String) { +private fun ErrorCard( + message: String, + isMobileOrMetered: Boolean = false, + onStart: () -> Unit = {}, +) { Card( modifier = Modifier.fillMaxWidth(), colors = @@ -389,6 +400,8 @@ private fun ErrorCard(message: String) { style = MaterialTheme.typography.bodyMedium, color = MaterialTheme.colorScheme.onErrorContainer, ) + Spacer(Modifier.height(10.dp)) + StartSyncButton(isMobileOrMetered, onStart) } } } @@ -418,7 +431,7 @@ private fun DestinationRelaysCard(activity: EventSync.LiveSyncActivity) { Spacer(Modifier.height(10.dp)) DestinationSection( label = stringRes(R.string.event_sync_outbox_relays), - relays = activity.outboxTargets, + relays = activity.outboxTargets.values, color = MaterialTheme.colorScheme.primary, ) } @@ -429,7 +442,7 @@ private fun DestinationRelaysCard(activity: EventSync.LiveSyncActivity) { Spacer(Modifier.height(10.dp)) DestinationSection( label = stringRes(R.string.event_sync_inbox_relays), - relays = activity.inboxTargets, + relays = activity.inboxTargets.values, color = MaterialTheme.colorScheme.secondary, ) } @@ -440,7 +453,7 @@ private fun DestinationRelaysCard(activity: EventSync.LiveSyncActivity) { Spacer(Modifier.height(10.dp)) DestinationSection( label = stringRes(R.string.event_sync_dm_relays), - relays = activity.dmTargets, + relays = activity.dmTargets.values, color = MaterialTheme.colorScheme.tertiary, ) } @@ -451,7 +464,7 @@ private fun DestinationRelaysCard(activity: EventSync.LiveSyncActivity) { @Composable private fun DestinationSection( label: String, - relays: List, + relays: Collection, color: androidx.compose.ui.graphics.Color, ) { Text( @@ -478,6 +491,8 @@ private fun DestinationRelayRow( info: EventSync.LiveSyncActivity.DestinationRelayInfo, color: androidx.compose.ui.graphics.Color, ) { + val eventsSent by info.eventsSent.collectAsStateWithLifecycle() + Row( modifier = Modifier @@ -498,70 +513,39 @@ private fun DestinationRelayRow( style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onSurface, maxLines = 1, - overflow = TextOverflow.Ellipsis, + overflow = TextOverflow.StartEllipsis, modifier = Modifier.weight(1f), ) - if (info.eventsSent > 0) { + if (eventsSent > 0) { + val eventsAccepted by info.eventsAccepted.collectAsStateWithLifecycle() Text( - text = stringRes(R.string.event_sync_log_recv, formatCount(info.eventsSent)), + text = stringRes(R.string.event_sync_log_sent, formatCount(eventsSent)), style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.onSurfaceVariant, + modifier = Modifier.weight(0.3f), + textAlign = TextAlign.End, + maxLines = 1, + overflow = TextOverflow.StartEllipsis, ) - Spacer(Modifier.width(8.dp)) Text( - text = stringRes(R.string.event_sync_log_new, formatCount(info.eventsAccepted)), + text = stringRes(R.string.event_sync_log_new, formatCount(eventsAccepted)), style = MaterialTheme.typography.bodySmall, - fontWeight = if (info.eventsAccepted > 0) FontWeight.SemiBold else FontWeight.Normal, - color = if (info.eventsAccepted > 0) color else MaterialTheme.colorScheme.onSurfaceVariant, + fontWeight = if (eventsAccepted > 0) FontWeight.SemiBold else FontWeight.Normal, + color = if (eventsAccepted > 0) color else MaterialTheme.colorScheme.onSurfaceVariant, + modifier = Modifier.weight(0.3f), + textAlign = TextAlign.End, + maxLines = 1, + overflow = TextOverflow.StartEllipsis, ) } } } -/** - * Scrollable log of recently completed relays, newest at the top. - * Uses a fixed-height inner scroll area so it doesn't compete with the outer scroll. - */ @Composable -private fun ActivityLogCard(completions: List) { - 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_activity_log, completions.size), - style = MaterialTheme.typography.titleSmall, - fontWeight = FontWeight.SemiBold, - ) - Spacer(Modifier.height(10.dp)) - - Box( - modifier = - Modifier - .fillMaxWidth() - .height(260.dp) - .verticalScroll(rememberScrollState()), - ) { - Column(verticalArrangement = Arrangement.spacedBy(2.dp)) { - completions.forEachIndexed { index, info -> - if (index > 0) { - HorizontalDivider( - color = MaterialTheme.colorScheme.outlineVariant.copy(alpha = 0.5f), - ) - } - ActivityLogRow(info = info) - } - } - } - } - } -} - -@Composable -private fun ActivityLogRow(info: EventSync.LiveSyncActivity.CompletedRelayInfo) { - val hasEvents = info.eventsFound > 0 +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) { MaterialTheme.colorScheme.primary @@ -595,32 +579,51 @@ private fun ActivityLogRow(info: EventSync.LiveSyncActivity.CompletedRelayInfo) style = MaterialTheme.typography.bodySmall, color = textColor, maxLines = 1, - overflow = TextOverflow.Ellipsis, - modifier = Modifier.weight(1f), + overflow = TextOverflow.StartEllipsis, + modifier = Modifier.weight(0.6f), ) if (hasEvents) { + val eventsAccepted by info.eventsAccepted.collectAsStateWithLifecycle() Text( - text = stringRes(R.string.event_sync_log_recv, formatCount(info.eventsFound)), + text = stringRes(R.string.event_sync_log_recv, formatCount(eventsFound)), style = MaterialTheme.typography.bodySmall, color = textColor, + modifier = Modifier.weight(0.2f), + maxLines = 1, + textAlign = TextAlign.End, + overflow = TextOverflow.StartEllipsis, ) - Spacer(Modifier.width(8.dp)) Text( - text = stringRes(R.string.event_sync_log_new, formatCount(info.eventsAccepted)), + text = stringRes(R.string.event_sync_log_new, formatCount(eventsAccepted)), style = MaterialTheme.typography.bodySmall, - fontWeight = if (info.eventsAccepted > 0) FontWeight.SemiBold else FontWeight.Normal, + fontWeight = if (eventsAccepted > 0) FontWeight.SemiBold else FontWeight.Normal, color = - if (info.eventsAccepted > 0) { + if (eventsAccepted > 0) { MaterialTheme.colorScheme.primary } else { MaterialTheme.colorScheme.onSurfaceVariant }, + maxLines = 1, + textAlign = TextAlign.End, + overflow = TextOverflow.StartEllipsis, + modifier = Modifier.weight(0.2f), ) } else { + val status by info.status.collectAsStateWithLifecycle() Text( - text = stringRes(R.string.event_sync_no_events), + text = + when (status) { + EventSync.LiveSyncActivity.ConnectionStatus.Connecting -> stringRes(R.string.event_sync_status_connecting) + EventSync.LiveSyncActivity.ConnectionStatus.Querying -> stringRes(R.string.event_sync_status_downloading) + is EventSync.LiveSyncActivity.ConnectionStatus.Error -> (status as EventSync.LiveSyncActivity.ConnectionStatus.Error).msg.ifBlank { stringRes(R.string.event_sync_status_error) } + EventSync.LiveSyncActivity.ConnectionStatus.Completed -> stringRes(R.string.event_sync_status_completed) + }, style = MaterialTheme.typography.bodySmall, color = textColor, + textAlign = TextAlign.End, + maxLines = 1, + overflow = TextOverflow.StartEllipsis, + modifier = Modifier.weight(0.45f), ) } } @@ -662,8 +665,7 @@ private fun NormalizedRelayUrl.displayHost(): String = /** Formats a count with K/M suffix for large numbers. */ private fun formatCount(n: Int): String = when { - n >= 1_000_000 -> "${n / 1_000_000}M" - n >= 1_000 -> "${n / 1_000}K" + n >= 1_000_000 -> "${n / 1_000}K" else -> n.toString() } @@ -671,19 +673,25 @@ private fun formatCount(n: Int): String = // Preview data // ------------------------------------------------------------------------- +private val previewRunning = + listOf( + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://relay.damus.io"), EventSync.LiveSyncActivity.ConnectionStatus.Querying, 1247, 891), + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://nos.lol"), EventSync.LiveSyncActivity.ConnectionStatus.Querying, 892, 45), + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://nos2.lol"), EventSync.LiveSyncActivity.ConnectionStatus.Connecting, 0, 0), + ) + private val previewCompletions = listOf( - EventSync.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://relay.damus.io"), 1247, 891), - EventSync.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://nos.lol"), 892, 45), - EventSync.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://relay.nostr.band"), 3500, 3498), - EventSync.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://slow.relay.example.com"), 0, 0), - EventSync.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://nostr.bitcoiner.social"), 15, 0), - EventSync.LiveSyncActivity.CompletedRelayInfo(NormalizedRelayUrl("wss://unreachable.relay.xyz"), 0, 0), + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://relay.nostr.band"), EventSync.LiveSyncActivity.ConnectionStatus.Completed, 3500, 3498), + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://slow.relay.example.com"), EventSync.LiveSyncActivity.ConnectionStatus.Completed, 0, 0), + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://nostr.bitcoiner.social"), EventSync.LiveSyncActivity.ConnectionStatus.Completed, 15, 0), + EventSync.LiveSyncActivity.SourceRelayInfo(NormalizedRelayUrl("wss://unreachable.relay.xyz"), EventSync.LiveSyncActivity.ConnectionStatus.Error("connection failed"), 0, 0), ) private val previewActivity = EventSync.LiveSyncActivity( - recentCompletions = previewCompletions, + runningRelays = previewRunning, + completedRelays = previewCompletions, outboxTargets = listOf( EventSync.LiveSyncActivity.DestinationRelayInfo(NormalizedRelayUrl("wss://outbox.nostr.com"), 1247, 891), @@ -704,6 +712,22 @@ private val previewActivity = // Previews // ------------------------------------------------------------------------- +@Composable +@Preview +fun IdleCardWifiPreview() { + ThemeComparisonColumn { + ExplanationCard(false, {}) + } +} + +@Composable +@Preview +fun IdleCardMobilePreview() { + ThemeComparisonColumn { + ExplanationCard(true, {}) + } +} + @Composable @Preview fun SyncProgressCardPreview() { @@ -715,21 +739,7 @@ fun SyncProgressCardPreview() { totalRelays = 1024, eventsSent = 4821, ), - ) - } -} - -@Composable -@Preview -fun PausedCardPreview() { - ThemeComparisonColumn { - PausedCard( - state = - EventSync.SyncState.Paused( - nextRelayIndex = 260, - totalRelays = 1024, - eventsSent = 3200, - ), + onCancel = {}, ) } } @@ -766,9 +776,23 @@ fun DestinationRelaysCardPreview() { } @Composable -@Preview -fun ActivityLogCardPreview() { - ThemeComparisonColumn { - ActivityLogCard(completions = previewCompletions) +@Preview(device = "spec:width=1800px,height=2340px,dpi=440") +fun EventScreenBodyPreview() { + ThemeComparisonRow { + EventScreenBody( + EventSync.SyncState.Idle, + EventSync.LiveSyncActivity(emptyList(), emptyList(), emptyList(), emptyList(), emptyList()), + ) + } +} + +@Composable +@Preview(device = "spec:width=1800px,height=2340px,dpi=440") +fun EventScreenBody2Preview() { + ThemeComparisonRow { + EventScreenBody( + EventSync.SyncState.Running(1247, 1024, 4821), + previewActivity, + ) } } diff --git a/amethyst/src/main/res/values/strings.xml b/amethyst/src/main/res/values/strings.xml index 3e693a2b0..91a96bcc7 100644 --- a/amethyst/src/main/res/values/strings.xml +++ b/amethyst/src/main/res/values/strings.xml @@ -1816,7 +1816,7 @@ Re-publish your events across all known relays to keep your outbox, inbox, and DM relays up to date. Requires Wi-Fi — this may use a lot of data. Open Relay Sync… What this does - This tool scans every relay your app has seen and redistributes your events to the correct destinations. It may transfer a large amount of data, so it is best run on Wi-Fi. + This tool scans every relay your app has seen and redistributes your events to the correct destinations: Download all events you authored and send them to your outbox relays. Download all events that mention you and send them to your inbox relays. Download all direct messages addressed to you and send them to your DM relays. @@ -1841,7 +1841,9 @@ Outbox Inbox DMs - Activity (%1$d relays) + Currently Checking (%1$d relays) + Finished (%1$d relays) + sent %1$s recv %1$s new %1$s no events @@ -1851,4 +1853,9 @@ profiles relay settings Last seen %1$s ago + + Connecting + Downloading + Error + Completed diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt index 668c1ff98..e4d11a98b 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt @@ -31,7 +31,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow -interface INostrClient { +interface INostrClient : AutoCloseable { fun connectedRelaysFlow(): StateFlow> fun availableRelaysFlow(): StateFlow> @@ -136,4 +136,6 @@ object EmptyNostrClient : INostrClient { override fun activeCounts(url: NormalizedRelayUrl): Map> = emptyMap() override fun activeOutboxCache(url: NormalizedRelayUrl): Set = emptySet() + + override fun close() {} } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index 8d40797e3..2fa082164 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -79,7 +79,8 @@ class NostrClient( private val websocketBuilder: WebsocketBuilder, private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()), ) : INostrClient, - IRelayClientListener { + IRelayClientListener, + AutoCloseable { private val relayPool: RelayPool = RelayPool(websocketBuilder, this) private val activeRequests: PoolRequests = PoolRequests() @@ -326,4 +327,8 @@ class NostrClient( override fun connectedRelaysFlow() = relayPool.connectedRelays override fun availableRelaysFlow() = relayPool.availableRelays + + override fun close() { + disconnect() + } } 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 d985c9005..e2438dff5 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 @@ -28,7 +28,6 @@ import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED import kotlinx.coroutines.ensureActive import kotlinx.coroutines.withTimeoutOrNull import kotlin.coroutines.coroutineContext @@ -63,6 +62,8 @@ suspend fun INostrClient.reqBypassingRelayLimits( // Track how many matching events each filter has received so far. val matchCountPerFilter = IntArray(filters.size) + val subId = newSubId() + while (true) { coroutineContext.ensureActive() @@ -75,9 +76,7 @@ suspend fun INostrClient.reqBypassingRelayLimits( if (remainingFilters.isEmpty()) break - val eventChannel = Channel(UNLIMITED) val doneChannel = Channel(Channel.CONFLATED) - val subId = newSubId() val activeFilters = if (until == null) { @@ -86,19 +85,37 @@ suspend fun INostrClient.reqBypassingRelayLimits( remainingFilters.map { it.copy(until = until) } } + var pageCount = 0 + var pageMinTs = Long.MAX_VALUE + val listener = object : IRequestListener { override fun onEvent( event: Event, isLive: Boolean, - relay: NormalizedRelayUrl, + relayInner: NormalizedRelayUrl, forFilters: List?, ) { - eventChannel.trySend(event) + onEvent(event) + pageCount++ + if (event.createdAt < pageMinTs) pageMinTs = event.createdAt + + // Count this event against every base filter it matches. + if (matchCountPerFilter.size == 1) { + // no need to run the match. + matchCountPerFilter[0]++ + } else { + for (i in filters.indices) { + val limit = filters[i].limit + if ((limit == null || matchCountPerFilter[i] < limit) && filters[i].match(event)) { + matchCountPerFilter[i]++ + } + } + } } override fun onEose( - relay: NormalizedRelayUrl, + relayInner: NormalizedRelayUrl, forFilters: List?, ) { doneChannel.trySend(Unit) @@ -106,49 +123,30 @@ suspend fun INostrClient.reqBypassingRelayLimits( override fun onClosed( message: String, - relay: NormalizedRelayUrl, + relayInner: NormalizedRelayUrl, forFilters: List?, ) { doneChannel.trySend(Unit) } override fun onCannotConnect( - relay: NormalizedRelayUrl, + relayInner: NormalizedRelayUrl, message: String, forFilters: List?, ) { - println("AABBCC $message") doneChannel.trySend(Unit) } } openReqSubscription(subId, mapOf(relay to activeFilters), listener) - withTimeoutOrNull(timeoutMs) { doneChannel.receive() } - close(subId) - eventChannel.close() - doneChannel.close() - var pageCount = 0 - var pageMinTs = Long.MAX_VALUE - for (event in eventChannel) { - onEvent(event) - pageCount++ - if (event.createdAt < pageMinTs) pageMinTs = event.createdAt - - // Count this event against every base filter it matches. - if (matchCountPerFilter.size == 1) { - // no need to run the match. - matchCountPerFilter[0]++ - } else { - for (i in filters.indices) { - val limit = filters[i].limit - if ((limit == null || matchCountPerFilter[i] < limit) && filters[i].match(event)) { - matchCountPerFilter[i]++ - } - } - } + withTimeoutOrNull(timeoutMs) { + doneChannel.receive() } + close(subId) + doneChannel.close() + if (pageCount == 0) break totalEvents += pageCount diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientSingleDownloadExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientSingleDownloadExt.kt index 1d894cd20..b797ab647 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientSingleDownloadExt.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientSingleDownloadExt.kt @@ -67,7 +67,7 @@ suspend fun INostrClient.downloadFirstEvent( subscriptionId: String = newSubId(), filters: Map>, ): Event? { - val resultChannel = Channel(UNLIMITED) + val resultChannel = Channel(UNLIMITED) val listener = object : IRequestListener { @@ -79,6 +79,29 @@ suspend fun INostrClient.downloadFirstEvent( ) { resultChannel.trySend(event) } + + override fun onCannotConnect( + relay: NormalizedRelayUrl, + message: String, + forFilters: List?, + ) { + resultChannel.trySend(null) + } + + override fun onClosed( + message: String, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + resultChannel.trySend(null) + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + resultChannel.trySend(null) + } } openReqSubscription(subscriptionId, filters, listener) diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutbox.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutbox.kt index 23ed96027..8f077132a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutbox.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutbox.kt @@ -28,26 +28,63 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.update +import kotlin.compareTo class PoolEventOutbox { private var eventOutbox = mapOf() val relays = MutableStateFlow(setOf()) - fun updateRelays() { - val myRelays = mutableSetOf() - eventOutbox.values.forEach { - myRelays.addAll(it.relaysLeft()) + fun needsToUpdateRelays(): Boolean { + val currentRelays = relays.value + + var relaysToRemoveCounter = 0 + + currentRelays.forEach { currentRelay -> + if (eventOutbox.values.none { currentRelay in it.relaysRemaining }) { + relaysToRemoveCounter++ + } } - if (relays.value != myRelays) { - relays.tryEmit(myRelays) + var relaysToAddCounter = 0 + eventOutbox.values.forEach { outboxState -> + if (outboxState.relaysRemaining.any { it !in currentRelays }) { + relaysToAddCounter++ + } + } + + return relaysToRemoveCounter > 0 || relaysToAddCounter > 0 + } + + fun updateRelays() { + if (needsToUpdateRelays()) { + relays.update { currentRelays -> + val relaysToRemove = mutableSetOf() + + currentRelays.forEach { currentRelay -> + if (eventOutbox.values.none { currentRelay in it.relaysRemaining }) { + relaysToRemove.add(currentRelay) + } + } + + val relaysToAdd = mutableSetOf() + eventOutbox.values.forEach { outboxState -> + outboxState.relaysRemaining.forEach { relay -> + if (relay !in relaysToAdd && relay !in currentRelays) { + relaysToAdd.add(relay) + } + } + } + + (currentRelays - relaysToRemove) + relaysToAdd + } } } fun activeOutboxCacheFor(url: NormalizedRelayUrl): Set { val myEvents = mutableSetOf() eventOutbox.forEach { (eventId, outboxCache) -> - if (url in outboxCache.relays) { + if (url in outboxCache.relaysRemaining) { myEvents.add(eventId) } } @@ -72,7 +109,12 @@ class PoolEventOutbox { id: HexKey, url: NormalizedRelayUrl, ) { - eventOutbox[id]?.newTry(url) + val waiting = eventOutbox[id] + waiting?.newTry(url) + if (waiting?.isDone() == true) { + eventOutbox = eventOutbox - waiting.event.id + updateRelays() + } } fun newResponse( @@ -84,15 +126,13 @@ class PoolEventOutbox { val waiting = eventOutbox[id] if (waiting != null) { waiting.newResponse(url, success, message) - clear() + if (waiting.isDone()) { + eventOutbox = eventOutbox - waiting.event.id + updateRelays() + } } } - fun clear() { - eventOutbox = eventOutbox.filter { !it.value.isDone() } - updateRelays() - } - // -------------------------- // State management functions // -------------------------- @@ -143,7 +183,7 @@ class PoolEventOutbox { errorMessage: String, ) { eventOutbox.forEach { - if (relay in it.value.relays) { + if (relay in it.value.relaysRemaining) { newResponse(it.key, relay, false, errorMessage) } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutboxState.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutboxState.kt index bb2407501..f75f3aa6a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutboxState.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/PoolEventOutboxState.kt @@ -26,35 +26,37 @@ import com.vitorpamplona.quartz.utils.TimeUtils class PoolEventOutboxState( val event: Event, - var relays: Set, + var relaysRemaining: Set, ) { - private var tries = mapOf() + private var failures = mapOf() fun updateRelays(newRelays: Set) { - relays = newRelays + relaysRemaining = newRelays } - fun isDone(url: NormalizedRelayUrl) = tries[url]?.isDone() ?: false + fun isDone() = relaysRemaining.isEmpty() - fun isDone() = relays.all { isDone(it) } + fun relaysLeft(): Set = relaysRemaining - fun relaysLeft(): Set = relays.filterTo(mutableSetOf()) { !isDone(it) } - - fun isSupposedToGo(url: NormalizedRelayUrl) = url in relays && !isDone(url) + fun isSupposedToGo(url: NormalizedRelayUrl) = url in relaysRemaining fun forEachUnsentEvent( url: NormalizedRelayUrl, run: (url: Event) -> Unit, ) = if (isSupposedToGo(url)) run(event) else null - fun remainingRelays() = relays.filterTo(mutableSetOf(), ::isSupposedToGo) + fun remainingRelays() = relaysRemaining fun newTry(url: NormalizedRelayUrl) { - val currentTries = tries[url] + val currentTries = failures[url] if (currentTries != null) { currentTries.addTriedTime(TimeUtils.now()) + if (currentTries.isDone()) { + relaysRemaining = relaysRemaining - url + failures = failures - url + } } else { - tries = tries + (url to Tries(listOf(TimeUtils.now()))) + failures = failures + (url to Tries(listOf(TimeUtils.now()))) } } @@ -63,38 +65,44 @@ class PoolEventOutboxState( success: Boolean, message: String, ) { - val currentTries = tries[url] - if (currentTries != null) { - currentTries.addResponse(Response(success, message)) + val currentTries = failures[url] + if (success || message.shouldDiscard()) { + relaysRemaining = relaysRemaining - url + failures = failures - url } else { - tries = tries + ( - url to - Tries( - listOf(TimeUtils.now() - 1), - listOf(Response(success, message)), - ) - ) + if (currentTries != null) { + currentTries.addResponse(message) + } else { + failures = failures + ( + url to + Tries( + listOf(TimeUtils.now() - 1), + listOf(message), + ) + ) + } } } + fun String.shouldDiscard() = + this.startsWith("replaced:") || + this.startsWith("pow:") || + this.startsWith("deleted:") || + this.startsWith("invalid:") + // Tries 3 times class Tries( var tries: List = listOf(), - var responses: List = listOf(), + var responses: List = listOf(), ) { - fun isDone() = responses.any { it.success } || responses.size > 2 || tries.size > 3 + fun isDone() = responses.size > 2 || tries.size > 3 - fun addResponse(r: Response) { - responses += r + fun addResponse(msg: String) { + responses += msg } fun addTriedTime(tried: Long) { tries += tried } } - - class Response( - val success: Boolean, - val message: String, - ) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/stats/RelayStats.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/stats/RelayStats.kt index fdd0c60ef..835499361 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/stats/RelayStats.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/stats/RelayStats.kt @@ -42,6 +42,8 @@ class RelayStats( override fun create(key: NormalizedRelayUrl): RelayStat = RelayStat() } + fun snapshot(): Map = innerCache.snapshot() + fun get(url: NormalizedRelayUrl): RelayStat = innerCache[url] ?: throw IllegalArgumentException("Should never happen") private val clientListener = diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip46RemoteSigner/signer/NostrSignerRemoteIsolationTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip46RemoteSigner/signer/NostrSignerRemoteIsolationTest.kt index 95432f61f..946b8fb2d 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip46RemoteSigner/signer/NostrSignerRemoteIsolationTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip46RemoteSigner/signer/NostrSignerRemoteIsolationTest.kt @@ -101,6 +101,8 @@ private class TrackingNostrClient : INostrClient { override fun activeCounts(url: NormalizedRelayUrl): Map> = emptyMap() override fun activeOutboxCache(url: NormalizedRelayUrl): Set = emptySet() + + override fun close() {} } /**