feat(federation): offline backends handling (epic)

logic/src/commonMain/kotlin/com/wire/kalium/logic/CoreFailure.kt

@@ -205,6 +205,21 @@ internal inline fun <T : Any> wrapStorageRequest(storageRequest: () -> T?): Eith
     }
 }
 
+/**
+ * Wrap a storage request with a custom error handler that let's delegate the error handling to the caller.
+ */
+@Suppress("TooGenericExceptionCaught")
+internal inline fun <T : Any> wrapStorageRequest(
+    noinline errorHandler: (Exception) -> Either<StorageFailure, T>,
+    storageRequest: () -> T?
+): Either<StorageFailure, T> {
+    return try {
+        storageRequest()?.let { data -> Either.Right(data) } ?: Either.Left(StorageFailure.DataNotFound)
+    } catch (exception: Exception) {
+        errorHandler(exception)
+    }
+}
+
 internal inline fun <T : Any> wrapStorageNullableRequest(storageRequest: () -> T?): Either<StorageFailure, T?> {
     return try {
         storageRequest().let { data -> Either.Right(data) }

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/ConversationMapper.kt

@@ -20,6 +20,7 @@ package com.wire.kalium.logic.data.conversation
 
 import com.wire.kalium.logic.data.connection.ConnectionStatusMapper
 import com.wire.kalium.logic.data.id.IdMapper
+import com.wire.kalium.logic.data.id.NetworkQualifiedId
 import com.wire.kalium.logic.data.id.TeamId
 import com.wire.kalium.logic.data.id.toApi
 import com.wire.kalium.logic.data.id.toDao
@@ -77,6 +78,7 @@ interface ConversationMapper {
     fun toApiModel(name: String?, members: List<UserId>, teamId: String?, options: ConversationOptions): CreateConversationRequest
 
     fun fromMigrationModel(conversation: Conversation): ConversationEntity
+    fun fromFailedGroupConversationToEntity(conversationId: NetworkQualifiedId): ConversationEntity
 }
 
 @Suppress("TooManyFunctions", "LongParameterList")
@@ -113,6 +115,7 @@ internal class ConversationMapperImpl(
         receiptMode = receiptModeMapper.fromApiToDaoModel(apiModel.receiptMode),
         messageTimer = apiModel.messageTimer,
         userMessageTimer = null, // user picked self deletion timer is only persisted locally
+        hasIncompleteMetadata = false
     )
 
     override fun fromApiModelToDaoModel(apiModel: ConvProtocol): Protocol = when (apiModel) {
@@ -357,6 +360,31 @@ internal class ConversationMapperImpl(
         )
     }
 
+    /**
+     * Default values and marked as [ConversationEntity.hasIncompleteMetadata] = true.
+     * So later we can re-fetch them.
+     */
+    override fun fromFailedGroupConversationToEntity(conversationId: NetworkQualifiedId): ConversationEntity = ConversationEntity(
+        id = conversationId.toDao(),
+        name = null,
+        type = ConversationEntity.Type.GROUP,
+        teamId = null,
+        protocolInfo = ProtocolInfo.Proteus,
+        mutedStatus = ConversationEntity.MutedStatus.ALL_ALLOWED,
+        mutedTime = 0,
+        removedBy = null,
+        creatorId = "",
+        lastNotificationDate = "1970-01-01T00:00:00.000Z".toInstant(),
+        lastModifiedDate = "1970-01-01T00:00:00.000Z".toInstant(),
+        lastReadDate = "1970-01-01T00:00:00.000Z".toInstant(),
+        access = emptyList(),
+        accessRole = emptyList(),
+        receiptMode = ConversationEntity.ReceiptMode.DISABLED,
+        messageTimer = null,
+        userMessageTimer = null,
+        hasIncompleteMetadata = true
+    )
+
     private fun ConversationResponse.getProtocolInfo(mlsGroupState: GroupState?): ProtocolInfo {
         return when (protocol) {
             ConvProtocol.MLS -> ProtocolInfo.MLS(

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/ConversationRepository.kt

@@ -26,6 +26,7 @@ import com.wire.kalium.logic.data.client.MLSClientProvider
 import com.wire.kalium.logic.data.id.ConversationId
 import com.wire.kalium.logic.data.id.GroupID
 import com.wire.kalium.logic.data.id.IdMapper
+import com.wire.kalium.logic.data.id.NetworkQualifiedId
 import com.wire.kalium.logic.data.id.QualifiedID
 import com.wire.kalium.logic.data.id.toApi
 import com.wire.kalium.logic.data.id.toCrypto
@@ -210,6 +211,7 @@ interface ConversationRepository {
     suspend fun getConversationUnreadEventsCount(conversationId: ConversationId): Either<StorageFailure, Long>
     suspend fun getUserSelfDeletionTimer(conversationId: ConversationId): Either<StorageFailure, SelfDeletionTimer?>
     suspend fun updateUserSelfDeletionTimer(conversationId: ConversationId, selfDeletionTimer: SelfDeletionTimer): Either<CoreFailure, Unit>
+    suspend fun syncConversationsWithoutMetadata(): Either<CoreFailure, Unit>
 }
 
 @Suppress("LongParameterList", "TooManyFunctions")
@@ -266,7 +268,8 @@ internal class ConversationDataSource internal constructor(
                 }.onSuccess { conversations ->
                     if (conversations.conversationsFailed.isNotEmpty()) {
                         kaliumLogger.withFeatureId(CONVERSATIONS)
-                            .d("Skipping ${conversations.conversationsFailed.size} conversations failed")
+                            .d("Handling ${conversations.conversationsFailed.size} conversations failed")
+                        persistIncompleteConversations(conversations.conversationsFailed)
                     }
                     if (conversations.conversationsNotFound.isNotEmpty()) {
                         kaliumLogger.withFeatureId(CONVERSATIONS)
@@ -751,6 +754,31 @@ internal class ConversationDataSource internal constructor(
         )
     }
 
+    override suspend fun syncConversationsWithoutMetadata(): Either<CoreFailure, Unit> = wrapStorageRequest {
+        val conversationsWithoutMetadata = conversationDAO.getConversationsWithoutMetadata()
+        if (conversationsWithoutMetadata.isNotEmpty()) {
+            kaliumLogger.d("Numbers of conversations to refresh: ${conversationsWithoutMetadata.size}")
+            val conversationsWithoutMetadataIds = conversationsWithoutMetadata.map { it.toApi() }
+            wrapApiRequest {
+                conversationApi.fetchConversationsListDetails(conversationsWithoutMetadataIds)
+            }.onSuccess {
+                persistConversations(it.conversationsFound, null)
+            }
+        }
+    }
+
+    private suspend fun persistIncompleteConversations(
+        conversationsFailed: List<NetworkQualifiedId>
+    ): Either<CoreFailure, Unit> {
+        return wrapStorageRequest {
+            if (conversationsFailed.isNotEmpty()) {
+                conversationDAO.insertConversations(conversationsFailed.map { conversationId ->
+                    conversationMapper.fromFailedGroupConversationToEntity(conversationId)
+                })
+            }
+        }
+    }
+
     companion object {
         const val DEFAULT_MEMBER_ROLE = "wire_member"
     }

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/NewGroupConversationSystemMessagesCreator.kt

@@ -149,7 +149,22 @@ internal class NewGroupConversationSystemMessagesCreatorImpl(
                     status = Message.Status.SENT,
                     visibility = Message.Visibility.VISIBLE
                 )
+            ).also { createFailedToAddSystemMessage(conversationResponse) }
+        }
+    }
+
+    private suspend fun createFailedToAddSystemMessage(conversationResponse: ConversationResponse) {
+        if (conversationResponse.failedToAdd.isNotEmpty()) {
+            val messageStartedWithFailedMembers = Message.System(
+                uuid4().toString(),
+                MessageContent.MemberChange.FailedToAdd(conversationResponse.failedToAdd.map { it.toModel() }),
+                conversationResponse.id.toModel(),
+                DateTimeUtil.currentIsoDateTimeString(),
+                selfUserId,
+                Message.Status.SENT,
+                Message.Visibility.VISIBLE
             )
+            persistMessage(messageStartedWithFailedMembers)
         }
     }
 }

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/message/Message.kt

@@ -20,10 +20,10 @@ package com.wire.kalium.logic.data.message
 
 import com.wire.kalium.logger.obfuscateDomain
 import com.wire.kalium.logger.obfuscateId
-import com.wire.kalium.util.serialization.toJsonElement
 import com.wire.kalium.logic.data.conversation.ClientId
 import com.wire.kalium.logic.data.id.ConversationId
 import com.wire.kalium.logic.data.user.UserId
+import com.wire.kalium.util.serialization.toJsonElement
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
 import kotlinx.serialization.encodeToString
@@ -80,7 +80,8 @@ sealed interface Message {
         val editStatus: EditStatus,
         val expirationData: ExpirationData? = null,
         val reactions: Reactions = Reactions.EMPTY,
-        val expectsReadConfirmation: Boolean = false
+        val expectsReadConfirmation: Boolean = false,
+        val deliveryStatus: DeliveryStatus = DeliveryStatus.CompleteDelivery
     ) : Sendable, Standalone {
 
         @Suppress("LongMethod")
@@ -133,7 +134,8 @@ sealed interface Message {
                 "visibility" to "$visibility",
                 "senderClientId" to senderClientId.value.obfuscateId(),
                 "editStatus" to editStatus.toLogMap(),
-                "expectsReadConfirmation" to "$expectsReadConfirmation"
+                "expectsReadConfirmation" to "$expectsReadConfirmation",
+                "deliveryStatus" to "$deliveryStatus"
             )
 
             properties.putAll(standardProperties)
@@ -456,3 +458,12 @@ enum class AssetType {
 
 typealias ReactionsCount = Map<String, Int>
 typealias UserReactions = Set<String>
+
+sealed class DeliveryStatus {
+    data class PartialDelivery(
+        val recipientsFailedWithNoClients: List<UserId>,
+        val recipientsFailedDelivery: List<UserId>
+    ) : DeliveryStatus()
+
+    object CompleteDelivery : DeliveryStatus()
+}

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/message/MessageMapper.kt

@@ -32,6 +32,7 @@ import com.wire.kalium.logic.data.notification.LocalNotificationMessageAuthor
 import com.wire.kalium.logic.data.user.UserId
 import com.wire.kalium.logic.di.MapperProvider
 import com.wire.kalium.persistence.dao.message.AssetTypeEntity
+import com.wire.kalium.persistence.dao.message.DeliveryStatusEntity
 import com.wire.kalium.persistence.dao.message.MessageEntity
 import com.wire.kalium.persistence.dao.message.MessageEntityContent
 import com.wire.kalium.persistence.dao.message.MessagePreviewEntity
@@ -105,6 +106,7 @@ class MessageMapperImpl(
         }
     }
 
+    @Suppress("ComplexMethod")
     override fun fromEntityToMessage(message: MessageEntity): Message.Standalone {
         val status = when (message.status) {
             MessageEntity.Status.PENDING -> Message.Status.PENDING
@@ -138,7 +140,14 @@ class MessageMapperImpl(
                     reactions = Message.Reactions(message.reactions.totalReactions, message.reactions.selfUserReactions),
                     senderUserName = message.senderName,
                     isSelfMessage = message.isSelfMessage,
-                    expectsReadConfirmation = message.expectsReadConfirmation
+                    expectsReadConfirmation = message.expectsReadConfirmation,
+                    deliveryStatus = when (val recipientsFailure = message.deliveryStatus) {
+                        is DeliveryStatusEntity.CompleteDelivery -> DeliveryStatus.CompleteDelivery
+                        is DeliveryStatusEntity.PartialDelivery -> DeliveryStatus.PartialDelivery(
+                            recipientsFailedWithNoClients = recipientsFailure.recipientsFailedWithNoClients.map { it.toModel() },
+                            recipientsFailedDelivery = recipientsFailure.recipientsFailedDelivery.map { it.toModel() }
+                        )
+                    },
                 )
 
             is MessageEntity.System -> Message.System(