override fun oneOnOneChatMessagesFromConversationDescending(
        oneOnOneChatConversationId: String
    ): Flow<State<List<OneOnOneChatMessage<ChatMessageContent>>>> =
        channelFlow<State<List<OneOnOneChatMessage<ChatMessageContent>>>> {
            println("PENTAGON: channelFlow start")
            send(State.Loading)

            val chatMessageHistoryDataState: State.Data<List<OneOnOneChatMessage<ChatMessageContent>>> = run {
                val chatMessageHistory: List<OneOnOneChatMessage<ChatMessageContent>> = run {
                    when (val getChatMessageHistoryResult =
                        getChatMessagesFromOneOnOneChatConversation(oneOnOneChatConversationId)) {

                        is Result.Failure -> {
                            offer(State.Error(getChatMessageHistoryResult.exception))
                            close()
                            return@channelFlow
                        }
                        is Result.Success -> getChatMessageHistoryResult.data
                    }
                }
                State.Data(chatMessageHistory)
            }
            send(chatMessageHistoryDataState)

            val chatMessageStatusUpdates: Flow<OneOnOneChatMessageStatusUpdate> =
                chatMessageStatusUpdatesFromConversation(oneOnOneChatConversationId)

            val incomingChatMessageResults: Flow<Result<OneOnOneChatMessage<ChatMessageContent>>> =
                incomingChatMessagesFromOneOnOneConversation(
                    quickbloxChatService = QBChatService.getInstance(),
                    oneOnOneChatConversationId = oneOnOneChatConversationId
                )

            val incomingFailedChatMessageResults = incomingChatMessageResults.filterIsInstance<Result.Failure>()

            val incomingSuccessfulChatMessageResults = incomingChatMessageResults
                .filterIsInstance<Result.Success<OneOnOneChatMessage<ChatMessageContent>>>()

            val currentChatMessages = chatMessageHistoryDataState.data.toMutableList()

            // Update state to error on incoming failed chat message
            incomingFailedChatMessageResults.onEach {
                if (isClosedForSend) return@onEach
                send(State.Error(it.exception))
                close()
                return@onEach
            }.launchIn(this)

            // Update state to data with new incoming message
            incomingSuccessfulChatMessageResults.onEach { incomingSuccessfulChatMessageResult ->
                if (isClosedForSend) return@onEach
                currentChatMessages.add(0, incomingSuccessfulChatMessageResult.data)
                send(State.Data(currentChatMessages))
            }.launchIn(this)

            // Update state to data with new sent message
            sentChatMessages.onEach { sentChatMessage ->
                if (isClosedForSend) return@onEach
                println("PENTAGON: sentChatMessages.onEach start")

                println("PENTAGON: sentChatMessages.onEach currentChatMessages.size=${currentChatMessages.size}")
                println("PENTAGON: adding sentChatMessage, content=${sentChatMessage.content}")
                currentChatMessages.add(0, sentChatMessage)
                println("PENTAGON: sending: currentChatMessages.size=${currentChatMessages.size}")
                send(State.Data(currentChatMessages))
                println("PENTAGON: sentChatMessages.onEach end")
            }.launchIn(this)

            // Update state on chat message status update
            chatMessageStatusUpdates.onEach { chatMessageStatusUpdate ->
                if (isClosedForSend) return@onEach
                val toBeUpdatedMessageIndex = currentChatMessages.indexOfFirst {
                    it.id == chatMessageStatusUpdate.oneOnOneChatMessageId
                }
                if (toBeUpdatedMessageIndex == -1) {
                    val exception = IllegalStateException(
                        "$chatMessageStatusUpdate is not applicable to $currentChatMessages. ID not found"
                    )
                    send(State.Error(exception))
                    close()
                    return@onEach
                }

                val updatedMessage = run {
                    val toBeUpdatedMessage = currentChatMessages[toBeUpdatedMessageIndex]
                    toBeUpdatedMessage.copy(status = chatMessageStatusUpdate.newStatus)
                }
                currentChatMessages[toBeUpdatedMessageIndex] = updatedMessage
                send(State.Data(currentChatMessages))
            }.launchIn(this)
            println("PENTAGON: channelFlow end")
        }.conflate()