diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt index 1ff37322..bc35c1fe 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/GiftWrapPayloadDao.kt @@ -13,8 +13,27 @@ interface GiftWrapPayloadDao { @Query("SELECT * FROM GiftWrapPayload") suspend fun getAllGiftWrapUnsigned(): List - @Query("SELECT * FROM GiftWrapPayload WHERE publicKey = :publicKey AND giftWrapSealId IS NULL") - fun observeUnsealedGiftWrapPayloads(publicKey: String): Flow + /** + * Everything this key has queued for sealing and not yet sealed, oldest first. + * + * A backlog, not a head, for the reason spelled out on + * `MarmotInnerEventDao.observeUnprocessedMarmotInnerEvents`: nothing takes a row off + * this queue but a successful seal, so one payload that cannot be sealed used to be + * the whole queue. That is not hypothetical here -- `sealGiftWrapPayload` refuses a + * payload belonging to an MLS room outright, and a Welcome whose joiner published no + * key package produces no wraps at all. Either one returns leaving `giftWrapSealId` + * null, and every payload behind it waited on a row that would never move. + * + * `id` breaks the tie because `createdAt` is stored to the second; there was no + * ordering here at all before, so which row was "the head" was whatever SQLite + * happened to return first. + */ + @Query( + "SELECT * FROM GiftWrapPayload " + + "WHERE publicKey = :publicKey AND giftWrapSealId IS NULL " + + "ORDER BY createdAt ASC, id ASC" + ) + fun observeUnsealedGiftWrapPayloads(publicKey: String): Flow> /** * Every payload of the given kinds this device holds for a room, oldest first. diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt index 37b43f32..35d1682d 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/dao/MarmotInnerEventDao.kt @@ -9,8 +9,30 @@ import kotlinx.coroutines.flow.Flow @Dao interface MarmotInnerEventDao { - @Query("SELECT * FROM MarmotInnerEvent WHERE publicKey = :publicKey AND marmotGroupEventId IS NULL ORDER BY createdAt ASC") - fun observeUnprocessedMarmotInnerEvents(publicKey: String): Flow + /** + * Everything this key has queued for the group and not yet sent, oldest first. + * + * A backlog, not a head. The only exit from this queue is a successful send: the + * row is stamped with the group event it became, in the same transaction that + * writes that event. So a row that cannot be sent does not leave, and while the + * notary was handed one row at a time that row *was* the queue -- a message that + * will never encrypt (a room whose MLS state is gone, a direct message to a key + * that is no longer a member) sat at the head and every message queued behind it + * waited on it forever. Not just in its own room, either: the filter is the + * sender's key, so one wedged room silenced every group this device belongs to. + * Nothing reports it; the messages simply stay unsealed, which is what they are. + * + * Handing over the whole backlog lets the caller fail one row without losing the + * rest. It must still work through them in order and one at a time: each send + * ratchets the room's MLS state forward and saves it, and the next send in that + * room is encrypted against what the last one left behind. + * + * `id` breaks the tie because `createdAt` is stored to the second and a burst of + * sends shares one. Ordering that is only ever "some row with this timestamp" + * would let two passes disagree about what comes next. + */ + @Query("SELECT * FROM MarmotInnerEvent WHERE publicKey = :publicKey AND marmotGroupEventId IS NULL ORDER BY createdAt ASC, id ASC") + fun observeUnprocessedMarmotInnerEvents(publicKey: String): Flow> @Upsert suspend fun upsert(marmotInnerEvent: MarmotInnerEvent) 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 80b04c00..5152230c 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 @@ -328,7 +328,7 @@ class DatabaseChatRepository( return null } - override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow { + override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow> { return database.giftWrapPayloadDao().observeUnsealedGiftWrapPayloads(publicKey) } @@ -349,8 +349,10 @@ class DatabaseChatRepository( // is the only way to reach them, and `MarmotOutboundDao.deliveryWelcome` queues one // here deliberately. Every Welcome carries its room's nostrGroupId, so a refusal // keyed on `mlsGroupState` alone catches all of them and no invite is ever - // delivered -- and because the unsealed queue is a single-row flow, the refused - // Welcome sits at its head and blocks every payload behind it too. + // delivered. The refusal below no longer strands the rest of the queue behind it + // -- the notary is handed the whole backlog and refuses one payload at a time -- + // but a refused Welcome is still an invite that never arrives, so the exception + // for it stays. if (giftWrapPayload.kind != WelcomeEvent.KIND) { val chatRoom = database.chatRoomDao().findChatRoomById(giftWrapPayload.chatRoomId) if (chatRoom?.chatRoom?.mlsGroupState != null) { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt index 8a644e57..e4cf72a6 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/database/repository/DatabaseMarmotRepository.kt @@ -4,6 +4,7 @@ import press.mantra.compose.database.MantraDatabase import press.mantra.compose.database.model.MarmotInnerEvent import press.mantra.compose.database.model.MarmotKeyPackageBundle import press.mantra.compose.database.model.UnsignedNostrEvent +import press.mantra.compose.exceptions.MarmotMissingChatGroupException import press.mantra.compose.extensions.toHex import press.mantra.compose.nostr.Relays import press.mantra.compose.repository.MarmotRepository @@ -120,7 +121,7 @@ class DatabaseMarmotRepository( return database.marmotKeyPackageBundleDao().getAllMarmotKeyPackageBundles(publicKey) } - override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow { + override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow> { return database.marmotInnerEventDao().observeUnprocessedMarmotInnerEvents(publicKey) } @@ -128,23 +129,36 @@ class DatabaseMarmotRepository( return database.marmotKeyPackageBundleDao().observeActiveMarmotKeyPackageBundle(publicKey) } + /** + * Both lookups used to be `?.let`, so a row for a room this device holds no MLS state + * for returned quietly, having done nothing. Nothing marks such a row, so it stayed + * queued and was handed back on the next pass, forever -- and while the queue was a + * single row, forever meant every message queued behind it too. Saying so is the same + * call `MarmotOutboundDao.inviteMemberToChatRoom` makes, for the same reason: there is + * no group to encrypt to, and reporting success is a lie the sender acts on. + */ override suspend fun encryptAndSendMarmotInnerEvent( marmotInnerEvent: MarmotInnerEvent, nostrSignerSync: NostrSignerSync ) { - database.chatRoomDao().findChatRoomById(marmotInnerEvent.chatRoomId)?.let { localChatRoom -> - localChatRoom.chatRoom.toMlsGroup()?.let { mlsGroup -> - logger.d("mlsGroup: $mlsGroup") + val localChatRoom = database.chatRoomDao().findChatRoomById(marmotInnerEvent.chatRoomId) + ?: throw MarmotMissingChatGroupException( + "Cannot send ${marmotInnerEvent.id}: no chat room ${marmotInnerEvent.chatRoomId}" + ) - database.marmotOutboundDao().encryptAndSendMarmotInnerEvent( - localChatRoom = localChatRoom, - mlsGroup = mlsGroup, - marmotInnerEvent = marmotInnerEvent, - nostrSignerSync = nostrSignerSync - ) - } + val mlsGroup = localChatRoom.chatRoom.toMlsGroup() + ?: throw MarmotMissingChatGroupException( + "Cannot send ${marmotInnerEvent.id}: ${marmotInnerEvent.chatRoomId} holds no MLS group state" + ) - } + logger.d("mlsGroup: $mlsGroup") + + database.marmotOutboundDao().encryptAndSendMarmotInnerEvent( + localChatRoom = localChatRoom, + mlsGroup = mlsGroup, + marmotInnerEvent = marmotInnerEvent, + nostrSignerSync = nostrSignerSync + ) } private fun generateKeyPackage( 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 57592dd4..dcec7bec 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/ChatRepository.kt @@ -108,7 +108,14 @@ interface ChatRepository { fun updateChatRoomSubject(publicKey: String, subject: String): ChatRoom? - suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow + /** + * The whole unsealed backlog for [publicKey], oldest first. + * + * See `GiftWrapPayloadDao.observeUnsealedGiftWrapPayloads`: a payload that cannot be + * sealed used to be the entire queue. Callers work through it and a refusal costs + * only the payload it refuses. + */ + suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow> suspend fun sealGiftWrapPayload(giftWrapPayload: GiftWrapPayload, nostrSignerSync: NostrSignerSync) @@ -241,7 +248,7 @@ interface ChatRepository { return null } - override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow { + override suspend fun observeUnsealedGiftWrapPayloads(publicKey: HexKey): Flow> { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt index 44deb99f..c9875665 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/repository/MarmotRepository.kt @@ -19,7 +19,14 @@ interface MarmotRepository { suspend fun getMarmotKeyPackageBundles(publicKey: HexKey): List - suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow + /** + * The whole outbound backlog for [publicKey], oldest first. + * + * See `MarmotInnerEventDao.observeUnprocessedMarmotInnerEvents` for why this is a + * list: a row that cannot be sent used to be the entire queue. Callers work + * through it in order, and a row that fails costs only itself. + */ + suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow> suspend fun observeActiveMarmotKeyPackageBundle(publicKey: HexKey): Flow @@ -46,7 +53,7 @@ interface MarmotRepository { return emptyList() } - override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow { + override suspend fun observeUnprocessedMarmotInnerEvents(publicKey: HexKey): Flow> { TODO("Not yet implemented") } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt index d87b080d..6875c636 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/NotaryViewModel.kt @@ -151,6 +151,16 @@ class NotaryViewModel( } + /** + * Seal everything queued for sealing, not just the oldest thing. + * + * Same shape as [observeUnprocessedMarmotInnerEvents] and a harder stop. A payload + * `sealGiftWrapPayload` declines to seal keeps `giftWrapSealId` null, so it came back + * as the head of the next emission -- and `GiftWrapPayload.equals` is a plain value + * comparison, so `distinctUntilChanged` read that as nothing having changed and + * dropped it. One refusal and nothing on this queue was collected again for the rest + * of the session, whatever was queued afterwards. + */ private suspend fun observeUnsealedGiftWrapPayloads( keyPair: KeyPair ) { @@ -160,8 +170,8 @@ class NotaryViewModel( logger.i { "observeUnsealedGiftWrapPayloads: ${keyPair.pubKey.toHexKey()}" } chatRepository.observeUnsealedGiftWrapPayloads( publicKey = keyPair.pubKey.toHexKey() - ).distinctUntilChanged().collect { giftWrapPayloadOrNull -> - giftWrapPayloadOrNull?.let { giftWrapPayload -> + ).collect { giftWrapPayloads -> + giftWrapPayloads.forEach { giftWrapPayload -> guardNotarization("gift wrap payload ${giftWrapPayload.id}") { logger.d("seal and deliver giftWrapPayload: $giftWrapPayload") @@ -174,6 +184,28 @@ class NotaryViewModel( } } + /** + * Encrypt and send everything a marmot room has queued, not just the oldest thing. + * + * The queue used to arrive one row at a time and the row only left it by being sent, + * so a message that could never be encrypted was a stopper: it came back at the head + * of every later emission and everything behind it -- in every room this key belongs + * to -- waited on a send that was never going to happen. Those messages show in the + * chat as unsealed, and stay that way. + * + * `distinctUntilChanged` went with it. A queue re-emits because its table changed, + * which is the signal to look again; comparing that emission against the last one + * asks "is this new work?" when the question is "is there work left?". Here it + * happened to suppress nothing -- `MarmotInnerEvent.equals` compares an `@Ignore`d + * per-instance Logger, so no two reads of a row are ever equal -- which is not a + * property to leave a queue standing on. Next door, where the payload's equals is an + * honest value comparison, it dropped every emission after the first refusal. + * + * Serially, in the order the query gives them: each send ratchets its room's MLS + * state forward and writes it back, and `encryptAndSendMarmotInnerEvent` reads that + * state fresh per row. Two sends for one room in parallel would encrypt from the same + * generation and one of them would be undecryptable. + */ private suspend fun observeUnprocessedMarmotInnerEvents( keyPair: KeyPair ) { @@ -183,8 +215,12 @@ class NotaryViewModel( logger.i { "observeUnprocessedMarmotInnerEvents: ${keyPair.pubKey.toHexKey()}" } marmotRepository.observeUnprocessedMarmotInnerEvents( publicKey = keyPair.pubKey.toHexKey() - ).distinctUntilChanged().collect { marmotInnerEventOrNull -> - marmotInnerEventOrNull?.let { marmotInnerEvent -> + ).collect { marmotInnerEvents -> + marmotInnerEvents.forEach { marmotInnerEvent -> + // Per row, so one that cannot be encrypted costs only itself. It stays + // queued and is tried again on the next pass -- a room whose state has + // not caught up yet is worth retrying, and one that never will is at + // least no longer standing in front of anybody. guardNotarization("marmot inner event ${marmotInnerEvent.id}") { logger.d("encrypt and broadcast: $marmotInnerEvent") diff --git a/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt b/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt new file mode 100644 index 00000000..56dab9c0 --- /dev/null +++ b/composeApp/src/jvmTest/kotlin/press/mantra/compose/database/dao/MarmotOutboundQueueJvmTest.kt @@ -0,0 +1,301 @@ +package press.mantra.compose.database.dao + +import androidx.room3.Room +import com.vitorpamplona.quartz.marmot.mip01Groups.MarmotGroupData +import com.vitorpamplona.quartz.marmot.mls.group.MlsGroup +import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray +import com.vitorpamplona.quartz.nip01Core.core.toHexKey +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nipC7Chats.ChatEvent +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking +import press.mantra.compose.database.MantraDatabase +import press.mantra.compose.database.builder.getRoomDatabase +import press.mantra.compose.database.model.ChatRoom +import press.mantra.compose.database.model.MarmotInnerEvent +import press.mantra.compose.database.model.NostrEvent +import press.mantra.compose.database.model.Profile +import press.mantra.compose.database.repository.DatabaseMarmotRepository +import press.mantra.compose.exceptions.MarmotMissingChatGroupException +import press.mantra.compose.extensions.toHex +import press.mantra.compose.nostr.Relays +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlin.time.Instant + +/** + * The outbound queue the notary drains, and what a message it cannot send does to the rest. + * + * Nothing takes a row off this queue but a successful send: `encryptAndSendMarmotInnerEvent` + * stamps the row with the group event it became, in the transaction that writes that event. + * So a row that can never be sent never leaves, and while the queue handed back one row at a + * time, that row was the queue. Every message behind it -- in every room the key belongs to, + * because the filter is the sender rather than the room -- waited on a send that was not + * coming, and showed in the chat as unsealed for as long as the user cared to look. + * + * These stand the two rooms side by side: one holding real MLS state, one holding none, which + * is the shape a room restored from an inbound gift wrap has. The unsendable row is queued + * *first* on purpose. Under the old queue it was the only row the notary ever saw. + */ +class MarmotOutboundQueueJvmTest { + + private val db: MantraDatabase = getRoomDatabase( + Room.inMemoryDatabaseBuilder() + ) + + @AfterTest + fun closeDb() = db.close() + + private val keyPair = KeyPair() + private val user = keyPair.pubKey.toHexKey() + private val signer = NostrSignerSync(keyPair) + + private val repository = DatabaseMarmotRepository( + database = db, + scope = CoroutineScope(Dispatchers.Default), + ) + + private val statelessRoomId = "a".repeat(64) + private val mlsRoomId = "b".repeat(64) + + /** Participant/ChatRoom rows point at a Profile, which points at the event it came from. */ + private suspend fun seedProfile() { + val nostrEventId = "c".repeat(64) + db.nostrEventDao().upsert( + NostrEvent( + id = nostrEventId, + pubKey = user, + kind = 0, + tags = emptyArray(), + content = "{}", + sig = "0".repeat(128), + ) + ) + db.profileDao().upsert(Profile(publicKey = user, userName = "user", nostrEventId = nostrEventId)) + } + + /** A room with `mlsGroupState = null`: there is no group here to encrypt anything to. */ + private suspend fun seedStatelessRoom() { + db.chatRoomDao().upsert( + ChatRoom( + id = statelessRoomId, + userPublicKey = user, + subject = "a room restored without its group", + description = null, + mlsGroupState = null, + ) + ) + } + + /** A room holding real MLS state, as a room this device created does. */ + private suspend fun seedMlsRoom() { + val mlsGroup = MlsGroup.create( + identity = user.hexToByteArray(), + initialExtensions = listOf( + MarmotGroupData.bootstrap( + nostrGroupId = mlsRoomId, + creatorPubKey = user, + outboxRelays = Relays.DefaultDMRelayList.map { it.url }, + ).toExtension() + ), + ) + db.chatRoomDao().upsert( + ChatRoom( + id = mlsRoomId, + userPublicKey = user, + subject = "a room this device created", + description = null, + mlsGroupState = mlsGroup.saveState().encodeTls().toHex(), + ) + ) + } + + private suspend fun queue(id: String, chatRoomId: String, text: String, atEpochSeconds: Long): MarmotInnerEvent { + val row = MarmotInnerEvent( + id = id, + publicKey = user, + chatRoomId = chatRoomId, + kind = ChatEvent.KIND, + tags = emptyArray(), + content = text, + createdAt = Instant.fromEpochSeconds(atEpochSeconds), + ) + db.marmotInnerEventDao().upsert(row) + return row + } + + private suspend fun backlog() = db.marmotInnerEventDao() + .observeUnprocessedMarmotInnerEvents(user) + .first() + + /** + * What the notary does with an emission, reduced to the part under test: every row, in + * order, each one's failure its own. Returns what failed, so a test can assert that a row + * was refused rather than quietly skipped. + */ + private suspend fun drain(): List = backlog().mapNotNull { row -> + runCatching { repository.encryptAndSendMarmotInnerEvent(row, signer) }.exceptionOrNull() + } + + @Test + fun `the queue hands back the whole backlog, oldest first`() = runBlocking { + seedProfile() + seedMlsRoom() + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "second", atEpochSeconds = 200) + queue(id = "1".repeat(64), chatRoomId = mlsRoomId, text = "first", atEpochSeconds = 100) + + assertEquals( + listOf("first", "second"), + backlog().map { it.content }, + "the notary is handed the backlog to work through, not just its head", + ) + } + + /** + * `createdAt` is stored to the second, so a burst of sends shares one. Without the `id` + * tiebreak the order within that second is SQLite's business, and two passes over the same + * backlog need not agree on it. + */ + @Test + fun `rows sharing a second come back in a stable order`() = runBlocking { + seedProfile() + seedMlsRoom() + queue(id = "f".repeat(64), chatRoomId = mlsRoomId, text = "f", atEpochSeconds = 100) + queue(id = "0".repeat(64), chatRoomId = mlsRoomId, text = "0", atEpochSeconds = 100) + queue(id = "7".repeat(64), chatRoomId = mlsRoomId, text = "7", atEpochSeconds = 100) + + assertEquals(listOf("0", "7", "f"), backlog().map { it.content }) + assertEquals(listOf("0", "7", "f"), backlog().map { it.content }, "the order changed between passes") + } + + /** + * Both lookups were `?.let`, so this returned having done nothing at all: no send, no + * error, and no mark on the row -- which is what made it a permanent head of the queue. + */ + @Test + fun `a message for a room with no mls state is refused rather than ignored`() = runBlocking { + seedProfile() + seedStatelessRoom() + val row = queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "nowhere to send this", atEpochSeconds = 100) + + assertFailsWith { + repository.encryptAndSendMarmotInnerEvent(row, signer) + } + } + + @Test + fun `a message for a room this device does not have is refused rather than ignored`() = runBlocking { + seedProfile() + val row = queue(id = "1".repeat(64), chatRoomId = "d".repeat(64), text = "no such room", atEpochSeconds = 100) + + assertFailsWith { + repository.encryptAndSendMarmotInnerEvent(row, signer) + } + } + + /** + * The one that matters. The unsendable row is the oldest, so it is what a queue that + * yields only its head would offer forever; the message behind it is in a different room + * and has everything it needs to go out. + */ + @Test + fun `a message that cannot be sent no longer holds up the ones behind it`() = runBlocking { + seedProfile() + seedStatelessRoom() + seedMlsRoom() + queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "the stopper", atEpochSeconds = 100) + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "sent anyway", atEpochSeconds = 200) + + val failures = drain() + + assertEquals(1, failures.size, "exactly the stopper should have failed") + assertTrue(failures.single() is MarmotMissingChatGroupException) + + val sent = assertNotNull(db.marmotInnerEventDao().getByChatRoomAndKinds(mlsRoomId, listOf(ChatEvent.KIND)).singleOrNull()) + assertNotNull(sent.marmotGroupEventId, "the message behind the stopper was never sent") + assertEquals( + listOf("the stopper"), + backlog().map { it.content }, + "the stopper stays queued and nothing else does", + ) + } + + /** + * Sent means handed to the broadcaster. `BroadcastNostrEventRequest` rows in "pending" are + * the only thing that puts a kind:445 on a relay, so a row that got a group event but no + * request is still a message nobody receives -- see 0211764. + */ + @Test + fun `a message past the stopper is queued for the relays`() = runBlocking { + seedProfile() + seedStatelessRoom() + seedMlsRoom() + queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "the stopper", atEpochSeconds = 100) + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "sent anyway", atEpochSeconds = 200) + + drain() + + val groupEventId = assertNotNull( + db.marmotInnerEventDao().getByChatRoomAndKinds(mlsRoomId, listOf(ChatEvent.KIND)).single().marmotGroupEventId + ) + val requests = db.broadcastNostrEventRequestDao().getAllBroadcastNostrEventRequests() + .filter { it.nostrEventId == groupEventId } + + assertEquals( + Relays.DefaultDMRelayList.size, + requests.size, + "one broadcast request per relay is what actually sends it", + ) + assertTrue(requests.all { it.status == "pending" }, "pending is the only status the broadcaster looks at") + } + + /** + * Two messages queued for the same room go out in order and each under its own generation: + * the send ratchets the room's MLS state forward and writes it back, and the next one is + * encrypted against what it left behind. A drain that read the room once and reused it + * would send both from the same generation, and the group could only read one of them. + */ + @Test + fun `two messages for one room each ratchet the group forward`() = runBlocking { + seedProfile() + seedMlsRoom() + val stateBefore = assertNotNull(db.chatRoomDao().findChatRoomById(mlsRoomId)).chatRoom.mlsGroupState + queue(id = "1".repeat(64), chatRoomId = mlsRoomId, text = "first", atEpochSeconds = 100) + queue(id = "2".repeat(64), chatRoomId = mlsRoomId, text = "second", atEpochSeconds = 200) + + assertEquals(emptyList(), drain()) + + assertEquals(emptyList(), backlog(), "both messages should have left the queue") + val sent = db.marmotInnerEventDao().getByChatRoomAndKinds(mlsRoomId, listOf(ChatEvent.KIND)) + assertEquals(2, sent.mapNotNull { it.marmotGroupEventId }.distinct().size, "both were sent as the same event") + assertTrue( + assertNotNull(db.chatRoomDao().findChatRoomById(mlsRoomId)).chatRoom.mlsGroupState != stateBefore, + "the ratchet never reached the database", + ) + } + + /** + * A refused send must leave nothing behind. `encryptAndSendMarmotInnerEvent` is one + * transaction, so a failure that had written a group event or a broadcast request would + * have the room believing it sent something the members will never see. + */ + @Test + fun `a refused send writes nothing`() = runBlocking { + seedProfile() + seedStatelessRoom() + val row = queue(id = "1".repeat(64), chatRoomId = statelessRoomId, text = "the stopper", atEpochSeconds = 100) + + runCatching { repository.encryptAndSendMarmotInnerEvent(row, signer) } + + assertNull(backlog().single().marmotGroupEventId, "the refused row was marked as sent") + assertEquals(emptyList(), db.broadcastNostrEventRequestDao().getAllBroadcastNostrEventRequests()) + } +}