From 1c8b280673f2ea5ecb88cf6286c8758e12f60176 Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Mon, 4 May 2026 21:18:52 +0200 Subject: [PATCH] Initial negentropic logic (with broadcasting logic) --- .../dao/BroadcastNostrEventRequestDao.kt | 4 ++ .../compose/database/dao/NostrEventDao.kt | 12 ++-- .../repository/DatabaseNostrRepository.kt | 6 ++ .../network/sockets/NostrIncomingMessage.kt | 14 +++++ .../sockets/NostrIncomingMessageExt.kt | 13 +++-- .../sockets/NostrIncomingMessageParser.kt | 52 ++++++++++++++++- .../compose/network/sockets/NostrVerb.kt | 9 +++ .../compose/repository/NostrRepository.kt | 6 ++ .../ui/view/model/FeedListViewModel.kt | 2 + .../compose/ui/view/model/HomeViewModel.kt | 14 ++++- .../ui/view/model/NavigationViewModel.kt | 56 +++++++++++++------ 11 files changed, 156 insertions(+), 32 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/BroadcastNostrEventRequestDao.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/BroadcastNostrEventRequestDao.kt index c8e1bd9b..c41f5fad 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/BroadcastNostrEventRequestDao.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/BroadcastNostrEventRequestDao.kt @@ -3,6 +3,7 @@ package ac.cord.auxiliary.compose.database.dao import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventRequest import ac.cord.auxiliary.compose.database.model.intermdiate.LocalBroadcastNostrEventRequest import androidx.room3.Dao +import androidx.room3.Insert import androidx.room3.Query import androidx.room3.Upsert import kotlinx.coroutines.flow.Flow @@ -20,4 +21,7 @@ interface BroadcastNostrEventRequestDao { @Upsert suspend fun upsert(broadcastNostrEventRequest: BroadcastNostrEventRequest) + + @Insert + suspend fun insert(broadcastNostrEventRequests: List) } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt index 6b410089..85789b4b 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt @@ -66,11 +66,11 @@ interface NostrEventDao { ): Flow> @Transaction - @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND createdAt > :since ORDER BY createdAt ASC") fun getNostrEvents(kinds: Array, since: Instant = GENESIS_AT): List @Transaction - @Query("SELECT * FROM NostrEvent WHERE content LIKE '%' || :search || '%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + @Query("SELECT * FROM NostrEvent WHERE content LIKE '%' || :search || '%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt ASC") fun getFilteredNostrEvents( kinds: Array, search: String, @@ -78,7 +78,7 @@ interface NostrEventDao { ): List @Transaction - @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND id in (:ids) AND createdAt > :since ORDER BY createdAt DESC") + @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND id in (:ids) AND createdAt > :since ORDER BY createdAt ASC") fun getFilteredNostrEvents( kinds: Array, ids: Array, @@ -86,7 +86,7 @@ interface NostrEventDao { ): List @Transaction - @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND pubKey in (:authors) AND createdAt > :since ORDER BY createdAt DESC LIMIT 50") + @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND pubKey in (:authors) AND createdAt > :since ORDER BY createdAt ASC") fun getAuthoredNostrEvents( kinds: Array, authors: Array, @@ -94,7 +94,7 @@ interface NostrEventDao { ): List @Transaction - @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :publicKey || '%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :publicKey || '%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt ASC") fun getPublicKeyMentionedNostrEvents( kinds: Array, publicKey: HexKey, @@ -102,7 +102,7 @@ interface NostrEventDao { ): List @Transaction - @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :eventId || '%reply%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :eventId || '%reply%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt ASC") fun getNostrEventReplies( kinds: Array, eventId: HexKey, 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 40a16844..2d46f392 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 @@ -653,4 +653,10 @@ class DatabaseNostrRepository( override suspend fun observePublicKeyRelays(publicKey: String): Flow> { return database.relayDao().observePublicKeyRelays(publicKey) } + + override suspend fun scheduleBroadcastNostrEventRequests(broadcastNostrEventRequests: List) { + return database.broadcastNostrEventRequestDao().insert( + broadcastNostrEventRequests + ) + } } diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessage.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessage.kt index 1301b1a6..83e8d306 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessage.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessage.kt @@ -39,6 +39,20 @@ sealed class NostrIncomingMessage { val subscriptionId: String, val nostrEvents: List = emptyList(), ) : NostrIncomingMessage() + + data class NegentropyMessage( + val subscriptionId: String, + val negentropyMessage: String, + ) : NostrIncomingMessage() + + data class NegentropyClose( + val subscriptionId: String + ) : NostrIncomingMessage() + + data class NegentropyError( + val subscriptionId: String, + val negentropyReason: String, + ) : NostrIncomingMessage() } fun NostrIncomingMessage?.verifyOrThrow(subscriptionId: String) { diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageExt.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageExt.kt index 8e007bc7..e512b4f4 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageExt.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageExt.kt @@ -5,11 +5,16 @@ import kotlinx.coroutines.flow.filter fun Flow.filterBySubscriptionId(id: String) = filter { + // Negentropy... + (it is NostrIncomingMessage.NegentropyMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.NegentropyClose && it.subscriptionId == id) || + (it is NostrIncomingMessage.NegentropyError && it.subscriptionId == id) || + // Events (it is NostrIncomingMessage.EventMessage && it.subscriptionId == id) || - (it is NostrIncomingMessage.EoseMessage && it.subscriptionId == id) || - (it is NostrIncomingMessage.CountMessage && it.subscriptionId == id) || - (it is NostrIncomingMessage.EventsMessage && it.subscriptionId == id) || - (it is NostrIncomingMessage.NoticeMessage) + (it is NostrIncomingMessage.EoseMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.CountMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.EventsMessage && it.subscriptionId == id) || + (it is NostrIncomingMessage.NoticeMessage) } fun Flow.filterByEventId(id: String) = diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageParser.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageParser.kt index 0e31622d..92e27c22 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageParser.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageParser.kt @@ -28,7 +28,10 @@ fun String.parseIncomingMessage(): NostrIncomingMessage? { NostrVerb.Incoming.AUTH -> jsonArray.takeAsAuthIncomingMessage() NostrVerb.Incoming.COUNT -> jsonArray.takeAsCountIncomingMessage() NostrVerb.Incoming.EVENTS -> jsonArray.takeAsEventsIncomingMessage() - else -> null + NostrVerb.Incoming.NEGENTROPY_MESSAGE -> jsonArray.takeAsNegentropyMessageIncomingMessage() + NostrVerb.Incoming.NEGENTROPY_CLOSE -> jsonArray.takeAsNegentropyCloseIncomingMessage() + NostrVerb.Incoming.NEGENTROPY_ERROR -> jsonArray.takeAsNegentropyErrorIncomingMessage() + null -> null } } catch (error: Exception) { Logger.withTag("String.parseIncomingMessage").w(error) { "Unable to parse incoming message." } @@ -137,6 +140,50 @@ private fun JsonArray.takeAsOkIncomingMessage(): NostrIncomingMessage? { } } +private fun JsonArray.takeAsNegentropyMessageIncomingMessage(): NostrIncomingMessage? { + Logger.withTag("NostrIncomingMessageParser").d("takeAsNegentropyMessageIncomingMessage") + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val negentropyMessage = elementAtOrNull(2)?.jsonPrimitive?.content + + return if (subscriptionId != null && negentropyMessage != null) { + Logger.withTag("NostrIncomingMessageParser").d("negentropyMessage: $negentropyMessage") + NostrIncomingMessage.NegentropyMessage( + subscriptionId = subscriptionId, + negentropyMessage = negentropyMessage + ) + } else { + Logger.withTag("NostrIncomingMessageParser").d("Couldn't consume message: $this") + null + } +} + +private fun JsonArray.takeAsNegentropyCloseIncomingMessage(): NostrIncomingMessage? { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + + return if (subscriptionId != null) { + NostrIncomingMessage.NegentropyClose( + subscriptionId = subscriptionId, + ) + } else { + null + } +} + + +private fun JsonArray.takeAsNegentropyErrorIncomingMessage(): NostrIncomingMessage? { + val subscriptionId = elementAtOrNull(1)?.toSubscriptionId() + val negentropyReason = elementAtOrNull(2)?.jsonPrimitive?.content + + return if (subscriptionId != null && negentropyReason != null) { + NostrIncomingMessage.NegentropyError( + subscriptionId = subscriptionId, + negentropyReason = negentropyReason + ) + } else { + null + } +} + private fun JsonElement.toIncomingMessageType(): NostrVerb.Incoming? { return when (this.jsonPrimitive.content) { "EVENT" -> NostrVerb.Incoming.EVENT @@ -146,6 +193,9 @@ private fun JsonElement.toIncomingMessageType(): NostrVerb.Incoming? { "COUNT" -> NostrVerb.Incoming.COUNT "EVENTS" -> NostrVerb.Incoming.EVENTS "NOTICE" -> NostrVerb.Incoming.NOTICE + "NEG-MSG" -> NostrVerb.Incoming.NEGENTROPY_MESSAGE + "NEG-CLOSE" -> NostrVerb.Incoming.NEGENTROPY_CLOSE + "NEG-ERR" -> NostrVerb.Incoming.NEGENTROPY_ERROR else -> { Logger.d("Unsupported incomingMessageType: ${this.jsonPrimitive.content}") null diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrVerb.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrVerb.kt index 749e6c6e..7bfa3ab9 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrVerb.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrVerb.kt @@ -45,5 +45,14 @@ internal sealed class NostrVerb { @SerialName("EVENTS") EVENTS, + + @SerialName("NEG-MSG") + NEGENTROPY_MESSAGE, + + @SerialName("NEG-CLOSE") + NEGENTROPY_CLOSE, + + @SerialName("NEG-ERR") + NEGENTROPY_ERROR, } } diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt index b4ba30f9..e9f34873 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt @@ -122,6 +122,8 @@ interface NostrRepository { suspend fun saveSearchQuery(query: String, synchronizationFilterArray: SynchronizationFilterArray? = null) + suspend fun scheduleBroadcastNostrEventRequests(broadcastNostrEventRequests: List) + companion object { val NO_OP_NOSTR_REPOSITORY = object : NostrRepository { override suspend fun observeProfile(publicKey: HexKey): Flow { @@ -292,6 +294,10 @@ interface NostrRepository { ) { TODO("Not yet implemented") } + + override suspend fun scheduleBroadcastNostrEventRequests(broadcastNostrEventRequests: List) { + TODO("Not yet implemented") + } } } } \ No newline at end of file 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 4de0b39d..cc0b01a1 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 @@ -39,6 +39,8 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.launch +import kotlin.time.Clock +import kotlin.time.Duration.Companion.days class FeedListViewModel( initialFeedListUIState: FeedListUIState, diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/HomeViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/HomeViewModel.kt index 66f34795..ad902d04 100755 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/HomeViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/HomeViewModel.kt @@ -29,6 +29,8 @@ import com.vitorpamplona.quartz.nipB7Blossom.BlossomServersEvent import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.launch +import kotlin.time.Clock +import kotlin.time.Duration.Companion.days enum class HomeScreenType { Following, Mentions, Messages @@ -75,6 +77,9 @@ class HomeViewModel( ): SynchronizationFilter { val publicKey = SeedManager.activePublicKey().toHexKey() + val threeDaysAgo = Clock.System.now().minus(3.days) + val now = Clock.System.now() + return when (homeScreenType) { HomeScreenType.Following -> { SynchronizationFilter( @@ -83,7 +88,8 @@ class HomeViewModel( TextNoteEvent.KIND, RepostEvent.KIND, ), - limit = 50 + since = threeDaysAgo, + until = now ) } HomeScreenType.Mentions -> { @@ -102,7 +108,8 @@ class HomeViewModel( tags = mapOf( Pair("p", listOf(publicKey)) ), - limit = 50 + since = threeDaysAgo, + until = now ) } HomeScreenType.Messages -> { @@ -113,7 +120,8 @@ class HomeViewModel( kinds = arrayOf( LongTextNoteEvent.KIND, ), - limit = 50 + since = threeDaysAgo, + until = now ) } } 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 d5194e37..8be94651 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 @@ -1,7 +1,7 @@ package ac.cord.auxiliary.compose.ui.view.model +import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventRequest import ac.cord.auxiliary.compose.database.model.NostrEvent -import ac.cord.auxiliary.compose.database.model.SynchronizeNostrEventRequest import ac.cord.auxiliary.compose.managers.SeedManager import ac.cord.auxiliary.compose.network.dto.RelayDTO import ac.cord.auxiliary.compose.network.relays.RelayPool.Companion.PUBLISH_TIMEOUT @@ -242,7 +242,6 @@ class NavigationViewModel( storage, ) - val negOpenCmd = NegOpenCmd( subId = negentropySynchronizeRequest.id, filter = Filter( @@ -258,14 +257,12 @@ class NavigationViewModel( ), initialMessage = negentropy.initiate().toHexString() ) - logger.d("negOpenCmd: $negOpenCmd") - + logger.d("negOpenCmd: ${negOpenCmd.initialMessage}") nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest) scope.launch(Dispatchers.IO) { try { - relaysSocketManager.sync( negOpenCmd, negentropySynchronizeRequest.relayURL @@ -305,19 +302,42 @@ class NavigationViewModel( negentropySynchronizeRequest.relayURL ) } - // TODO: Handle negentropy error... -// nostrRepository.queueSynchronizeNostrEvent( -// Relays.eventPublishRelaySet.take(1).map { normalizedRelay -> // TODO: Sync from all the publish relays... -// SynchronizeNostrEventRequest( -// purpose = "feed", -// synchronizationFilters = arrayOf( -// synchronizationFilter -// ), -// relayURL = normalizedRelay.url, -// level = 0, -// ) -// } -// ) + is NostrIncomingMessage.NegentropyError -> { + // TODO: Handle negentropy error... +// nostrRepository.queueSynchronizeNostrEvent( +// Relays.eventPublishRelaySet.take(1).map { normalizedRelay -> // TODO: Sync from all the publish relays... +// SynchronizeNostrEventRequest( +// purpose = "feed", +// synchronizationFilters = arrayOf( +// synchronizationFilter +// ), +// relayURL = normalizedRelay.url, +// level = 0, +// ) +// } +// ) + } + 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 }}") + + val broadcastNostrEventRequests = result.sendIds.map { nostrEventId -> + BroadcastNostrEventRequest( + nostrEventId = nostrEventId.toHexString(), + relayURL = negentropySynchronizeRequest.relayURL + ) + } + nostrRepository.scheduleBroadcastNostrEventRequests( + broadcastNostrEventRequests + ) + } + else -> { logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage") }