Create logical vs concrete relay pool

2026-04-19 01:46:02 +02:00
parent 87bd722960
commit 56ba03c2b3
6 changed files with 151 additions and 80 deletions
--- 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()


--- 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<RelayDTO> = emptyList()
-        private set
+    val relays: MutableSet<RelayDTO> = mutableSetOf()

    @VisibleForTesting
    var socketClients = listOf<NostrSocketClient>()
@@ -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<RelayDTO>) {
+        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<RelayDTO>) {
+        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<RelayDTO> = 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)
        }
    }

--- 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<RelayDTO>()
+
+    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<RelayDTO>?, walletRelays: List<RelayDTO>?) {
+    private suspend fun updateRelayPools(regularRelays: List<RelayDTO>?) {
        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<RelayDTO>) {
-        val customPool = buildRelayPool()
-        customPool.changeRelays(relays = relays)
-        customPool.publishEvent(nostrEvent = nostrEvent, cachingProxyEnabled = isCachingProxyEnabled())
-        customPool.closePool()
-    }
+    suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set<RelayDTO>) {
+        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)
 }
--- 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<RelayDTO>): 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
    }
--- 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<SignInRoute> {
            SignInToProfileScreen(
-                nostrRepository = nostrRepository
+                nostrRepository = databaseNostrRepository
            )
        }
        composable<UnsignedProfileRoute> { backStackEntry ->
@@ -267,7 +268,7 @@ fun AuxNavHost(
            UnqueuedProfileSynchronizationScreen(
                unsignedNostrEventId = route.unsignedNostrEventId,
                profilePublicKey = route.publicKey,
-                nostrRepository = nostrRepository
+                nostrRepository = databaseNostrRepository
            )
        }
        composable<UnindexedProfileRoute> {
@@ -320,7 +321,7 @@ fun AuxNavHost(
                        route = SearchRoute
                    )
                },
-                nostrRepository = nostrRepository
+                nostrRepository = databaseNostrRepository
            )
        }
        composable<BlankRoute> {
@@ -334,7 +335,7 @@ fun AuxNavHost(
        composable<SearchRoute> {
            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()
                },
--- 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
+//                                )
+//                            }
+//                        }
+//                    )
                }
            }
        }