From 56ba03c2b3b02cbeed4877a0fb30e9f7abba3044 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sun, 19 Apr 2026 01:46:02 +0200 Subject: [PATCH] Create logical vs concrete relay pool --- .../compose/network/relays/FallbackRelays.kt | 2 +- .../aux/compose/network/relays/RelayPool.kt | 55 +++++++++++-- .../network/relays/RelaysSocketManager.kt | 81 +++++++++---------- .../repository/NostrPublisherRepository.kt | 4 +- .../ui/composable/navigation/AuxNavHost.kt | 21 ++--- .../ui/view/model/NavigationViewModel.kt | 68 ++++++++++++---- 6 files changed, 151 insertions(+), 80 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt index 471b4ee9..e9a26b43 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/FallbackRelays.kt @@ -11,6 +11,6 @@ val FALLBACK_RELAYS = listOf( "wss://nos.lol", "wss://offchain.pub", "wss://nostr.bitcoiner.social", -).map { it.toRelayDTO() } +).map { it.toRelayDTO() }.toSet() diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt index b12757d5..818d616c 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt @@ -47,8 +47,7 @@ class RelayPool( private val scope = CoroutineScope(Dispatchers.IO) - var relays: List = emptyList() - private set + val relays: MutableSet = mutableSetOf() @VisibleForTesting var socketClients = listOf() @@ -88,7 +87,41 @@ class RelayPool( updateRelayStatus(url = client.socketUrl, connected = false) scope.launch { client.close() } } - this.relays = relays + this.relays.clear() + this.relays.addAll(relays) + } + + fun removeRelays(relays: Set) { + val relayUrls = relays.map { it.url } + + val toRemoveSocketClients = socketClients.filter { it.socketUrl in relayUrls } + + val newSocketClients = socketClients.toMutableList().apply { + removeAll(toRemoveSocketClients) + } + + socketClients = newSocketClients + toRemoveSocketClients.forEach { client -> + updateRelayStatus(url = client.socketUrl, connected = false) + scope.launch { client.close() } + } + this.relays.removeAll(relays) + } + + fun addRelays(relays: Set) { + val existingRelayUrls = socketClients.map { it.socketUrl } + val newRelayUrls = relays.map { it.url } + + val toAddRelayUrls = newRelayUrls.filter { it !in existingRelayUrls } + val toAddSocketClients = relays.filter { it.url in toAddRelayUrls }.mapAsNostrSocketClient() + + val newSocketClients = socketClients.toMutableList().apply { + addAll(toAddSocketClients) + } + + socketClients = newSocketClients + + this.relays.addAll(relays) } fun closePool() { @@ -97,7 +130,7 @@ class RelayPool( scope.launch { client.close() } } socketClients = emptyList() - relays = emptyList() + relays.clear() } fun hasRelays() = relays.isNotEmpty() @@ -118,11 +151,17 @@ class RelayPool( } @Throws(NostrPublishException::class) - suspend fun publishEvent(nostrEvent: NostrEvent, cachingProxyEnabled: Boolean = false) { - if (cachingProxyEnabled) { - handleBroadcastEventThroughCachingProxy(relays.map { it.url }, nostrEvent) - } else { + suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set = emptySet()) { + val relayUrls = relays.map { it.url } + + if (relayUrls.isEmpty()) { handlePublishEventToRelays(socketClients, nostrEvent) + } else { + addRelays(relays) + + val filteredSocketClients = socketClients.filter { relayUrls.contains(it.socketUrl) } + + handlePublishEventToRelays(filteredSocketClients, nostrEvent) } } diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt index 7187063a..0bbfd72a 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt @@ -36,24 +36,26 @@ class RelaysSocketManager constructor( private var relaysObserverJob: Job? = null - private fun buildRelayPool() = - RelayPool( - nostrSocketClientFactory = nostrSocketClientFactory, - cachingImportRepository = cachingImportRepository, - ) - private val userRelaysPool: RelayPool = buildRelayPool() - private val nwcRelaysPool: RelayPool = buildRelayPool() - private val fallbackRelaysPool: RelayPool = buildRelayPool() + /** + * Concrete relay pool + */ + private val relayPool = RelayPool( + nostrSocketClientFactory = nostrSocketClientFactory, + cachingImportRepository = cachingImportRepository, + ) - val userRelayPoolStatus = userRelaysPool.relayPoolStatus + /** + * Logical relayPool + */ + private val userRelays = mutableSetOf() + + val userRelayPoolStatus = relayPool.relayPoolStatus init { - initFallbackRelaysPool() observeActiveUserId() } - private fun initFallbackRelaysPool() = fallbackRelaysPool.changeRelays(FALLBACK_RELAYS) private fun observeActiveUserId() = scope.launch { @@ -79,68 +81,61 @@ class RelaysSocketManager constructor( scope.launch { try { relayRepository.observePublicKeyRelays(publicKey = publicKey).collect { relays -> - val userRelays = relays.filter { it.type != "nwc" }.map { it.mapToRelayDTO() } - val nwcRelays = relays.filter { it.type == "nwc" }.map { it.mapToRelayDTO() } - updateRelayPools(regularRelays = userRelays, walletRelays = nwcRelays) + val userRelays = relays.filter { it.type == "user" }.map { it.mapToRelayDTO() } + updateRelayPools(regularRelays = userRelays) } } catch (error: CancellationException) { logger.w(throwable = error) { "Relay observation cancelled" } } } - private suspend fun updateRelayPools(regularRelays: List?, walletRelays: List?) { + private suspend fun updateRelayPools(regularRelays: List?) { relayPoolsMutex.withLock { - val userRelaysChanged = userRelaysPool.relays != regularRelays + val userRelaysChanged = relayPool.relays != regularRelays if (userRelaysChanged && !regularRelays.isNullOrEmpty()) { - userRelaysPool.changeRelays(relays = regularRelays) - } - - val nwcRelaysChanged = nwcRelaysPool.relays != walletRelays - if (nwcRelaysChanged && !walletRelays.isNullOrEmpty()) { - nwcRelaysPool.changeRelays(relays = walletRelays) + relayPool.changeRelays(relays = regularRelays) } } } private suspend fun clearRelayPools() = relayPoolsMutex.withLock { - userRelaysPool.closePool() - nwcRelaysPool.closePool() + relayPool.closePool() } @Throws(NostrPublishException::class) suspend fun publishEvent(nostrEvent: NostrEvent) { - if (userRelaysPool.hasRelays()) { - userRelaysPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + if (userRelays.isNotEmpty()) { + relayPool.publishEvent( + nostrEvent = nostrEvent, + relays = userRelays + ) } else { - fallbackRelaysPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + relayPool.publishEvent( + nostrEvent = nostrEvent, + relays = FALLBACK_RELAYS + ) } } @Throws(NostrPublishException::class) - suspend fun publishEvent(nostrEvent: NostrEvent, relays: List) { - val customPool = buildRelayPool() - customPool.changeRelays(relays = relays) - customPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) - customPool.closePool() - } + suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set) { + relayPool.publishEvent( + nostrEvent = nostrEvent, + relays = relays + ) - @Throws(NostrPublishException::class) - suspend fun publishNwcEvent(nostrEvent: NostrEvent) { - if (!nwcRelaysPool.hasRelays()) { - throw NostrPublishException(cause = IllegalStateException("nwc relay not found")) - } - - nwcRelaysPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled()) + // TODO: Close relays and remove them from relayPool +// customPool.closePool() } fun tryConnectingToAllUserRelays() { - userRelaysPool.relays.forEach { + relayPool.relays.forEach { scope.launch { - userRelaysPool.tryConnectingToRelay(it.url) + relayPool.tryConnectingToRelay(it.url) } } } - suspend fun tryConnectingToUserRelay(url: String) = userRelaysPool.tryConnectingToRelay(url) + suspend fun tryConnectingToUserRelay(url: String) = relayPool.tryConnectingToRelay(url) } diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrPublisherRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrPublisherRepository.kt index a9adf8a7..4c1fe7b0 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrPublisherRepository.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrPublisherRepository.kt @@ -44,7 +44,7 @@ class NostrPublisherRepository( runCatching { relaysSocketManager.publishEvent( nostrEvent = signedNostrEvent, - relays = outboxRelays.map { RelayDTO(url = it, read = false, write = true) }, + relays = outboxRelays.map { RelayDTO(url = it, read = false, write = true) }.toSet(), ) }.onFailure { error -> logger.w(throwable = error) { "Failed to publish to outbox relays." } @@ -66,7 +66,7 @@ class NostrPublisherRepository( @Throws(NostrPublishException::class, SignatureException::class) suspend fun publishRelayList(userId: String, relays: List): NostrEvent { val signedNostrEvent = nostrNotary.signRelayListMetadata(userId = userId, relays = relays).unwrapOrThrow() - relaysSocketManager.publishEvent(nostrEvent = signedNostrEvent, relays = relays) + relaysSocketManager.publishEvent(nostrEvent = signedNostrEvent, relays = relays.toSet()) importEvent(signedNostrEvent) return signedNostrEvent } diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/composable/navigation/AuxNavHost.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/composable/navigation/AuxNavHost.kt index c6041b88..9b304b54 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/composable/navigation/AuxNavHost.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/composable/navigation/AuxNavHost.kt @@ -74,7 +74,7 @@ fun AuxNavHost( val databaseManager = DatabaseManager(auxGlobal) - val nostrRepository = DatabaseNostrRepository( + val databaseNostrRepository = DatabaseNostrRepository( database = databaseManager.auxDatabase ) val searchRepository = DatabaseSearchRepository( @@ -93,7 +93,8 @@ fun AuxNavHost( val navigationViewModel: NavigationViewModel = viewModel ( factory = NavigationViewModel.factory( initialNavigationUIState = NavigationUIState.Loading, - nostrRepository = nostrRepository, + nostrRepository = databaseNostrRepository, + relayRepository = databaseNostrRepository, scope = applicationIOScope ) ) @@ -233,7 +234,7 @@ fun AuxNavHost( popUpTo(0) } }, - nostrRepository = nostrRepository + nostrRepository = databaseNostrRepository ) } composable< WriteNewNoteRoute> { @@ -243,12 +244,12 @@ fun AuxNavHost( navController.popBackStack() } }, - nostrRepository = nostrRepository + nostrRepository = databaseNostrRepository ) } composable { SignInToProfileScreen( - nostrRepository = nostrRepository + nostrRepository = databaseNostrRepository ) } composable { backStackEntry -> @@ -267,7 +268,7 @@ fun AuxNavHost( UnqueuedProfileSynchronizationScreen( unsignedNostrEventId = route.unsignedNostrEventId, profilePublicKey = route.publicKey, - nostrRepository = nostrRepository + nostrRepository = databaseNostrRepository ) } composable { @@ -320,7 +321,7 @@ fun AuxNavHost( route = SearchRoute ) }, - nostrRepository = nostrRepository + nostrRepository = databaseNostrRepository ) } composable { @@ -334,7 +335,7 @@ fun AuxNavHost( composable { SearchScreen( initialSearchUIState = SearchUIState.Prompt, - nostrRepository = nostrRepository, + nostrRepository = databaseNostrRepository, searchRepository = searchRepository, onNavigateToProfile = { route -> navController.navigate( @@ -353,7 +354,7 @@ fun AuxNavHost( SearchResultScreen( searchQuery = route.query, - nostrRepository = nostrRepository, + nostrRepository = databaseNostrRepository, searchRepository = searchRepository, onNavigateBack = { navController.popBackStack() @@ -371,7 +372,7 @@ fun AuxNavHost( NostrEventDetailScreen( initialNostrEventDetailUIState = NostrEventDetailUIState.Loading, nostrEventId = route.nostrEventId, - nostrRepository = nostrRepository, + nostrRepository = databaseNostrRepository, onNavigateBack = { navController.popBackStack() }, diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt index 2a766c2f..f60a9cec 100644 --- a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt @@ -3,10 +3,13 @@ package ac.aux.compose.ui.view.model import ac.aux.compose.database.model.NostrEvent import ac.aux.compose.managers.SeedManager import ac.aux.compose.network.NostrEventBroadcaster +import ac.aux.compose.network.dto.RelayDTO import ac.aux.compose.network.relays.RelaysSocketManager import ac.aux.compose.network.sockets.NostrSocketClientFactory import ac.aux.compose.nostr.Relays +import ac.aux.compose.repository.CachingImportRepository import ac.aux.compose.repository.NostrRepository +import ac.aux.compose.repository.RelayRepository import ac.aux.compose.ui.view.state.NavigationUIState import androidx.lifecycle.ViewModel import androidx.lifecycle.ViewModelProvider @@ -49,21 +52,32 @@ import kotlin.time.Instant class NavigationViewModel( initialNavigationUIState: NavigationUIState, val nostrRepository: NostrRepository, + val relayRepository: RelayRepository, val scope: CoroutineScope, ): ViewModel() { + val relaysSocketManager = RelaysSocketManager( + nostrSocketClientFactory = NostrSocketClientFactory, + cachingImportRepository = CachingImportRepository.NO_OP_CACHING_IMPORT_REPOSITORY, + relayRepository = relayRepository + ) + + + companion object { private const val TAG = "NavigationViewModel" fun factory( initialNavigationUIState: NavigationUIState, nostrRepository: NostrRepository, + relayRepository: RelayRepository, scope: CoroutineScope, ): ViewModelProvider.Factory = viewModelFactory { initializer { NavigationViewModel( initialNavigationUIState = initialNavigationUIState, nostrRepository = nostrRepository, + relayRepository = relayRepository, scope = scope ) } @@ -290,23 +304,45 @@ class NavigationViewModel( scope.launch(Dispatchers.IO) { nostrRepository.observePendingBroadcastNostrEventRequests().collect { localBroadcastNostrEventRequests -> localBroadcastNostrEventRequests.forEach { localBroadcastNostrEventRequest -> - nostrEventBroadcaster.broadcastEvent( - localBroadcastNostrEventRequest = localBroadcastNostrEventRequest, - onBroadcastRequestProcessed = { broadcastNostrEventRequest -> - scope.launch(Dispatchers.IO) { - nostrRepository.broadcastProcessed( - broadcastNostrEventRequest - ) - } - }, - onBroadcastReceipt = { broadcastNostrEventReceipt -> - scope.launch(Dispatchers.IO) { - nostrRepository.saveBroadcastReceipt( - broadcastNostrEventReceipt - ) - } - } + + if (localBroadcastNostrEventRequest.nostrEvent.broadcastedAt == null) { + relaysSocketManager.publishEvent(localBroadcastNostrEventRequest.nostrEvent) + } + + scope.launch(Dispatchers.IO) { + nostrRepository.broadcastProcessed( + localBroadcastNostrEventRequest.broadcastNostrEventRequest + ) + } + // Broadcast to the intended relay... + relaysSocketManager.publishEvent( + localBroadcastNostrEventRequest.nostrEvent, + setOf( + RelayDTO( + localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL, + write = true, + read = true + ) + ) ) + +// nostrEventBroadcaster.broadcastEvent( +// localBroadcastNostrEventRequest = localBroadcastNostrEventRequest, +// onBroadcastRequestProcessed = { broadcastNostrEventRequest -> +// scope.launch(Dispatchers.IO) { +// nostrRepository.broadcastProcessed( +// broadcastNostrEventRequest +// ) +// } +// }, +// onBroadcastReceipt = { broadcastNostrEventReceipt -> +// scope.launch(Dispatchers.IO) { +// nostrRepository.saveBroadcastReceipt( +// broadcastNostrEventReceipt +// ) +// } +// } +// ) } } }