diff --git a/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json b/composeApp/schemas/ac.aux.compose.database.AuxDatabase/1.json index 73f6a3e9..6b4acd40 100644 --- 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')" ] } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/AuxDatabase.kt index 681728db..236c13fe 100644 --- 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 diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/NostrDao.kt index 33f195f4..c9210521 100644 --- 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 + ) { + 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(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 + ) { 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 - ) - ) - } } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventRequestDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventRequestDao.kt new file mode 100644 index 00000000..c10d2049 --- /dev/null +++ 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 + + @Upsert + fun upsert(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventResultDao.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/dao/SynchronizeNostrEventResultDao.kt new file mode 100644 index 00000000..b8430827 --- /dev/null +++ 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) +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventRequest.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventRequest.kt new file mode 100644 index 00000000..959139d0 --- /dev/null +++ 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 \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventResult.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/model/SynchronizeNostrEventResult.kt new file mode 100644 index 00000000..0982835e --- /dev/null +++ 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 \ No newline at end of file diff --git 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 new file mode 100644 index 00000000..6f8876ce --- /dev/null +++ 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? +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt index f41e97e1..1beb1d3c 100644 --- 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 { + 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 + ) { + database.nostrDao().storeNostrEvent( + nostrEvent, + synchronizeNostrEventRequest, + synchronizationRelayURLs = synchronizationRelayURLs + ) + } + } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt index c92f3c3c..71f5a51d 100644 --- 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> + suspend fun observePendingSynchronizeNostrEventRequests(): Flow + 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 + ) + companion object { val NO_OP_NOSTR_REPOSITORY = object : NostrRepository { override suspend fun observeProfile(publicKey: HexKey): Flow { @@ -62,6 +73,10 @@ interface NostrRepository { TODO("Not yet implemented") } + override suspend fun observePendingSynchronizeNostrEventRequests(): Flow { + 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 + ) { + TODO("Not yet implemented") + } } } } \ No newline at end of file diff --git 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 index 5d86d7a6..73418b31 100644 --- 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() + 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() + + 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(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(