From acea87e932c624e79d199ee5e46ba5fcaccff9a6 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Mon, 4 May 2026 22:09:10 +0200 Subject: [PATCH] Close negentropy sync --- .../compose/network/relays/RelayPool.kt | 25 ++++++++++++++++++- .../network/relays/RelaysSocketManager.kt | 21 +++++++++++----- .../ui/view/model/NavigationViewModel.kt | 22 ++++++++++------ 3 files changed, 53 insertions(+), 15 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelayPool.kt index 68f14a98..98b9f57c 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelayPool.kt @@ -34,6 +34,7 @@ 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 com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.flow.Flow @@ -210,7 +211,7 @@ class RelayPool( } } - suspend fun sync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow { + suspend fun negentropySync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow { addRelaysIfMissing( setOf( NormalizedRelayUrl(relayUrl).url.toRelayDTO() @@ -256,6 +257,28 @@ class RelayPool( } } + suspend fun closeNegentropySync(negCloseCmd: NegCloseCmd, 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 closeNegentropySubscription = OptimizedJsonMapper.toJson(negCloseCmd) + + if (nostrSocketClient == null) { + throw NetworkException("$relayUrl is not connected") + } + coroutineScope { + with(nostrSocketClient) { + sendMESSAGE(closeNegentropySubscription) + } + } + } + private fun Flow.transformWhileEventsAreIncoming() = transformWhile { emit(it) diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelaysSocketManager.kt index d8206166..5da9112b 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelaysSocketManager.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/relays/RelaysSocketManager.kt @@ -21,6 +21,7 @@ import ac.cord.auxiliary.compose.managers.toHex import ac.cord.auxiliary.compose.network.sockets.NostrIncomingMessage import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd +import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import kotlinx.coroutines.flow.Flow @@ -144,12 +145,6 @@ class RelaysSocketManager constructor( ) } - suspend fun sync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow { - return relayPool.sync( - negOpenCmd = negOpenCmd, - relayUrl = relayUrl - ) - } suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) { return relayPool.closeQuery( @@ -157,4 +152,18 @@ class RelaysSocketManager constructor( relayUrl = relayUrl ) } + + suspend fun negentropySync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow { + return relayPool.negentropySync( + negOpenCmd = negOpenCmd, + relayUrl = relayUrl + ) + } + + suspend fun closeNegentropySync(negCloseCmd: NegCloseCmd, relayUrl: String) { + return relayPool.closeNegentropySync( + negCloseCmd = negCloseCmd, + relayUrl = relayUrl + ) + } } diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt index 914fa296..19a0bcbc 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt @@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync import com.vitorpamplona.quartz.nip51Lists.encryption.PrivateTagsInContent +import com.vitorpamplona.quartz.nip77Negentropy.NegCloseCmd import com.vitorpamplona.quartz.nip77Negentropy.NegOpenCmd import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -41,7 +42,6 @@ import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.getAndUpdate import kotlinx.coroutines.flow.timeout import kotlinx.coroutines.launch -import kotlin.time.Clock import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Instant @@ -263,7 +263,11 @@ class NavigationViewModel( scope.launch(Dispatchers.IO) { try { - relaysSocketManager.sync( + val negCloseCmd = NegCloseCmd( + subId = negentropySynchronizeRequest.id, + ) + + relaysSocketManager.negentropySync( negOpenCmd, negentropySynchronizeRequest.relayURL ).collect { nostrIncomingMessage -> @@ -293,17 +297,14 @@ class NavigationViewModel( } is NostrIncomingMessage.EoseMessage -> { logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage") - val closeCommand = CloseCmd( - subId = negentropySynchronizeRequest.id, - ) - relaysSocketManager.closeQuery( - closeCommand, + relaysSocketManager.closeNegentropySync( + negCloseCmd, negentropySynchronizeRequest.relayURL ) } is NostrIncomingMessage.NegentropyError -> { - logger.d("Negentropy Erroy (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}") + logger.d("Negentropy Error (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}") val since = negentropySynchronizeRequest.synchronizationFilter.since // TODO: Update since -> until based on what we have in the local db... val until = negentropySynchronizeRequest.synchronizationFilter.until @@ -346,6 +347,11 @@ class NavigationViewModel( nostrRepository.scheduleBroadcastNostrEventRequests( broadcastNostrEventRequests ) + + relaysSocketManager.closeNegentropySync( + negCloseCmd, + negentropySynchronizeRequest.relayURL + ) } else -> { logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage")