Synchronization workflow

2026-03-29 22:22:16 +02:00
parent da62bede43
commit daa6cb11a7
11 changed files with 609 additions and 16 deletions
--- a/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json
+++ b/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json
@@ -2,7 +2,7 @@
  "formatVersion": 1,
  "database": {
    "version": 1,
-    "identityHash": "866bd337c819985dae51a248851b3dd3",
+    "identityHash": "75e5c80b2c3183c8a4fa300627ffb5db",
    "entities": [
      {
        "tableName": "BroadcastNostrEventReceipt",
@@ -871,6 +871,202 @@
          }
        ]
      },
+      {
+        "tableName": "SynchronizeNostrEventRequest",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `status` TEXT NOT NULL, `relayURL` TEXT NOT NULL, `eventIds` TEXT, `authorPublicKeys` TEXT, `kinds` TEXT, `tagName` TEXT, `since` INTEGER, `until` INTEGER, `limit` INTEGER, `nostrEventId` TEXT, `unsignedNostrEventId` INTEGER, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, PRIMARY KEY(`id`), FOREIGN KEY(`nostrEventId`) REFERENCES `NostrEvent`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE , FOREIGN KEY(`unsignedNostrEventId`) REFERENCES `UnsignedNostrEvent`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )",
+        "fields": [
+          {
+            "fieldPath": "id",
+            "columnName": "id",
+            "affinity": "TEXT",
+            "notNull": true
+          },
+          {
+            "fieldPath": "status",
+            "columnName": "status",
+            "affinity": "TEXT",
+            "notNull": true
+          },
+          {
+            "fieldPath": "relayURL",
+            "columnName": "relayURL",
+            "affinity": "TEXT",
+            "notNull": true
+          },
+          {
+            "fieldPath": "eventIds",
+            "columnName": "eventIds",
+            "affinity": "TEXT"
+          },
+          {
+            "fieldPath": "authorPublicKeys",
+            "columnName": "authorPublicKeys",
+            "affinity": "TEXT"
+          },
+          {
+            "fieldPath": "kinds",
+            "columnName": "kinds",
+            "affinity": "TEXT"
+          },
+          {
+            "fieldPath": "tagName",
+            "columnName": "tagName",
+            "affinity": "TEXT"
+          },
+          {
+            "fieldPath": "since",
+            "columnName": "since",
+            "affinity": "INTEGER"
+          },
+          {
+            "fieldPath": "until",
+            "columnName": "until",
+            "affinity": "INTEGER"
+          },
+          {
+            "fieldPath": "limit",
+            "columnName": "limit",
+            "affinity": "INTEGER"
+          },
+          {
+            "fieldPath": "nostrEventId",
+            "columnName": "nostrEventId",
+            "affinity": "TEXT"
+          },
+          {
+            "fieldPath": "unsignedNostrEventId",
+            "columnName": "unsignedNostrEventId",
+            "affinity": "INTEGER"
+          },
+          {
+            "fieldPath": "createdAt",
+            "columnName": "createdAt",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "updatedAt",
+            "columnName": "updatedAt",
+            "affinity": "INTEGER",
+            "notNull": true
+          }
+        ],
+        "primaryKey": {
+          "autoGenerate": false,
+          "columnNames": [
+            "id"
+          ]
+        },
+        "indices": [
+          {
+            "name": "index_SynchronizeNostrEventRequest_nostrEventId",
+            "unique": false,
+            "columnNames": [
+              "nostrEventId"
+            ],
+            "orders": [],
+            "createSql": "CREATE INDEX IF NOT EXISTS `index_SynchronizeNostrEventRequest_nostrEventId` ON `${TABLE_NAME}` (`nostrEventId`)"
+          },
+          {
+            "name": "index_SynchronizeNostrEventRequest_status",
+            "unique": false,
+            "columnNames": [
+              "status"
+            ],
+            "orders": [],
+            "createSql": "CREATE INDEX IF NOT EXISTS `index_SynchronizeNostrEventRequest_status` ON `${TABLE_NAME}` (`status`)"
+          }
+        ],
+        "foreignKeys": [
+          {
+            "table": "NostrEvent",
+            "onDelete": "CASCADE",
+            "onUpdate": "NO ACTION",
+            "columns": [
+              "nostrEventId"
+            ],
+            "referencedColumns": [
+              "id"
+            ]
+          },
+          {
+            "table": "UnsignedNostrEvent",
+            "onDelete": "CASCADE",
+            "onUpdate": "NO ACTION",
+            "columns": [
+              "unsignedNostrEventId"
+            ],
+            "referencedColumns": [
+              "id"
+            ]
+          }
+        ]
+      },
+      {
+        "tableName": "SynchronizeNostrEventResult",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `relayURL` TEXT NOT NULL, `nostrEventId` TEXT, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, PRIMARY KEY(`id`), FOREIGN KEY(`nostrEventId`) REFERENCES `NostrEvent`(`id`) ON UPDATE NO ACTION ON DELETE CASCADE )",
+        "fields": [
+          {
+            "fieldPath": "id",
+            "columnName": "id",
+            "affinity": "TEXT",
+            "notNull": true
+          },
+          {
+            "fieldPath": "relayURL",
+            "columnName": "relayURL",
+            "affinity": "TEXT",
+            "notNull": true
+          },
+          {
+            "fieldPath": "nostrEventId",
+            "columnName": "nostrEventId",
+            "affinity": "TEXT"
+          },
+          {
+            "fieldPath": "createdAt",
+            "columnName": "createdAt",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "updatedAt",
+            "columnName": "updatedAt",
+            "affinity": "INTEGER",
+            "notNull": true
+          }
+        ],
+        "primaryKey": {
+          "autoGenerate": false,
+          "columnNames": [
+            "id"
+          ]
+        },
+        "indices": [
+          {
+            "name": "index_SynchronizeNostrEventResult_nostrEventId",
+            "unique": false,
+            "columnNames": [
+              "nostrEventId"
+            ],
+            "orders": [],
+            "createSql": "CREATE INDEX IF NOT EXISTS `index_SynchronizeNostrEventResult_nostrEventId` ON `${TABLE_NAME}` (`nostrEventId`)"
+          }
+        ],
+        "foreignKeys": [
+          {
+            "table": "NostrEvent",
+            "onDelete": "CASCADE",
+            "onUpdate": "NO ACTION",
+            "columns": [
+              "nostrEventId"
+            ],
+            "referencedColumns": [
+              "id"
+            ]
+          }
+        ]
+      },
      {
        "tableName": "UnsignedNostrEvent",
        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `pubKey` TEXT NOT NULL, `kind` INTEGER NOT NULL, `tags` TEXT NOT NULL, `content` TEXT NOT NULL, `signedAt` INTEGER, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, `savedAt` INTEGER NOT NULL, `deletedAt` INTEGER, `broadcastedAt` INTEGER)",
