Fix WebSocket real-time messaging

This commit is contained in:
2025-12-17 16:43:22 +03:00
Unverified
parent dd4f6f5552
commit a59672103b
7 changed files with 149 additions and 46 deletions
@@ -111,7 +111,7 @@ object ApiClient {
}
// WebSocket send helpers
suspend fun sendMessage(content: String, replyToId: Int? = null) {
suspend fun sendMessage(content: String, replyToId: Int? = null, clientMessageId: String? = null) {
val token = token ?: throw IllegalStateException("Not authenticated")
WebSocketManager.send(
WebSocketMessage(
@@ -123,7 +123,8 @@ object ApiClient {
data = json.encodeToJsonElement(
WebSocketSendMessageRequest(
content = content,
reply_to_id = replyToId
reply_to_id = replyToId,
client_message_id = clientMessageId
)
)
)
@@ -57,7 +57,9 @@ data class Message(
val username: String,
val profile_picture: String? = null,
val verified: Boolean? = null,
val reply_to: Message? = null
val reply_to: Message? = null,
val client_message_id: String? = null,
val reactions: List<ReactionData>? = null
) {
val utcTimestamp = "${timestamp}Z"
}
@@ -140,7 +142,8 @@ data class UpdatesMessage(
@Serializable
data class WebSocketSendMessageRequest(
val content: String,
val reply_to_id: Int? = null
val reply_to_id: Int? = null,
val client_message_id: String? = null
)
@Serializable
@@ -0,0 +1,39 @@
package ru.fromchat.api
import kotlinx.serialization.Serializable
@Serializable
data class WebSocketUpdatesData(
val seq: Int,
val updates: List<WebSocketMessage>
)
@Serializable
data class ReactionUpdateData(
val message_id: Int,
val emoji: String,
val action: String,
val user_id: Int,
val username: String,
val reactions: List<ReactionData>
)
@Serializable
data class ReactionData(
val emoji: String,
val count: Int,
val users: List<ReactionUser>
)
@Serializable
data class ReactionUser(
val id: Int,
val username: String
)
@Serializable
data class TypingUpdateData(
val userId: Int,
val username: String
)
@@ -201,7 +201,7 @@ abstract class ChatPanel(
// Retry sending
try {
// Extract content from message
sendMessage(message.content, message.reply_to?.id)
sendMessage(message.content, message.reply_to?.id, message.client_message_id)
} catch (e: Exception) {
timeoutJob.cancel()
pendingMessages.remove(tempId)
@@ -276,7 +276,7 @@ abstract class ChatPanel(
// Actually send the message
try {
sendMessage(content, replyToId)
sendMessage(content, replyToId, tempId)
// Message sent successfully - will be updated when WebSocket confirms
} catch (error: Exception) {
// Remove the temporary message from display
@@ -297,7 +297,7 @@ abstract class ChatPanel(
}
// Abstract methods to implement
abstract suspend fun sendMessage(content: String, replyToId: Int?)
abstract suspend fun sendMessage(content: String, replyToId: Int?, clientMessageId: String?)
abstract suspend fun loadMessages()
abstract suspend fun loadMoreMessages()
abstract suspend fun handleWebSocketMessage(message: WebSocketMessage)
@@ -5,7 +5,10 @@ import kotlinx.serialization.json.decodeFromJsonElement
import ru.fromchat.api.ApiClient
import ru.fromchat.api.Message
import ru.fromchat.api.MessageDeletedData
import ru.fromchat.api.ReactionUpdateData
import ru.fromchat.api.TypingUpdateData
import ru.fromchat.api.WebSocketMessage
import ru.fromchat.api.WebSocketUpdatesData
import ru.fromchat.core.Logger
class PublicChatPanel(
@@ -24,8 +27,14 @@ class PublicChatPanel(
updateState { it.copy(title = chatName) }
}
override suspend fun sendMessage(content: String, replyToId: Int?) {
ApiClient.sendMessage(content, replyToId)
private fun handleReactionUpdate(reactionUpdate: ReactionUpdateData) {
updateMessage(reactionUpdate.message_id) { message ->
message.copy(reactions = reactionUpdate.reactions)
}
}
override suspend fun sendMessage(content: String, replyToId: Int?, clientMessageId: String?) {
ApiClient.sendMessage(content, replyToId, clientMessageId)
}
override suspend fun loadMessages() {
@@ -75,57 +84,73 @@ class PublicChatPanel(
}
}
override suspend fun handleWebSocketMessage(message: WebSocketMessage) {
private suspend fun handleSingleUpdate(updateMessage: WebSocketMessage) {
val json = ApiClient.json
Logger.d("PublicChatPanel", "Handling WebSocket message: type=${message.type}")
when (message.type) {
when (updateMessage.type) {
"newMessage" -> {
val data = message.data ?: return
// Data is directly a Message, not wrapped
val data = updateMessage.data ?: return
val newMsg = json.decodeFromJsonElement<Message>(data)
Logger.d("PublicChatPanel", "New message received: id=${newMsg.id}, content=${newMsg.content.take(50)}")
// Check if this is a confirmation of a message we sent
val isOurMessage = newMsg.user_id == currentUserId
if (isOurMessage) {
// This is our message being confirmed, find the temp message and replace it
val tempMessages = _state.messages.filter { it.id < 0 }
for (tempMsg in tempMessages) {
if (tempMsg.content == newMsg.content) {
// Replace temp message with confirmed
updateState { currentState ->
currentState.copy(
messages = currentState.messages.map { msg ->
if (msg.id < 0 && msg.content == newMsg.content) {
newMsg
} else {
msg
}
}
)
}
return
}
}
if (newMsg.client_message_id != null && newMsg.user_id == currentUserId) {
handleMessageConfirmed(newMsg.client_message_id, newMsg)
} else {
addMessage(newMsg)
}
addMessage(newMsg)
}
"messageEdited" -> {
val data = message.data ?: return
// Data is directly a Message
val data = updateMessage.data ?: return
val editedMsg = json.decodeFromJsonElement<Message>(data)
updateMessage(editedMsg.id) { editedMsg }
}
"messageDeleted" -> {
val data = message.data ?: return
// Data is { message_id: Int }
val data = updateMessage.data ?: return
val deletedData = json.decodeFromJsonElement<MessageDeletedData>(data)
removeMessage(deletedData.message_id)
}
"typing" -> {
// Typing status is handled in ChatScreen
"reactionUpdate" -> {
val data = updateMessage.data ?: return
val reactionUpdate = json.decodeFromJsonElement<ReactionUpdateData>(data)
handleReactionUpdate(reactionUpdate)
}
"typing" -> {
val data = updateMessage.data ?: return
val typingData = json.decodeFromJsonElement<TypingUpdateData>(data)
typingHandler.handleTypingEvent(typingData.userId, typingData.username)
}
"stopTyping" -> {
val data = updateMessage.data ?: return
val typingData = json.decodeFromJsonElement<TypingUpdateData>(data)
typingHandler.handleStopTypingEvent(typingData.userId)
}
"statusUpdate" -> {
// Handled in ChatScreen or by global WebSocketManager listeners
}
"suspended" -> {
// Handled by global WebSocketManager listeners or shown as a toast
}
"account_deleted" -> {
// Handled by global WebSocketManager listeners or shown as a toast
}
else -> {
Logger.w("PublicChatPanel", "Unhandled WebSocket update type: ${updateMessage.type}")
}
}
}
override suspend fun handleWebSocketMessage(message: WebSocketMessage) {
Logger.d("PublicChatPanel", "Handling raw WebSocket message: type=${message.type}")
if (message.type == "updates") {
val json = ApiClient.json
val data = message.data ?: return
val updatesData = json.decodeFromJsonElement<WebSocketUpdatesData>(data)
Logger.d("PublicChatPanel", "Received ${updatesData.updates.size} batched updates (seq: ${updatesData.seq})")
updatesData.updates.forEach { update ->
handleSingleUpdate(update)
}
} else {
// Fallback for non-batched messages (legacy or direct signals)
handleSingleUpdate(message)
}
}
@@ -3,8 +3,13 @@ package ru.fromchat.ui.chat
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
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 ru.fromchat.api.ApiClient
import kotlin.time.Duration.Companion.seconds
/**
* Interface for handling typing indicators
@@ -12,8 +17,16 @@ import ru.fromchat.api.ApiClient
interface TypingHandler {
fun sendTyping()
fun stopTyping()
fun handleTypingEvent(userId: Int, username: String)
fun handleStopTypingEvent(userId: Int)
val typingUsers: StateFlow<List<TypingUser>>
}
data class TypingUser(
val userId: Int,
val username: String
)
/**
* Typing handler for public chat using WebSocket
*/
@@ -21,6 +34,8 @@ class PublicChatTypingHandler(
private val scope: CoroutineScope
) : TypingHandler {
private var stopTypingJob: Job? = null
private val _typingUsers = MutableStateFlow<List<TypingUser>>(emptyList())
override val typingUsers = _typingUsers.asStateFlow()
override fun sendTyping() {
scope.launch {
@@ -36,7 +51,7 @@ class PublicChatTypingHandler(
// Schedule stop typing after delay
stopTypingJob = scope.launch {
delay(3000) // 3 seconds
delay(3.seconds) // 3 seconds
stopTyping()
}
}
@@ -52,6 +67,21 @@ class PublicChatTypingHandler(
}
}
}
override fun handleTypingEvent(userId: Int, username: String) {
_typingUsers.update { currentUsers ->
if (currentUsers.none { it.userId == userId }) {
currentUsers + TypingUser(userId, username)
} else {
currentUsers
}
}
}
override fun handleStopTypingEvent(userId: Int) {
_typingUsers.update { currentUsers ->
currentUsers.filter { it.userId != userId }
}
}
}