From feaaf31b92548e8e0050d06ba93e155000cff722 Mon Sep 17 00:00:00 2001 From: Jacob Persson <7156+typfel@users.noreply.github.com> Date: Mon, 18 Sep 2023 16:53:26 +0200 Subject: [PATCH] fix: improve stability of MLS 1-1 conversations (#2063) * fix: don't fail migration query if messages has already been copied * fix: don't fail the slow sync on a non-recoverable error when resolving 1-1s * fix: don't fail the slow sync when etablishing 1-1 fails due to missing key packages * fix: re-use existing mls group if it exists * fix: establish 1-1 also with other self clients * test: add missing test for establishing 1-1 * chore: fix detekt --- .../kalium/logic/CoreCryptoExceptionMapper.kt | 1 + .../com/wire/kalium/logic/CoreFailure.kt | 2 + .../conversation/MLSConversationRepository.kt | 13 +++- .../JoinExistingMLSConversationUseCase.kt | 2 +- .../JoinExistingMLSConversationsUseCase.kt | 13 ++++ .../conversation/mls/OneOnOneResolver.kt | 25 +++++--- ...OrCreateOneToOneConversationUseCaseTest.kt | 2 +- .../JoinExistingMLSConversationUseCaseTest.kt | 59 ++++++++++++++----- ...serveConversationListDetailsUseCaseTest.kt | 6 +- .../MLSOneOnOneConversationResolverTest.kt | 2 +- .../conversation/mls/OneOnOneMigratorTest.kt | 4 +- .../logic/framework/TestConversation.kt | 4 +- .../framework/TestConversationDetails.kt | 2 +- .../com/wire/kalium/persistence/Messages.sq | 2 +- 14 files changed, 101 insertions(+), 36 deletions(-) diff --git a/logic/src/commonJvmAndroid/kotlin/com/wire/kalium/logic/CoreCryptoExceptionMapper.kt b/logic/src/commonJvmAndroid/kotlin/com/wire/kalium/logic/CoreCryptoExceptionMapper.kt index 2e2783af545..97cd5ceac2e 100644 --- a/logic/src/commonJvmAndroid/kotlin/com/wire/kalium/logic/CoreCryptoExceptionMapper.kt +++ b/logic/src/commonJvmAndroid/kotlin/com/wire/kalium/logic/CoreCryptoExceptionMapper.kt @@ -27,6 +27,7 @@ actual fun mapMLSException(exception: Exception): MLSFailure = is CryptoError.DuplicateMessage -> MLSFailure.DuplicateMessage is CryptoError.SelfCommitIgnored -> MLSFailure.SelfCommitIgnored is CryptoError.UnmergedPendingGroup -> MLSFailure.UnmergedPendingGroup + is CryptoError.ConversationAlreadyExists -> MLSFailure.ConversationAlreadyExists else -> MLSFailure.Generic(exception) } } else { diff --git a/logic/src/commonMain/kotlin/com/wire/kalium/logic/CoreFailure.kt b/logic/src/commonMain/kotlin/com/wire/kalium/logic/CoreFailure.kt index fec4e188df8..f604af891d7 100644 --- a/logic/src/commonMain/kotlin/com/wire/kalium/logic/CoreFailure.kt +++ b/logic/src/commonMain/kotlin/com/wire/kalium/logic/CoreFailure.kt @@ -177,6 +177,8 @@ interface MLSFailure : CoreFailure { object UnmergedPendingGroup : MLSFailure + object ConversationAlreadyExists : MLSFailure + object ConversationDoesNotSupportMLS : MLSFailure class Generic(internal val exception: Exception) : MLSFailure { diff --git a/logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/MLSConversationRepository.kt b/logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/MLSConversationRepository.kt index 100254cfb85..f07a9fbd3fd 100644 --- a/logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/MLSConversationRepository.kt +++ b/logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/MLSConversationRepository.kt @@ -23,6 +23,7 @@ import com.wire.kalium.cryptography.CryptoQualifiedClientId import com.wire.kalium.cryptography.CryptoQualifiedID import com.wire.kalium.logger.obfuscateId import com.wire.kalium.logic.CoreFailure +import com.wire.kalium.logic.MLSFailure import com.wire.kalium.logic.NetworkFailure import com.wire.kalium.logic.data.client.MLSClientProvider import com.wire.kalium.logic.data.event.Event @@ -479,6 +480,12 @@ internal class MLSConversationDataSource( idMapper.toCryptoModel(groupID), publicKeys.map { mlsPublicKeysMapper.toCrypto(it) } ) + }.flatMapLeft { + if (it is MLSFailure.ConversationAlreadyExists) { + Either.Right(Unit) + } else { + Either.Left(it) + } } }.flatMap { internalAddMemberToMLSGroup(groupID, members, retryOnStaleMessage = false).onFailure { @@ -566,9 +573,13 @@ internal class MLSConversationDataSource( kaliumLogger.w("Discarding the failed commit.") return mlsClientProvider.getMLSClient().flatMap { mlsClient -> - wrapMLSRequest { + @Suppress("TooGenericExceptionCaught") + try { mlsClient.clearPendingCommit(idMapper.toCryptoModel(groupID)) + } catch (error: Throwable) { + kaliumLogger.e("Discarding pending commit failed: $error") } + Either.Right(Unit) } } } diff --git a/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCase.kt b/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCase.kt index 5eff9c9bf8f..9007ba1cecb 100644 --- a/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCase.kt +++ b/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCase.kt @@ -149,7 +149,7 @@ internal class JoinExistingMLSConversationUseCaseImpl( conversationRepository.getConversationMembers(conversation.id).flatMap { members -> mlsConversationRepository.establishMLSGroup( protocol.groupId, - listOf(members.first()) + members ) } } diff --git a/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationsUseCase.kt b/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationsUseCase.kt index 686bfbc71d6..b34a9a42eb7 100644 --- a/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationsUseCase.kt +++ b/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationsUseCase.kt @@ -25,6 +25,7 @@ import com.wire.kalium.logic.data.conversation.ConversationRepository import com.wire.kalium.logic.featureFlags.FeatureSupport import com.wire.kalium.logic.functional.Either import com.wire.kalium.logic.functional.flatMap +import com.wire.kalium.logic.functional.flatMapLeft import com.wire.kalium.logic.functional.foldToEitherWhileRight import com.wire.kalium.logic.functional.getOrElse import com.wire.kalium.logic.kaliumLogger @@ -57,6 +58,18 @@ internal class JoinExistingMLSConversationsUseCaseImpl( return pendingConversations.map { conversation -> joinExistingMLSConversationUseCase(conversation.id) + .flatMapLeft { + if (it is CoreFailure.NoKeyPackagesAvailable) { + kaliumLogger.w( + "Failed to establish mls group for ${conversation.id.toLogString()} " + + "since some participants are out of key packages, skipping." + ) + Either.Right(Unit) + } else { + Either.Left(it) + } + + } }.foldToEitherWhileRight(Unit) { value, _ -> value } diff --git a/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneResolver.kt b/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneResolver.kt index f6a899cb78e..1c8f7d43241 100644 --- a/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneResolver.kt +++ b/logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneResolver.kt @@ -18,6 +18,7 @@ package com.wire.kalium.logic.feature.conversation.mls import com.wire.kalium.logic.CoreFailure +import com.wire.kalium.logic.NetworkFailure import com.wire.kalium.logic.StorageFailure import com.wire.kalium.logic.data.id.ConversationId import com.wire.kalium.logic.data.sync.IncrementalSyncRepository @@ -67,7 +68,23 @@ internal class OneOnOneResolverImpl( val usersWithOneOnOne = userRepository.getUsersWithOneOnOneConversation() kaliumLogger.i("Resolving one-on-one protocol for ${usersWithOneOnOne.size} user(s)") return usersWithOneOnOne.foldToEitherWhileRight(Unit) { item, _ -> - resolveOneOnOneConversationWithUser(item).map { } + resolveOneOnOneConversationWithUser(item).flatMapLeft { + when (it) { + is CoreFailure.NoKeyPackagesAvailable, + is NetworkFailure.ServerMiscommunication, + is NetworkFailure.FederatedBackendFailure, + is CoreFailure.NoCommonProtocolFound + -> { + kaliumLogger.e("Resolving one-on-one failed $it, skipping") + Either.Right(Unit) + } + + else -> { + kaliumLogger.e("Resolving one-on-one failed $it, retrying") + Either.Left(it) + } + } + }.map { } } } @@ -91,12 +108,6 @@ internal class OneOnOneResolverImpl( SupportedProtocol.PROTEUS -> oneOnOneMigrator.migrateToProteus(user) SupportedProtocol.MLS -> oneOnOneMigrator.migrateToMLS(user) } - }.flatMapLeft { - if (it is CoreFailure.NoCommonProtocolFound) { - // TODO mark conversation as read only - Either.Right(Unit) - } - Either.Left(it) } } } diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/GetOrCreateOneToOneConversationUseCaseTest.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/GetOrCreateOneToOneConversationUseCaseTest.kt index 5d6f5e4f789..4015859c8b3 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/GetOrCreateOneToOneConversationUseCaseTest.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/GetOrCreateOneToOneConversationUseCaseTest.kt @@ -123,6 +123,6 @@ class GetOrCreateOneToOneConversationUseCaseTest { private companion object { val OTHER_USER = TestUser.OTHER val OTHER_USER_ID = OTHER_USER.id - val CONVERSATION = TestConversation.ONE_ON_ONE + val CONVERSATION = TestConversation.ONE_ON_ONE() } } diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCaseTest.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCaseTest.kt index 40ca4891ed8..49b60592245 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCaseTest.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/JoinExistingMLSConversationUseCaseTest.kt @@ -27,11 +27,11 @@ import com.wire.kalium.logic.data.conversation.DecryptedMessageBundle import com.wire.kalium.logic.data.conversation.MLSConversationRepository import com.wire.kalium.logic.data.id.ConversationId import com.wire.kalium.logic.data.id.GroupID +import com.wire.kalium.logic.data.user.UserId import com.wire.kalium.logic.featureFlags.FeatureSupport import com.wire.kalium.logic.framework.TestConversation +import com.wire.kalium.logic.framework.TestUser import com.wire.kalium.logic.functional.Either -import com.wire.kalium.logic.sync.receiver.conversation.message.MLSMessageUnpacker -import com.wire.kalium.logic.sync.receiver.conversation.message.MessageUnpackResult import com.wire.kalium.logic.util.shouldFail import com.wire.kalium.logic.util.shouldSucceed import com.wire.kalium.network.api.base.authenticated.conversation.ConversationApi @@ -109,7 +109,7 @@ class JoinExistingMLSConversationUseCaseTest { } @Test - fun givenGroupConversationWithZeroEpoch_whenInvokingUseCase_ThenDoNotEstablishGroup() = + fun givenGroupConversationWithZeroEpoch_whenInvokingUseCase_ThenDoNotEstablishMlsGroup() = runTest { val (arrangement, joinExistingMLSConversationsUseCase) = Arrangement() .withIsMLSSupported(true) @@ -127,7 +127,7 @@ class JoinExistingMLSConversationUseCaseTest { } @Test - fun givenSelfConversationWithZeroEpoch_whenInvokingUseCase_ThenEstablishGroup() = + fun givenSelfConversationWithZeroEpoch_whenInvokingUseCase_ThenEstablishMlsGroup() = runTest { val (arrangement, joinExistingMLSConversationsUseCase) = Arrangement() .withIsMLSSupported(true) @@ -144,6 +144,26 @@ class JoinExistingMLSConversationUseCaseTest { .wasInvoked(once) } + @Test + fun givenOneOnOneConversationWithZeroEpoch_whenInvokingUseCase_ThenEstablishMlsGroup() = + runTest { + val members = listOf(TestUser.USER_ID, TestUser.OTHER_USER_ID) + val (arrangement, joinExistingMLSConversationsUseCase) = Arrangement() + .withIsMLSSupported(true) + .withHasRegisteredMLSClient(true) + .withGetConversationsByIdSuccessful(Arrangement.MLS_UNESTABLISHED_ONE_ONE_ONE_CONVERSATION) + .withGetConversationMembersSuccessful(members) + .withEstablishMLSGroupSuccessful() + .arrange() + + joinExistingMLSConversationsUseCase(Arrangement.MLS_UNESTABLISHED_ONE_ONE_ONE_CONVERSATION.id).shouldSucceed() + + verify(arrangement.mlsConversationRepository) + .suspendFunction(arrangement.mlsConversationRepository::establishMLSGroup) + .with(eq(Arrangement.GROUP_ID_ONE_ON_ONE), eq(members)) + .wasInvoked(once) + } + @Test fun givenOutOfDateEpochFailure_whenInvokingUseCase_ThenRetryWithNewEpoch() = runTest { val (arrangement, joinExistingMLSConversationsUseCase) = Arrangement() @@ -200,16 +220,12 @@ class JoinExistingMLSConversationUseCaseTest { @Mock val mlsConversationRepository = mock(classOf()) - @Mock - val mlsMessageUnpacker = mock(classOf()) - fun arrange() = this to JoinExistingMLSConversationUseCaseImpl( featureSupport, conversationApi, clientRepository, conversationRepository, - mlsConversationRepository, - mlsMessageUnpacker + mlsConversationRepository ) @Suppress("MaxLineLength") @@ -228,6 +244,13 @@ class JoinExistingMLSConversationUseCaseTest { .then { Either.Right(Unit) } } + fun withGetConversationMembersSuccessful(members: List) = apply { + given(conversationRepository) + .suspendFunction(conversationRepository::getConversationMembers) + .whenInvokedWith(anything()) + .then { Either.Right(members) } + } + fun withEstablishMLSGroupSuccessful() = apply { given(mlsConversationRepository) .suspendFunction(mlsConversationRepository::establishMLSGroup) @@ -270,13 +293,6 @@ class JoinExistingMLSConversationUseCaseTest { .thenReturn(Either.Right(result)) } - fun withUnpackMlsBundleSuccessful() = apply { - given(mlsMessageUnpacker) - .suspendFunction(mlsMessageUnpacker::unpackMlsBundle) - .whenInvokedWith(anything()) - .thenReturn(MessageUnpackResult.HandshakeMessage) - } - companion object { val PUBLIC_GROUP_STATE = "public_group_state".encodeToByteArray() @@ -303,6 +319,7 @@ class JoinExistingMLSConversationUseCaseTest { val GROUP_ID1 = GroupID("group1") val GROUP_ID2 = GroupID("group2") val GROUP_ID3 = GroupID("group3") + val GROUP_ID_ONE_ON_ONE = GroupID("group-one-on-ne") val GROUP_ID_SELF = GroupID("group-self") val MLS_CONVERSATION1 = TestConversation.GROUP( @@ -344,6 +361,16 @@ class JoinExistingMLSConversationUseCaseTest { cipherSuite = Conversation.CipherSuite.MLS_128_DHKEMX25519_AES128GCM_SHA256_Ed25519 ) ).copy(id = ConversationId("self", "domain")) + + val MLS_UNESTABLISHED_ONE_ONE_ONE_CONVERSATION = TestConversation.ONE_ON_ONE( + Conversation.ProtocolInfo.MLS( + GROUP_ID_ONE_ON_ONE, + Conversation.ProtocolInfo.MLSCapable.GroupState.PENDING_JOIN, + epoch = 0UL, + keyingMaterialLastUpdate = DateTimeUtil.currentInstant(), + cipherSuite = Conversation.CipherSuite.MLS_128_DHKEMX25519_AES128GCM_SHA256_Ed25519 + ) + ).copy(id = ConversationId("one-on-one", "domain")) } } } diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/ObserveConversationListDetailsUseCaseTest.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/ObserveConversationListDetailsUseCaseTest.kt index 05fd44e0560..84b839fcc9c 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/ObserveConversationListDetailsUseCaseTest.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/ObserveConversationListDetailsUseCaseTest.kt @@ -179,7 +179,7 @@ class ObserveConversationListDetailsUseCaseTest { @Test fun givenSomeConversationsDetailsAreUpdated_whenObservingDetailsList_thenTheUpdateIsPropagatedThroughTheFlow() = runTest { // Given - val oneOnOneConversation = TestConversation.ONE_ON_ONE + val oneOnOneConversation = TestConversation.ONE_ON_ONE() val groupConversation = TestConversation.GROUP() val conversations = listOf(groupConversation, oneOnOneConversation) val fetchArchivedConversations = false @@ -342,9 +342,9 @@ class ObserveConversationListDetailsUseCaseTest { @Test fun givenConversationDetailsFailure_whenObservingDetailsList_thenIgnoreConversationWithFailure() = runTest { // Given - val successConversation = TestConversation.ONE_ON_ONE.copy(id = ConversationId("successId", "domain")) + val successConversation = TestConversation.ONE_ON_ONE().copy(id = ConversationId("successId", "domain")) val successConversationDetails = TestConversationDetails.CONVERSATION_ONE_ONE.copy(conversation = successConversation) - val failureConversation = TestConversation.ONE_ON_ONE.copy(id = ConversationId("failedId", "domain")) + val failureConversation = TestConversation.ONE_ON_ONE().copy(id = ConversationId("failedId", "domain")) val fetchArchivedConversations = false val (_, observeConversationsUseCase) = Arrangement() diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/MLSOneOnOneConversationResolverTest.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/MLSOneOnOneConversationResolverTest.kt index 638a1a587c8..4eca8bb46e2 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/MLSOneOnOneConversationResolverTest.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/MLSOneOnOneConversationResolverTest.kt @@ -152,7 +152,7 @@ class MLSOneOnOneConversationResolverTest { private companion object { private val userId = TestUser.USER_ID - private val CONVERSATION_ONE_ON_ONE_PROTEUS = TestConversation.ONE_ON_ONE.copy( + private val CONVERSATION_ONE_ON_ONE_PROTEUS = TestConversation.ONE_ON_ONE().copy( id = ConversationId("one-on-one-proteus", "test"), protocol = Conversation.ProtocolInfo.Proteus, ) diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneMigratorTest.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneMigratorTest.kt index 6d88e704b72..3585bdaeff9 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneMigratorTest.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/mls/OneOnOneMigratorTest.kt @@ -93,7 +93,7 @@ class OneOnOneMigratorTest { val (arrangement, oneOneMigrator) = arrange { withGetOneOnOneConversationsWithOtherUserReturning(Either.Right(emptyList())) - withCreateGroupConversationReturning(Either.Right(TestConversation.ONE_ON_ONE)) + withCreateGroupConversationReturning(Either.Right(TestConversation.ONE_ON_ONE())) withUpdateOneOnOneConversationReturning(Either.Right(Unit)) } @@ -107,7 +107,7 @@ class OneOnOneMigratorTest { verify(arrangement.userRepository) .suspendFunction(arrangement.userRepository::updateActiveOneOnOneConversation) - .with(eq(TestUser.OTHER.id), eq(TestConversation.ONE_ON_ONE.id)) + .with(eq(TestUser.OTHER.id), eq(TestConversation.ONE_ON_ONE().id)) .wasInvoked() } diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversation.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversation.kt index c2d35bd1b01..ea6423ffbe3 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversation.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversation.kt @@ -57,12 +57,12 @@ object TestConversation { val ID = ConversationId(conversationValue, conversationDomain) fun id(suffix: Int = 0) = ConversationId("${conversationValue}_$suffix", conversationDomain) - val ONE_ON_ONE = Conversation( + fun ONE_ON_ONE(protocolInfo: ProtocolInfo = ProtocolInfo.Proteus) = Conversation( ID.copy(value = "1O1 ID"), "ONE_ON_ONE Name", Conversation.Type.ONE_ON_ONE, TestTeam.TEAM_ID, - ProtocolInfo.Proteus, + protocolInfo, MutedConversationStatus.AllAllowed, null, null, diff --git a/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversationDetails.kt b/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversationDetails.kt index 8791d0b3256..02ac4d426f2 100644 --- a/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversationDetails.kt +++ b/logic/src/commonTest/kotlin/com/wire/kalium/logic/framework/TestConversationDetails.kt @@ -38,7 +38,7 @@ object TestConversationDetails { ) val CONVERSATION_ONE_ONE = ConversationDetails.OneOne( - TestConversation.ONE_ON_ONE, + TestConversation.ONE_ON_ONE(), TestUser.OTHER, LegalHoldStatus.DISABLED, UserType.EXTERNAL, diff --git a/persistence/src/commonMain/db_user/com/wire/kalium/persistence/Messages.sq b/persistence/src/commonMain/db_user/com/wire/kalium/persistence/Messages.sq index 84a1b25da71..cb110bb1f79 100644 --- a/persistence/src/commonMain/db_user/com/wire/kalium/persistence/Messages.sq +++ b/persistence/src/commonMain/db_user/com/wire/kalium/persistence/Messages.sq @@ -508,6 +508,6 @@ INSERT OR IGNORE INTO MessageRecipientFailure(message_id, conversation_id, recip VALUES(?, ?, ?, ?); moveMessages: -UPDATE Message +UPDATE OR REPLACE Message SET conversation_id = :to WHERE conversation_id = :from;