diff --git a/composeApp/build.gradle.kts b/composeApp/build.gradle.kts index 71501a81..a951896c 100644 --- a/composeApp/build.gradle.kts +++ b/composeApp/build.gradle.kts @@ -121,6 +121,10 @@ kotlin { } commonTest.dependencies { implementation(libs.kotlin.test) + // runTest: the DAO, model and relay layers are all suspending, and there is no + // runBlocking in a common source set — so anything worth asserting about them + // needs a coroutine, and a scheduler, to assert it in. + implementation(libs.kotlinx.coroutinesTest) } jvmMain.dependencies { implementation(compose.desktop.currentOs) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt index daa45b01..d0657be0 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/ChatMessageDao.kt @@ -30,6 +30,26 @@ interface ChatMessageDao { @Query("SELECT COUNT(*) FROM ChatMessage WHERE chatRoomId = :chatRoomId AND senderPublicKey = :senderPublicKey") fun countChatMessagesBySenderPublicKey(chatRoomId: HexKey, senderPublicKey: HexKey): Int + /** + * The room's group events that already have something to show for them. + * + * One query so a replay can skip the events it has no business touching without + * a lookup per event -- `marmotGroupEventId` carries no index, so each of those + * is a scan of the whole table. + * + * [unresolvedTypes] is [press.mantra.compose.database.model.ChatMessage.UNRESOLVED_MARMOT_TYPES]: + * lines that stand in for a message still to come rather than reporting one. + */ + @Query( + "SELECT marmotGroupEventId FROM ChatMessage " + + "WHERE chatRoomId = :chatRoomId AND marmotGroupEventId IS NOT NULL " + + "AND messageType NOT IN (:unresolvedTypes)" + ) + suspend fun getResolvedMarmotGroupEventIds( + chatRoomId: String, + unresolvedTypes: Collection + ): List + @Upsert suspend fun upsert(chatMessage: press.mantra.compose.database.model.ChatMessage): Long } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt index 0aaba9be..f3344e5b 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrDao.kt @@ -19,6 +19,7 @@ import press.mantra.compose.database.model.Participant import press.mantra.compose.database.model.Post import press.mantra.compose.database.model.Profile import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.MarmotReindexReport import press.mantra.compose.database.model.types.SynchronizationFilter import press.mantra.compose.exceptions.GiftWrapImpersonationException import press.mantra.compose.exceptions.GiftWrapSealDecryptionException @@ -35,12 +36,15 @@ import press.mantra.compose.nostr.frost.FrostSigningEvents import press.mantra.compose.managers.FrostSigningManager import press.mantra.compose.managers.MlsGroupCache import press.mantra.compose.managers.MarmotInboundManager +import press.mantra.compose.managers.MarmotReindexSweep import co.touchlab.kermit.Logger import kotlinx.coroutines.CancellationException import com.vitorpamplona.quartz.marmot.GroupEventResult import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageEvent import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageRelayListEvent import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent +import com.vitorpamplona.quartz.marmot.mip03GroupMessages.CommitOrdering +import com.vitorpamplona.quartz.marmot.mip03GroupMessages.GroupEvent import com.vitorpamplona.quartz.marmot.mls.codec.TlsReader import com.vitorpamplona.quartz.marmot.mls.group.MlsGroup import com.vitorpamplona.quartz.marmot.mls.messages.KeyPackageBundle @@ -386,114 +390,11 @@ abstract class NostrDao( logger.d("groupEvent: ${groupEvent.toJson()}") groupEvent.groupId()?.let { chatRoomId -> - val localChatRoom = database.chatRoomDao().findChatRoomById(chatRoomId) - - if (localChatRoom != null) { - // Through the cache rather than rebuilt here, so the secret - // tree's skipped-generation keys survive from one message to - // the next. Two events published in the same instant arrive in - // whatever order the relay feels like, and rebuilding between - // them loses the earlier one for good -- see MlsGroupCache. - val handled = MlsGroupCache.withGroup( - chatRoomId = chatRoomId, - storedStateHex = localChatRoom.chatRoom.mlsGroupState, - build = { localChatRoom.chatRoom.toMlsGroup() }, - save = { stateHex -> - database.chatRoomDao().upsert( - localChatRoom.chatRoom.copy(mlsGroupState = stateHex) - ) - } - ) { mlsGroup -> - val memberPubkeys = mlsGroup.members().mapNotNull { (leafIndex, leafNode) -> - - val pubkey = when (val cred = leafNode.credential) { - is Credential.Basic -> cred.identity.toHexKey() - else -> null - } - - logger.d("leafIndex=$leafIndex pubKey=$pubkey") - pubkey - } - - if (memberPubkeys.contains(activeKeyPair.pubKey.toHex())) { - database.marmotGroupEventDao().upsert( - MarmotGroupEvent( - id = groupEvent.id, - userPublicKey = activeKeyPair.pubKey.toHex(), - publicKey = groupEvent.pubKey, - encryptedContent = groupEvent.encryptedContent(), - chatRoomId = chatRoomId, - createdAt = Instant.fromEpochSeconds(groupEvent.createdAt), - expiresAt = groupEvent.expiration()?.let { Instant.fromEpochSeconds(it) }, - signature = groupEvent.sig - ) - ) - MarmotInboundManager.processGroupEvent( - database = database, - activeKeyPair = activeKeyPair, - localChatRoom = localChatRoom, - mlsGroup = mlsGroup, - groupEvent = groupEvent - )?.let { groupEventResult -> - logger.d("groupEventResult: $groupEventResult") - - when(groupEventResult) { - is GroupEventResult.CommitProcessed -> { - MarmotInboundManager.processGroupMembershipChanges( - database = database, - mlsGroup = mlsGroup, - localChatRoom = localChatRoom - ) - } - else -> { - logger.i("No app logic to handle: $groupEventResult") - } - } - - ChatMessage.fromGroupEventResult( - database = database, - activeKeyPair = activeKeyPair, - groupEvent = groupEvent, - groupEventResult = groupEventResult, - )?.let { chatMessage -> - logger.d("chatMessage: $chatMessage") - database.chatMessageDao().upsert( - chatMessage - ) - } - - // A FROST signing message for this group. Driven from - // here rather than from ChatMessage because the - // manager needs the room to publish its own replies - // into, and because it writes its transcript lines - // itself. The manager is idempotent, so a redelivered - // message re-runs a step it has already taken. - if (groupEventResult is GroupEventResult.ApplicationMessage) { - Event.fromJsonOrNull(groupEventResult.innerEventJson) - ?.takeIf { FrostSigningEvents.isFrostSigningKind(it.kind) } - ?.let { innerEvent -> - FrostSigningManager.processSigningPayload( - database = database, - localChatRoom = localChatRoom, - innerEvent = innerEvent, - userPublicKey = activeKeyPair.pubKey.toHex() - ) - } - } - } - } else { - throw MarmotNotMemberOfChatGroupException("We are not a member of the chat room ${localChatRoom.chatRoom.id}") - } - } - - // Null means the room has no usable group state, which is what - // a failed toMlsGroup() meant before the cache existed. - if (handled == null) { - throw MarmotMissingNostrGroupDataExtension("Couldn't find chatRoom for $nostrEvent") - } - } else { - throw MarmotMissingChatGroupException("Couldn't find chatRoom for $nostrEvent") - } + indexMarmotGroupEvent( + groupEvent = groupEvent, + chatRoomId = chatRoomId, + activeKeyPair = activeKeyPair, + ) } } @@ -1279,6 +1180,236 @@ abstract class NostrDao( } } + /** + * Reads one kind:445 against the room's MLS group and files whatever it turns + * out to be: a message, a commit that advances the group, an entity for the + * library, or a placeholder saying it could not be read yet. + * + * Its own function so a replay can run exactly what a first delivery ran -- + * see [reindexMarmotGroupEvents]. Throws for a room this device cannot process + * the event against at all, which the caller decides what to do about: a first + * delivery lets it roll back its transaction, a replay logs it and moves on. + */ + private suspend fun indexMarmotGroupEvent( + groupEvent: GroupEvent, + chatRoomId: String, + activeKeyPair: KeyPair, + ) { + val localChatRoom = database.chatRoomDao().findChatRoomById(chatRoomId) + ?: throw MarmotMissingChatGroupException("Couldn't find chatRoom for ${groupEvent.id}") + + // Through the cache rather than rebuilt here, so the secret + // tree's skipped-generation keys survive from one message to + // the next. Two events published in the same instant arrive in + // whatever order the relay feels like, and rebuilding between + // them loses the earlier one for good -- see MlsGroupCache. + val handled = MlsGroupCache.withGroup( + chatRoomId = chatRoomId, + storedStateHex = localChatRoom.chatRoom.mlsGroupState, + build = { localChatRoom.chatRoom.toMlsGroup() }, + save = { stateHex -> + database.chatRoomDao().upsert( + localChatRoom.chatRoom.copy(mlsGroupState = stateHex) + ) + } + ) { mlsGroup -> + val memberPubkeys = mlsGroup.members().mapNotNull { (leafIndex, leafNode) -> + + val pubkey = when (val cred = leafNode.credential) { + is Credential.Basic -> cred.identity.toHexKey() + else -> null + } + + logger.d("leafIndex=$leafIndex pubKey=$pubkey") + pubkey + } + + if (memberPubkeys.contains(activeKeyPair.pubKey.toHex())) { + database.marmotGroupEventDao().upsert( + MarmotGroupEvent( + id = groupEvent.id, + userPublicKey = activeKeyPair.pubKey.toHex(), + publicKey = groupEvent.pubKey, + encryptedContent = groupEvent.encryptedContent(), + chatRoomId = chatRoomId, + createdAt = Instant.fromEpochSeconds(groupEvent.createdAt), + expiresAt = groupEvent.expiration()?.let { Instant.fromEpochSeconds(it) }, + signature = groupEvent.sig + ) + ) + MarmotInboundManager.processGroupEvent( + database = database, + activeKeyPair = activeKeyPair, + localChatRoom = localChatRoom, + mlsGroup = mlsGroup, + groupEvent = groupEvent + )?.let { groupEventResult -> + logger.d("groupEventResult: $groupEventResult") + + when(groupEventResult) { + is GroupEventResult.CommitProcessed -> { + MarmotInboundManager.processGroupMembershipChanges( + database = database, + mlsGroup = mlsGroup, + localChatRoom = localChatRoom + ) + } + else -> { + logger.i("No app logic to handle: $groupEventResult") + } + } + + ChatMessage.fromGroupEventResult( + database = database, + activeKeyPair = activeKeyPair, + groupEvent = groupEvent, + groupEventResult = groupEventResult, + )?.let { chatMessage -> + logger.d("chatMessage: $chatMessage") + persistMarmotChatMessage(chatMessage) + } + + // A FROST signing message for this group. Driven from + // here rather than from ChatMessage because the + // manager needs the room to publish its own replies + // into, and because it writes its transcript lines + // itself. The manager is idempotent, so a redelivered + // message re-runs a step it has already taken. + if (groupEventResult is GroupEventResult.ApplicationMessage) { + Event.fromJsonOrNull(groupEventResult.innerEventJson) + ?.takeIf { FrostSigningEvents.isFrostSigningKind(it.kind) } + ?.let { innerEvent -> + FrostSigningManager.processSigningPayload( + database = database, + localChatRoom = localChatRoom, + innerEvent = innerEvent, + userPublicKey = activeKeyPair.pubKey.toHex() + ) + } + } + } + } else { + throw MarmotNotMemberOfChatGroupException("We are not a member of the chat room ${localChatRoom.chatRoom.id}") + } + } + + // Null means the room has no usable group state, which is what + // a failed toMlsGroup() meant before the cache existed. + if (handled == null) { + throw MarmotMissingNostrGroupDataExtension("Couldn't find chatRoom for ${groupEvent.id}") + } + } + + /** + * Files the chat line a kind:445 produced, replacing the line that event + * already has rather than adding a second one. The rule for which of the two + * wins is [ChatMessage.reconcileMarmotLine]. + */ + private suspend fun persistMarmotChatMessage(chatMessage: ChatMessage) { + val existing = chatMessage.marmotGroupEventId?.let { + database.chatMessageDao().getChatMessagesByMarmotGroupEventId(it) + } + + val row = ChatMessage.reconcileMarmotLine(fresh = chatMessage, existing = existing) + + if (row == null) { + logger.d("Keeping ${existing?.messageType} line for ${chatMessage.marmotGroupEventId}") + return + } + + database.chatMessageDao().upsert(row) + } + + /** + * Reads a room's stored kind:445 events again, oldest first. + * + * Relays impose no ordering, so a group event can arrive before the group is + * able to read it: an application message encrypted under an epoch whose commit + * has not landed yet, or a commit for an epoch ahead of the local one. Both are + * stored and then dropped -- `MarmotInboundManager` refuses to apply an + * out-of-epoch commit precisely so it does not half-mutate the group -- and + * nothing revisits them once the missing event fills the gap. This is the + * revisit, driven from the group's detail screen. + * + * Only events with nothing to show for them are replayed: no chat line at all, + * or one of [ChatMessage.UNRESOLVED_MARMOT_TYPES]. Everything already read is + * left untouched, which is what keeps the replay from disturbing a room where + * there was nothing wrong. Within that set a replay is safe to repeat: the ids + * of every row it writes are derived from the events themselves, commits that + * are behind the current epoch are rejected as duplicates before they can touch + * the group, and the one row with a generated id is reconciled by + * [persistMarmotChatMessage]. + * + * Passes repeat while a pass recovers something -- see [MarmotReindexSweep] -- + * because created_at order is not necessarily epoch order, and a commit + * recovered by one pass can be what lets the next read the messages that were + * waiting on it. + * + * What this cannot do is recover a message whose key is gone: an application + * message the ratchet has already advanced past, or one from an epoch that + * predates this device joining. Those stay unreadable however often they are + * replayed. + */ + open suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + activeKeyPair: KeyPair, + ): MarmotReindexReport { + val userPublicKey = activeKeyPair.pubKey.toHex() + + // The query's LIKE only narrows; the h tag is what decides the room. Sorted + // by CommitOrdering's own comparator rather than left in createdAt order, so + // that where two commits compete for an epoch the first one replayed is the + // one every other member also picked. + val groupEvents = database.nostrEventDao() + .getMarmotGroupNostrEventsByChatRoomId(chatRoomId) + .mapNotNull { storedNostrEvent -> + storedNostrEvent + .toGroupEvent(userPublicKey = userPublicKey) + ?.takeIf { it.groupId() == chatRoomId } + } + .sortedWith(CommitOrdering.comparator) + + logger.i("Reindex $chatRoomId: ${groupEvents.size} stored group event(s)") + + // Held commits are what a replay is most often for, and they are only ever + // cleared by one applying -- see MarmotInboundManager.forgetPendingCommits. + MarmotInboundManager.forgetPendingCommits(chatRoomId) + + val resolved = database.chatMessageDao().getResolvedMarmotGroupEventIds( + chatRoomId = chatRoomId, + unresolvedTypes = ChatMessage.UNRESOLVED_MARMOT_TYPES, + ).toSet() + + return MarmotReindexSweep.run( + stored = groupEvents.size, + unresolved = groupEvents.filterNot { it.id in resolved }, + replay = { groupEvent -> + indexMarmotGroupEvent( + groupEvent = groupEvent, + chatRoomId = chatRoomId, + activeKeyPair = activeKeyPair, + ) + }, + isUnresolved = { groupEvent -> isUnresolvedMarmotGroupEvent(groupEvent.id) }, + ).also { + logger.i("Reindex $chatRoomId: $it") + } + } + + /** + * Whether one group event still has nothing to show for it, asked again after + * a replay to see whether that replay achieved anything. + * + * Some events can never leave this state and will be swept every time: a commit + * rejected as a duplicate writes no line, and neither does a commit this device + * sent, whose line was written against the send rather than the event. Both cost + * one refused decrypt per sweep, which is cheaper than a bookkeeping column that + * would have to be migrated in. + */ + private suspend fun isUnresolvedMarmotGroupEvent(groupEventId: String): Boolean { + val chatMessage = database.chatMessageDao().getChatMessagesByMarmotGroupEventId(groupEventId) + return chatMessage == null || chatMessage.messageType in ChatMessage.UNRESOLVED_MARMOT_TYPES + } /** * Files an arriving NIP-17 chat message so it shows up in the room's feed. @@ -1425,4 +1556,5 @@ abstract class NostrDao( } } } + } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt index 1066c814..f354e547 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/NostrEventDao.kt @@ -143,6 +143,26 @@ interface NostrEventDao { now: Instant = Clock.System.now(), ): List + /** + * Every kind:445 held locally that mentions [chatRoomId], oldest first. + * + * The `LIKE` is a prefilter over the serialised tags, not the test: it can match + * a group id that happens to appear in some other tag, so callers must confirm + * the event's own `h` tag before treating it as this room's. Ordered ascending + * because a replay has to apply commits in the order they were sent. + * + * Deliberately not a join on `MarmotGroupEvent`: an event whose indexing failed + * part way may have been stored without ever reaching that table, and those are + * exactly the ones worth replaying. + */ + @Transaction + @Query( + "SELECT * FROM NostrEvent WHERE kind = ${GroupEvent.KIND} " + + "AND tags LIKE '%' || :chatRoomId || '%' " + + "ORDER BY createdAt ASC" + ) + fun getMarmotGroupNostrEventsByChatRoomId(chatRoomId: HexKey): List + @Transaction @Query("SELECT * FROM NostrEvent WHERE kind in (:kinds) AND pubKey in (:authors) AND createdAt > :since ORDER BY createdAt ASC LIMIT :limit") fun getAuthoredNostrEvents( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt index feba4e8c..f6df2e83 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/ChatMessage.kt @@ -249,6 +249,69 @@ data class ChatMessage( TYPE_DKG_FAILED, ) + /** + * A kind:445 arrived that the group could not read at the time. + * + * Both are placeholders for a message still to come: the outer layer was + * encrypted under an epoch whose exporter secret we did not hold yet, or + * the commit is one of several competing for an epoch and none has been + * applied. Neither says anything a member wrote. + */ + const val TYPE_UNDECRYPTABLE_OUTER_LAYER = "undecryptableOuterLayer" + const val TYPE_PENDING_COMMIT = "pendingCommit" + + /** + * The lines a reindex is allowed to replace. + * + * A group event that left one of these behind was never read, so replaying + * it can only improve on what is there. Every other line -- a message, an + * applied commit, anything this device sent -- is the final word on its + * group event and a replay must leave it alone. + */ + val UNRESOLVED_MARMOT_TYPES = setOf( + TYPE_UNDECRYPTABLE_OUTER_LAYER, + TYPE_PENDING_COMMIT, + ) + + /** + * The row to write for a group event that was just read, given the line that + * event already has. Null means leave what is there alone. + * + * [ChatMessage.id] is autogenerated, so writing [fresh] as it comes always + * inserts. That is right the first time a group event is read and wrong every + * time after: a replay would leave the room showing each recovered message + * twice, once as "Undecryptable Message" and once as itself. A group event has + * exactly one line and a deterministic id, so [existing] is what makes a + * re-read an update instead. + * + * A placeholder never overwrites a line that says something. A group event + * this device sent already has its line, written on the way out, and a replay + * that cannot read back our own ratchet-consumed message must not replace it + * with "Undecryptable Message". + * + * [viewedAt] and [savedAt] stay with [existing]: when a line was first seen is + * a fact about the reader, not about the event, and a re-read is not a new + * arrival. Its id stays too, so the broadcast and nostr-event relations + * pointing at that row keep pointing at it. + */ + fun reconcileMarmotLine( + fresh: ChatMessage, + existing: ChatMessage?, + ): ChatMessage? { + if (existing == null) return fresh + + val freshIsPlaceholder = fresh.messageType in UNRESOLVED_MARMOT_TYPES + val existingIsPlaceholder = existing.messageType in UNRESOLVED_MARMOT_TYPES + + if (freshIsPlaceholder && !existingIsPlaceholder) return null + + return fresh.copy( + id = existing.id, + savedAt = existing.savedAt, + viewedAt = existing.viewedAt, + ) + } + suspend fun fromGroupEventResult( database: MantraDatabase, activeKeyPair: KeyPair, @@ -326,7 +389,7 @@ data class ChatMessage( giftWrapPayloadId = null, marmotGroupEventId = groupEvent.id, marmotInnerEventId = null, - messageType = "pendingCommit", + messageType = TYPE_PENDING_COMMIT, senderPublicKey = groupEvent.pubKey, isUserMessage = activeKeyPair.pubKey.toHex() == groupEvent.pubKey, chatRoomId = groupEventResult.groupId, @@ -392,7 +455,7 @@ data class ChatMessage( giftWrapPayloadId = null, marmotGroupEventId = groupEvent.id, marmotInnerEventId = null, - messageType = "undecryptableOuterLayer", + messageType = TYPE_UNDECRYPTABLE_OUTER_LAYER, senderPublicKey = groupEvent.pubKey, isUserMessage = activeKeyPair.pubKey.toHex() == groupEvent.pubKey, chatRoomId = groupEventResult.groupId, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt new file mode 100644 index 00000000..0933e6da --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/model/types/MarmotReindexReport.kt @@ -0,0 +1,27 @@ +package press.mantra.compose.database.model.types + +/** + * What a replay of a room's stored kind:445 events did. + * + * Nothing here is persisted -- it exists so the screen that asked for the replay + * can say what happened rather than leaving the user to guess from the message + * list whether anything moved. + * + * @param stored every kind:445 held locally for the room. + * @param unresolved how many of those had nothing to show for them, or only a + * placeholder line -- the ones a replay was allowed to touch. + * @param recovered how many of [unresolved] came out with something to show: + * a message, an applied commit, an entity added to the group's library. + * @param failed how many threw while being replayed. Expected to be non-zero + * on a room with events from before this device joined, whose epoch secrets it + * never held and never will. + */ +data class MarmotReindexReport( + val stored: Int = 0, + val unresolved: Int = 0, + val recovered: Int = 0, + val failed: Int = 0, +) { + /** Nothing was left to try, so the replay was a no-op. */ + val isNoOp: Boolean get() = unresolved == 0 +} diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt index b12e3723..663336a0 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseChatRepository.kt @@ -10,9 +10,11 @@ import press.mantra.compose.database.model.MarmotKeyPackage import press.mantra.compose.database.model.Participant import press.mantra.compose.database.model.intermdiate.LocalChatMessage import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.MarmotReindexReport import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.marmot.mip02Welcome.WelcomeEvent import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray import com.vitorpamplona.quartz.nip01Core.core.Kind import com.vitorpamplona.quartz.nip01Core.crypto.EventHasher import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair @@ -407,6 +409,22 @@ class DatabaseChatRepository( onCompletion.invoke() } + /** + * Only the public key is needed: reading a group event is done with the MLS + * group's own secrets, held in the room's stored state, and the identity is + * used for nothing but deciding whether this device is a member and whether a + * line is its own. So a read-only [KeyPair] is enough, and the nostr private + * key stays where it is. + */ + override suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + userPublicKey: HexKey + ): MarmotReindexReport = + database.nostrDao().reindexMarmotGroupEvents( + chatRoomId = chatRoomId, + activeKeyPair = KeyPair(pubKey = userPublicKey.hexToByteArray()), + ) + companion object { const val TAG = "DatabaseChatRepository" } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt index c7be3dba..f2f7ffa6 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotInboundManager.kt @@ -413,6 +413,31 @@ object MarmotInboundManager { } } + /** + * Forget the commits being held for [nostrGroupId], so a replay decides its + * epochs again from the events it is about to feed back in. + * + * [handleCommitEvent] holds a commit whenever a second one turns up for an + * epoch, and only [CommitOrdering.EpochCommitTracker.clearEpoch] on a + * successful apply ever empties that. Nothing applies while more than one is + * held, so the set is a dead end: replaying those commits without clearing + * first just adds them again and leaves every one of them pending, and the + * list grows with each attempt. + * + * Safe to drop because the tracker is a scratch pad, not a record -- the + * commits themselves are on disk, and a replay hands them back oldest first, + * which is the order [CommitOrdering.comparator] picks a winner in. + */ + suspend fun forgetPendingCommits(nostrGroupId: HexKey) { + commitTracker + .pendingGroupEpochs() + .filter { it.groupId == nostrGroupId } + .forEach { pending -> + logger.d("Forgetting pending commits for $nostrGroupId epoch ${pending.epoch}") + commitTracker.clearEpoch(pending.groupId, pending.epoch) + } + } + private suspend fun applyCommit( database: MantraDatabase, mlsGroup: MlsGroup, diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt new file mode 100644 index 00000000..47071e5d --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/MarmotReindexSweep.kt @@ -0,0 +1,95 @@ +package press.mantra.compose.managers + +import co.touchlab.kermit.Logger +import kotlinx.coroutines.CancellationException +import press.mantra.compose.database.model.types.MarmotReindexReport + +/** + * Reads a backlog of group events again, repeatedly, until repeating stops helping. + * + * The whole reason a replay is worth anything is that the events depend on each + * other: a commit nobody could apply is what a room's later messages were waiting + * on, and applying it makes them readable. So one pass in order is not enough -- + * a pass that recovers something has changed what the next pass can do. + * + * Ordering is the caller's business; this only decides how many times to go round. + * A pass that recovers nothing ends it, because everything left is waiting on + * something no later pass will produce either. + * + * Kept apart from the database so the stopping rule can be read, and tested, on + * its own -- it is the part with something to get wrong. + */ +object MarmotReindexSweep { + private const val TAG = "MarmotReindexSweep" + + private val logger = Logger.withTag(TAG) + + /** + * How many times a sweep will go round. + * + * A pass that recovers nothing already ends the sweep, so this only bounds the + * pathological shape where every pass recovers exactly one event and the sweep + * turns quadratic. A chain of five events each waiting on the last is already + * far past what relay reordering produces. + */ + const val DEFAULT_MAX_PASSES = 5 + + /** + * @param stored how many group events the room holds in total, for the report. + * @param unresolved those with nothing to show for them, in the order to replay + * them. Anything already read must be left out: a replay is only ever allowed + * to touch events it cannot make worse. + * @param replay reads one event again. Its failures are expected -- a message + * whose key is gone throws every time -- so they are logged, not raised. + * @param isUnresolved asked again after [replay] to see whether it achieved + * anything. This, rather than [replay]'s return, is what counts a recovery: + * an event is recovered when the room has something to show for it, which is + * a fact about the database and not about the call that just ran. + */ + suspend fun run( + stored: Int, + unresolved: List, + maxPasses: Int = DEFAULT_MAX_PASSES, + replay: suspend (T) -> Unit, + isUnresolved: suspend (T) -> Boolean, + ): MarmotReindexReport { + var remaining = unresolved + var recovered = 0 + var pass = 0 + + while (remaining.isNotEmpty() && pass < maxPasses) { + pass++ + val stillUnresolved = mutableListOf() + + for (candidate in remaining) { + try { + replay(candidate) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.w("Replay of $candidate failed", e) + } + + if (isUnresolved(candidate)) { + stillUnresolved.add(candidate) + } else { + recovered++ + } + } + + val madeProgress = stillUnresolved.size < remaining.size + remaining = stillUnresolved + + if (!madeProgress) break + } + + return MarmotReindexReport( + stored = stored, + unresolved = unresolved.size, + recovered = recovered, + failed = remaining.size, + ).also { + logger.i("Swept ${unresolved.size} unresolved event(s) in $pass pass(es): $it") + } + } +} diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt index 4669fae9..34174c20 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt @@ -6,6 +6,7 @@ import press.mantra.compose.database.model.MarmotKeyPackage import press.mantra.compose.database.model.Participant import press.mantra.compose.database.model.intermdiate.LocalChatMessage import press.mantra.compose.database.model.intermdiate.LocalChatRoom +import press.mantra.compose.database.model.types.MarmotReindexReport import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.Kind import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync @@ -106,6 +107,19 @@ interface ChatRepository { onCompletion: () -> Unit ) + /** + * Reads the room's stored marmot group events again, so ones the group could + * not make sense of when they arrived get another chance now that the events + * they were waiting on have landed. + * + * Only events with nothing to show for them are replayed, so a room where + * nothing went wrong is left exactly as it was. + */ + suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + userPublicKey: HexKey + ): MarmotReindexReport + companion object { val NO_OP_CHAT_REPOSITORY = object: ChatRepository { override suspend fun observeChatRoomListByPublicKey(publicKey: String): Flow> { @@ -232,6 +246,11 @@ interface ChatRepository { ) { TODO("Not yet implemented") } + + override suspend fun reindexMarmotGroupEvents( + chatRoomId: String, + userPublicKey: HexKey + ): MarmotReindexReport = MarmotReindexReport() } } } \ No newline at end of file diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt index ba444544..bc19ed8e 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/ChatRoomDetailScreen.kt @@ -7,12 +7,14 @@ import androidx.compose.foundation.layout.fillMaxSize import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.height import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.layout.size import androidx.compose.foundation.layout.width import androidx.compose.foundation.lazy.LazyColumn import androidx.compose.foundation.lazy.items import androidx.compose.material.icons.Icons import androidx.compose.material.icons.filled.Key import androidx.compose.material.icons.filled.AccountTree +import androidx.compose.material.icons.filled.Autorenew import androidx.compose.material.icons.filled.ChevronRight import androidx.compose.material.icons.filled.DeleteForever import androidx.compose.material.icons.filled.LibraryBooks @@ -23,6 +25,7 @@ import androidx.compose.material.icons.filled.Unsubscribe import androidx.compose.material.icons.filled.WaterfallChart import androidx.compose.material3.ButtonDefaults import androidx.compose.material3.Card +import androidx.compose.material3.CircularProgressIndicator import androidx.compose.material3.ExperimentalMaterial3Api import androidx.compose.material3.ExperimentalMaterial3ExpressiveApi import androidx.compose.material3.HorizontalDivider @@ -493,6 +496,22 @@ fun ChatRoomDetailScreen( Text("Delete Group") } } + + // Only MLS rooms have group events to read again. A NIP-17 + // room's messages are gift wraps, which carry no ordering + // for anything to go wrong with. + if (chatRoomDetailUIState.localChatRoom.chatRoom.mlsGroupState != null) { + item { + HorizontalDivider() + } + + item { + ReindexMarmotGroupEventsButton( + reindexState = chatRoomDetailViewModel.reindexState, + onReindex = chatRoomDetailViewModel::reindexMarmotGroupEvents + ) + } + } } } } @@ -539,6 +558,77 @@ fun ChatRoomDetailScreen( } } +/** + * Reads the room's stored marmot group events again. + * + * Sits at the very bottom, below leaving and deleting, because it is a repair + * rather than something anyone should need in the ordinary course of using the + * group. It says what it found afterwards -- a replay that recovered nothing + * looks identical to one that was never pressed, and the difference between + * "nothing was broken" and "nothing could be fixed" is worth telling. + */ +@Composable +private fun ReindexMarmotGroupEventsButton( + reindexState: ChatRoomDetailViewModel.ReindexState, + onReindex: () -> Unit +) { + Column( + horizontalAlignment = Alignment.CenterHorizontally, + verticalArrangement = Arrangement.spacedBy(5.dp) + ) { + TextButton( + enabled = reindexState !is ChatRoomDetailViewModel.ReindexState.Running, + onClick = onReindex + ) { + if (reindexState is ChatRoomDetailViewModel.ReindexState.Running) { + CircularProgressIndicator( + modifier = Modifier.size(18.dp), + strokeWidth = 2.dp + ) + } else { + Icon( + Icons.Default.Autorenew, + contentDescription = "Reindex group events" + ) + } + + Spacer( + modifier = Modifier.width(10.dp) + ) + + Text("Reindex Events") + } + + when (reindexState) { + is ChatRoomDetailViewModel.ReindexState.Done -> { + val report = reindexState.report + Text( + text = when { + report.isNoOp -> "Nothing to reindex · ${report.stored} event(s) all read" + report.recovered > 0 && report.failed > 0 -> + "Recovered ${report.recovered} of ${report.unresolved} · ${report.failed} still unreadable" + report.recovered > 0 -> "Recovered ${report.recovered} of ${report.unresolved} event(s)" + else -> "${report.failed} event(s) still unreadable" + }, + style = MaterialTheme.typography.bodySmall, + textAlign = TextAlign.Center + ) + } + + is ChatRoomDetailViewModel.ReindexState.Failed -> { + Text( + text = reindexState.message, + color = MaterialTheme.colorScheme.error, + style = MaterialTheme.typography.bodySmall, + textAlign = TextAlign.Center + ) + } + + else -> Unit + } + } +} + @Preview @Composable private fun ChatRoomMessagingScreenPreview() { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt index 9e1d7068..b2cff93a 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/ChatRoomDetailViewModel.kt @@ -18,9 +18,11 @@ import press.mantra.compose.ui.composable.navigation.routes.Route import press.mantra.compose.ui.view.state.ChatRoomDetailUIState import co.touchlab.kermit.Logger import com.vitorpamplona.quartz.nip01Core.core.HexKey +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.launch +import press.mantra.compose.database.model.types.MarmotReindexReport class ChatRoomDetailViewModel( val chatRoomId: String, @@ -35,6 +37,26 @@ class ChatRoomDetailViewModel( var chatRoomDetailUIState: ChatRoomDetailUIState by mutableStateOf(initialChatRoomDetailUIState) private set + /** + * Where the room's replay has got to, for the button that started it. + * + * Held here rather than in [ChatRoomDetailUIState] because a replay does not + * change what the screen is showing -- the message list is observed elsewhere + * and updates itself -- it only changes what the button has to say for itself. + */ + var reindexState: ReindexState by mutableStateOf(ReindexState.Idle) + private set + + sealed interface ReindexState { + data object Idle : ReindexState + + data object Running : ReindexState + + data class Done(val report: MarmotReindexReport) : ReindexState + + data class Failed(val message: String) : ReindexState + } + private val logger = Logger.withTag(TAG) @@ -81,6 +103,39 @@ class ChatRoomDetailViewModel( } } + /** + * Reads the room's stored marmot group events again. + * + * Nostr relays impose no ordering, so a group event can arrive before the group + * can read it -- a message from an epoch whose commit has not landed, a commit + * for an epoch ahead of the local one. Nothing revisits those once the gap + * fills, which is what this is for. + * + * Only events with nothing to show for them are replayed, so pressing this on a + * room where nothing went wrong changes nothing. The feed is observed, so + * anything recovered appears in the message list without a refresh. + */ + fun reindexMarmotGroupEvents() { + if (reindexState is ReindexState.Running) return + + reindexState = ReindexState.Running + viewModelScope.launch(Dispatchers.IO) { + reindexState = try { + ReindexState.Done( + chatRepository.reindexMarmotGroupEvents( + chatRoomId = chatRoomId, + userPublicKey = activeUserPublicKey, + ) + ) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + logger.e("Failed to reindex $chatRoomId", e) + ReindexState.Failed(e.message ?: "Reindex failed") + } + } + } + fun softDeleteGroup( localChatRoom: LocalChatRoom, onPopBackToRoute: (Route) -> Unit diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt new file mode 100644 index 00000000..cd898372 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotChatLineReconciliationTest.kt @@ -0,0 +1,161 @@ +package press.mantra.compose.database.model + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertSame +import kotlin.time.Instant + +/** + * What happens to a room's chat line when its group event is read a second time. + * + * [ChatMessage.id] is autogenerated, so writing a freshly built line always + * inserts. That is correct exactly once. Reading the same kind:445 again -- which + * is what a reindex does -- would otherwise leave the room showing the recovered + * message twice, once as the placeholder that was written when it could not be + * read and once as itself. And the other way round matters just as much: a group + * event this device sent already has its line, and a replay that cannot read our + * own ratchet-consumed message must not replace it with "Undecryptable Message". + * + * Both directions are decided by [ChatMessage.reconcileMarmotLine], which is why + * they are asserted rather than left to the shape of the calling code. + */ +class MarmotChatLineReconciliationTest { + + private val groupEventId = "a".repeat(64) + private val sender = "b".repeat(64) + private val room = "c".repeat(64) + + private fun line( + id: Long = 0, + messageType: String, + content: String, + viewedAt: Instant? = null, + savedAt: Instant = Instant.fromEpochSeconds(1_000), + ) = ChatMessage( + id = id, + senderPublicKey = sender, + isUserMessage = false, + giftWrapPayloadId = null, + marmotGroupEventId = groupEventId, + marmotInnerEventId = null, + chatRoomId = room, + content = content, + messageType = messageType, + savedAt = savedAt, + viewedAt = viewedAt, + ) + + @Test + fun `a group event read for the first time is written as it comes`() { + val fresh = line(messageType = "message", content = "hello") + + assertSame(fresh, ChatMessage.reconcileMarmotLine(fresh = fresh, existing = null)) + } + + @Test + fun `a recovered message replaces the placeholder rather than joining it`() { + val placeholder = line( + id = 42, + messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER, + content = "Undecryptable Message", + ) + val recovered = line(messageType = "message", content = "hello") + + val row = ChatMessage.reconcileMarmotLine(fresh = recovered, existing = placeholder) + + assertEquals(42, row?.id, "reusing the row id is what makes this an update, not a second line") + assertEquals("hello", row?.content) + assertEquals("message", row?.messageType) + } + + @Test + fun `a placeholder never overwrites a line that says something`() { + val sent = line(id = 7, messageType = "message", content = "hello") + val placeholder = line( + messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER, + content = "Undecryptable Message", + ) + + assertNull( + ChatMessage.reconcileMarmotLine(fresh = placeholder, existing = sent), + "a replay that cannot read our own message must leave its line alone", + ) + } + + @Test + fun `a commit still waiting may replace the placeholder it already wrote`() { + val pending = line( + id = 3, + messageType = ChatMessage.TYPE_PENDING_COMMIT, + content = "Pending Commit in epoch 4", + ) + val stillPending = line( + messageType = ChatMessage.TYPE_PENDING_COMMIT, + content = "Pending Commit in epoch 5", + ) + + val row = ChatMessage.reconcileMarmotLine(fresh = stillPending, existing = pending) + + assertEquals(3, row?.id, "one placeholder replacing another is still one line") + assertEquals("Pending Commit in epoch 5", row?.content) + } + + /** + * When a line was first seen, and when it was first stored, are facts about + * the reader rather than about the event. Reading the event again is not the + * message arriving again, so neither may be reset. + */ + @Test + fun `replacing a line keeps when it was first saved and seen`() { + val seenAt = Instant.fromEpochSeconds(2_000) + val storedAt = Instant.fromEpochSeconds(1_500) + + val placeholder = line( + id = 9, + messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER, + content = "Undecryptable Message", + viewedAt = seenAt, + savedAt = storedAt, + ) + val recovered = line( + messageType = "message", + content = "hello", + viewedAt = null, + savedAt = Instant.fromEpochSeconds(9_999), + ) + + val row = ChatMessage.reconcileMarmotLine(fresh = recovered, existing = placeholder) + + assertEquals(seenAt, row?.viewedAt, "a re-read must not mark a read message unread") + assertEquals(storedAt, row?.savedAt) + } + + /** + * The lines a replay is allowed to replace are exactly the ones that stand in + * for a message still to come. Adding a type to the transcript without adding + * it here silently makes those events unrecoverable; adding one here that + * reports something real makes them overwritable. + */ + @Test + fun `only the placeholder types count as unresolved`() { + assertEquals( + setOf(ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER, ChatMessage.TYPE_PENDING_COMMIT), + ChatMessage.UNRESOLVED_MARMOT_TYPES, + ) + + listOf("message", "processedCommit", "proposalStaged", "artifact", "dialect") + .forEach { messageType -> + assertNull( + ChatMessage.reconcileMarmotLine( + fresh = line( + messageType = ChatMessage.TYPE_UNDECRYPTABLE_OUTER_LAYER, + content = "Undecryptable Message", + ), + existing = line(id = 1, messageType = messageType, content = "something"), + ), + "a $messageType line reports something and must survive a replay", + ) + } + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt new file mode 100644 index 00000000..509d803b --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/database/model/MarmotGroupEventRoomFilterTest.kt @@ -0,0 +1,85 @@ +package press.mantra.compose.database.model + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotEquals +import kotlin.test.assertNull +import kotlin.time.Instant + +/** + * Which room a stored kind:445 belongs to. + * + * `NostrEventDao.getMarmotGroupNostrEventsByChatRoomId` finds a room's group + * events with `tags LIKE '%' || :chatRoomId || '%'`, because the tags are one + * serialised column and there is nothing better to match on. That is a prefilter + * and not a test: a group id can appear in a tag that is not the `h` tag, and an + * event matched that way belongs to a different room entirely. Feeding one of + * those to the replay would read another group's event against this group's MLS + * state. + * + * So the reindex confirms `groupId()` in Kotlin after the query. These pin the + * cases that confirmation is there for. + */ +class MarmotGroupEventRoomFilterTest { + + private val room = "a".repeat(64) + private val otherRoom = "d".repeat(64) + private val ephemeralSender = "e".repeat(64) + + private fun storedGroupEvent( + tags: Array>, + kind: Int = 445, + ) = NostrEvent( + id = "1".repeat(64), + pubKey = ephemeralSender, + kind = kind, + tags = tags, + content = "bm9uY2UrY2lwaGVydGV4dA==", + sig = "f".repeat(128), + createdAt = Instant.fromEpochSeconds(1_700_000_000), + ) + + @Test + fun `the h tag is what names the room`() { + val stored = storedGroupEvent(arrayOf(arrayOf("h", room))) + + assertEquals(room, stored.toGroupEvent(userPublicKey = ephemeralSender)?.groupId()) + } + + /** + * The case the `LIKE` cannot tell apart: this event is another room's, and + * only mentions ours in a tag that says nothing about routing. + */ + @Test + fun `an event mentioning the room outside its h tag belongs to the other room`() { + val stored = storedGroupEvent( + arrayOf( + arrayOf("h", otherRoom), + arrayOf("q", room), + ) + ) + + val groupId = stored.toGroupEvent(userPublicKey = ephemeralSender)?.groupId() + + assertEquals(otherRoom, groupId) + assertNotEquals(room, groupId, "the LIKE would match this event; groupId() is what rejects it") + } + + @Test + fun `an event with no h tag names no room`() { + val stored = storedGroupEvent(arrayOf(arrayOf("q", room))) + + assertNull(stored.toGroupEvent(userPublicKey = ephemeralSender)?.groupId()) + } + + /** + * Only kind:445 is a group event. The query pins the kind in SQL, and this is + * the other half of that: nothing else can be read as one by mistake. + */ + @Test + fun `an event of another kind is not a group event`() { + val stored = storedGroupEvent(tags = arrayOf(arrayOf("h", room)), kind = 9) + + assertNull(stored.toGroupEvent(userPublicKey = ephemeralSender)) + } +} diff --git a/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt new file mode 100644 index 00000000..51a0f4b0 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/press/mantra/compose/managers/MarmotReindexSweepTest.kt @@ -0,0 +1,173 @@ +package press.mantra.compose.managers + +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * The stopping rule a reindex sweep runs on. + * + * A replay is worth doing at all because the events depend on each other: the + * commit nobody could apply is what the room's later messages were waiting on. + * That is also what makes one ordered pass insufficient, and what makes "keep + * going" dangerous — a room full of events whose keys are gone would be swept + * forever. Both halves of that are pinned here. + * + * The sweep never sees a database. What counts as recovered is whatever the + * `isUnresolved` probe says afterwards, so these fakes are the real contract and + * not a stand-in for one. + */ +class MarmotReindexSweepTest { + + /** + * A backlog where each event is unlocked by the one before it, which is the + * shape a chain of out-of-order commits arrives in. + * + * [unlockedBy] is the event that has to be replayed successfully first; + * null means it can be read straight away. + */ + private class Backlog(unlocks: Map) { + private val unlockedBy = unlocks + private val resolved = mutableSetOf() + + /** Every event replayed, in order, across every pass. */ + val replayed = mutableListOf() + + suspend fun replay(id: String) { + replayed.add(id) + val blocker = unlockedBy.getValue(id) + if (blocker == null || blocker in resolved) { + resolved.add(id) + } + } + + suspend fun isUnresolved(id: String): Boolean = id !in resolved + } + + @Test + fun `a chain of events unlocking each other is read in as many passes as it takes`() = runTest { + // c waits on b, b waits on a. One pass in this order recovers only a. + val backlog = Backlog(mapOf("c" to "b", "b" to "a", "a" to null)) + + val report = MarmotReindexSweep.run( + stored = 10, + unresolved = listOf("c", "b", "a"), + replay = backlog::replay, + isUnresolved = backlog::isUnresolved, + ) + + assertEquals(3, report.recovered, "every event in the chain should have been recovered") + assertEquals(0, report.failed) + assertEquals(3, report.unresolved, "unresolved is what the sweep started with") + assertEquals(10, report.stored, "stored is reported as handed in") + } + + @Test + fun `an event that can never be read stops the sweep after one pass`() = runTest { + // Nothing unlocks these: the keys are gone, replaying achieves nothing. + val backlog = Backlog(mapOf("a" to "never", "b" to "never")) + + val report = MarmotReindexSweep.run( + stored = 2, + unresolved = listOf("a", "b"), + replay = backlog::replay, + isUnresolved = backlog::isUnresolved, + ) + + assertEquals(listOf("a", "b"), backlog.replayed, "a pass that recovers nothing must not repeat") + assertEquals(0, report.recovered) + assertEquals(2, report.failed) + } + + @Test + fun `events that can be read are still recovered alongside ones that cannot`() = runTest { + val backlog = Backlog(mapOf("stuck" to "never", "b" to "a", "a" to null)) + + val report = MarmotReindexSweep.run( + stored = 3, + unresolved = listOf("stuck", "b", "a"), + replay = backlog::replay, + isUnresolved = backlog::isUnresolved, + ) + + assertEquals(2, report.recovered) + assertEquals(1, report.failed, "the unreadable one is reported, not retried forever") + assertEquals(report.unresolved, report.recovered + report.failed, "every event is accounted for") + } + + /** + * The pass cap only bites on the shape where each pass recovers exactly one + * event, which is the only way the sweep can go quadratic. Six events in a + * chain replayed worst-first need six passes and get five. + */ + @Test + fun `the pass cap bounds a chain longer than it`() = runTest { + val chain = listOf("f", "e", "d", "c", "b", "a") + val backlog = Backlog( + mapOf("f" to "e", "e" to "d", "d" to "c", "c" to "b", "b" to "a", "a" to null) + ) + + val report = MarmotReindexSweep.run( + stored = chain.size, + unresolved = chain, + replay = backlog::replay, + isUnresolved = backlog::isUnresolved, + ) + + assertEquals(MarmotReindexSweep.DEFAULT_MAX_PASSES, report.recovered) + assertEquals(1, report.failed, "what the cap cut short is reported as still unread") + } + + @Test + fun `a replay that throws is survived by the rest of the sweep`() = runTest { + val resolved = mutableSetOf() + + val report = MarmotReindexSweep.run( + stored = 3, + unresolved = listOf("throws", "a", "b"), + replay = { id -> + if (id == "throws") error("no key for this epoch") + resolved.add(id) + }, + isUnresolved = { id -> id !in resolved }, + ) + + assertEquals(2, report.recovered, "a throw must not abandon the events behind it") + assertEquals(1, report.failed) + } + + @Test + fun `nothing unresolved does no work at all`() = runTest { + var replays = 0 + + val report = MarmotReindexSweep.run( + stored = 40, + unresolved = emptyList(), + replay = { replays++ }, + isUnresolved = { true }, + ) + + assertEquals(0, replays, "a room where nothing went wrong must be left alone") + assertTrue(report.isNoOp) + assertEquals(40, report.stored) + } + + /** + * The probe, not the replay call, is what counts a recovery: a group event can + * be replayed without error and still leave the room with nothing to show for + * it -- a commit rejected as a duplicate does exactly that. + */ + @Test + fun `a replay that raises nothing but resolves nothing is not counted as recovered`() = runTest { + val report = MarmotReindexSweep.run( + stored = 1, + unresolved = listOf("duplicateCommit"), + replay = { /* applies cleanly, writes no line */ }, + isUnresolved = { true }, + ) + + assertEquals(0, report.recovered) + assertEquals(1, report.failed) + } +} diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index e51ecfde..9d846a9f 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -63,6 +63,7 @@ kermit = { module = "co.touchlab:kermit", version.ref = "kermit" } kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" } kotlin-testJunit = { module = "org.jetbrains.kotlin:kotlin-test-junit", version.ref = "kotlin" } kotlinx-coroutinesSwing = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-swing", version.ref = "kotlinx-coroutines" } +kotlinx-coroutinesTest = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines" } kotlinx-datetime = { module = "org.jetbrains.kotlinx:kotlinx-datetime", version.ref = "kotlinx-datetime" } kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinxSerialization" } kotlinx-serialization-cbor = { module = "org.jetbrains.kotlinx:kotlinx-serialization-cbor", version.ref = "kotlinxSerialization" }