@@ -1134,7 +1330,7 @@
    ],
    "setupQueries": [
      "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)",
-      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '866bd337c819985dae51a248851b3dd3')"
+      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '75e5c80b2c3183c8a4fa300627ffb5db')"
    ]
  }
 }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt
@@ -9,6 +9,8 @@ import ac.aux.compose.database.dao.PostDao
 import ac.aux.compose.database.dao.ProfileDao
 import ac.aux.compose.database.dao.ReactionDao
 import ac.aux.compose.database.dao.RepostDao
+import ac.aux.compose.database.dao.SynchronizeNostrEventRequestDao
+import ac.aux.compose.database.dao.SynchronizeNostrEventResultDao
 import ac.aux.compose.database.dao.UnsignedNostrEventDao
 import ac.aux.compose.database.dao.ZapDao
 import ac.aux.compose.database.model.BroadcastNostrEventReceipt
@@ -18,6 +20,8 @@ import ac.aux.compose.database.model.Post
 import ac.aux.compose.database.model.Profile
 import ac.aux.compose.database.model.Reaction
 import ac.aux.compose.database.model.Repost
+import ac.aux.compose.database.model.SynchronizeNostrEventRequest
+import ac.aux.compose.database.model.SynchronizeNostrEventResult
 import ac.aux.compose.database.model.UnsignedNostrEvent
 import ac.aux.compose.database.model.Zap
 import androidx.room.Database
@@ -35,6 +39,8 @@ import androidx.room.useWriterConnection
        Profile::class,
        Reaction::class,
        Repost::class,
+        SynchronizeNostrEventRequest::class,
+        SynchronizeNostrEventResult::class,
        UnsignedNostrEvent::class,
        Zap::class
    ],
