From 500f8db340b6333595b87ab1902c7a59c92870d8 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sun, 19 Apr 2026 21:26:36 +0200 Subject: [PATCH] Subscription management --- .../aux/compose/network/relays/RelayPool.kt | 65 +++++++++---------- .../network/relays/RelaysSocketManager.kt | 11 +++- .../ui/view/model/NavigationViewModel.kt | 53 +++++++++++---- .../compose/managers/DataStoreManager.ios.kt | 33 ---------- .../compose/managers/DataStoreManager.jvm.kt | 15 ----- 5 files changed, 84 insertions(+), 93 deletions(-) delete mode 100644 composeApp/src/iosMain/kotlin/ac/aux/compose/managers/DataStoreManager.ios.kt delete mode 100644 composeApp/src/jvmMain/kotlin/ac/aux/compose/managers/DataStoreManager.jvm.kt 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 9d899bfd..3648a998 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 @@ -28,20 +28,16 @@ import ac.aux.compose.network.sockets.SocketConnectionClosedCallback import ac.aux.compose.network.sockets.SocketConnectionOpenedCallback import ac.aux.compose.network.sockets.filterByEventId import ac.aux.compose.network.sockets.filterBySubscriptionId -import ac.aux.compose.network.sockets.verifyOrThrow import ac.aux.compose.repository.CachingImportRepository import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.normalizer.displayUrl -import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.toList import kotlinx.coroutines.flow.transformWhile -import kotlin.time.Duration.Companion.seconds /** * As seen in Primal @@ -121,7 +117,7 @@ class RelayPool( this.relays.removeAll(relays) } - fun addRelays(relays: Set) { + fun addRelaysIfMissing(relays: Set) { val existingRelayUrls = socketClients.map { it.socketUrl } val newRelayUrls = relays.map { it.url } @@ -171,7 +167,7 @@ class RelayPool( if (relayUrls.isEmpty()) { handlePublishEventToRelays(socketClients, nostrEvent) } else { - addRelays(relays) + addRelaysIfMissing(relays) val filteredSocketClients = socketClients.filter { relayUrls.contains(it.socketUrl) } @@ -179,9 +175,8 @@ class RelayPool( } } - @OptIn(FlowPreview::class) - suspend fun query(reqCommand: ReqCmd, relayUrl: String): Pair> { - addRelays( + suspend fun query(reqCommand: ReqCmd, relayUrl: String): Flow { + addRelaysIfMissing( setOf( NormalizedRelayUrl(relayUrl).url.toRelayDTO() ) @@ -196,11 +191,33 @@ class RelayPool( throw NetworkException("$relayUrl is not connected") } return coroutineScope { - val deferredQueryResult = async { nostrSocketClient.collectQueryResult(reqCommand.subId) } + val eventFlow = nostrSocketClient.queryAsFlow(reqCommand.subId) with(nostrSocketClient) { sendMESSAGE(filterRequest) } - deferredQueryResult.await() + eventFlow + } + } + + suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) { + addRelaysIfMissing( + setOf( + NormalizedRelayUrl(relayUrl).url.toRelayDTO() + ) + ) + + logger.d("socketClients: ${socketClients.map { it.socketUrl }}") + val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + + val closeSubscription = OptimizedJsonMapper.toJson(closeCmd) + + if (nostrSocketClient == null) { + throw NetworkException("$relayUrl is not connected") + } + coroutineScope { + with(nostrSocketClient) { + sendMESSAGE(closeSubscription) + } } } @@ -211,28 +228,10 @@ class RelayPool( } @OptIn(FlowPreview::class) - private suspend fun NostrSocketClient.collectQueryResult(subscriptionId: String): Pair> { - val messages = this.incomingMessages + private suspend fun NostrSocketClient.queryAsFlow(subscriptionId: String): Flow { + return this.incomingMessages .filterBySubscriptionId(id = subscriptionId) - .transformWhileEventsAreIncoming() - .timeout(15.seconds) - .toList() - - val terminationMessage = messages.lastOrNull() - terminationMessage.verifyOrThrow(subscriptionId) - checkNotNull(terminationMessage) - - val eventMessages = messages.filterIsInstance() - val eventsMessage = messages.filterIsInstance() - - val allNostrEvents = eventMessages.mapNotNull { it.nostrEvent } + - eventsMessage.map { it.nostrEvents }.flatten() - - - return Pair( - terminationMessage, - allNostrEvents, - ) +// .transformWhileEventsAreIncoming() } @OptIn(FlowPreview::class) 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 75006279..3cabc34a 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 @@ -21,7 +21,9 @@ import ac.aux.compose.managers.SeedManager import ac.aux.compose.managers.bech32ToHexOrNull import ac.aux.compose.managers.toHex import ac.aux.compose.network.sockets.NostrIncomingMessage +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import kotlinx.coroutines.flow.Flow /** @@ -139,10 +141,17 @@ class RelaysSocketManager constructor( suspend fun tryConnectingToUserRelay(url: String) = relayPool.tryConnectingToRelay(url) - suspend fun query(reqCommand: ReqCmd, relayUrl: String): Pair> { + suspend fun query(reqCommand: ReqCmd, relayUrl: String): Flow { return relayPool.query( reqCommand = reqCommand, relayUrl = relayUrl ) } + + suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) { + return relayPool.closeQuery( + closeCmd = closeCmd, + relayUrl = relayUrl + ) + } } 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 5953ddba..38c5b8c7 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 @@ -5,6 +5,7 @@ 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.NostrIncomingMessage import ac.aux.compose.network.sockets.NostrSocketClientFactory import ac.aux.compose.nostr.Relays import ac.aux.compose.repository.CachingImportRepository @@ -19,6 +20,7 @@ import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper import com.vitorpamplona.quartz.nip01Core.core.toHexKey +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync @@ -169,26 +171,55 @@ class NavigationViewModel( } ) + nostrRepository.synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest) - scope.launch { + scope.launch(Dispatchers.IO) { try { - val result = relaysSocketManager.query( + relaysSocketManager.query( reqCommand, synchronizeNostrEventRequest.relayURL - ) + ).collect { nostrIncomingMessage -> + when (nostrIncomingMessage) { + is NostrIncomingMessage.EventMessage -> { + logger.d("Import message: $nostrIncomingMessage") + nostrIncomingMessage.nostrEvent?.let { + nostrRepository.saveNostrEvent( + nostrEvent = it, + synchronizeNostrEventRequest, + synchronizationRelayURLs = Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url } + synchronizeNostrEventRequest.relayURL + ) + } - logger.d("Result: $result") + } + is NostrIncomingMessage.EventsMessage -> { + logger.d("Import messages: $nostrIncomingMessage") - result.second.forEach { nostrEvent -> - nostrRepository.saveNostrEvent( - nostrEvent = nostrEvent, - synchronizeNostrEventRequest, - synchronizationRelayURLs = Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url } + synchronizeNostrEventRequest.relayURL - ) + nostrIncomingMessage.nostrEvents.forEach { nostrEvent -> + nostrRepository.saveNostrEvent( + nostrEvent = nostrEvent, + synchronizeNostrEventRequest, + synchronizationRelayURLs = Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url } + synchronizeNostrEventRequest.relayURL + ) + } + } + is NostrIncomingMessage.EoseMessage -> { + logger.d("Sync request has been successfully processed (${synchronizeNostrEventRequest.relayURL}): $nostrIncomingMessage") + val closeCommand = CloseCmd( + subId = synchronizeNostrEventRequest.id, + ) + + relaysSocketManager.closeQuery( + closeCommand, + synchronizeNostrEventRequest.relayURL + ) + } + else -> { + logger.d("Unhandled message: $nostrIncomingMessage") + } + } } - nostrRepository.synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest) } catch (e: Throwable) { logger.e("Failed to sync", e) } diff --git a/composeApp/src/iosMain/kotlin/ac/aux/compose/managers/DataStoreManager.ios.kt b/composeApp/src/iosMain/kotlin/ac/aux/compose/managers/DataStoreManager.ios.kt deleted file mode 100644 index ff8d2cc9..00000000 --- a/composeApp/src/iosMain/kotlin/ac/aux/compose/managers/DataStoreManager.ios.kt +++ /dev/null @@ -1,33 +0,0 @@ -package ac.aux.compose.managers - -import ac.aux.compose.PlatformContext -import androidx.datastore.core.Storage -import androidx.datastore.core.okio.OkioSerializer -import androidx.datastore.core.okio.OkioStorage -import okio.FileSystem -import okio.Path.Companion.toPath -import platform.Foundation.NSDocumentDirectory -import platform.Foundation.NSFileManager -import platform.Foundation.NSURL -import platform.Foundation.NSUserDomainMask - -actual fun computeDatastoreStorage( - platformContext: PlatformContext, - dataStoreFileName: String -): Storage> { - return OkioStorage( - fileSystem = FileSystem.SYSTEM, - serializer = OkioSerializer>, - producePath = { - val documentDirectory: NSURL? = NSFileManager.defaultManager.URLsForDirectory( - directory = NSDocumentDirectory, - inDomains = NSUserDomainMask, - appropriateForURL = null, - create = false, - error = null - ) - - (requireNotNull(documentDirectory).path + "/$dataStoreFileName").toPath() - } - ) -} \ No newline at end of file diff --git a/composeApp/src/jvmMain/kotlin/ac/aux/compose/managers/DataStoreManager.jvm.kt b/composeApp/src/jvmMain/kotlin/ac/aux/compose/managers/DataStoreManager.jvm.kt deleted file mode 100644 index c462fe5a..00000000 --- a/composeApp/src/jvmMain/kotlin/ac/aux/compose/managers/DataStoreManager.jvm.kt +++ /dev/null @@ -1,15 +0,0 @@ -package ac.aux.compose.managers - -import ac.aux.compose.PlatformContext -import androidx.datastore.core.Storage -import io.ktor.client.plugins.cache.storage.FileStorage - -actual fun computeDatastoreStorage( - platformContext: PlatformContext, - dataStoreFileName: String -): Storage> { - FileStorage( - directory = "" - ) - TODO("Not yet implemented") -} \ No newline at end of file