diff --git a/composeApp/schemas/ac.cord.auxiliary.compose.database.AuxDatabase/1.json b/composeApp/schemas/ac.cord.auxiliary.compose.database.AuxDatabase/1.json index 7f8b64b1..059071f1 100644 --- a/composeApp/schemas/ac.cord.auxiliary.compose.database.AuxDatabase/1.json +++ b/composeApp/schemas/ac.cord.auxiliary.compose.database.AuxDatabase/1.json @@ -2,7 +2,7 @@ "formatVersion": 1, "database": { "version": 1, - "identityHash": "2e56a98cfa2734b3c9f8945e5384f7e7", + "identityHash": "52ac6d568351e25cbde397f09d3fecd5", "entities": [ { "tableName": "BroadcastNostrEventReceipt", @@ -473,6 +473,171 @@ } ] }, + { + "tableName": "NegentropySynchronizeRequest", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `purpose` TEXT NOT NULL, `status` TEXT NOT NULL, `relayURL` TEXT NOT NULL, `isRecommendedRelay` INTEGER NOT NULL, `level` INTEGER NOT NULL, `synchronizationFilter` 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": "purpose", + "columnName": "purpose", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "status", + "columnName": "status", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "relayURL", + "columnName": "relayURL", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "isRecommendedRelay", + "columnName": "isRecommendedRelay", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "level", + "columnName": "level", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "synchronizationFilter", + "columnName": "synchronizationFilter", + "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_NegentropySynchronizeRequest_nostrEventId", + "unique": false, + "columnNames": [ + "nostrEventId" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_NegentropySynchronizeRequest_nostrEventId` ON `${TABLE_NAME}` (`nostrEventId`)" + }, + { + "name": "index_NegentropySynchronizeRequest_status", + "unique": false, + "columnNames": [ + "status" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_NegentropySynchronizeRequest_status` ON `${TABLE_NAME}` (`status`)" + } + ], + "foreignKeys": [ + { + "table": "NostrEvent", + "onDelete": "CASCADE", + "onUpdate": "NO ACTION", + "columns": [ + "nostrEventId" + ], + "referencedColumns": [ + "id" + ] + } + ] + }, + { + "tableName": "NegentropySynchronizeResult", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `relayURL` TEXT NOT NULL, `status` TEXT NOT NULL, `negentropySynchronizeRequestId` TEXT NOT NULL, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, PRIMARY KEY(`id`), FOREIGN KEY(`negentropySynchronizeRequestId`) REFERENCES `SynchronizeNostrEventResult`(`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": "status", + "columnName": "status", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "negentropySynchronizeRequestId", + "columnName": "negentropySynchronizeRequestId", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "createdAt", + "columnName": "createdAt", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "updatedAt", + "columnName": "updatedAt", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "id" + ] + }, + "foreignKeys": [ + { + "table": "SynchronizeNostrEventResult", + "onDelete": "CASCADE", + "onUpdate": "NO ACTION", + "columns": [ + "negentropySynchronizeRequestId" + ], + "referencedColumns": [ + "id" + ] + } + ] + }, { "tableName": "NostrEvent", "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` TEXT NOT NULL, `pubKey` TEXT NOT NULL, `kind` INTEGER NOT NULL, `tags` TEXT NOT NULL, `content` TEXT NOT NULL, `sig` TEXT NOT NULL, `quotedNostrEventId` TEXT, `quotedAuthorPublicKey` TEXT, `inReplyToNostrEventId` TEXT, `inReplyToAuthorPublicKey` TEXT, `inReplyToRootNostrEventId` TEXT, `inReplyToRootAuthorPublicKey` TEXT, `repostedNostrEventId` TEXT, `repostedAuthorPublicKey` TEXT, `unsignedNostrEventId` INTEGER, `createdAt` INTEGER NOT NULL, `updatedAt` INTEGER NOT NULL, `savedAt` INTEGER NOT NULL, `deletedAt` INTEGER, `broadcastedAt` INTEGER, PRIMARY KEY(`id`))", @@ -1822,7 +1987,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, '2e56a98cfa2734b3c9f8945e5384f7e7')" + "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '52ac6d568351e25cbde397f09d3fecd5')" ] } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/AuxDatabase.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/AuxDatabase.kt index cd98075e..f8b5f495 100644 --- 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 diff --git 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 index 4b18c559..24e68e3d 100644 --- 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(it) + } + } catch (e: Throwable) { + logger.e("Failed to convert to SynchronizationFilter $value", e) + return null + } + } \ No newline at end of file diff --git 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 new file mode 100644 index 00000000..ad70e337 --- /dev/null +++ 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 + + @Query("SELECT COUNT(*) FROM NegentropySynchronizeRequest WHERE purpose = :purpose AND status IN (:status) AND createdAt > :createdAt") + fun observeNegentropySynchronizeRequestByPurposeAndStatusCount(purpose: String, status: List, createdAt: Instant = Clock.System.now()): Flow + + @Upsert + fun upsert(negentropySynchronizeRequest: NegentropySynchronizeRequest) + + @Insert + fun insert(negentropySynchronizeRequests: List) +} \ No newline at end of file diff --git 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 new file mode 100644 index 00000000..c7fa0980 --- /dev/null +++ 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) +} \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt index fd3c337f..6b410089 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/dao/NostrEventDao.kt @@ -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> + @Transaction + @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + fun getNostrEvents(kinds: Array, since: Instant = GENESIS_AT): List + + @Transaction + @Query("SELECT * FROM NostrEvent WHERE content LIKE '%' || :search || '%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + fun getFilteredNostrEvents( + kinds: Array, + search: String, + since: Instant = GENESIS_AT + ): List + + @Transaction + @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND id in (:ids) AND createdAt > :since ORDER BY createdAt DESC") + fun getFilteredNostrEvents( + kinds: Array, + ids: Array, + since: Instant = GENESIS_AT + ): List + + @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, + authors: Array, + since: Instant = GENESIS_AT + ): List + + @Transaction + @Query("SELECT * FROM NostrEvent WHERE tags LIKE '%' || :publicKey || '%' AND kind in (:kinds) AND createdAt > :since ORDER BY createdAt DESC") + fun getPublicKeyMentionedNostrEvents( + kinds: Array, + publicKey: HexKey, + since: Instant = GENESIS_AT + ): List + + @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, + eventId: HexKey, + since: Instant + ): List + @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) ORDER BY savedAt DESC") fun getAllNostrEvents(kinds: Array): List diff --git 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 new file mode 100644 index 00000000..3c0ea0e8 --- /dev/null +++ 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 { + +} \ No newline at end of file diff --git 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 new file mode 100644 index 00000000..9477d539 --- /dev/null +++ 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 \ No newline at end of file diff --git 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 index c1d33dfa..47be292c 100644 --- 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( id = id, pubKey = pubKey, diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt index 290ffa1d..6badd13c 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/database/repository/DatabaseNostrRepository.kt @@ -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 { + 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 + ) { + 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> { - return when { synchronizationFilter.kinds != null && synchronizationFilter.search != null -> { logger.d("observeFilteredNostrEvents: $synchronizationFilter") @@ -517,6 +548,62 @@ class DatabaseNostrRepository( } } + override suspend fun getNostrFeedIds(synchronizationFilters: Array): List { + 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 { return database.nostrEventDao().observeNostrEventById(nostrEventId) } diff --git 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 index c2258242..99cf1a9f 100644 --- 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() = diff --git 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 index c686f3cb..671f4c41 100644 --- 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 { + 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( diff --git 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 index d772e6cf..d8206166 100644 --- 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 { + return relayPool.sync( + negOpenCmd = negOpenCmd, + relayUrl = relayUrl + ) + } + suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) { return relayPool.closeQuery( closeCmd = closeCmd, diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/nostr/Relays.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/nostr/Relays.kt index c4b3b5c2..58347877 100644 --- 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) diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt index ecbda582..9593130a 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/repository/NostrRepository.kt @@ -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 + suspend fun observePendingNegentropySynchronizeRequests(): Flow + 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 ) + suspend fun saveNostrEvent( + nostrEvent: NostrEvent, + negentropySynchronizeRequest: NegentropySynchronizeRequest, + synchronizationRelayURLs: List + ) + suspend fun queueSynchronizeNostrEvent( synchronizeNostrEventRequests: List, ) @@ -90,6 +100,8 @@ interface NostrRepository { suspend fun observeNostrFeed(synchronizationFilter: SynchronizationFilter): Flow> + suspend fun getNostrFeedIds(synchronizationFilters: Array): List + suspend fun observeLocalNostrEventById(nostrEventId: String): Flow suspend fun searchableProfiles(): List @@ -135,6 +147,10 @@ interface NostrRepository { TODO("Not yet implemented") } + override suspend fun observePendingNegentropySynchronizeRequests(): Flow { + 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 + ) { + TODO("Not yet implemented") + } + override suspend fun queueSynchronizeNostrEvent(synchronizeNostrEventRequests: List) { TODO("Not yet implemented") } @@ -218,6 +246,10 @@ interface NostrRepository { TODO("Not yet implemented") } + override suspend fun getNostrFeedIds(synchronizationFilters: Array): List { + TODO("Not yet implemented") + } + override suspend fun observeLocalNostrEventById(nostrEventId: String): Flow { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt index d0d2401d..81830d0f 100644 --- a/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/ac/cord/auxiliary/compose/ui/view/model/NavigationViewModel.kt @@ -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) + } + } } } } diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 92eed254..0dd4852c 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -26,6 +26,7 @@ material3 = "1.10.0-alpha05" materialIconsCore = "1.7.3" materialIconsExtended = "1.7.3" navigationCompose = "2.9.2" +negentropyKmp = "v1.0.2" okhttp = "5.3.2" okio = "3.16.4" pagingCommon = "3.5.0-alpha01"