Subscription management

2026-04-19 21:26:36 +02:00
parent 403b245880
commit 500f8db340
5 changed files with 84 additions and 93 deletions
--- 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<RelayDTO>) {
+    fun addRelaysIfMissing(relays: Set<RelayDTO>) {
        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<NostrIncomingMessage, List<NostrEvent>> {
-        addRelays(
+    suspend fun query(reqCommand: ReqCmd, relayUrl: String): Flow<NostrIncomingMessage> {
+        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<NostrIncomingMessage, List<NostrEvent>> {
-        val messages = this.incomingMessages
+    private suspend fun NostrSocketClient.queryAsFlow(subscriptionId: String): Flow<NostrIncomingMessage> {
+        return this.incomingMessages
            .filterBySubscriptionId(id = subscriptionId)
-            .transformWhileEventsAreIncoming()
-            .timeout(15.seconds)
-            .toList()
-
-        val terminationMessage = messages.lastOrNull()
-        terminationMessage.verifyOrThrow(subscriptionId)
-        checkNotNull(terminationMessage)
-
-        val eventMessages = messages.filterIsInstance<NostrIncomingMessage.EventMessage>()
-        val eventsMessage = messages.filterIsInstance<NostrIncomingMessage.EventsMessage>()
-
-        val allNostrEvents = eventMessages.mapNotNull { it.nostrEvent } +
-                eventsMessage.map { it.nostrEvents }.flatten()
-
-
-        return Pair(
-            terminationMessage,
-            allNostrEvents,
-        )
+//            .transformWhileEventsAreIncoming()
    }

    @OptIn(FlowPreview::class)
--- 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<NostrIncomingMessage, List<NostrEvent>> {
+    suspend fun query(reqCommand: ReqCmd, relayUrl: String): Flow<NostrIncomingMessage> {
        return relayPool.query(
            reqCommand = reqCommand,
            relayUrl = relayUrl
        )
    }
+
+    suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) {
+        return relayPool.closeQuery(
+            closeCmd = closeCmd,
+            relayUrl = relayUrl
+        )
+    }
 }
--- 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)
                        }
--- a/composeApp/src/iosMain/kotlin/ac/aux/compose/managers/DataStoreManager.ios.kt
+++ b/composeApp/src/iosMain/kotlin/ac/aux/compose/managers/DataStoreManager.ios.kt
@@ -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<Set<CredentialsManager.Credential>> {
-    return OkioStorage(
-        fileSystem = FileSystem.SYSTEM,
-        serializer = OkioSerializer<Set<CredentialsManager.Credential>>,
-        producePath = {
-            val documentDirectory: NSURL? = NSFileManager.defaultManager.URLsForDirectory(
-                directory = NSDocumentDirectory,
-                inDomains = NSUserDomainMask,
-                appropriateForURL = null,
-                create = false,
-                error = null
-            )
-
-            (requireNotNull(documentDirectory).path + "/$dataStoreFileName").toPath()
-        }
-    )
-}
--- a/composeApp/src/jvmMain/kotlin/ac/aux/compose/managers/DataStoreManager.jvm.kt
+++ b/composeApp/src/jvmMain/kotlin/ac/aux/compose/managers/DataStoreManager.jvm.kt
@@ -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<Set<CredentialsManager.Credential>> {
-    FileStorage(
-        directory = ""
-    )
-    TODO("Not yet implemented")
-}