Introduce foundation for negentropy logic.

2026-05-04 18:12:11 +02:00
parent 1b619cc29d
commit e8a42b0123
17 changed files with 666 additions and 14 deletions
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/AuxDatabase.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/AuxDatabase.kt
@@ -6,6 +6,8 @@ import ac.cord.auxiliary.compose.database.dao.BroadcastNostrEventRequestDao
 import ac.cord.auxiliary.compose.database.dao.ConnectionDao
 import ac.cord.auxiliary.compose.database.dao.InReplyToRelationDao
 import ac.cord.auxiliary.compose.database.dao.MentionDao
+import ac.cord.auxiliary.compose.database.dao.NegentropySynchronizeRequestDao
+import ac.cord.auxiliary.compose.database.dao.NegentropySynchronizeResultDao
 import ac.cord.auxiliary.compose.database.dao.NostrDao
 import ac.cord.auxiliary.compose.database.dao.NostrEventDao
 import ac.cord.auxiliary.compose.database.dao.PostDao
@@ -24,6 +26,8 @@ import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventRequest
 import ac.cord.auxiliary.compose.database.model.Connection
 import ac.cord.auxiliary.compose.database.model.InReplyToRelation
 import ac.cord.auxiliary.compose.database.model.Mention
+import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeRequest
+import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeResult
 import ac.cord.auxiliary.compose.database.model.NostrEvent
 import ac.cord.auxiliary.compose.database.model.Post
 import ac.cord.auxiliary.compose.database.model.Profile
@@ -52,6 +56,8 @@ val GENESIS_AT = Instant.fromEpochMilliseconds(1231006505000L)
        Connection::class,
        InReplyToRelation::class,
        Mention::class,
+        NegentropySynchronizeRequest::class,
+        NegentropySynchronizeResult::class,
        NostrEvent::class,
        Post::class,
        Profile::class,
@@ -77,6 +83,10 @@ abstract class AuxDatabase: RoomDatabase() {

    abstract fun mentionDao(): MentionDao

+    abstract fun negentropySynchronizeRequestDao(): NegentropySynchronizeRequestDao
+
+    abstract fun negentropySynchronizeResultDao(): NegentropySynchronizeResultDao
+
    abstract fun nostrDao(): NostrDao

    abstract fun nostrEventDao(): NostrEventDao
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/converters/AuxConverters.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/converters/AuxConverters.kt
@@ -95,4 +95,25 @@ class AuxConverters {
        logger.e("Failed to convert to FilterArray $value", e)
        return null
    }
+
+    @TypeConverter
+    fun fromSynchronizationFilter(value: SynchronizationFilter?): String? = try {
+        return value?.let {
+            Json.encodeToString(value)
+        }
+    } catch (e: Throwable) {
+        logger.e("Failed to convert from SynchronizationFilter $value", e)
+        return null
+    }
+
+    @TypeConverter
+    fun toSynchronizationFilter(value: String?): SynchronizationFilter? = try {
+        return value?.let {
+            Json.decodeFromString<SynchronizationFilter>(it)
+        }
+    } catch (e: Throwable) {
+        logger.e("Failed to convert to SynchronizationFilter $value", e)
+        return null
+    }
+
 }
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NegentropySynchronizeRequestDao.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NegentropySynchronizeRequestDao.kt
@@ -0,0 +1,26 @@
+package ac.cord.auxiliary.compose.database.dao
+
+import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeRequest
+import ac.cord.auxiliary.compose.database.model.SynchronizeNostrEventRequest
+import androidx.room3.Dao
+import androidx.room3.Insert
+import androidx.room3.Query
+import androidx.room3.Upsert
+import kotlinx.coroutines.flow.Flow
+import kotlin.time.Clock
+import kotlin.time.Instant
+
+@Dao
+interface NegentropySynchronizeRequestDao {
+    @Query("SELECT * FROM NegentropySynchronizeRequest WHERE status = :status AND createdAt > :createdAt")
+    fun observeNegentropySynchronizeRequestsByStatus(status: String, createdAt: Instant = Clock.System.now()): Flow<NegentropySynchronizeRequest?>
+
+    @Query("SELECT COUNT(*) FROM NegentropySynchronizeRequest WHERE purpose = :purpose AND status IN (:status) AND createdAt > :createdAt")
+    fun observeNegentropySynchronizeRequestByPurposeAndStatusCount(purpose: String, status: List<String>, createdAt: Instant = Clock.System.now()): Flow<Int>
+
+    @Upsert
+    fun upsert(negentropySynchronizeRequest: NegentropySynchronizeRequest)
+
+    @Insert
+    fun insert(negentropySynchronizeRequests: List<NegentropySynchronizeRequest>)
+}
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NegentropySynchronizeResultDao.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NegentropySynchronizeResultDao.kt
@@ -0,0 +1,16 @@
+package ac.cord.auxiliary.compose.database.dao
+
+import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeResult
+import androidx.room3.Dao
+import androidx.room3.Insert
+import androidx.room3.Query
+import androidx.room3.Upsert
+import kotlinx.coroutines.flow.Flow
+import kotlin.time.Clock
+import kotlin.time.Instant
+
+@Dao
+interface NegentropySynchronizeResultDao {
+    @Upsert
+    fun upsert(negentropySynchronizeResult: NegentropySynchronizeResult)
+}
--- 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
@@ -1,5 +1,6 @@
 package ac.cord.auxiliary.compose.database.dao

