TDLib Handlers refactor

build.gradle.kts

@@ -46,6 +46,12 @@ val appAssemblyVariants =
         }
     }
 
+val verifyBeforeAssemble =
+    providers.gradleProperty("verifyBeforeAssemble")
+        .map { it.equals("true", ignoreCase = true) }
+        .orElse(false)
+        .get()
+
 appAssemblyVariants.forEach { variant ->
     val verifyTask = tasks.register("verify${variant.name}BeforeAssemble") {
         group = "verification"
@@ -59,8 +65,10 @@ appAssemblyVariants.forEach { variant ->
         )
     }
 
-    project(":app").tasks.matching { it.name == "assemble${variant.name}" }.configureEach {
-        dependsOn(verifyTask)
+    if (verifyBeforeAssemble) {
+        project(":app").tasks.matching { it.name == "assemble${variant.name}" }.configureEach {
+            dependsOn(verifyTask)
+        }
     }
 }
 

data/build.gradle.kts

@@ -103,6 +103,7 @@ dependencies {
 
     implementation(libs.androidx.room.runtime)
     implementation(libs.androidx.room.ktx)
+    implementation(libs.androidx.work.runtime.ktx)
     ksp(libs.androidx.room.compiler)
 
     testImplementation(libs.junit)

data/src/firebase/java/org/monogram/data/service/FcmPushService.kt

@@ -3,23 +3,17 @@ package org.monogram.data.service
 import com.google.firebase.messaging.FirebaseMessagingService
 import com.google.firebase.messaging.RemoteMessage
 import org.koin.android.ext.android.inject
-import org.monogram.data.di.TdNotificationManager
-import org.monogram.data.gateway.TelegramGateway
-import org.monogram.data.push.PushSyncTrigger
+import org.monogram.data.push.PushProcessingCoordinator
 import org.monogram.domain.repository.AppPreferencesProvider
 
 class FcmPushService : FirebaseMessagingService() {
-    private val gateway: TelegramGateway by inject()
     private val appPreferences: AppPreferencesProvider by inject()
-    private val notificationManager: TdNotificationManager by inject()
-    private val pushSyncTrigger: PushSyncTrigger by inject()
+    private val pushCoordinator: PushProcessingCoordinator by inject()
     private val delegate by lazy {
         BaseFcmPushService(
             context = this,
-            gateway = gateway,
             appPreferences = appPreferences,
-            notificationManager = notificationManager,
-            pushSyncTrigger = pushSyncTrigger
+            pushCoordinator = pushCoordinator
         )
     }
 

data/src/main/java/org/monogram/data/datasource/cache/ChatLocalDataSource.kt

@@ -74,6 +74,59 @@ interface ChatLocalDataSource {
     )
 
     suspend fun deleteMessage(chatId: Long, messageId: Long)
+
+    suspend fun deleteMessages(chatId: Long, messageIds: List<Long>) {
+        messageIds.forEach { messageId -> deleteMessage(chatId, messageId) }
+    }
+
+    suspend fun applyMessageCacheMutations(mutations: List<MessageCacheMutation>) {
+        mutations.forEach { mutation ->
+            when (mutation) {
+                is MessageCacheMutation.Persist -> replaceMessage(mutation.message)
+                is MessageCacheMutation.ReplaceId -> replaceMessageId(
+                    mutation.chatId,
+                    mutation.oldMessageId,
+                    mutation.message
+                )
+
+                is MessageCacheMutation.UpdateContent -> updateMessageContent(
+                    mutation.chatId,
+                    mutation.messageId,
+                    mutation.content,
+                    mutation.contentType,
+                    mutation.contentMeta,
+                    mutation.mediaFileId,
+                    mutation.mediaPath,
+                    mutation.editDate
+                )
+
+                is MessageCacheMutation.UpdateInteraction -> updateInteractionInfo(
+                    mutation.chatId,
+                    mutation.messageId,
+                    mutation.viewCount,
+                    mutation.forwardCount,
+                    mutation.replyCount
+                )
+
+                is MessageCacheMutation.MarkRead -> markAsRead(
+                    mutation.chatId,
+                    mutation.upToMessageId
+                )
+
+                is MessageCacheMutation.DeleteMessages -> deleteMessages(
+                    mutation.chatId,
+                    mutation.messageIds
+                )
+
+                is MessageCacheMutation.UpdateMediaPath -> updateMediaPath(
+                    mutation.chatId,
+                    mutation.messageId,
+                    mutation.fileId,
+                    mutation.path
+                )
+            }
+        }
+    }
     suspend fun clearMessagesForChat(chatId: Long)
 
     suspend fun getChatFullInfo(chatId: Long): ChatFullInfoEntity?

data/src/main/java/org/monogram/data/datasource/cache/MessageCacheMutation.kt

@@ -0,0 +1,43 @@
+package org.monogram.data.datasource.cache
+
+import org.monogram.data.db.model.MessageEntity
+
+sealed interface MessageCacheMutation {
+    data class Persist(val message: MessageEntity) : MessageCacheMutation
+
+    data class ReplaceId(
+        val chatId: Long,
+        val oldMessageId: Long,
+        val message: MessageEntity
+    ) : MessageCacheMutation
+
+    data class UpdateContent(
+        val chatId: Long,
+        val messageId: Long,
+        val content: String,
+        val contentType: String,
+        val contentMeta: String?,
+        val mediaFileId: Int,
+        val mediaPath: String?,
+        val editDate: Int
+    ) : MessageCacheMutation
+
+    data class UpdateInteraction(
+        val chatId: Long,
+        val messageId: Long,
+        val viewCount: Int,
+        val forwardCount: Int,
+        val replyCount: Int
+    ) : MessageCacheMutation
+
+    data class MarkRead(val chatId: Long, val upToMessageId: Long) : MessageCacheMutation
+
+    data class DeleteMessages(val chatId: Long, val messageIds: List<Long>) : MessageCacheMutation
+
+    data class UpdateMediaPath(
+        val chatId: Long,
+        val messageId: Long,
+        val fileId: Int,
+        val path: String
+    ) : MessageCacheMutation
+}

data/src/main/java/org/monogram/data/datasource/cache/MessageCacheWriter.kt

@@ -0,0 +1,103 @@
+package org.monogram.data.datasource.cache
+
+import android.util.Log
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.launch
+import java.util.concurrent.atomic.AtomicInteger
+
+internal class MessageCacheWriter(
+    scope: CoroutineScope,
+    private val applyBatch: suspend (List<MessageCacheMutation>) -> Unit,
+    private val maxBatchSize: Int = DEFAULT_MAX_BATCH_SIZE
+) {
+    data class Stats(
+        val batches: Long = 0,
+        val mutations: Long = 0,
+        val lastBatchSize: Int = 0,
+        val lastLatencyMs: Long = 0,
+        val failures: Long = 0,
+        val retries: Long = 0,
+        val enqueued: Long = 0,
+        val pending: Int = 0,
+        val dropped: Long = 0
+    )
+
+    private val mutations = Channel<MessageCacheMutation>(Channel.UNLIMITED)
+    private val pending = AtomicInteger()
+    private val _stats = MutableStateFlow(Stats())
+    val stats: StateFlow<Stats> = _stats.asStateFlow()
+
+    init {
+        scope.launch {
+            for (first in mutations) {
+                pending.decrementAndGet()
+                val batch = ArrayList<MessageCacheMutation>(maxBatchSize)
+                batch += first
+                while (batch.size < maxBatchSize) {
+                    mutations.tryReceive().getOrNull()?.let { mutation ->
+                        pending.decrementAndGet()
+                        batch += mutation
+                    } ?: break
+                }
+
+                val startedAt = System.nanoTime()
+                var retries = 0L
+                var failure: Throwable? = null
+                var attempt = 0
+                do {
+                    failure = runCatching { applyBatch(batch) }.exceptionOrNull()
+                    attempt++
+                    if (failure != null && attempt < MAX_ATTEMPTS) {
+                        retries++
+                        delay(RETRY_DELAY_MS)
+                    }
+                } while (failure != null && attempt < MAX_ATTEMPTS)
+                if (failure != null) Log.e(
+                    TAG,
+                    "Message cache batch failed: size=${batch.size}",
+                    failure
+                )
+                _stats.update { previous ->
+                    previous.copy(
+                        batches = previous.batches + 1,
+                        mutations = previous.mutations + batch.size,
+                        lastBatchSize = batch.size,
+                        lastLatencyMs = (System.nanoTime() - startedAt) / 1_000_000,
+                        failures = previous.failures + if (failure == null) 0 else 1,
+                        retries = previous.retries + retries,
+                        pending = pending.get()
+                    )
+                }
+            }
+        }
+    }
+
+    fun enqueue(mutation: MessageCacheMutation): Boolean {
+        pending.incrementAndGet()
+        if (mutations.trySend(mutation).isSuccess) {
+            _stats.update { stats ->
+                stats.copy(enqueued = stats.enqueued + 1, pending = pending.get())
+            }
+            return true
+        }
+
+        pending.decrementAndGet()
+        _stats.update { stats ->
+            stats.copy(dropped = stats.dropped + 1, pending = pending.get())
+        }
+        return false
+    }
+
+    private companion object {
+        const val TAG = "MessageCacheWriter"
+        const val DEFAULT_MAX_BATCH_SIZE = 64
+        const val MAX_ATTEMPTS = 2
+        const val RETRY_DELAY_MS = 50L
+    }
+}

data/src/main/java/org/monogram/data/datasource/cache/RoomChatLocalDataSource.kt

@@ -104,9 +104,22 @@ class RoomChatLocalDataSource(
     override suspend fun getMessagesByIds(chatId: Long, messageIds: List<Long>) =
         if (messageIds.isEmpty()) emptyList() else messageDao.getMessagesByIds(chatId, messageIds)
 
-    override suspend fun insertMessage(message: MessageEntity) = messageDao.insertMessage(message)
+    override suspend fun insertMessage(message: MessageEntity) {
+        database.withTransaction {
+            messageDao.insertMessage(message)
+            cleanupMessageCache(message.chatId)
+        }
+    }
 
-    override suspend fun insertMessages(messages: List<MessageEntity>) = messageDao.insertMessages(messages)
+    override suspend fun insertMessages(messages: List<MessageEntity>) {
+        if (messages.isEmpty()) return
+        database.withTransaction {
+            messageDao.insertMessages(messages)
+            for (chatId in messages.asSequence().map(MessageEntity::chatId).distinct()) {
+                cleanupMessageCache(chatId)
+            }
+        }
+    }
 
     override suspend fun replaceMessage(message: MessageEntity) = messageDao.insertMessage(message)
 
@@ -163,6 +176,81 @@ class RoomChatLocalDataSource(
     override suspend fun deleteMessage(chatId: Long, messageId: Long) =
         messageDao.deleteMessage(chatId, messageId)
 
+    override suspend fun deleteMessages(chatId: Long, messageIds: List<Long>) {
+        if (messageIds.isNotEmpty()) {
+            messageDao.deleteMessages(chatId, messageIds)
+        }
+    }
+
+    override suspend fun applyMessageCacheMutations(mutations: List<MessageCacheMutation>) {
+        if (mutations.isEmpty()) return
+
+        database.withTransaction {
+            mutations.forEach { mutation ->
+                when (mutation) {
+                    is MessageCacheMutation.Persist -> messageDao.insertMessage(mutation.message)
+                    is MessageCacheMutation.ReplaceId -> {
+                        messageDao.deleteMessage(mutation.chatId, mutation.oldMessageId)
+                        messageDao.insertMessage(mutation.message)
+                    }
+
+                    is MessageCacheMutation.UpdateContent -> messageDao.updateContent(
+                        mutation.chatId,
+                        mutation.messageId,
+                        mutation.content,
+                        mutation.contentType,
+                        mutation.contentMeta,
+                        0,
+                        null,
+                        mutation.editDate
+                    )
+
+                    is MessageCacheMutation.UpdateInteraction -> messageDao.updateInteractionInfo(
+                        mutation.chatId,
+                        mutation.messageId,
+                        mutation.viewCount,
+                        mutation.forwardCount,
+                        mutation.replyCount
+                    )
+
+                    is MessageCacheMutation.MarkRead -> messageDao.markAsRead(
+                        mutation.chatId,
+                        mutation.upToMessageId
+                    )
+
+                    is MessageCacheMutation.DeleteMessages -> {
+                        if (mutation.messageIds.isNotEmpty()) {
+                            messageDao.deleteMessages(mutation.chatId, mutation.messageIds)
+                        }
+                    }
+
+                    is MessageCacheMutation.UpdateMediaPath -> Unit
+                }
+            }
+            for (chatId in mutations.asSequence().map(::mutationChatId).distinct()) {
+                cleanupMessageCache(chatId)
+            }
+        }
+    }
+
+    private suspend fun cleanupMessageCache(chatId: Long) {
+        messageDao.cleanupChat(
+            chatId = chatId,
+            keepCount = MESSAGE_CACHE_ROWS_PER_CHAT,
+            olderThan = System.currentTimeMillis() - MESSAGE_CACHE_TTL_MS
+        )
+    }
+
+    private fun mutationChatId(mutation: MessageCacheMutation): Long = when (mutation) {
+        is MessageCacheMutation.Persist -> mutation.message.chatId
+        is MessageCacheMutation.ReplaceId -> mutation.chatId
+        is MessageCacheMutation.UpdateContent -> mutation.chatId
+        is MessageCacheMutation.UpdateInteraction -> mutation.chatId
+        is MessageCacheMutation.MarkRead -> mutation.chatId
+        is MessageCacheMutation.DeleteMessages -> mutation.chatId
+        is MessageCacheMutation.UpdateMediaPath -> mutation.chatId
+    }
+
     override suspend fun clearMessagesForChat(chatId: Long) = messageDao.clearMessagesForChat(chatId)
 
     override suspend fun getChatFullInfo(chatId: Long): ChatFullInfoEntity? = chatFullInfoDao.getChatFullInfo(chatId)
@@ -186,4 +274,9 @@ class RoomChatLocalDataSource(
             messageDao.deleteExpired(timestamp)
         }
     }
-}
+
+    private companion object {
+        const val MESSAGE_CACHE_ROWS_PER_CHAT = 1_000
+        const val MESSAGE_CACHE_TTL_MS = 90L * 24 * 60 * 60 * 1_000
+    }
+}