From 12e71c1bc69e450c0ff7a924c484e08a3d6d7b1f Mon Sep 17 00:00:00 2001 From: infi Date: Wed, 29 Jul 2026 06:03:01 +0200 Subject: [PATCH] feat: adjust event propagation to avoid android primitives (+perf) KMP sure is coming --- app/src/main/java/chat/stoat/api/StoatAPI.kt | 88 +++++---- .../chat/stoat/api/realtime/RealtimeSocket.kt | 71 ++++--- .../chat/views/channel/ChannelScreen.kt | 4 - .../views/channel/ChannelScreenViewModel.kt | 187 ++++++++++++++---- 4 files changed, 239 insertions(+), 111 deletions(-) diff --git a/app/src/main/java/chat/stoat/api/StoatAPI.kt b/app/src/main/java/chat/stoat/api/StoatAPI.kt index a9175d8e..2d31b418 100644 --- a/app/src/main/java/chat/stoat/api/StoatAPI.kt +++ b/app/src/main/java/chat/stoat/api/StoatAPI.kt @@ -1,7 +1,5 @@ package chat.stoat.api -import android.os.Handler -import android.os.Looper import android.util.Log import androidx.compose.runtime.mutableStateMapOf import chat.stoat.BuildConfig @@ -38,22 +36,29 @@ import io.ktor.client.plugins.websocket.WebSockets import io.ktor.client.request.header import io.ktor.serialization.kotlinx.json.json import io.sentry.Sentry +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.DelicateCoroutinesApi import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.newSingleThreadContext -import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withContext import kotlinx.serialization.ExperimentalSerializationApi import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable import kotlinx.serialization.cbor.Cbor import kotlinx.serialization.json.Json +import logcat.LogPriority +import logcat.asLog +import logcat.logcat import java.net.SocketException +import kotlin.time.Duration.Companion.seconds import chat.stoat.core.model.schemas.Channel as ChannelSchema fun String.api(): String { @@ -132,10 +137,13 @@ val StoatHttp = HttpClient(OkHttp) { } } -val mainHandler = Handler(Looper.getMainLooper()) - object StoatAPI { const val TOKEN_HEADER_NAME = "x-session-token" + private const val WS_EVENT_BUFFER_CAPACITY = + 128 // arbitrary -- should be adjusted if too much gets dropped... + private val INITIAL_RECONNECT_DELAY = 1.seconds + private val MAX_RECONNECT_DELAY = 30.seconds + private val PING_INTERVAL = 30.seconds // Same interval as the web clients (/revolt.js) val userCache = mutableStateMapOf() val serverCache = mutableStateMapOf() @@ -159,10 +167,11 @@ object StoatAPI { val realtimeContext = newSingleThreadContext("RealtimeContext") val wsFrameChannel = MutableSharedFlow( replay = 0, - extraBufferCapacity = Int.MAX_VALUE, + extraBufferCapacity = WS_EVENT_BUFFER_CAPACITY, ) private var socketCoroutine: Job? = null + private var pingCoroutine: Job? = null private var openForLocalHydration = true @@ -183,28 +192,36 @@ object StoatAPI { @OptIn(ExperimentalCoroutinesApi::class) suspend fun connectWS() { + socketCoroutine?.cancelAndJoin() + RealtimeSocket.updateDisconnectionState(DisconnectionState.Reconnecting) + val token = sessionToken socketCoroutine = CoroutineScope(Dispatchers.IO).launch { - try { - withContext(realtimeContext) { - try { - RealtimeSocket.connect(sessionToken) - } catch (e: SocketException) { - Log.d("RevoltAPI", "Socket closed, probably no big deal /// " + e.message) - RealtimeSocket.updateDisconnectionState(DisconnectionState.Disconnected) - } catch (e: Exception) { - Log.e("RevoltAPI", "WebSocket error", e) - RealtimeSocket.updateDisconnectionState(DisconnectionState.Disconnected) - } - } - } catch (e: Exception) { + var reconnectDelay = INITIAL_RECONNECT_DELAY + while (isActive && sessionToken == token) { try { - if (e is InterruptedException) { - Log.d("RevoltAPI", "Socket interrupted") - } else { - Log.e("RevoltAPI", "WebSocket error", e) + withContext(realtimeContext) { + RealtimeSocket.connect(token) } - RealtimeSocket.updateDisconnectionState(DisconnectionState.Disconnected) + reconnectDelay = INITIAL_RECONNECT_DELAY + } catch (e: CancellationException) { + throw e + } catch (e: SocketException) { + logcat { "WebSocket closed: ${e.message}" } } catch (e: Exception) { + logcat(LogPriority.ERROR) { "WebSocket error:\n${e.asLog()}" } + } + + if (!isActive || sessionToken != token) break + + try { + RealtimeSocket.updateDisconnectionState(DisconnectionState.Reconnecting) + delay(reconnectDelay) + reconnectDelay = + (reconnectDelay * 2).coerceAtMost(MAX_RECONNECT_DELAY) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + RealtimeSocket.updateDisconnectionState(DisconnectionState.Disconnected) Sentry.captureMessage("Error in socket error handling: $e") } } @@ -214,17 +231,20 @@ object StoatAPI { private suspend fun startSocketOps() { connectWS() - // Send a ping every roughly 30 seconds else the socket dies - // Same interval as the web clients (/revolt.js) - // Note: This will run even if the socket is closed (sendPing will just exit early) - mainHandler.post(object : Runnable { - override fun run() { - runBlocking { + // Send a ping every roughly PING_INTERVAL else the socket dies + pingCoroutine?.cancel() + pingCoroutine = CoroutineScope(Dispatchers.IO).launch { + while (isActive) { + delay(PING_INTERVAL) + try { RealtimeSocket.sendPing() + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { "Failed to ping WebSocket:\n${e.asLog()}" } } - mainHandler.postDelayed(this, 30 * 1000) } - }) + } } suspend fun initialize() { @@ -259,7 +279,7 @@ object StoatAPI { unreads.clear() socketCoroutine?.cancel() - mainHandler.removeCallbacksAndMessages(null) + pingCoroutine?.cancel() clearPersistentCache() } @@ -372,4 +392,4 @@ data class RateLimitResponse(@SerialName("retry_after") val retryAfter: Int) { internal const val NO_RETRY_AFTER = Int.MIN_VALUE class HitRateLimitException(retryAfter: Int = NO_RETRY_AFTER) : - Exception(if (retryAfter == NO_RETRY_AFTER) "Hit rate limit" else "Hit rate limit, retry after ${retryAfter}ms") \ No newline at end of file + Exception(if (retryAfter == NO_RETRY_AFTER) "Hit rate limit" else "Hit rate limit, retry after ${retryAfter}ms") diff --git a/app/src/main/java/chat/stoat/api/realtime/RealtimeSocket.kt b/app/src/main/java/chat/stoat/api/realtime/RealtimeSocket.kt index c49ebca0..27738260 100644 --- a/app/src/main/java/chat/stoat/api/realtime/RealtimeSocket.kt +++ b/app/src/main/java/chat/stoat/api/realtime/RealtimeSocket.kt @@ -97,46 +97,55 @@ object RealtimeSocket { socket?.close(CloseReason(CloseReason.Codes.NORMAL, "Reconnecting to websocket.")) - StoatHttp.ws(STOAT_WEBSOCKET) { - socket = this + var activeSocket: WebSocketSession? = null + try { + StoatHttp.ws(STOAT_WEBSOCKET) { + activeSocket = this + socket = this - Log.d("RealtimeSocket", "Connected to websocket.") - updateDisconnectionState(DisconnectionState.Connected) - pushReconnectEvent() + logcat { "Connected to websocket." } + updateDisconnectionState(DisconnectionState.Connected) + pushReconnectEvent() - // Send authorization frame - val authFrame = AuthorizationFrame("Authenticate", token) - val authFrameString = - StoatJson.encodeToString(AuthorizationFrame.serializer(), authFrame) + // Send authorization frame + val authFrame = AuthorizationFrame("Authenticate", token) + val authFrameString = + StoatJson.encodeToString(AuthorizationFrame.serializer(), authFrame) - Log.d( - "RealtimeSocket", - "Sending authorization frame: ${ - authFrameString.replace( - token, - "X".repeat(token.length) - ) - }" - ) - send(StoatJson.encodeToString(AuthorizationFrame.serializer(), authFrame)) + logcat { + "Sending authorization frame: ${ + authFrameString.replace( + token, + "X".repeat(token.length) + ) + }" + } + send(StoatJson.encodeToString(AuthorizationFrame.serializer(), authFrame)) - incoming.consumeEach { frame -> - if (frame is Frame.Text) { - val frameString = frame.readText() - try { - val frameType = - StoatJson.decodeFromString(AnyFrame.serializer(), frameString).type + incoming.consumeEach { frame -> + if (frame is Frame.Text) { + val frameString = frame.readText() + try { + val frameType = + StoatJson.decodeFromString(AnyFrame.serializer(), frameString).type - handleFrame(frameType, frameString) - } catch (e: CancellationException) { - throw e - } catch (e: Exception) { - logcat(LogPriority.ERROR) { - "Failed to handle frame: $frameString\n" + e.asLog() + handleFrame(frameType, frameString) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { + "Failed to handle frame: $frameString\n" + e.asLog() + } } } } } + } finally { + if (activeSocket == null || socket === activeSocket) { + socket = null + updateDisconnectionState(DisconnectionState.Disconnected) + logcat { "WebSocket disconnected." } + } } } diff --git a/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreen.kt b/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreen.kt index deec0468..367164ea 100644 --- a/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreen.kt +++ b/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreen.kt @@ -229,10 +229,6 @@ fun ChannelScreen( val resources = LocalResources.current val config = LocalConfiguration.current - LaunchedEffect(Unit) { - viewModel.listenToWsEvents() - } - DisposableEffect(Unit) { val job = scope.launch { viewModel.listenToUiCallbacks() } diff --git a/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreenViewModel.kt b/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreenViewModel.kt index e4e9410f..63eaa61c 100644 --- a/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreenViewModel.kt +++ b/app/src/main/java/chat/stoat/screens/chat/views/channel/ChannelScreenViewModel.kt @@ -40,7 +40,6 @@ import chat.stoat.api.routes.microservices.autumn.MAX_ATTACHMENTS_PER_MESSAGE import chat.stoat.api.routes.microservices.autumn.uploadToAutumn import chat.stoat.api.routes.server.fetchMember import chat.stoat.api.routes.user.addUserIfUnknown -import chat.stoat.api.routes.user.fetchUser import chat.stoat.api.settings.GeoStateProvider import chat.stoat.callbacks.Action import chat.stoat.callbacks.ActionChannel @@ -63,6 +62,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.flow.catch +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onEach @@ -75,12 +75,18 @@ import logcat.LogPriority import logcat.asLog import logcat.logcat import java.time.ZoneId +import kotlin.time.Duration.Companion.seconds class ChannelScreenViewModel( private val kvStorage: KVStorage, ) : ViewModel() { + companion object { + private val TYPING_INDICATOR_TIMEOUT = 10.seconds + } + var items = mutableStateListOf() var typingUsers = mutableStateListOf() + private val typingExpiryJobs = mutableMapOf() var channelId by mutableStateOf(null) val channel: Channel? @@ -137,6 +143,9 @@ class ChannelScreenViewModel( viewModelScope.launch { keyboardHeight = kvStorage.getInt("keyboardHeight") ?: 900 // reasonable default for now } + viewModelScope.launch { + listenToWsEvents() + } } private var loadMessagesJob: Job? = null @@ -145,6 +154,9 @@ class ChannelScreenViewModel( fun switchChannel(id: String) { // Reset state this.loadMessagesJob?.cancel() + this.stopTypingJob?.cancel() + stopTyping(channelId) + clearAllTypingUsers() requestSequence++ this.channelId = id this.items = mutableStateListOf(ChannelScreenItem.Loading) @@ -212,6 +224,31 @@ class ChannelScreenViewModel( } } + private fun refreshTypingUser(userId: String) { + if (userId == StoatAPI.selfId) return + + if (!typingUsers.contains(userId)) { + typingUsers.add(userId) + } + typingExpiryJobs.remove(userId)?.cancel() + typingExpiryJobs[userId] = viewModelScope.launch { + delay(TYPING_INDICATOR_TIMEOUT) + typingUsers.remove(userId) + typingExpiryJobs.remove(userId) + } + } + + private fun clearTypingUser(userId: String) { + typingExpiryJobs.remove(userId)?.cancel() + typingUsers.remove(userId) + } + + private fun clearAllTypingUsers() { + typingExpiryJobs.values.forEach(Job::cancel) + typingExpiryJobs.clear() + typingUsers.clear() + } + private suspend fun denyMessageFieldIfNeeded() { if (channel == null) return @@ -258,16 +295,19 @@ class ChannelScreenViewModel( private fun startTyping() { if (editingMessage != null) return + val targetChannelId = channel?.id ?: return if (lastSentBeginTyping != null) { val diff = Clock.System.now() - lastSentBeginTyping!! if (diff.inWholeSeconds < 1) return } viewModelScope.launch { - withContext(StoatAPI.realtimeContext) { - channel?.id?.let { - RealtimeSocket.beginTyping(it) - } + try { + RealtimeSocket.beginTyping(targetChannelId) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { "Failed to begin typing:\n${e.asLog()}" } } } @@ -278,18 +318,24 @@ class ChannelScreenViewModel( private fun queueStopTyping() { stopTypingJob = viewModelScope.launch { - delay(5000) + delay(5.seconds) + stopTypingJob = null stopTyping() } } - private fun stopTyping() { + private fun stopTyping(targetChannelId: String? = channel?.id) { + lastSentBeginTyping = null if (editingMessage != null) return + if (targetChannelId == null) return + viewModelScope.launch { - withContext(StoatAPI.realtimeContext) { - channel?.id?.let { - RealtimeSocket.endTyping(it) - } + try { + RealtimeSocket.endTyping(targetChannelId) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { "Failed to end typing:\n${e.asLog()}" } } } } @@ -373,6 +419,7 @@ class ChannelScreenViewModel( stopTypingJob?.cancel() queueStopTyping() } else { + stopTypingJob?.cancel() stopTyping() } } @@ -785,12 +832,74 @@ class ChannelScreenViewModel( ackChannel(channel?.id ?: return, messageId) } - suspend fun listenToWsEvents() { - withContext(StoatAPI.realtimeContext) { - StoatAPI.wsFrameChannel.onEach { + private fun hydrateIncomingMessage(message: MessageFrame, expectedChannelId: String) { + val userId = message.author + val serverId = channel?.server + + if (userId != null) { + viewModelScope.launch { + try { + addUserIfUnknown(userId) + if (serverId != null && !StoatAPI.members.hasMember(serverId, userId)) { + fetchMember(serverId, userId) + } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { + "Failed to hydrate message author:\n${e.asLog()}" + } + } + } + } + + val messageId = message.id + if (messageId != null) { + viewModelScope.launch { + try { + ackChannel(expectedChannelId, messageId) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { "Failed to ack message:\n${e.asLog()}" } + } + } + } + + if (messageId != null && message.system == null && !message.content.isNullOrBlank()) { + viewModelScope.launch { + val ast = try { + withContext(Dispatchers.Default) { parseAst(message.content) } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { + "Failed to parse incoming message:\n${e.asLog()}" + } + return@launch + } + if (channelId != expectedChannelId) return@launch + + val index = items.indexOfFirst { item -> + item is ChannelScreenItem.RegularMessage && item.message.id == messageId + } + val current = items.getOrNull(index) as? ChannelScreenItem.RegularMessage + ?: return@launch + if (current.message.content == message.content) { + items[index] = current.copy(mdAst = ast) + } + } + } + } + + private suspend fun listenToWsEvents() { + StoatAPI.wsFrameChannel.onEach { + try { when (it) { is MessageFrame -> { if (it.channel != channel?.id) return@onEach + it.author?.let(::clearTypingUser) + // If we already have the message we are just catching up on the WebSocket connection. Skip if (items.any { m -> (m is ChannelScreenItem.RegularMessage && m.message.id == it.id) || (m is ChannelScreenItem.SystemMessage && m.message.id == it.id) }) return@onEach it.id?.let { messageId -> StoatAPI.messageCache[messageId] = it } @@ -800,25 +909,10 @@ class ChannelScreenViewModel( return@onEach } - it.author?.let { userId -> - if (StoatAPI.userCache[userId] == null) { - StoatAPI.userCache[userId] = fetchUser(userId) - } - } - channel?.server?.let { serverId -> - try { - it.author?.let { userId -> - fetchMember(serverId, userId) - } - } catch (e: Exception) { - Log.e("ChannelScreenViewModel", "Failed to fetch member", e) - } - } - if (didInitialChannelFetch) { // this check is so that we don't end up with a message that arrives at the same time as the initial fetch in front of the loading indicator val newItem = when { it.system != null -> ChannelScreenItem.SystemMessage(it) - else -> ChannelScreenItem.RegularMessage(it, parseAst(it.content)) + else -> ChannelScreenItem.RegularMessage(it, null) } updateItems(listOf(newItem) + items.filter { m -> if (m is ChannelScreenItem.ProspectiveMessage) { @@ -829,7 +923,7 @@ class ChannelScreenViewModel( }) } - it.id?.let { mid -> ackMessage(mid) } + hydrateIncomingMessage(it, channel?.id ?: return@onEach) } is MessageDeleteFrame -> { @@ -949,18 +1043,26 @@ class ChannelScreenViewModel( is ChannelStartTypingFrame -> { if (it.id != channel?.id) return@onEach - if (typingUsers.contains(it.user)) return@onEach if (it.user == StoatAPI.selfId) return@onEach - addUserIfUnknown(it.user) - typingUsers.add(it.user) + refreshTypingUser(it.user) + viewModelScope.launch { + try { + addUserIfUnknown(it.user) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { + "Failed to hydrate typing user:\n${e.asLog()}" + } + } + } } is ChannelStopTypingFrame -> { if (it.id != channel?.id) return@onEach - if (!typingUsers.contains(it.user)) return@onEach - typingUsers.remove(it.user) + clearTypingUser(it.user) } is ChannelDeleteFrame -> { @@ -982,14 +1084,15 @@ class ChannelScreenViewModel( } else { loadLatest(markLastAsRead = true) } - typingUsers.clear() - listenToWsEvents() + clearAllTypingUsers() } } - }.catch { - Log.e("ChannelScreen", "Failed to receive WS frame", it) - }.launchIn(this) - } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + logcat(LogPriority.ERROR) { "Failed to receive WS frame:\n${e.asLog()}" } + } + }.collect() } suspend fun listenToUiCallbacks() {