@@ -52,6 +58,10 @@ abstract class AuxDatabase: RoomDatabase() {
    abstract fun reactionDao(): ReactionDao
    abstract fun repostDao(): RepostDao

+    abstract fun synchronizeNostrEventRequestDao(): SynchronizeNostrEventRequestDao
+
+    abstract fun synchronizeNostrEventResultDao(): SynchronizeNostrEventResultDao
+
    abstract fun unsignedNostrEventDao(): UnsignedNostrEventDao
    abstract fun zapDao(): ZapDao

--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt
@@ -5,6 +5,8 @@ import ac.aux.compose.database.model.BroadcastNostrEventRequest
 import ac.aux.compose.database.model.NostrEvent
 import ac.aux.compose.database.model.Post
 import ac.aux.compose.database.model.Profile
+import ac.aux.compose.database.model.SynchronizeNostrEventRequest
+import ac.aux.compose.database.model.SynchronizeNostrEventResult
 import ac.aux.compose.database.model.UnsignedNostrEvent
 import androidx.room.Dao
 import androidx.room.Embedded
@@ -59,13 +61,87 @@ abstract class NostrDao(

        database.nostrEventDao().upsert(nostrEvent)

+        indexNostrEvent(
+            nostrEvent = nostrEvent,
+            synchronizationRelayURLs = relayURLs
+        )
+
+        relayURLs.forEach { relayURL ->
+            database.broadcastNostrEventRequestDao().upsert(
+                BroadcastNostrEventRequest(
+                    nostrEventId = nostrEvent.id,
+                    unsignedNostrEventId = nostrEvent.unsignedNostrEventId,
+                    relayURL = relayURL
+                )
+            )
+        }
+    }
+
+    @Transaction
+    open suspend fun storeNostrEvent(
+        nostrEvent: NostrEvent,
+        synchronizeNostrEventRequest: SynchronizeNostrEventRequest,
+        synchronizationRelayURLs: List<String>
+    ) {
+        logger.i("Store Nostr Event: $nostrEvent")
+        // Find or create profile with the pubKey... if not found submit a sync request...
+        val profile = database.profileDao().getProfileByPublicKey(nostrEvent.pubKey)
+
+        if (profile == null) {
+            val placeHolderProfile = Profile(
+                publicKey = nostrEvent.pubKey,
+                nostrEventId = nostrEvent.pubKey // This is illegal... because the nostrEventId should be the one with all the profile information
+            )
+            val placeHolderProfileNostrEvent = NostrEvent(
+                id = placeHolderProfile.publicKey,
+                kind = 30024,
+                pubKey = placeHolderProfile.publicKey,
+                content = placeHolderProfile.publicKey,
+                createdAt = Clock.System.now(),
+                sig = placeHolderProfile.publicKey,
+                tags = emptyArray(),
+            )
+            database.nostrEventDao().upsert(placeHolderProfileNostrEvent)
+            database.profileDao().upsert(placeHolderProfile)
+
+            synchronizationRelayURLs.forEach { synchronizationRelayURL ->
+                database.synchronizeNostrEventRequestDao().upsert(
+                    SynchronizeNostrEventRequest(
+                        authorPublicKeys = listOf<String>(nostrEvent.pubKey).joinToString(","),
+                        relayURL = synchronizationRelayURL
+                    )
+                )
+            }
+        }
+
+        database.nostrEventDao().upsert(nostrEvent)
+
+        // Index nostrEvent
+        indexNostrEvent(
+            nostrEvent = nostrEvent,
+            synchronizationRelayURLs = synchronizationRelayURLs
+        )
+
+        // Save synchronizationResult
+        database.synchronizeNostrEventResultDao().upsert(
+            SynchronizeNostrEventResult(
+                nostrEventId = nostrEvent.id,
+                relayURL = synchronizeNostrEventRequest.relayURL
+            )
+        )
+    }
+
+    private suspend fun indexNostrEvent(
+        nostrEvent: NostrEvent,
+        synchronizationRelayURLs: List<String>
+    ) {
        nostrEvent.toPost()?.let { post ->
            if (post.replyToId != null) {
                // Find or create placeHolder note that is replyTo
                val replyingToPost = database.postDao().getPostById(post.replyToId)

                if (replyingToPost == null) {
-                    // PlaceHolder and sync
+                    // Persist PlaceHolder and sync
                    database.postDao().insert(
                        Post(
                            id = post.replyToId,
@@ -74,6 +150,16 @@ abstract class NostrDao(
                            profilePublicKey = nostrEvent.pubKey // Will get overwriting by sync,
                        )
                    )
+
+                    // Request a sync for the post being replied to
+                    synchronizationRelayURLs.forEach { synchronizationRelayURL ->
+                        database.synchronizeNostrEventRequestDao().upsert(
+                            SynchronizeNostrEventRequest(
+                                eventIds = listOf(post.replyToId).joinToString(","),
+                                relayURL = synchronizationRelayURL
+                            )
+                        )
+                    }
                }
            }

@@ -90,7 +176,15 @@ abstract class NostrDao(
                            profilePublicKey = nostrEvent.pubKey // Will get overwriting by sync,
                        )
                    )
-                    // Sync repostId
+                    // Request a sync for the post being reposted
+                    synchronizationRelayURLs.forEach { synchronizationRelayURL ->
+                        database.synchronizeNostrEventRequestDao().upsert(
+                            SynchronizeNostrEventRequest(
+                                eventIds = listOf(post.repostId).joinToString(","),
+                                relayURL = synchronizationRelayURL
+                            )
+                        )
+                    }
                }
            }
            database.postDao().insert(post)
@@ -111,15 +205,5 @@ abstract class NostrDao(
        nostrEvent.toZap()?.let { zap ->
            database.zapDao().upsert(zap)
        }
-
-        relayURLs.forEach { relayURL ->
-            database.broadcastNostrEventRequestDao().upsert(
-                BroadcastNostrEventRequest(
-                    nostrEventId = nostrEvent.id,
-                    unsignedNostrEventId = nostrEvent.unsignedNostrEventId,
-                    relayURL = relayURL
-                )
-            )
-        }
    }
 }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventRequestDao.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventRequestDao.kt