+import ac.cord.auxiliary.compose.database.GENESIS_AT
 import ac.cord.auxiliary.compose.database.model.NostrEvent
 import ac.cord.auxiliary.compose.database.model.intermdiate.LocalNostrEvent
 import androidx.room3.Dao
@@ -64,6 +65,50 @@ interface NostrEventDao {
        query: RoomRawQuery
    ): Flow<List<LocalNostrEvent>>

+    @Transaction
+    @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC")
+    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")
+    fun  getFilteredNostrEvents(
+        kinds: Array<Kind>,
+        search: String,
+        since: Instant = GENESIS_AT
+    ): List<NostrEvent>
+
+    @Transaction
+    @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND id in (:ids) AND createdAt > :since ORDER BY createdAt DESC")
+    fun  getFilteredNostrEvents(
+        kinds: Array<Kind>,
+        ids: Array<HexKey>,
+        since: Instant = GENESIS_AT
+    ): List<NostrEvent>
+
+    @Transaction
+    @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND pubKey in (:authors)  AND createdAt > :since ORDER BY createdAt DESC LIMIT 50")
+    fun  getAuthoredNostrEvents(
+        kinds: Array<Kind>,
+        authors: Array<HexKey>,
+        since: Instant = GENESIS_AT
+    ): List<NostrEvent>
+
+    @Transaction
+    @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :publicKey || '%' AND kind in (:kinds)  AND createdAt > :since ORDER BY createdAt DESC")
+    fun  getPublicKeyMentionedNostrEvents(
+        kinds: Array<Kind>,
+        publicKey: HexKey,
+        since: Instant = GENESIS_AT
+    ): List<NostrEvent>
+
+    @Transaction
+    @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :eventId || '%reply%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC")
+    fun  getNostrEventReplies(
+        kinds: Array<Kind>,
+        eventId: HexKey,
+        since: Instant
+    ): List<NostrEvent>
+
    @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) ORDER BY savedAt DESC")
    fun  getAllNostrEvents(kinds: Array<Kind>): List<NostrEvent>

