Enhance TDLib updates and support for more authentication methods

app/src/main/java/org/monogram/app/MainActivity.kt

@@ -1,6 +1,5 @@
 package org.monogram.app
 
-import android.app.ForegroundServiceStartNotAllowedException
 import android.content.Intent
 import android.os.Build
 import android.os.Bundle
@@ -148,8 +147,6 @@ class MainActivity : FragmentActivity() {
             } else {
                 startService(intent)
             }
-        } catch (_: ForegroundServiceStartNotAllowedException) {
-            Log.w(TAG, "Foreground notification service start was blocked by the system")
         } catch (error: IllegalStateException) {
             Log.w(TAG, "Foreground notification service start failed", error)
         }

data/src/main/java/org/monogram/data/datasource/remote/SettingsRemoteDataSource.kt

@@ -11,6 +11,7 @@ interface SettingsRemoteDataSource {
     suspend fun getStorageStatisticsFast(): TdApi.StorageStatisticsFast?
     suspend fun getNetworkStatistics(): TdApi.NetworkStatistics?
     suspend fun getOption(name: String): TdApi.OptionValue?
+    suspend fun getPremiumLimit(limitType: TdApi.PremiumLimitType): TdApi.PremiumLimit?
     suspend fun getChatNotificationSettingsExceptions(
         scope: TdApi.NotificationSettingsScope,
         compareSound: Boolean

data/src/main/java/org/monogram/data/datasource/remote/TdAuthRemoteDataSource.kt

@@ -19,14 +19,14 @@ class TdAuthRemoteDataSource(
         val settings = TdApi.PhoneNumberAuthenticationSettings().apply {
             isCurrentPhoneNumber = false
             allowFlashCall = false
-            allowMissedCall = false
+            allowMissedCall = true
             allowSmsRetrieverApi = false
         }
         gateway.execute(TdApi.SetAuthenticationPhoneNumber(phone, settings))
     }
 
     override suspend fun resendCode() {
-        gateway.execute(TdApi.ResendAuthenticationCode())
+        gateway.execute(TdApi.ResendAuthenticationCode(TdApi.ResendCodeReasonUserRequest()))
     }
 
     override suspend fun setAuthCode(code: String) {

data/src/main/java/org/monogram/data/datasource/remote/TdMessageRemoteDataSource.kt

@@ -58,6 +58,7 @@ import org.monogram.domain.models.MessageSendOptions
 import org.monogram.domain.models.MessageUploadProgressEvent
 import org.monogram.domain.models.MessageViewerModel
 import org.monogram.domain.models.PollDraft
+import org.monogram.domain.models.TdLibLimits
 import org.monogram.domain.models.UserModel
 import org.monogram.domain.models.webapp.ThemeParams
 import org.monogram.domain.models.webapp.WebAppInfoModel
@@ -68,6 +69,7 @@ import org.monogram.domain.repository.PollRepository
 import org.monogram.domain.repository.ReadUpdate
 import org.monogram.domain.repository.RichTextParseMode
 import org.monogram.domain.repository.SearchChatMessagesResult
+import org.monogram.domain.repository.TdLibLimitsRepository
 import org.monogram.domain.repository.UserRepository
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
@@ -84,19 +86,31 @@ class TdMessageRemoteDataSource(
     private val webPageMapper: WebPageMapper,
     private val draftLinkPreviewResolver: DraftLinkPreviewResolver,
     private val dispatcherProvider: DispatcherProvider,
-    val scope: CoroutineScope
+    val scope: CoroutineScope,
+    private val tdLibLimitsRepository: TdLibLimitsRepository
 ) : MessageRemoteDataSource {
 
     private val chatRequests = ConcurrentHashMap<Long, Deferred<TdApi.Chat?>>()
     private val messageRequests = ConcurrentHashMap<Pair<Long, Long>, Deferred<TdApi.Message?>>()
     private val refreshJobs = ConcurrentHashMap<Pair<Long, Long>, Job>()
     private val missingMessageCooldownUntil = ConcurrentHashMap<Pair<Long, Long>, Long>()
     private val sendQueue = Channel<suspend () -> Unit>(Channel.BUFFERED)
-    override val newMessageFlow = MutableSharedFlow<MessageModel>()
-    override val messageEditedFlow = MutableSharedFlow<MessageModel>()
+    // These are fed from `scope.launch { ... }` inside update handling. With the default
+    // arguments (replay 0, no buffer, SUSPEND) a MutableSharedFlow is a rendezvous channel:
+    // every emit parks until *all* subscribers have taken the value, which piled emitters up
+    // without bound during bursts. OrderedEventFlow.enqueue is non-suspending and lossless,
+    // so the event streams go through it; progress ticks are conflatable and get an explicit
+    // bounded buffer instead.
+    private val newMessages = OrderedEventFlow<MessageModel>(scope)
+    override val newMessageFlow = newMessages.events
+    private val messageEdits = OrderedEventFlow<MessageModel>(scope)
+    override val messageEditedFlow = messageEdits.events
     private val messageReads = OrderedEventFlow<ReadUpdate>(scope)
     override val messageReadFlow = messageReads.events
-    override val messageUploadProgressFlow = MutableSharedFlow<MessageUploadProgressEvent>()
+    override val messageUploadProgressFlow = MutableSharedFlow<MessageUploadProgressEvent>(
+        extraBufferCapacity = 64,
+        onBufferOverflow = BufferOverflow.DROP_OLDEST
+    )
     private val fileDownloads = OrderedEventFlow<FileDownloadEvent>(scope)
     override val fileDownloadFlow = fileDownloads.events
     private val messageDownloads = OrderedEventFlow<MessageDownloadEvent>(scope)
@@ -151,10 +165,26 @@ class TdMessageRemoteDataSource(
             throw e
         } catch (e: Exception) {
             Log.e("TdMessageRemote", "Error executing ${function.javaClass.simpleName}", e)
+            if (e.isLikelyLimitViolation()) {
+                scope.launch {
+                    runCatching { tdLibLimitsRepository.refresh() }
+                }
+            }
             null
         }
     }
 
+    private fun Throwable.isLikelyLimitViolation(): Boolean {
+        val error = (this as? TdLibException)?.error ?: return false
+        if (error.code !in 400..499) return false
+        val message = error.message.orEmpty().lowercase()
+        return "too long" in message ||
+                "length" in message ||
+                "limit" in message ||
+                "maximum" in message ||
+                "max_" in message
+    }
+
 
     override suspend fun getMessage(chatId: Long, messageId: Long): TdApi.Message? {
         cache.getMessage(chatId, messageId)?.let { return it }
@@ -841,7 +871,11 @@ class TdMessageRemoteDataSource(
         val replyTo = if (replyToMsgId != null && replyToMsgId != 0L) TdApi.InputMessageReplyToMessage(replyToMsgId, null, 0, "") else null
         val topicId = resolveTopicId(chatId, threadId)
         var lastMessage: TdApi.Message? = null
-        explodeTextContent(content, MAX_TEXT_MESSAGE_CODE_POINTS).forEach { messageContent ->
+        explodeTextContent(
+            content,
+            tdLibLimitsRepository.limits.value.messageTextLengthMax
+                ?: TdLibLimits.DEFAULT_MESSAGE_TEXT_LENGTH_MAX
+        ).forEach { messageContent ->
             val req = TdApi.SendMessage().apply {
                 this.chatId = chatId
                 this.topicId = topicId
@@ -1992,7 +2026,7 @@ class TdMessageRemoteDataSource(
                 scope.launch(dispatcherProvider.io) {
                     try {
                         val model = mapMessageToModel(message)
-                        newMessageFlow.emit(model)
+                        newMessages.enqueue(model)
                     } catch (e: Exception) { Log.e("TdMessageRemote", "Error mapping NewMessage", e) }
                 }
             }
@@ -2039,7 +2073,7 @@ class TdMessageRemoteDataSource(
                             errorCode = update.error?.code ?: 0
                         )
                     )
-                    messageEditedFlow.emit(model)
+                    messageEdits.enqueue(model)
                 }
             }
             is TdApi.UpdateMessageContent -> {
@@ -2216,7 +2250,7 @@ class TdMessageRemoteDataSource(
         if (messageId == 0L) return
         val msg = cache.getMessage(chatId, messageId) ?: return
         val model = mapMessageToModel(msg)
-        messageEditedFlow.emit(model)
+        messageEdits.enqueue(model)
     }
 
     private suspend fun mapMessageToModel(message: TdApi.Message): MessageModel {
@@ -2445,7 +2479,7 @@ class TdMessageRemoteDataSource(
                 delay(150)
                 val msg = cache.getMessage(chatId, messageId) ?: return@launch
                 try {
-                    messageEditedFlow.emit(mapMessageToModel(msg))
+                    messageEdits.enqueue(mapMessageToModel(msg))
                 } catch (e: CancellationException) {
                     throw e
                 } catch (e: Exception) {
@@ -2492,6 +2526,5 @@ class TdMessageRemoteDataSource(
     private companion object {
         private val MISSING_MESSAGE_COOLDOWN_MS = TimeUnit.MINUTES.toMillis(2)
         private const val DRAFT_LINK_PREVIEW_TAG = "DraftLinkPreview"
-        private const val MAX_TEXT_MESSAGE_CODE_POINTS = 4096
     }
 }

data/src/main/java/org/monogram/data/datasource/remote/TdSettingsRemoteDataSource.kt

@@ -83,6 +83,9 @@ class TdSettingsRemoteDataSource(
     override suspend fun getOption(name: String): TdApi.OptionValue? =
         coRunCatching { gateway.execute(TdApi.GetOption(name)) }.getOrNull()
 
+    override suspend fun getPremiumLimit(limitType: TdApi.PremiumLimitType): TdApi.PremiumLimit? =
+        coRunCatching { gateway.execute(TdApi.GetPremiumLimit(limitType)) }.getOrNull()
+
     override suspend fun getArchiveChatListSettings(): TdApi.ArchiveChatListSettings? =
         coRunCatching { gateway.execute(TdApi.GetArchiveChatListSettings()) }.getOrNull()
 

data/src/main/java/org/monogram/data/di/TdLibClient.kt

@@ -1,9 +1,8 @@
 package org.monogram.data.di
 
 import android.util.Log
-import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -16,16 +15,21 @@ import org.monogram.data.BuildConfig
 import org.monogram.data.gateway.TdLibException
 import org.monogram.data.gateway.isExpectedProxyFailure
 import java.util.concurrent.ConcurrentHashMap
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.EmptyCoroutineContext
 import kotlin.coroutines.resume
 
 internal class TdLibClient {
     private val TAG = "TdLibClient"
     private val retryAfterUntilMsByScope = ConcurrentHashMap<String, Long>()
-    private val _updates = MutableSharedFlow<TdApi.Update>(
-        replay = 3,
-        extraBufferCapacity = 64,
-        onBufferOverflow = BufferOverflow.DROP_OLDEST
-    )
+
+    /**
+     * Authoritative ingestion for TDLib updates.
+     *
+     * Consumers that own durable state subscribe with [lane] (lossless, ordered);
+     * consumers that only render subscribe to [updates] (conflating).
+     */
+    private val pipeline = TdUpdatePipeline()
 
     private val _isAuthenticated = MutableStateFlow(false)
     val isAuthenticated = _isAuthenticated.asStateFlow()
@@ -44,17 +48,40 @@ internal class TdLibClient {
         }
     }
 
-    val updates: SharedFlow<TdApi.Update> = _updates
+    /**
+     * Observation stream. Conflates under load, so it must not be used to drive durable
+     * state; use [lane] for that.
+     */
+    val updates: SharedFlow<TdApi.Update> = pipeline.updates
+
+    /**
+     * Lossless, strictly ordered subscription for consumers that own durable state:
+     * Room writes, [org.monogram.data.chats.ChatCache] mutation, or TDLib requests.
+     *
+     * Register during startup — updates delivered before the lane exists are not replayed.
+     */
+    fun lane(
+        name: String,
+        scope: CoroutineScope,
+        context: CoroutineContext = EmptyCoroutineContext,
+        filter: (TdApi.Update) -> Boolean = { true },
+        handler: suspend (TdApi.Update) -> Unit,
+    ): TdUpdatePipeline.Lane = pipeline.lane(name, scope, context, filter, handler)
+
+    /** Diagnostics: ingest/lane backlogs, processed counts and handler failures. */
+    fun updateMetrics(): String = pipeline.metrics()
 
     private val client = Client.create(
         { result ->
             if (result is TdApi.Update) {
+                // Kept on the callback thread on purpose: these gate sendSuspend, so they
+                // must not depend on the update pipeline making progress.
                 if (result is TdApi.UpdateAuthorizationState) {
                     val state = result.authorizationState
                     _isInitialized.value = state !is TdApi.AuthorizationStateWaitTdlibParameters
                     _isAuthenticated.value = state is TdApi.AuthorizationStateReady
                 }
-                _updates.tryEmit(result)
+                pipeline.submit(result)
             }
         },
         { error ->

data/src/main/java/org/monogram/data/di/TdNotificationManager.kt

@@ -167,14 +167,13 @@ class TdNotificationManager(
             }
         }
 
-        scope.launch {
-            updates.all.collect { update ->
-                runCatching {
-                    handleCoreUpdate(update)
-                }.onFailure {
-                    Log.e(TAG, "Failed to handle update ${update.javaClass.simpleName}", it)
-                }
-            }
+        // handleCoreUpdate issues TDLib requests inline (getChat, membership checks), so
+        // this consumer is orders of magnitude slower than the update rate and would be
+        // the first to be conflated away on the observation flow. updateNotificationGroup
+        // carries added/removed deltas and updateActiveNotifications arrives exactly once
+        // before them, so none of it may be dropped or reordered.
+        updates.lane(name = "notifications", scope = scope) { update ->
+            handleCoreUpdate(update)
         }
 
         scope.launch {