diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt index 5cbb1fb120..4a19906c82 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt @@ -10,6 +10,7 @@ package com.nextcloud.talk.chat.data import android.os.Bundle import com.nextcloud.talk.chat.data.io.LifecycleAwareManager import com.nextcloud.talk.chat.data.model.ChatMessage +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.data.database.model.ChatMessageEntity import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.models.domain.ConversationModel @@ -44,6 +45,13 @@ interface ChatMessageRepository : LifecycleAwareManager { val incomingMessageFlow: Flow + /** + * Emits only for call-related system messages that just genuinely arrived (long poll, chat + * relay/signaling, insurance request) — never for one merely present in a re-loaded/paginated + * window of already-known history. See [ChatMessageSyncer.Events.onCallSystemMessage]. + */ + val callSystemMessageFlow: Flow + val isLoadingFlow: Flow // /** diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt index 53509965d8..10f60f1715 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt @@ -66,6 +66,17 @@ class ChatMessageSyncer @Inject constructor( get() = user.id!! } + /** + * A call-related system message that arrived as a genuinely new message (see [Events.onCallSystemMessage]), + * as opposed to one merely present in a re-loaded/paginated window of already-known history. + */ + sealed interface CallSystemMessageEvent { + data class CallStarted(val actorDisplayName: String, val actorType: String, val actorId: String) : + CallSystemMessageEvent + + data class CallEnded(val systemMessageType: ChatMessage.SystemMessageType) : CallSystemMessageEvent + } + /** * Side effects that only matter while a chat is on screen. Background callers can pass [NO_EVENTS]. */ @@ -85,6 +96,10 @@ class ChatMessageSyncer @Inject constructor( suspend fun onIncomingMessagesFromOthers() { // no-op by default } + + suspend fun onCallSystemMessage(event: CallSystemMessageEvent) { + // no-op by default + } } /** @@ -493,10 +508,28 @@ class ChatMessageSyncer @Inject constructor( val queriedMessageId = fieldMap["lastKnownMessageId"] val lookIntoFuture = fieldMap["lookIntoFuture"] == 1 + val includeLastKnown = fieldMap["includeLastKnown"] == 1 + val requestedLimit = fieldMap["limit"] ?: DEFAULT_MESSAGES_LIMIT + // An unconditional "give me the newest messages" fetch (no anchor) always reaches the true + // current tail of the conversation, regardless of how many messages came back. Any other + // fetch — including a lookIntoFuture one — only reaches the tail if the server returned + // fewer than requested: a full page means more of a (potentially large, e.g. after the + // in-memory HTTP-sync anchor was lost on an app restart) backlog remains beyond this batch, + // so nothing in it may be trusted as reflecting live call state yet. Computed against the + // actual result below, in handleSuccessfulPull. + val unconditionalNewestFetch = includeLastKnown && queriedMessageId == null return when (val result = pullMessagesFlow(target, fieldMap).first()) { is ChatPullResult.Success -> - handleSuccessfulPull(target, result, queriedMessageId, lookIntoFuture, events) + handleSuccessfulPull( + target, + result, + queriedMessageId, + lookIntoFuture, + unconditionalNewestFetch, + requestedLimit, + events + ) is ChatPullResult.NotModified -> { Log.d(TAG, "Server returned NOT_MODIFIED, nothing to update") @@ -523,22 +556,33 @@ class ChatMessageSyncer @Inject constructor( } } + @Suppress("LongParameterList") private suspend fun handleSuccessfulPull( target: SyncTarget, result: ChatPullResult.Success, queriedMessageId: Int?, lookIntoFuture: Boolean, + unconditionalNewestFetch: Boolean, + requestedLimit: Int, events: Events ): SyncOutcome { events.onLastCommonReadChanged(result.lastCommonRead) val hasHistory = getHasHistory(HTTP_CODE_OK, lookIntoFuture) + // A batch as large as requestedLimit means more of a (potentially large) backlog remains + // beyond it — e.g. the in-memory HTTP-sync anchor was lost on an app restart, so an insurance + // request restarts far behind and now has to crawl forward through everything in between. + // A call system message in such a batch is old backlog, not live state, and must not be + // trusted until a batch actually reaches the tail (fewer messages than requested). + val reflectsCurrentTail = unconditionalNewestFetch || result.messages.size < requestedLimit + Log.d( TAG, "internalConv=${target.internalConversationId} statusCode=$HTTP_CODE_OK " + "lookIntoFuture=$lookIntoFuture hasHistory=$hasHistory " + - "queriedMessageId=$queriedMessageId" + "queriedMessageId=$queriedMessageId reflectsCurrentTail=$reflectsCurrentTail " + + "batchSize=${result.messages.size} requestedLimit=$requestedLimit" ) val blockContainingQueriedMessage: ChatBlockEntity? = getBlockOfMessage(target, queriedMessageId) @@ -555,6 +599,7 @@ class ChatMessageSyncer @Inject constructor( result.messages, blockContainingQueriedMessage, lookIntoFuture, + reflectsCurrentTail, hasHistory, events ) @@ -582,6 +627,7 @@ class ChatMessageSyncer @Inject constructor( chatMessagesJson: List, blockContainingQueriedMessage: ChatBlockEntity?, lookIntoFuture: Boolean, + reflectsCurrentTail: Boolean, hasHistory: Boolean, events: Events ): List { @@ -589,6 +635,7 @@ class ChatMessageSyncer @Inject constructor( target, chatMessagesJson, emitOnIncoming = lookIntoFuture, + reflectsCurrentTail = reflectsCurrentTail, events = events ) @@ -641,9 +688,10 @@ class ChatMessageSyncer @Inject constructor( target: SyncTarget, chatMessages: List, emitOnIncoming: Boolean = false, + reflectsCurrentTail: Boolean = emitOnIncoming, events: Events = NO_EVENTS ): List { - handleSystemMessagesThatAffectDatabase(target, chatMessages, events) + handleSystemMessagesThatAffectDatabase(target, chatMessages, reflectsCurrentTail, events) val chatMessageEntities = chatMessages.map { it.asEntity(target.accountId) @@ -687,6 +735,7 @@ class ChatMessageSyncer @Inject constructor( private suspend fun handleSystemMessagesThatAffectDatabase( target: SyncTarget, messagesJson: List, + reflectsCurrentTail: Boolean, events: Events ) { var needsRoomRefresh = false @@ -726,6 +775,41 @@ class ChatMessageSyncer @Inject constructor( } } if (needsRoomRefresh) events.onRoomRefreshNeeded() + + if (reflectsCurrentTail) { + reportLastDecisiveCallSystemMessage(messagesJson, events) + } + } + + /** + * Reports only the chronologically last call-related message in [messagesJson] — never one per + * message in array order. The API does not guarantee ascending order across every fetch shape: + * an unconditional "give me the newest messages" fetch, in particular, returns newest-first, so + * iterating in array order and treating the last-processed message as authoritative would report + * the OLDEST message in the batch as the current call state. Only called when the batch is + * confirmed to reflect the current tail of the conversation — see [handleSuccessfulPull]. + */ + private suspend fun reportLastDecisiveCallSystemMessage(messagesJson: List, events: Events) { + val lastDecisiveCallMessage = messagesJson + .filter { it.systemMessageType in DECISIVE_CALL_SYSTEM_MESSAGE_TYPES } + .maxByOrNull { it.id } + ?: return + + when (lastDecisiveCallMessage.systemMessageType) { + ChatMessage.SystemMessageType.CALL_STARTED -> + events.onCallSystemMessage( + CallSystemMessageEvent.CallStarted( + actorDisplayName = lastDecisiveCallMessage.actorDisplayName.orEmpty(), + actorType = lastDecisiveCallMessage.actorType.orEmpty(), + actorId = lastDecisiveCallMessage.actorId.orEmpty() + ) + ) + + else -> + events.onCallSystemMessage( + CallSystemMessageEvent.CallEnded(lastDecisiveCallMessage.systemMessageType!!) + ) + } } // the parent message is always the newest state, no matter how old the system message is. @@ -831,6 +915,14 @@ class ChatMessageSyncer @Inject constructor( val NO_EVENTS: Events = object : Events {} + private val DECISIVE_CALL_SYSTEM_MESSAGE_TYPES = setOf( + ChatMessage.SystemMessageType.CALL_STARTED, + ChatMessage.SystemMessageType.CALL_ENDED, + ChatMessage.SystemMessageType.CALL_ENDED_EVERYONE, + ChatMessage.SystemMessageType.CALL_MISSED, + ChatMessage.SystemMessageType.CALL_TRIED + ) + private val NOTHING_SYNCED = SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null) private val SYNC_FAILED = SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null, syncFailed = true) diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt index 98594e6d85..e364967cc8 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt @@ -108,6 +108,12 @@ class OfflineFirstChatRepository @Inject constructor( private val _incomingMessageFlow: MutableSharedFlow = MutableSharedFlow() + override val callSystemMessageFlow: Flow + get() = _callSystemMessageFlow + + private val _callSystemMessageFlow: + MutableSharedFlow = MutableSharedFlow(extraBufferCapacity = 4) + override val isLoadingFlow: Flow get() = _isLoadingFlow @@ -171,6 +177,10 @@ class OfflineFirstChatRepository @Inject constructor( override suspend fun onIncomingMessagesFromOthers() { _incomingMessageFlow.emit(Unit) } + + override suspend fun onCallSystemMessage(event: ChatMessageSyncer.CallSystemMessageEvent) { + _callSystemMessageFlow.emit(event) + } } override suspend fun loadInitialMessages(withNetworkParams: Bundle) { @@ -783,7 +793,12 @@ class OfflineFirstChatRepository @Inject constructor( chatMessages: List, emitOnIncoming: Boolean = false ): List = - syncer.persistChatMessagesAndHandleSystemMessages(syncTarget, chatMessages, emitOnIncoming, syncEvents) + syncer.persistChatMessagesAndHandleSystemMessages( + target = syncTarget, + chatMessages = chatMessages, + emitOnIncoming = emitOnIncoming, + events = syncEvents + ) override fun observeLatestMessages(internalConversationId: String): Flow> = chatBlocksDao diff --git a/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt b/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt index 996346fe5d..fd4facf9d8 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt @@ -25,6 +25,7 @@ import com.nextcloud.talk.chat.data.io.AudioFocusRequestManager import com.nextcloud.talk.chat.data.io.MediaPlayerManager import com.nextcloud.talk.chat.data.io.MediaRecorderManager import com.nextcloud.talk.chat.data.model.ChatMessage +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource import com.nextcloud.talk.chat.ui.model.ChatMessageUi import com.nextcloud.talk.chat.ui.model.MessageTypeContent @@ -565,6 +566,7 @@ class ChatViewModel @AssistedInject constructor( observePinnedMessage() observeRoomRefresh() observeIncomingMessages() + observeCallSystemMessages() } fun enterSearchMode() { @@ -1023,16 +1025,9 @@ class ChatViewModel @AssistedInject constructor( val shouldShow: Boolean ) - private val _lastCallSystemMessage = MutableStateFlow(null) + private val _lastCallSystemMessage = MutableStateFlow(null) - val lastCallSystemMessage = _lastCallSystemMessage.map { msg -> - CallStartedIndicatorData( - msg?.actorDisplayName ?: "", - msg?.actorType ?: "", - msg?.actorId ?: "", - msg != null - ) - } + val lastCallSystemMessage = _lastCallSystemMessage.map { it ?: CallStartedIndicatorData("", "", "", false) } private val _callEndedSystemMessage = MutableSharedFlow(extraBufferCapacity = 1) val callEndedSystemMessage: SharedFlow @@ -1375,20 +1370,7 @@ class ChatViewModel @AssistedInject constructor( val chatMessageMap = chatMessageList.associateBy { it.jsonMessageId }.toMutableMap() val chatMessageIterator = chatMessageMap.iterator() - chatMessageList.lastOrNull { - it.systemMessageType in - listOf( - ChatMessage.SystemMessageType.CALL_STARTED, - ChatMessage.SystemMessageType.CALL_JOINED, - ChatMessage.SystemMessageType.CALL_LEFT, - ChatMessage.SystemMessageType.CALL_ENDED, - ChatMessage.SystemMessageType.CALL_TRIED, - ChatMessage.SystemMessageType.CALL_ENDED_EVERYONE, - ChatMessage.SystemMessageType.CALL_MISSED - ) - }?.let { callMessage -> - processCallSystemMessage(callMessage) - } + seedInitialCallStateIfNeeded(chatMessageList) while (chatMessageIterator.hasNext()) { val currentMessage = chatMessageIterator.next() @@ -1401,30 +1383,78 @@ class ChatViewModel @AssistedInject constructor( } private var hasSeenInitialCallSystemMessage = false - private var lastNotifiedCallEndedMessageId: Int? = null - private fun processCallSystemMessage(recent: ChatMessage) { - val isInitialSnapshot = !hasSeenInitialCallSystemMessage + /** + * Seeds the call-started banner exactly once per opened chat, from the tail window that + * [messagesFlow] always emits first (before any anchor/search navigation can switch it to a + * bounded history window) — so a call already running when the chat is opened still shows the + * banner immediately, without waiting for a live event. Every later update goes exclusively + * through [observeCallSystemMessages]; this seed never re-runs, so it can't misfire from a + * history window loaded afterwards. + */ + private fun seedInitialCallStateIfNeeded(chatMessageList: List) { + if (hasSeenInitialCallSystemMessage) { + return + } hasSeenInitialCallSystemMessage = true - when (recent.systemMessageType) { - ChatMessage.SystemMessageType.CALL_STARTED -> { - _lastCallSystemMessage.tryEmit(recent) - } - ChatMessage.SystemMessageType.CALL_ENDED, - ChatMessage.SystemMessageType.CALL_ENDED_EVERYONE -> { - _lastCallSystemMessage.tryEmit(null) - if (!isInitialSnapshot && lastNotifiedCallEndedMessageId != recent.jsonMessageId) { - _callEndedSystemMessage.tryEmit(recent.systemMessageType!!) + val callMessages = chatMessageList.filter { + it.systemMessageType in + listOf( + ChatMessage.SystemMessageType.CALL_STARTED, + ChatMessage.SystemMessageType.CALL_ENDED, + ChatMessage.SystemMessageType.CALL_ENDED_EVERYONE, + ChatMessage.SystemMessageType.CALL_MISSED, + ChatMessage.SystemMessageType.CALL_TRIED + ) + } + // maxByOrNull, not lastOrNull: chatMessageList isn't guaranteed ascending for every source + // (e.g. an unconditional "newest messages" fetch can return newest-first), so the last array + // element isn't reliably the most recent one. + val lastDecisiveCallMessage = callMessages.maxByOrNull { it.jsonMessageId } ?: return + + if (lastDecisiveCallMessage.systemMessageType == ChatMessage.SystemMessageType.CALL_STARTED) { + _lastCallSystemMessage.value = CallStartedIndicatorData( + actorDisplayName = lastDecisiveCallMessage.actorDisplayName ?: "", + actorType = lastDecisiveCallMessage.actorType ?: "", + actorId = lastDecisiveCallMessage.actorId ?: "", + shouldShow = true + ) + } + } + + /** + * Drives the call-started banner and the call-ended notification from + * [ChatMessageRepository.callSystemMessageFlow], which only emits for call system messages that + * just genuinely arrived (see [ChatMessageSyncer.Events.onCallSystemMessage]) — never for one + * merely present in a re-loaded/paginated window of already-known history (e.g. jumping to an old + * message). Deriving call state from whatever history window happened to be loaded is what used + * to make the banner reappear for calls long since ended. + */ + private fun observeCallSystemMessages() { + chatRepository.callSystemMessageFlow + .onEach { event -> + when (event) { + is ChatMessageSyncer.CallSystemMessageEvent.CallStarted -> { + _lastCallSystemMessage.value = CallStartedIndicatorData( + actorDisplayName = event.actorDisplayName, + actorType = event.actorType, + actorId = event.actorId, + shouldShow = true + ) + } + + is ChatMessageSyncer.CallSystemMessageEvent.CallEnded -> { + _lastCallSystemMessage.value = null + if (event.systemMessageType == ChatMessage.SystemMessageType.CALL_ENDED || + event.systemMessageType == ChatMessage.SystemMessageType.CALL_ENDED_EVERYONE + ) { + _callEndedSystemMessage.tryEmit(event.systemMessageType) + } + } } - lastNotifiedCallEndedMessageId = recent.jsonMessageId - } - ChatMessage.SystemMessageType.CALL_MISSED, - ChatMessage.SystemMessageType.CALL_TRIED -> { - _lastCallSystemMessage.tryEmit(null) } - else -> {} - } + .launchIn(viewModelScope) } private fun isInfoMessageAboutDeletion(currentMessage: MutableMap.MutableEntry): Boolean =