@@ -0,0 +1,16 @@
+package ac.aux.compose.database.dao
+
+import ac.aux.compose.database.model.SynchronizeNostrEventRequest
+import androidx.room.Dao
+import androidx.room.Query
+import androidx.room.Upsert
+import kotlinx.coroutines.flow.Flow
+
+@Dao
+interface SynchronizeNostrEventRequestDao {
+    @Query("SELECT * FROM SynchronizeNostrEventRequest WHERE status = :status")
+    fun observeSynchronizeNostrEventRequestsByStatus(status: String): Flow<SynchronizeNostrEventRequest?>
+
+    @Upsert
+    fun upsert(synchronizeNostrEventRequest: SynchronizeNostrEventRequest)
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventResultDao.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventResultDao.kt
@@ -0,0 +1,14 @@
+package ac.aux.compose.database.dao
+
+import ac.aux.compose.database.model.SynchronizeNostrEventRequest
+import ac.aux.compose.database.model.SynchronizeNostrEventResult
+import androidx.room.Dao
+import androidx.room.Query
+import androidx.room.Upsert
+import kotlinx.coroutines.flow.Flow
+
+@Dao
+interface SynchronizeNostrEventResultDao {
+    @Upsert
+    fun upsert(synchronizeNostrEventResult: SynchronizeNostrEventResult)
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventRequest.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventRequest.kt
@@ -0,0 +1,55 @@
+package ac.aux.compose.database.model
+
+import ac.aux.compose.database.model.traits.NostrEventEntity
+import ac.aux.compose.database.model.traits.OptionalNostrEventEntity
+import ac.aux.compose.database.model.traits.TimestampedEntity
+import ac.aux.compose.database.model.traits.UnsignedNostrEventEntity
+import androidx.room.Entity
+import androidx.room.ForeignKey
+import androidx.room.Index
+import androidx.room.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.CASCADE,
+        ),
+        ForeignKey(
+            entity = UnsignedNostrEvent::class,
+            parentColumns = ["id"],
+            childColumns = ["unsignedNostrEventId"],
+            onDelete = ForeignKey.CASCADE,
+        ),
+    ],
+    indices = [
+        Index("nostrEventId"),
+        Index("status"),
+    ],
+)
+data class SynchronizeNostrEventRequest(
+    @PrimaryKey
+    val id: String = Uuid.generateV4().toHexDashString(),
+    val status: String = "pending",
+    val relayURL: String,
+    val eventIds: String? = null, // TODO: EventIdArray
+    val authorPublicKeys: String? = null, // TODO: PublicKeyArray
+    val kinds: String? = null, // TODO: KindArray
+    val tagName: String? = null,
+    val since: Instant? = null,
+    val until: Instant? = null,
+    val limit: Long? = null,
+
+    override val nostrEventId: HexKey? = null,
+    override val unsignedNostrEventId: Long? = null,
+    override val createdAt: Instant = Clock.System.now(),
+    override val updatedAt: Instant = createdAt,
+): OptionalNostrEventEntity, UnsignedNostrEventEntity, TimestampedEntity
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventResult.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventResult.kt
@@ -0,0 +1,40 @@
+package ac.aux.compose.database.model
+
+import ac.aux.compose.database.model.traits.NostrEventEntity
+import ac.aux.compose.database.model.traits.OptionalNostrEventEntity
+import ac.aux.compose.database.model.traits.TimestampedEntity
+import ac.aux.compose.database.model.traits.UnsignedNostrEventEntity
+import androidx.room.Entity
+import androidx.room.ForeignKey
+import androidx.room.Index
+import androidx.room.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.CASCADE,
+        ),
+    ],
+    indices = [
+        Index("nostrEventId"),
+    ],
+)
+data class SynchronizeNostrEventResult(
+    @PrimaryKey
+    val id: String = Uuid.generateV4().toHexDashString(),
+    val relayURL: String,
+
+    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/aux/compose/database/model/traits/OptionalNostrEventEntity.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/traits/OptionalNostrEventEntity.kt
@@ -0,0 +1,7 @@
+package ac.aux.compose.database.model.traits
+
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+
+interface OptionalNostrEventEntity {
+    val nostrEventId: HexKey?
+}
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt
@@ -5,6 +5,7 @@ import ac.aux.compose.database.model.BroadcastNostrEventReceipt
 import ac.aux.compose.database.model.BroadcastNostrEventRequest
 import ac.aux.compose.database.model.NostrEvent
 import ac.aux.compose.database.model.Profile
+import ac.aux.compose.database.model.SynchronizeNostrEventRequest
 import ac.aux.compose.database.model.UnsignedNostrEvent
 import ac.aux.compose.database.model.intermdiate.LocalBroadcastNostrEventRequest
 import ac.aux.compose.database.model.intermdiate.LocalProfile
@@ -39,6 +40,10 @@ class DatabaseNostrRepository(
        return database.broadcastNostrEventRequestDao().observeBroadcastNostrEventRequestsByStatus("pending")
    }

+    override suspend fun observePendingSynchronizeNostrEventRequests(): Flow<SynchronizeNostrEventRequest?> {
+        return database.synchronizeNostrEventRequestDao().observeSynchronizeNostrEventRequestsByStatus("pending")
+    }
+
    override suspend fun createNewProfile(
        publicKey: HexKey,
        name: String?,
@@ -144,10 +149,33 @@ class DatabaseNostrRepository(
        logger.d("Processed: $broadcastNostrEventRequest")
    }

+    override suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) {
+        logger.i("Update local reference: $synchronizeNostrEventRequest")
+        database.synchronizeNostrEventRequestDao().upsert(
+            synchronizeNostrEventRequest.copy(
+                status = "sent",
+                updatedAt = Clock.System.now()
+            )
+        )
+        logger.i("Processed: $synchronizeNostrEventRequest")
+    }
+
    override suspend fun saveBroadcastReceipt(broadcastNostrEventReceipt: BroadcastNostrEventReceipt) {
        database.broadcastNostrEventReceiptDao().upsert(
            broadcastNostrEventReceipt
        )
    }

+    override suspend fun saveNostrEvent(
+        nostrEvent: NostrEvent,
+        synchronizeNostrEventRequest: SynchronizeNostrEventRequest,
+        synchronizationRelayURLs: List<String>
+    ) {
+        database.nostrDao().storeNostrEvent(
+            nostrEvent,
+            synchronizeNostrEventRequest,
+            synchronizationRelayURLs = synchronizationRelayURLs
+        )
+    }
+
 }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt
@@ -4,6 +4,7 @@ import ac.aux.compose.database.model.BroadcastNostrEventReceipt
 import ac.aux.compose.database.model.BroadcastNostrEventRequest
 import ac.aux.compose.database.model.NostrEvent
 import ac.aux.compose.database.model.Profile
+import ac.aux.compose.database.model.SynchronizeNostrEventRequest
 import ac.aux.compose.database.model.UnsignedNostrEvent
 import ac.aux.compose.database.model.intermdiate.LocalBroadcastNostrEventRequest
 import ac.aux.compose.database.model.intermdiate.LocalProfile
@@ -17,6 +18,8 @@ interface NostrRepository {

    suspend fun observePendingBroadcastNostrEventRequests(): Flow<List<LocalBroadcastNostrEventRequest>>

+    suspend fun observePendingSynchronizeNostrEventRequests(): Flow<SynchronizeNostrEventRequest?>
+
    suspend fun createNewProfile(
        publicKey: HexKey,
        name: String?,
@@ -46,8 +49,16 @@ interface NostrRepository {

    suspend fun broadcastProcessed(broadcastNostrEventRequest: BroadcastNostrEventRequest)

+    suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest)
+
    suspend fun saveBroadcastReceipt(broadcastNostrEventReceipt: BroadcastNostrEventReceipt)

+    suspend fun saveNostrEvent(
+        nostrEvent: NostrEvent,
+        synchronizeNostrEventRequest: SynchronizeNostrEventRequest,
+        synchronizationRelayURLs: List<String>
+    )
+
    companion object {
        val NO_OP_NOSTR_REPOSITORY = object : NostrRepository {
            override suspend fun observeProfile(publicKey: HexKey): Flow<LocalProfile?> {
@@ -62,6 +73,10 @@ interface NostrRepository {
                TODO("Not yet implemented")
            }

+            override suspend fun observePendingSynchronizeNostrEventRequests(): Flow<SynchronizeNostrEventRequest?> {
+                TODO("Not yet implemented")
+            }
+
            override suspend fun createNewProfile(
                publicKey: HexKey,
                name: String?,
@@ -99,9 +114,21 @@ interface NostrRepository {
                TODO("Not yet implemented")
            }

+            override suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) {
+                TODO("Not yet implemented")
+            }
+
            override suspend fun saveBroadcastReceipt(broadcastNostrEventReceipt: BroadcastNostrEventReceipt) {
                TODO("Not yet implemented")
            }
+
+            override suspend fun saveNostrEvent(
+                nostrEvent: NostrEvent,
+                synchronizeNostrEventRequest: SynchronizeNostrEventRequest,
+                synchronizationRelayURLs: List<String>
+            ) {
+                TODO("Not yet implemented")
+            }
        }
    }
 }
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt
@@ -1,6 +1,5 @@
 package ac.aux.compose.ui.view.model

-import ac.aux.compose.database.model.BroadcastNostrEventReceipt
 import ac.aux.compose.database.model.NostrEvent
 import ac.aux.compose.managers.SeedManager
 import ac.aux.compose.network.NostrEventBroadcaster
@@ -35,6 +34,10 @@ import kotlinx.coroutines.flow.getAndUpdate
 import kotlinx.coroutines.launch
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import kotlin.collections.set
 import kotlin.time.Instant

 class NavigationViewModel(
@@ -63,6 +66,8 @@ class NavigationViewModel(

    private val logger = Logger.withTag(TAG)

+    private val syncingJobs = mutableMapOf<String, Job>()
+
    val httpClient = HttpClient() {
        install(WebSockets) {
            contentConverter = KotlinxWebsocketSerializationConverter(Json {
@@ -129,14 +134,125 @@ class NavigationViewModel(
    }

    private fun observeSyncNostrEventRequests() {
+        logger.i { "observeSyncNostrEventRequests" }

+        scope.launch(Dispatchers.IO) {
+            nostrRepository.observePendingSynchronizeNostrEventRequests().collect { synchronizeNostrEventRequestOrNull ->
+                synchronizeNostrEventRequestOrNull?.let { synchronizeNostrEventRequest ->
+                    val filterContent = mutableMapOf<String, JsonElement>()
+
+                    synchronizeNostrEventRequest.eventIds?.let { eventIds ->
+                        filterContent.put("ids", JsonPrimitive(eventIds))
+                    }
+                    synchronizeNostrEventRequest.authorPublicKeys?.let { authorPublicKeys ->
+                        filterContent.put("authors", JsonPrimitive(authorPublicKeys))
+                    }
+                    synchronizeNostrEventRequest.kinds?.let { kinds ->
+                        filterContent.put("kinds", JsonPrimitive(kinds))
+                    }
+                    synchronizeNostrEventRequest.since?.let { since ->
+                        filterContent.put("since", JsonPrimitive(since.toEpochMilliseconds()))
+                    }
+                    synchronizeNostrEventRequest.until?.let { until ->
+                        filterContent.put("until", JsonPrimitive(until.toEpochMilliseconds()))
+                    }
+                    synchronizeNostrEventRequest.limit?.let { limit ->
+                        filterContent.put("limit", JsonPrimitive(limit))
+                    }
+
+                    val filter = JsonObject(
+                        filterContent
+                    )
+                    logger.d("Syncing: $filter")
+                    scope.launch {
+                        val webSocketSession = httpClient.webSocketSession(
+                            urlString = synchronizeNostrEventRequest.relayURL
+                        )
+
+                        syncingJobs[synchronizeNostrEventRequest.relayURL] = scope.launch {
+                            webSocketSession.send(
+                                frame = Frame.Text("[\"REQ\",${synchronizeNostrEventRequest.id},${filter}]")
+                            )
+
+                            nostrRepository.synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest)
+
+                            webSocketSession.incoming.consumeAsFlow().collect { frame ->
+                                when(frame) {
+                                    is Frame.Text -> {
+                                        frame.readText().let { text ->
+                                            try {
+                                                logger.d("WebSocket Read Text: $text")
+                                                val result = Json.decodeFromString<JsonArray>(text)
+
+                                                if (result.firstOrNull()?.text == "EVENT") {
+                                                    result.getOrNull(1)?.text?.let { subscriptionId ->
+                                                        if (subscriptionId == synchronizeNostrEventRequest.id) {
+
+                                                            result.getOrNull(2)?.text?.let { jsonText ->
+                                                                val event = Event.fromJson(
+                                                                    jsonText
+                                                                )
+
+                                                                nostrRepository.saveNostrEvent(
+                                                                    nostrEvent = NostrEvent(
+                                                                        id = event.id,
+                                                                        pubKey = event.pubKey,
+                                                                        kind = event.kind,
+                                                                        content = event.content,
+                                                                        tags = event.tags,
+                                                                        createdAt = Instant.fromEpochMilliseconds(event.createdAt),
+                                                                        sig = event.sig
+                                                                    ),
+                                                                    synchronizeNostrEventRequest,
+                                                                    synchronizationRelayURLs = Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
+                                                                )
+
+                                                            }
+                                                        }
+                                                    }
+                                                } else {
+                                                    logger.d("Unsupported: $text")
+                                                }
+                                            } catch (e: Throwable) {
+                                                logger.e("Error processing receipt: $text", e)
+                                            }
+                                        }
+                                        // Syncing doesn't necessary have to close after result..
+                                        webSocketSession.close()
+                                        syncingJobs.remove(synchronizeNostrEventRequest.relayURL)?.cancel()
+                                    }
+                                    is Frame.Close -> {
+                                        logger.d("Close $frame")
+                                    }
+                                    is Frame.Ping -> {
+                                        logger.d("Ping: $frame")
+                                    }
+                                    is Frame.Pong -> {
+                                        logger.d("Pong: $frame")
+                                    }
+                                    is Frame.Binary -> {
+                                        logger.d("Binary Message: $frame")
+                                    }
+                                    else -> {
+                                        logger.d("Unsupported frame: $frame")
+                                    }
+                                }
+                            }
+
+                            logger.i("Websocket exit")
+                        }
+
+                        syncingJobs[synchronizeNostrEventRequest.relayURL]?.join()
+                    }
+                }
+            }
+        }
    }

    private fun observePendingBroadcastNostrEventRequests() {
        logger.i { "observePendingBroadcastNostrEventRequests" }

        scope.launch(Dispatchers.IO) {
-
            nostrRepository.observePendingBroadcastNostrEventRequests().collect { localBroadcastNostrEventRequests ->
                localBroadcastNostrEventRequests.forEach { localBroadcastNostrEventRequest ->
                    nostrEventBroadcaster.broadcastEvent(