mantra/mantra-kmp
1
0
Fork 0
Code Issues 2 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

Initial negentropic logic (with broadcasting logic)

Browse Source
This commit is contained in:
Kgothatso Ngako
2026-05-04 21:18:52 +02:00
parent 4fbbfc72f7
commit 1c8b280673
11 changed files with 156 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/BroadcastNostrEventRequestDao.kt
View File

@@ -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<BroadcastNostrEventRequest>)
}

12
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt
View File

@@ -66,11 +66,11 @@ interface NostrEventDao {
): Flow<List<LocalNostrEvent>>
@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<Kind>, since: Instant = GENESIS_AT): List<NostrEvent>
@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<Kind>,
search: String,
@@ -78,7 +78,7 @@ interface NostrEventDao {
): List<NostrEvent>
@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<Kind>,
ids: Array<HexKey>,
@@ -86,7 +86,7 @@ interface NostrEventDao {
): List<NostrEvent>
@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<Kind>,
authors: Array<HexKey>,
@@ -94,7 +94,7 @@ interface NostrEventDao {
): List<NostrEvent>
@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<Kind>,
publicKey: HexKey,
@@ -102,7 +102,7 @@ interface NostrEventDao {
): List<NostrEvent>
@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<Kind>,
eventId: HexKey,

6
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt
View File

@@ -653,4 +653,10 @@ class DatabaseNostrRepository(
override suspend fun observePublicKeyRelays(publicKey: String): Flow<List<Relay>> {
return database.relayDao().observePublicKeyRelays(publicKey)
}
override suspend fun scheduleBroadcastNostrEventRequests(broadcastNostrEventRequests: List<BroadcastNostrEventRequest>) {
return database.broadcastNostrEventRequestDao().insert(
broadcastNostrEventRequests
)
}
}

14
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessage.kt
View File

@@ -39,6 +39,20 @@ sealed class NostrIncomingMessage {
val subscriptionId: String,
val nostrEvents: List<NostrEvent> = 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) {

13
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageExt.kt
View File

@@ -5,11 +5,16 @@ import kotlinx.coroutines.flow.filter
fun Flow<NostrIncomingMessage>.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<NostrIncomingMessage>.filterByEventId(id: String) =

52
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrIncomingMessageParser.kt
View File

@@ -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

9
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/sockets/NostrVerb.kt
View File

@@ -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,
}
}

6
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt
View File

@@ -122,6 +122,8 @@ interface NostrRepository {
suspend fun saveSearchQuery(query: String, synchronizationFilterArray: SynchronizationFilterArray? = null)
suspend fun scheduleBroadcastNostrEventRequests(broadcastNostrEventRequests: List<BroadcastNostrEventRequest>)
companion object {
val NO_OP_NOSTR_REPOSITORY = object : NostrRepository {
override suspend fun observeProfile(publicKey: HexKey): Flow<LocalAccount?> {
@@ -292,6 +294,10 @@ interface NostrRepository {
) {
TODO("Not yet implemented")
}
override suspend fun scheduleBroadcastNostrEventRequests(broadcastNostrEventRequests: List<BroadcastNostrEventRequest>) {
TODO("Not yet implemented")
}
}
}
}

2
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/FeedListViewModel.kt
View File

@@ -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,

14
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/HomeViewModel.kt
View File

@@ -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
)
}
}

56
composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt
View File

@@ -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")
}