--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/NegentropySynchronizeRequest.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/NegentropySynchronizeRequest.kt
@@ -0,0 +1,48 @@
+package ac.cord.auxiliary.compose.database.model
+
+import ac.cord.auxiliary.compose.database.model.traits.OptionalNostrEventEntity
+import ac.cord.auxiliary.compose.database.model.traits.TimestampedEntity
+import ac.cord.auxiliary.compose.database.model.types.SynchronizationFilter
+import androidx.room3.Entity
+import androidx.room3.ForeignKey
+import androidx.room3.Index
+import androidx.room3.PrimaryKey
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import kotlin.time.Clock
+import kotlin.time.Instant
+import kotlin.uuid.ExperimentalUuidApi
+import kotlin.uuid.Uuid
+
+@OptIn(ExperimentalUuidApi::class)
+@Entity(
+    foreignKeys = [
+        ForeignKey(
+            entity = NostrEvent::class,
+            parentColumns = ["id"],
+            childColumns = ["nostrEventId"],
+            onDelete = ForeignKey.Companion.CASCADE,
+        ),
+    ],
+    indices = [
+        Index("nostrEventId"),
+        Index("status"),
+    ],
+)
+data class NegentropySynchronizeRequest(
+    @PrimaryKey
+    val id: String = Uuid.Companion.generateV4().toHexDashString(),
+    val purpose: String,
+    val status: String = "pending",
+    val relayURL: String,
+    val isRecommendedRelay: Boolean = false,
+
+    val level: Int,
+
+    val synchronizationFilter: SynchronizationFilter,
+
+    override val nostrEventId: HexKey? = null,
+    override val createdAt: Instant = Clock.System.now(),
+    override val updatedAt: Instant = createdAt,
+): OptionalNostrEventEntity, TimestampedEntity {
+
+}
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/NegentropySynchronizeResult.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/NegentropySynchronizeResult.kt
@@ -0,0 +1,36 @@
+package ac.cord.auxiliary.compose.database.model
+
+import ac.cord.auxiliary.compose.database.model.traits.OptionalNostrEventEntity
+import ac.cord.auxiliary.compose.database.model.traits.TimestampedEntity
+import androidx.room3.Entity
+import androidx.room3.ForeignKey
+import androidx.room3.Index
+import androidx.room3.PrimaryKey
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+import kotlin.time.Clock
+import kotlin.time.Instant
+import kotlin.uuid.ExperimentalUuidApi
+import kotlin.uuid.Uuid
+
+@OptIn(ExperimentalUuidApi::class)
+@Entity(
+    foreignKeys = [
+        ForeignKey(
+            entity = SynchronizeNostrEventResult::class,
+            parentColumns = ["id"],
+            childColumns = ["negentropySynchronizeRequestId"],
+            onDelete = ForeignKey.CASCADE,
+        ),
+    ],
+)
+data class NegentropySynchronizeResult(
+    @PrimaryKey
+    val id: String = Uuid.generateV4().toHexDashString(),
+    val relayURL: String,
+    val status: String = "pending",
+
+    val negentropySynchronizeRequestId: String,
+
+    override val createdAt: Instant = Clock.System.now(),
+    override val updatedAt: Instant = createdAt,
+): TimestampedEntity
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/NostrEvent.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/model/NostrEvent.kt
@@ -296,7 +296,7 @@ data class NostrEvent(
    }

    fun toInReplyToRelation(): InReplyToRelation? = try {
-        if (kind == RepostEvent.KIND) {
+        if (kind == TextNoteEvent.KIND) {
            val textNoteEvent = EventFactory.create<TextNoteEvent>(
                id = id,
                pubKey = pubKey,
--- 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
@@ -5,6 +5,7 @@ import ac.cord.auxiliary.compose.database.GENESIS_AT
 import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventReceipt
 import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventRequest
 import ac.cord.auxiliary.compose.database.model.Connection
+import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeRequest
 import ac.cord.auxiliary.compose.database.model.NostrEvent
 import ac.cord.auxiliary.compose.database.model.Profile
 import ac.cord.auxiliary.compose.database.model.RecentSearch
@@ -106,6 +107,10 @@ class DatabaseNostrRepository(
        return database.synchronizeNostrEventRequestDao().observeSynchronizeNostrEventRequestsByStatus("pending")
    }

+    override suspend fun observePendingNegentropySynchronizeRequests(): Flow<NegentropySynchronizeRequest?> {
+        return database.negentropySynchronizeRequestDao().observeNegentropySynchronizeRequestsByStatus("pending")
+    }
+
    override suspend fun createNewProfile(
        publicKey: HexKey,
        name: String?,
@@ -404,6 +409,15 @@ class DatabaseNostrRepository(
        )
    }

+    override suspend fun negentropySynchronizeRequestProcessed(negentropySynchronizeRequest: NegentropySynchronizeRequest) {
+        database.negentropySynchronizeRequestDao().upsert(
+            negentropySynchronizeRequest.copy(
+                status = "sent",
+                updatedAt = Clock.System.now()
+            )
+        )
+    }
+
    override suspend fun saveBroadcastReceipt(broadcastNostrEventReceipt: BroadcastNostrEventReceipt) {
        database.broadcastNostrEventReceiptDao().upsert(
            broadcastNostrEventReceipt
@@ -430,7 +444,25 @@ class DatabaseNostrRepository(
                )
            )
        }
-        logger.d("done: ${nostrEvent.id}")
+    }
+
+    override suspend fun saveNostrEvent(
+        nostrEvent: NostrEvent,
+        negentropySynchronizeRequest: NegentropySynchronizeRequest,
+        synchronizationRelayURLs: List<String>
+    ) {
+        logger.d("saveNostrEvent: $nostrEvent")
+        database.nostrDao().storeNostrEvent(
+            nostrEvent,
+            synchronizationRelayURLs = synchronizationRelayURLs,
+            level = negentropySynchronizeRequest.level
+        )
+
+        database.negentropySynchronizeRequestDao().upsert(
+            negentropySynchronizeRequest.copy(
+                status = "processed"
+            )
+        )
    }

    override suspend fun queueSynchronizeNostrEvent(
@@ -468,7 +500,6 @@ class DatabaseNostrRepository(
    }

    override suspend fun observeNostrFeed(synchronizationFilter: SynchronizationFilter): Flow<List<LocalNostrEvent>> {
-
        return when {
            synchronizationFilter.kinds != null && synchronizationFilter.search != null -> {
                logger.d("observeFilteredNostrEvents: $synchronizationFilter")
@@ -517,6 +548,62 @@ class DatabaseNostrRepository(
        }
    }

+    override suspend fun getNostrFeedIds(synchronizationFilters: Array<SynchronizationFilter>): List<NostrEvent> {
+        if (synchronizationFilters.size == 1) {
+            val synchronizationFilter = synchronizationFilters.first()
+
+            return when {
+                synchronizationFilter.kinds != null && synchronizationFilter.search != null -> {
+                    logger.d("observeFilteredNostrEvents: $synchronizationFilter")
+                    database.nostrEventDao().getFilteredNostrEvents(
+                        kinds = synchronizationFilter.kinds,
+                        search = synchronizationFilter.search
+                    )
+                }
+                synchronizationFilter.kinds != null && synchronizationFilter.ids != null -> {
+                    logger.d("observeFilteredNostrEvents: $synchronizationFilter")
+                    database.nostrEventDao().getFilteredNostrEvents(
+                        kinds = synchronizationFilter.kinds,
+                        ids = synchronizationFilter.ids
+                    )
+                }
+                synchronizationFilter.kinds != null && synchronizationFilter.authors != null -> {
+                    logger.d("observeAuthoredNostrEvents: $synchronizationFilter")
+                    database.nostrEventDao().getAuthoredNostrEvents(
+                        kinds = synchronizationFilter.kinds,
+                        authors = synchronizationFilter.authors
+                    )
+                }
+                synchronizationFilter.kinds != null && synchronizationFilter.since != null &&  synchronizationFilter.tags?.contains("e") == true && synchronizationFilter.tags["e"]?.first()?.isNotEmpty() == true -> {
+                    logger.d("observePublicKeyMentionedNostrEvents: $synchronizationFilter")
+                    database.nostrEventDao().getNostrEventReplies(
+                        kinds = synchronizationFilter.kinds,
+                        eventId = synchronizationFilter.tags["e"]?.first()!!,
+                        since = synchronizationFilter.since
+                    )
+                }
+                synchronizationFilter.kinds != null && synchronizationFilter.tags?.contains("p") == true && synchronizationFilter.tags["p"]?.first()?.isNotEmpty() == true -> {
+                    logger.d("observePublicKeyMentionedNostrEvents: $synchronizationFilter")
+                    database.nostrEventDao().getPublicKeyMentionedNostrEvents(
+                        kinds = synchronizationFilter.kinds,
+                        publicKey = synchronizationFilter.tags["p"]?.first()!!
+                    )
+                }
+                else -> {
+                    logger.d("NostrEvent: $synchronizationFilter")
+                    database.nostrEventDao().getNostrEvents(
+                        kinds = arrayOf(
+                            TextNoteEvent.KIND,
+                        )
+                    )
+                }
+            }
+        } else {
+            logger.w("Multi synchronizationFilters not yet supported")
+            return emptyList()
+        }
+    }
+
    override suspend fun observeLocalNostrEventById(nostrEventId: String): Flow<LocalNostrEvent?> {
        return database.nostrEventDao().observeNostrEventById(nostrEventId)
    }
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/dto/RelayDTO.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/network/dto/RelayDTO.kt
@@ -8,8 +8,27 @@ import kotlinx.serialization.Serializable
 data class RelayDTO(
    val url: String,
    val read: Boolean,
-    val write: Boolean,
+    val write: Boolean
 ) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other == null || this::class != other::class) return false
+
+        other as RelayDTO
+
+        if (read != other.read) return false
+        if (write != other.write) return false
+        if (url != other.url) return false
+
+        return true
+    }
+
+    override fun hashCode(): Int {
+        var result = read.hashCode()
+        result = 31 * result + write.hashCode()
+        result = 31 * result + url.hashCode()
+        return result
+    }
 }

 fun Relay.mapToRelayDTO() =
--- 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.NegOpenCmd
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.transformWhile
@@ -126,14 +127,17 @@ class RelayPool(
        logger.d("toAddRelayUrls: $toAddRelayUrls")
        val toAddSocketClients = relays.filter { it.url in toAddRelayUrls }.mapAsNostrSocketClient()
        logger.d("toAddSocketClients: ${toAddSocketClients.map { it.socketUrl }}" )
-        val newSocketClients = socketClients.toMutableSet().apply {
-            addAll(toAddSocketClients)
+        if (toAddSocketClients.isNotEmpty()) {
+            val newSocketClients = socketClients.toMutableSet().apply {
+                addAll(toAddSocketClients)
+            }
+            logger.d("newSocketClients: ${newSocketClients.map { it.socketUrl }}")
+
+            socketClients = newSocketClients
+
+            this.relays.addAll(relays)
        }
-        logger.d("newSocketClients: ${newSocketClients.map { it.socketUrl }}")

-        socketClients = newSocketClients
-
-        this.relays.addAll(relays)
    }

    fun closePool() {
@@ -201,6 +205,30 @@ class RelayPool(
        }
    }

+    suspend fun sync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<NostrIncomingMessage> {
+        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 negentropySyncRequest = OptimizedJsonMapper.toJson(negOpenCmd)
+
+        if (nostrSocketClient == null) {
+            throw NetworkException("$relayUrl is not connected")
+        }
+        return coroutineScope {
+            val eventFlow =  nostrSocketClient.queryAsFlow(negOpenCmd.subId)
+            with(nostrSocketClient) {
+                sendMESSAGE(negentropySyncRequest)
+            }
+            eventFlow
+        }
+    }
+
    suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) {
        addRelaysIfMissing(
            setOf(
--- 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.NegOpenCmd
 import kotlinx.coroutines.flow.Flow


@@ -143,6 +144,13 @@ class RelaysSocketManager constructor(
        )
    }

+    suspend fun sync(negOpenCmd: NegOpenCmd, relayUrl: String): Flow<NostrIncomingMessage> {
+        return relayPool.sync(
+            negOpenCmd = negOpenCmd,
+            relayUrl = relayUrl
+        )
+    }
+
    suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) {
        return relayPool.closeQuery(
            closeCmd = closeCmd,
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/nostr/Relays.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/nostr/Relays.kt
@@ -35,7 +35,7 @@ object Relays {

    val bootstrapInboxRelaySet = setOf(damus, primal, mom, nos, bitcoiner, oxtr, yabu)
    val eventFinderRelaySet = setOf(wine, damus, mom, primal, nos, bitcoiner, oxtr)
-    val eventPublishRelaySet = setOf(primal, mom, nos, bitcoiner, oxtr)
+    val eventPublishRelaySet = setOf(damus, mom, nos, primal, bitcoiner, oxtr)


    val DefaultNIP65RelaySet = setOf(mom, nos, bitcoiner)
--- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt
+++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt
@@ -3,6 +3,7 @@ package ac.cord.auxiliary.compose.repository
 import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventReceipt
 import ac.cord.auxiliary.compose.database.model.BroadcastNostrEventRequest
 import ac.cord.auxiliary.compose.database.model.Connection
+import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeRequest
 import ac.cord.auxiliary.compose.database.model.NostrEvent
 import ac.cord.auxiliary.compose.database.model.Profile
 import ac.cord.auxiliary.compose.database.model.RecentSearch
@@ -18,7 +19,6 @@ import ac.cord.auxiliary.compose.database.model.typealiases.SynchronizationFilte
 import ac.cord.auxiliary.compose.database.model.types.SynchronizationFilter
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.core.Kind
-import fr.acinq.secp256k1.Hex
 import kotlinx.coroutines.flow.Flow

 interface NostrRepository {
@@ -34,6 +34,8 @@ interface NostrRepository {

    suspend fun observePendingSynchronizeNostrEventRequests(): Flow<SynchronizeNostrEventRequest?>

+    suspend fun observePendingNegentropySynchronizeRequests(): Flow<NegentropySynchronizeRequest?>
+
    suspend fun createNewProfile(
        publicKey: HexKey,
        name: String?,
@@ -69,6 +71,8 @@ interface NostrRepository {

    suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest)

+    suspend fun negentropySynchronizeRequestProcessed(negentropySynchronizeRequest: NegentropySynchronizeRequest)
+
    suspend fun saveBroadcastReceipt(broadcastNostrEventReceipt: BroadcastNostrEventReceipt)

    suspend fun saveNostrEvent(
@@ -77,6 +81,12 @@ interface NostrRepository {
        synchronizationRelayURLs: List<String>
    )

+    suspend fun saveNostrEvent(
+        nostrEvent: NostrEvent,
+        negentropySynchronizeRequest: NegentropySynchronizeRequest,
+        synchronizationRelayURLs: List<String>
+    )
+
    suspend fun queueSynchronizeNostrEvent(
        synchronizeNostrEventRequests: List<SynchronizeNostrEventRequest>,
    )
@@ -90,6 +100,8 @@ interface NostrRepository {

    suspend fun observeNostrFeed(synchronizationFilter: SynchronizationFilter): Flow<List<LocalNostrEvent>>

+    suspend fun getNostrFeedIds(synchronizationFilters: Array<SynchronizationFilter>): List<NostrEvent>
+
    suspend fun observeLocalNostrEventById(nostrEventId: String): Flow<LocalNostrEvent?>

    suspend fun searchableProfiles(): List<Profile>
@@ -135,6 +147,10 @@ interface NostrRepository {
                TODO("Not yet implemented")
            }

+            override suspend fun observePendingNegentropySynchronizeRequests(): Flow<NegentropySynchronizeRequest?> {
+                TODO("Not yet implemented")
+            }
+
            override suspend fun createNewProfile(
                publicKey: HexKey,
                name: String?,
@@ -183,6 +199,10 @@ interface NostrRepository {
                TODO("Not yet implemented")
            }

+            override suspend fun negentropySynchronizeRequestProcessed(negentropySynchronizeRequest: NegentropySynchronizeRequest) {
+                TODO("Not yet implemented")
+            }
+
            override suspend fun saveBroadcastReceipt(broadcastNostrEventReceipt: BroadcastNostrEventReceipt) {
                TODO("Not yet implemented")
            }
@@ -195,6 +215,14 @@ interface NostrRepository {
                TODO("Not yet implemented")
            }

+            override suspend fun saveNostrEvent(
+                nostrEvent: NostrEvent,
+                negentropySynchronizeRequest: NegentropySynchronizeRequest,
+                synchronizationRelayURLs: List<String>
+            ) {
+                TODO("Not yet implemented")
+            }
+
            override suspend fun queueSynchronizeNostrEvent(synchronizeNostrEventRequests: List<SynchronizeNostrEventRequest>) {
                TODO("Not yet implemented")
            }
@@ -218,6 +246,10 @@ interface NostrRepository {
                TODO("Not yet implemented")
            }

+            override suspend fun getNostrFeedIds(synchronizationFilters: Array<SynchronizationFilter>): List<NostrEvent> {
+                TODO("Not yet implemented")
+            }
+
            override suspend fun observeLocalNostrEventById(nostrEventId: String): Flow<LocalNostrEvent?> {
                TODO("Not yet implemented")
            }
--- 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
@@ -17,6 +17,8 @@ import androidx.lifecycle.ViewModelProvider
 import androidx.lifecycle.viewmodel.initializer
 import androidx.lifecycle.viewmodel.viewModelFactory
 import co.touchlab.kermit.Logger
+import com.vitorpamplona.negentropy.Negentropy
+import com.vitorpamplona.negentropy.storage.StorageVector
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.toHexKey
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
@@ -24,6 +26,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.NegOpenCmd
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.FlowPreview
@@ -84,6 +87,7 @@ class NavigationViewModel(
        observePendingBroadcastNostrEventRequests()
        observeProfile()
        observePendingSyncNostrEventRequests()
+        observePendingNegentropySynchronizeRequests()
    }


@@ -202,8 +206,114 @@ class NavigationViewModel(
                            logger.e("Failed to sync", e)
                        }
                    }
+                }
+            }
+        }
+    }
+
+    private fun observePendingNegentropySynchronizeRequests() {
+        logger.i { "observePendingNegentropySynchronizeRequests" }
+
+        scope.launch(Dispatchers.IO) {
+            nostrRepository.observePendingNegentropySynchronizeRequests().distinctUntilChanged().collect { negentropySynchronizeRequestOrNull ->
+                negentropySynchronizeRequestOrNull?.let { negentropySynchronizeRequest ->
+                    logger.i("synchronizeNostrEventRequest: $negentropySynchronizeRequest")
+                    // TODO: Negentropy...
+
+                    val events = nostrRepository.getNostrFeedIds(
+                        arrayOf(
+                            negentropySynchronizeRequest.synchronizationFilter
+                        )
+                    )
+                    logger.d("Events: ${events.size}")
+                    val storage = StorageVector().apply {
+                        events.forEach { event ->
+                            insert(
+                                event.createdAt.toEpochMilliseconds(),
+                                event.id
+                            )
+                        }
+
+                        seal()
+                    }
+                    logger.d("Negentropy Storage: $storage")
+                    val negentropy = Negentropy(
+                        storage,
+                    )


+                    val negOpenCmd = NegOpenCmd(
+                        subId = negentropySynchronizeRequest.id,
+                        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 = negentropySynchronizeRequest.synchronizationFilter.limit,
+                            search = negentropySynchronizeRequest.synchronizationFilter.search
+                        ),
+                        initialMessage = negentropy.initiate().toHexString()
+                    )
+                    logger.d("negOpenCmd: $negOpenCmd")
+
+
+                    nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest)
+
+                    scope.launch(Dispatchers.IO) {
+                        try {
+
+                            relaysSocketManager.sync(
+                                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")
+
+                                        nostrIncomingMessage.nostrEvents.forEach { nostrEvent ->
+                                            nostrRepository.saveNostrEvent(
+                                                nostrEvent = nostrEvent,
+                                                negentropySynchronizeRequest,
+                                                synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
+                                            )
+                                        }
+                                    }
+                                    is NostrIncomingMessage.EoseMessage -> {
+                                        logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage")
+                                        val closeCommand = CloseCmd(
+                                            subId = negentropySynchronizeRequest.id,
+                                        )
+
+                                        relaysSocketManager.closeQuery(
+                                            closeCommand,
+                                            negentropySynchronizeRequest.relayURL
+                                        )
+                                    }
+                                    else -> {
+                                        logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage")
+                                    }
+                                }
+                            }
+
+                        } catch (e: Throwable) {
+                            logger.e("Failed to sync", e)
+                        }
+                    }
                }
            }
        }