From f188548ea2c91f08f5de1893f486c1c4c709e575 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Tue, 5 May 2026 14:33:27 +0200 Subject: [PATCH] Lock it up --- .../compose/database/model/Profile.kt | 29 ++ .../repository/DatabaseNostrRepository.kt | 3 - .../compose/network/relays/RelayPool.kt | 5 - .../widgets/detail/TextNoteEventDetail.kt | 3 +- .../ui/view/model/FeedListViewModel.kt | 2 +- .../ui/view/model/FollowersListViewModel.kt | 2 +- .../ui/view/model/FollowingListViewModel.kt | 2 +- .../ui/view/model/InReplyToViewModel.kt | 21 +- .../ui/view/model/NavigationViewModel.kt | 262 +++++++++--------- 9 files changed, 188 insertions(+), 141 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/Profile.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/Profile.kt index 636f934f..413f6f19 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/Profile.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/Profile.kt @@ -6,6 +6,7 @@ import ac.cord.auxiliary.compose.database.model.traits.NostrEventEntity import ac.cord.auxiliary.compose.database.model.traits.SoftDeletableEntity import ac.cord.auxiliary.compose.database.model.traits.TimestampedEntity import ac.cord.auxiliary.compose.database.model.traits.UserViewableEntity +import ac.cord.auxiliary.compose.repository.NostrRepository import ac.cord.auxiliary.compose.ui.composable.navigation.routes.NostrEventDetailRoute import ac.cord.auxiliary.compose.ui.composable.navigation.routes.Route import ac.cord.auxiliary.compose.ui.composable.widgets.profile.ProfileAvatar @@ -31,6 +32,8 @@ import androidx.room3.Entity import androidx.room3.ForeignKey import androidx.room3.Index import androidx.room3.PrimaryKey +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip51Lists.relayLists.RelayFeedsListEvent import kotlin.time.Clock import kotlin.time.Instant @@ -100,6 +103,32 @@ data class Profile( } } + suspend fun profilePublicRelays( + nostrRepository: NostrRepository + ): List? { + val relays = nostrRepository.getNostrEvent( + publicKey, + kind = RelayFeedsListEvent.KIND + )?.let { relayFeedNostrEvent -> + val relayFeedEvent = RelayFeedsListEvent( + id = relayFeedNostrEvent.id, + pubKey = relayFeedNostrEvent.pubKey, + content = relayFeedNostrEvent.content, + tags = relayFeedNostrEvent.tags, + sig = relayFeedNostrEvent.sig, + createdAt = relayFeedNostrEvent.createdAt.epochSeconds + ) + + relayFeedEvent.publicRelays() + } + + if (relays?.isNotEmpty() == true) { + return relays + } + + return null + } + @Composable fun RenderAsListItem( onNavigateToEvent: (Route) -> Unit diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt index 630665ca..6362ec38 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt @@ -390,17 +390,14 @@ class DatabaseNostrRepository( broadcastNostrEventRequest: BroadcastNostrEventRequest, status: String ) { - logger.d("Update local reference: $broadcastNostrEventRequest") database.broadcastNostrEventRequestDao().upsert( broadcastNostrEventRequest.copy( status = status ) ) - logger.d("$status: $broadcastNostrEventRequest") } override suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) { - logger.i("Update local reference: $synchronizeNostrEventRequest") database.synchronizeNostrEventRequestDao().upsert( synchronizeNostrEventRequest.copy( status = "sent", 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 98b9f57c..2281a718 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 @@ -124,15 +124,11 @@ class RelayPool( suspend fun addRelaysIfMissing(relays: Set) { relayMutex.withLock { - logger.d("addRelaysIfMissing: ${relays.map { it.url }}") val existingRelayUrls = socketClients.map { NormalizedRelayUrl(it.socketUrl).displayUrl() } - logger.d("Existing Relays: $existingRelayUrls") val newRelayUrls = relays.map { it.url } val toAddRelayUrls = newRelayUrls.filter { NormalizedRelayUrl(it).displayUrl() !in existingRelayUrls }.toSet() - logger.d("toAddRelayUrls: $toAddRelayUrls") val toAddSocketClients = relays.filter { it.url in toAddRelayUrls }.mapAsNostrSocketClient() - logger.d("toAddSocketClients: ${toAddSocketClients.map { it.socketUrl }}" ) if (toAddSocketClients.isNotEmpty()) { val newSocketClients = socketClients.toMutableSet().apply { addAll(toAddSocketClients) @@ -242,7 +238,6 @@ class RelayPool( ) ) - logger.d("socketClients: ${socketClients.map { it.socketUrl }}") val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } val closeSubscription = OptimizedJsonMapper.toJson(closeCmd) diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/composable/widgets/detail/TextNoteEventDetail.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/composable/widgets/detail/TextNoteEventDetail.kt index 2d56e6b3..8da4b479 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/composable/widgets/detail/TextNoteEventDetail.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/composable/widgets/detail/TextNoteEventDetail.kt @@ -79,7 +79,8 @@ fun TextNoteEventDetail( val inReplyToNostrEventFeedListViewModel: InReplyToViewModel = viewModel( key = "replies-${localNostrEvent.nostrEvent.id}", factory = InReplyToViewModel.factory( - nostrEventId = localNostrEvent.nostrEvent.id, + nostrEvent = localNostrEvent.nostrEvent, + profile = localNostrEvent.profile, initialFeedListUIState = FeedListUIState.Loading, nostrRepository = nostrRepository, createdAt = localNostrEvent.nostrEvent.createdAt diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FeedListViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FeedListViewModel.kt index 4ecbc26e..cda290ab 100755 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FeedListViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FeedListViewModel.kt @@ -53,7 +53,7 @@ class FeedListViewModel( val isSynchronizationPending: MutableState = mutableStateOf(false) - val negentropySynchronizeRequests = Relays.negentropicRelaySet.shuffled().take(1).map { normalizedRelayUrl -> + val negentropySynchronizeRequests = Relays.negentropicRelaySet.shuffled().map { normalizedRelayUrl -> NegentropySynchronizeRequest( id = NegentropySynchronizeRequest.computeId( normalizedRelayUrl.url, diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowersListViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowersListViewModel.kt index 7e67053a..5b8656ad 100755 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowersListViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowersListViewModel.kt @@ -76,7 +76,7 @@ class FollowersListViewModel( viewModelScope.launch(Dispatchers.IO) { // Sync Notifications... might want to also run this in the background nostrRepository.queueNegentropySynchronizeRequest( - Relays.eventPublishRelaySet.take(1).map { normalizedRelayUrl -> + Relays.negentropicRelaySet.shuffled().map { normalizedRelayUrl -> NegentropySynchronizeRequest( id = NegentropySynchronizeRequest.computeId( relayURL = normalizedRelayUrl.url, diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowingListViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowingListViewModel.kt index 3acbe147..28228af2 100755 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowingListViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FollowingListViewModel.kt @@ -75,7 +75,7 @@ class FollowingListViewModel( viewModelScope.launch(Dispatchers.IO) { // Sync Notifications... might want to also run this in the background nostrRepository.queueNegentropySynchronizeRequest( - Relays.eventPublishRelaySet.take(1).map { normalizedRelayUrl -> + Relays.negentropicRelaySet.shuffled().map { normalizedRelayUrl -> NegentropySynchronizeRequest( id = NegentropySynchronizeRequest.computeId( relayURL = normalizedRelayUrl.url, diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/InReplyToViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/InReplyToViewModel.kt index 447ad0e8..e9273bbd 100755 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/InReplyToViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/InReplyToViewModel.kt @@ -1,6 +1,9 @@ package ac.cord.auxiliary.compose.ui.view.model import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeRequest +import ac.cord.auxiliary.compose.database.model.NostrEvent +import ac.cord.auxiliary.compose.database.model.NostrEventRelay +import ac.cord.auxiliary.compose.database.model.Profile import ac.cord.auxiliary.compose.database.model.SynchronizeNostrEventRequest import ac.cord.auxiliary.compose.database.model.types.SynchronizationFilter import ac.cord.auxiliary.compose.nostr.Relays @@ -37,6 +40,7 @@ import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent import com.vitorpamplona.quartz.nip18Reposts.RepostEvent import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent +import com.vitorpamplona.quartz.nip51Lists.relayLists.RelayFeedsListEvent import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.flow.distinctUntilChanged @@ -44,7 +48,8 @@ import kotlinx.coroutines.launch import kotlin.time.Instant class InReplyToViewModel( - nostrEventId: String, + val nostrEvent: NostrEvent, + val profile: Profile?, initialFeedListUIState: FeedListUIState, val synchronizationFilter: SynchronizationFilter, val nostrRepository: NostrRepository @@ -79,8 +84,12 @@ class InReplyToViewModel( logger.d("scheduleSynchronization") viewModelScope.launch(Dispatchers.IO) { // Sync Notifications... might want to also run this in the background + val publicRelays = profile?.profilePublicRelays( + nostrRepository + ) ?: Relays.negentropicRelaySet + nostrRepository.queueNegentropySynchronizeRequest( - Relays.eventPublishRelaySet.take(1).map { normalizedRelayUrl -> + publicRelays.take(3).map { normalizedRelayUrl -> NegentropySynchronizeRequest( id = NegentropySynchronizeRequest.computeId( relayURL = normalizedRelayUrl.url, @@ -193,7 +202,8 @@ class InReplyToViewModel( const val TAG = "InReplyToViewModel" fun factory( - nostrEventId: String, + nostrEvent: NostrEvent, + profile: Profile?, createdAt: Instant, initialFeedListUIState: FeedListUIState = FeedListUIState.Loading, synchronizationFilter: SynchronizationFilter = SynchronizationFilter( @@ -204,7 +214,7 @@ class InReplyToViewModel( ), tags = mapOf( "e" to listOf( - nostrEventId + nostrEvent.id ) ), since = createdAt @@ -214,7 +224,8 @@ class InReplyToViewModel( initializer { InReplyToViewModel( initialFeedListUIState = initialFeedListUIState, - nostrEventId = nostrEventId, + nostrEvent = nostrEvent, + profile = profile, synchronizationFilter = synchronizationFilter, nostrRepository = nostrRepository ) 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 fff549bc..ce2e781f 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 @@ -43,6 +43,8 @@ import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.getAndUpdate import kotlinx.coroutines.flow.timeout import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Instant @@ -62,6 +64,8 @@ class NavigationViewModel( companion object { private const val TAG = "NavigationViewModel" + private val mutex = Mutex() + fun factory( initialNavigationUIState: NavigationUIState, nostrRepository: NostrRepository, @@ -221,168 +225,178 @@ class NavigationViewModel( scope.launch(Dispatchers.IO) { nostrRepository.observePendingNegentropySynchronizeRequests().distinctUntilChanged().collect { negentropySynchronizeRequestOrNull -> negentropySynchronizeRequestOrNull?.let { negentropySynchronizeRequest -> - logger.i("negentropySynchronizeRequest: $negentropySynchronizeRequest") + mutex.withLock { + logger.i("negentropySynchronizeRequest: $negentropySynchronizeRequest") - val events = nostrRepository.getNostrFeedIds( - arrayOf( - negentropySynchronizeRequest.synchronizationFilter - ), - applyLimits = false - ) - logger.d("Events: ${events.size}") - val storage = StorageVector().apply { - events.forEach { event -> - insert( - event.createdAt.toEpochMilliseconds(), - event.id - ) + val events = nostrRepository.getNostrFeedIds( + arrayOf( + negentropySynchronizeRequest.synchronizationFilter + ), + applyLimits = false + ) + logger.d("Events: ${events.size}") + val storage = StorageVector().apply { + events.forEach { event -> + insert( + event.createdAt.toEpochMilliseconds(), + event.id + ) + } + + seal() } + val negentropy = Negentropy( + storage, + ) - seal() - } - val negentropy = Negentropy( - storage, - ) + val negOpenCmd = NegOpenCmd( + subId = negentropySynchronizeRequest.uuid, + filter = Filter( + ids = negentropySynchronizeRequest.synchronizationFilter.ids?.toList(), + authors = negentropySynchronizeRequest.synchronizationFilter.authors?.toList(), + kinds = negentropySynchronizeRequest.synchronizationFilter.kinds?.toList(), + tags = negentropySynchronizeRequest.synchronizationFilter.tags, + tagsAll = negentropySynchronizeRequest.synchronizationFilter.tagsAll, + since = negentropySynchronizeRequest.synchronizationFilter.since?.epochSeconds, + until = negentropySynchronizeRequest.synchronizationFilter.until?.epochSeconds, + limit = null, // We don't do limits when performing negentropy... + search = negentropySynchronizeRequest.synchronizationFilter.search + ), + initialMessage = negentropy.initiate().toHexString() + ) + logger.d("negOpenCmd: ${negOpenCmd.initialMessage}") - val negOpenCmd = NegOpenCmd( - subId = negentropySynchronizeRequest.uuid, - filter = Filter( - ids = negentropySynchronizeRequest.synchronizationFilter.ids?.toList(), - authors = negentropySynchronizeRequest.synchronizationFilter.authors?.toList(), - kinds = negentropySynchronizeRequest.synchronizationFilter.kinds?.toList(), - tags = negentropySynchronizeRequest.synchronizationFilter.tags, - tagsAll = negentropySynchronizeRequest.synchronizationFilter.tagsAll, - since = negentropySynchronizeRequest.synchronizationFilter.since?.epochSeconds, - until = negentropySynchronizeRequest.synchronizationFilter.until?.epochSeconds, - limit = null, // We don't do limits when performing negentropy... - search = negentropySynchronizeRequest.synchronizationFilter.search - ), - initialMessage = negentropy.initiate().toHexString() - ) - logger.d("negOpenCmd: ${negOpenCmd.initialMessage}") + nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest) - nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest) + scope.launch(Dispatchers.IO) { + try { + val negCloseCmd = NegCloseCmd( + subId = negentropySynchronizeRequest.uuid, + ) - scope.launch(Dispatchers.IO) { - try { - val negCloseCmd = NegCloseCmd( - subId = negentropySynchronizeRequest.uuid, - ) + relaysSocketManager.negentropySync( + negOpenCmd, + negentropySynchronizeRequest.relayURL + ).collect { nostrIncomingMessage -> + when (nostrIncomingMessage) { + is NostrIncomingMessage.EventMessage -> { + scope.launch(Dispatchers.IO) { + logger.d("Import message: $nostrIncomingMessage") + nostrIncomingMessage.nostrEvent?.let { + nostrRepository.saveNostrEvent( + nostrEvent = it, + negentropySynchronizeRequest, + synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url } + ) + } + } + } + is NostrIncomingMessage.EventsMessage -> { + logger.d("Import messages: $nostrIncomingMessage") - relaysSocketManager.negentropySync( - negOpenCmd, - negentropySynchronizeRequest.relayURL - ).collect { nostrIncomingMessage -> - when (nostrIncomingMessage) { - is NostrIncomingMessage.EventMessage -> { - scope.launch(Dispatchers.IO) { - logger.d("Import message: $nostrIncomingMessage") - nostrIncomingMessage.nostrEvent?.let { + nostrIncomingMessage.nostrEvents.forEach { nostrEvent -> nostrRepository.saveNostrEvent( - nostrEvent = it, + nostrEvent = nostrEvent, negentropySynchronizeRequest, synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url } ) } } - } - is NostrIncomingMessage.EventsMessage -> { - logger.d("Import messages: $nostrIncomingMessage") + is NostrIncomingMessage.EoseMessage -> { + logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage") - nostrIncomingMessage.nostrEvents.forEach { nostrEvent -> - nostrRepository.saveNostrEvent( - nostrEvent = nostrEvent, - negentropySynchronizeRequest, - synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url } + relaysSocketManager.closeNegentropySync( + negCloseCmd, + negentropySynchronizeRequest.relayURL ) + + return@collect } - } - is NostrIncomingMessage.EoseMessage -> { - logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage") + is NostrIncomingMessage.NegentropyError -> { + logger.d("Negentropy Error (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}") - relaysSocketManager.closeNegentropySync( - negCloseCmd, - negentropySynchronizeRequest.relayURL - ) - } - is NostrIncomingMessage.NegentropyError -> { - logger.d("Negentropy Error (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}") + // TODO: Don't schedule a sync when on mobile internet. + val since = negentropySynchronizeRequest.synchronizationFilter.since // TODO: Update since -> until based on what we have in the local db... + val until = negentropySynchronizeRequest.synchronizationFilter.until - // TODO: Don't schedule a sync when on mobile internet. - val since = negentropySynchronizeRequest.synchronizationFilter.since // TODO: Update since -> until based on what we have in the local db... - val until = negentropySynchronizeRequest.synchronizationFilter.until - - val synchronizationFilter = negentropySynchronizeRequest.synchronizationFilter.copy( - since = since, - until = until - ) - - nostrRepository.queueSynchronizeNostrEvent( - listOf( - SynchronizeNostrEventRequest( - purpose = negentropySynchronizeRequest.purpose, - synchronizationFilters = arrayOf( - synchronizationFilter - ), - relayURL = negentropySynchronizeRequest.relayURL, - level = negentropySynchronizeRequest.level, - ) + val synchronizationFilter = negentropySynchronizeRequest.synchronizationFilter.copy( + since = since, + until = until ) - ) - } - is NostrIncomingMessage.NegentropyMessage -> { - logger.d("NegentropyMessage: ${nostrIncomingMessage.negentropyMessage}") - val result = negentropy.reconcile( - nostrIncomingMessage.negentropyMessage.hexToByteArray() - ) - logger.d("NeedIds: ${result.needIds.map { it.toHexString() }}") - logger.d("SendIds: ${result.sendIds.map { it.toHexString() }}") - logger.d("EventsIds: ${events.map { it.id }}") - logger.d("Timestamp: ${events.map { it.createdAt.epochSeconds }}") - - if (result.needIds.isNotEmpty()) { - // Schedule a sync from this relay... nostrRepository.queueSynchronizeNostrEvent( listOf( SynchronizeNostrEventRequest( purpose = negentropySynchronizeRequest.purpose, synchronizationFilters = arrayOf( - SynchronizationFilter( - ids = result.needIds.map { it.toHexString() }.toTypedArray() - ) + synchronizationFilter ), relayURL = negentropySynchronizeRequest.relayURL, level = negentropySynchronizeRequest.level, ) ) ) + return@collect } - val broadcastNostrEventRequests = result.sendIds.map { nostrEventId -> - BroadcastNostrEventRequest( - nostrEventId = nostrEventId.toHexString(), - relayURL = negentropySynchronizeRequest.relayURL - ) - } - nostrRepository.scheduleBroadcastNostrEventRequests( - broadcastNostrEventRequests - ) + is NostrIncomingMessage.NegentropyMessage -> { + logger.d("NegentropyMessage: ${nostrIncomingMessage.negentropyMessage}") - relaysSocketManager.closeNegentropySync( - negCloseCmd, - negentropySynchronizeRequest.relayURL - ) - } - else -> { - logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage") + val result = negentropy.reconcile( + nostrIncomingMessage.negentropyMessage.hexToByteArray() + ) + logger.d("NeedIds: ${result.needIds.map { it.toHexString() }}") + logger.d("SendIds: ${result.sendIds.map { it.toHexString() }}") + logger.d("EventsIds: ${events.map { it.id }}") + logger.d("Timestamp: ${events.map { it.createdAt.epochSeconds }}") + + if (result.needIds.isNotEmpty()) { + // Schedule a sync from this relay... + nostrRepository.queueSynchronizeNostrEvent( + listOf( + SynchronizeNostrEventRequest( + purpose = negentropySynchronizeRequest.purpose, + synchronizationFilters = arrayOf( + SynchronizationFilter( + ids = result.needIds.map { it.toHexString() }.toTypedArray() + ) + ), + relayURL = negentropySynchronizeRequest.relayURL, + level = negentropySynchronizeRequest.level, + ) + ) + ) + } + val broadcastNostrEventRequests = result.sendIds.map { nostrEventId -> + BroadcastNostrEventRequest( + nostrEventId = nostrEventId.toHexString(), + relayURL = negentropySynchronizeRequest.relayURL + ) + } + nostrRepository.scheduleBroadcastNostrEventRequests( + broadcastNostrEventRequests + ) + + relaysSocketManager.closeNegentropySync( + negCloseCmd, + negentropySynchronizeRequest.relayURL + ) + + return@collect + } + else -> { + logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage") + } } } + logger.d("Queried Sync") + } catch (e: Throwable) { + logger.e("Failed to sync", e) } - - } catch (e: Throwable) { - logger.e("Failed to sync", e) } + + logger.i("After launch") } + logger.i("Lock released") } } }