Fix message update issues

Signed-off-by: denis0001-dev <denis0001.dev@ya.ru>
This commit is contained in:
2026-07-20 19:02:27 +03:00
Unverified
parent 5b745f3f9c
commit 79f6ab57eb
32 changed files with 984 additions and 286 deletions
+1 -1
View File
@@ -21,7 +21,7 @@
android:networkSecurityConfig="@xml/network_security_config">
<meta-data
android:name="com.google.firebase.messaging.default_notification_icon"
android:resource="@drawable/logo" />
android:resource="@drawable/ic_stat_fromchat" />
<activity
android:name=".MainActivity"
android:exported="true"
@@ -27,7 +27,11 @@ class FromChatFirebaseMessagingService : FirebaseMessagingService() {
val fallbackMessageId = pushData["message_id"]?.toIntOrNull()
?: pushData["dm_id"]?.toIntOrNull()
val senderId = pushData["sender_id"]?.toIntOrNull()
val sender = pushData["sender_username"] ?: remoteMessage.data["senderUsername"]
val sender = pushData["sender_display_name"]
?.takeIf { it.isNotBlank() }
?: pushData["sender_username"]
?: remoteMessage.data["senderUsername"]
?: remoteMessage.data["senderDisplayName"]
val title = remoteMessage.notification?.title ?: pushData["title"] ?: "FromChat"
val body = remoteMessage.notification?.body ?: pushData["body"] ?: "New message"
val messageType = pushData["type"] ?: "public_message"
@@ -305,10 +305,11 @@ object NotificationHelper {
val senderName = when {
envelopeId == dmMessageId && !dmSenderName.isNullOrBlank() -> dmSenderName
!envelope.senderUsername.isNullOrBlank() -> envelope.senderUsername
!envelope.senderDisplayName.isNullOrBlank() -> envelope.senderDisplayName
else -> ProfileCache.get(envelope.senderId)
?.visibleDisplayName(currentUserId)
?.takeIf { it.isNotBlank() }
?: envelope.senderUsername
}.orEmpty()
val dmConversationUserId = envelope.senderId
val notificationBody = buildChatListPreviewFromEnvelope(
@@ -403,7 +404,7 @@ object NotificationHelper {
notify(
SUMMARY_NOTIFICATION_ID,
NotificationCompat.Builder(context, CHANNEL_ID)
.setSmallIcon(R.drawable.logo)
.setSmallIcon(NotificationSmallIcon.resId(context))
.setContentTitle(title)
.setContentText(body)
.setStyle(
@@ -506,7 +507,7 @@ object NotificationHelper {
notify(
SUMMARY_NOTIFICATION_ID,
NotificationCompat.Builder(context, CHANNEL_ID)
.setSmallIcon(R.drawable.logo)
.setSmallIcon(NotificationSmallIcon.resId(context))
.setStyle(
NotificationCompat.MessagingStyle(
Person.Builder().setName("FromChat").build()
@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="utf-8"?>
<vector
xmlns:android="http://schemas.android.com/apk/res/android"
android:width="24dp"
android:height="24dp"
android:viewportWidth="1000"
android:viewportHeight="1000">
<path
android:fillColor="#FFFFFFFF"
android:pathData="M494,221.5c43.2,0 85.8,0.1 128.4,-0 42.7,-0.1 81.5,12.2 114.8,39.8 30.5,25.2 49.7,58 57,97.6 1.8,9.4 2.6,19.1 2.6,28.7 0.3,47.8 0,95.6 0.1,143.4 0.1,32.6 -9.9,61.9 -28.3,88.2 -21.2,30.3 -48.8,52.2 -83.2,64.2 -12.9,4.5 -26.9,7.5 -40.5,7.6 -95.6,0.6 -191.2,0.2 -286.8,0.6 -5.2,0 -11.3,2.8 -15.6,6.2 -22.2,17.2 -43.8,35.1 -65.8,52.5 -10.9,8.6 -21.9,17.4 -33.8,24.4 -14.4,8.4 -31.4,2.4 -37.5,-12.1 -1.8,-4.1 -2.1,-9.2 -2.1,-13.8 -0.1,-97.8 -0.2,-195.6 -0.2,-293.4 0,-24.8 -0.9,-49.6 0.3,-74.4 1.8,-37.5 14.7,-71 39.1,-99.2 31.7,-36.6 71.7,-57 119.4,-59.2 38.1,-1.7 76.3,-0.9 114.5,-1.1 5.6,-0 11.1,0 17.3,0M330.4,269.1c-16,5.5 -30.2,14.3 -43,25.6 -30.2,26.8 -46.7,60.7 -46.8,101.7 -0.4,108 -0.2,216 -0.2,324 0,2.5 0.2,4.9 0.4,8.1 1.5,-0.8 2.4,-1.1 3.1,-1.6 22.5,-18 45.5,-35.4 67.3,-54.3 15.9,-13.9 33.5,-19.1 54,-19 88.1,0.4 176.1,0.1 264.2,0.2 22.8,0 44,-6.1 63.9,-17 22.3,-12.2 40.4,-29.1 52.5,-52.1 11.1,-21.2 14.6,-44 14.3,-67.8 -0.4,-42.7 0.2,-85.3 -1.3,-128 -1.1,-34.2 -15.6,-63.3 -39.1,-87.4 -24.5,-25.1 -54,-39.6 -88.9,-39.8 -84,-0.4 -168.1,-0.1 -252.1,-0.2 -16.3,0 -32.1,2 -48.4,7.7h0Z" />
<path
android:fillColor="#FFFFFFFF"
android:pathData="M401.4,512.8c-30.4,8.6 -59,-5.7 -73.3,-29.9 -13.2,-22.3 -11,-51.9 4.9,-72.2 23.3,-29.7 68.1,-35.5 96.6,-8.4 23.3,22.2 27,58.5 7,85.5 -8.9,12.1 -20.5,20.5 -35.2,25.1h0Z" />
<path
android:fillColor="#FFFFFFFF"
android:pathData="M484.5,471.4c-9.1,-6.2 -12.4,-14.2 -9.9,-22.8 2,-7 10.4,-13.8 20,-13.7 39.3,0.4 78.7,0.2 118,0.2h7.3c0.3,-0.5 0.5,-1 0.8,-1.5 -3.5,-3.7 -6.9,-7.5 -10.4,-11.2 -7.9,-8.4 -8.3,-20 -0.9,-27.5 8.2,-8.2 21.2,-8.7 29.1,-0.7 15.2,15.3 30.3,30.7 45.3,46.1 8.1,8.4 8.4,20.7 0.3,29.1 -16.3,16.7 -32.8,33.1 -49.3,49.5 -6.1,6.1 -18.3,5.5 -24.8,-0.8 -6.9,-6.7 -8,-17.9 -2,-25 4.5,-5.3 9.7,-10.1 14.5,-15.1 1.1,-1.2 2.2,-2.4 4.2,-4.6 -3.3,-0.2 -5.3,-0.4 -7.3,-0.4 -42,0 -84.1,0.1 -126.1,-0 -2.8,0 -5.6,-0.9 -8.8,-1.5h-0Z" />
</vector>
@@ -0,0 +1,14 @@
package ru.fromchat.notifications
import android.content.Context
/** White silhouette drawable for status-bar / notification small icons. */
object NotificationSmallIcon {
private const val DRAWABLE_NAME = "ic_stat_fromchat"
fun resId(context: Context): Int {
val id = context.resources.getIdentifier(DRAWABLE_NAME, "drawable", context.packageName)
check(id != 0) { "Missing drawable/$DRAWABLE_NAME in application resources" }
return id
}
}
@@ -1,6 +1,5 @@
package ru.fromchat.ui.calls
import android.R
import android.app.Notification
import android.app.NotificationChannel
import android.app.NotificationManager
@@ -16,6 +15,7 @@ import androidx.core.app.Person
import androidx.core.app.ServiceCompat
import androidx.core.content.ContextCompat
import ru.fromchat.api.calls.CallStore
import ru.fromchat.notifications.NotificationSmallIcon
/**
* Foreground call session: keeps camera / mic eligible in background.
@@ -60,12 +60,6 @@ class CallForegroundService : Service() {
ensureActiveCallChannel(nm, channelLabel)
val smallIcon = try {
packageManager.getApplicationInfo(packageName, 0).icon
} catch (_: Exception) {
R.drawable.sym_call_outgoing
}
val hangUpPi = PendingIntent.getService(
this,
RC_HANG_UP,
@@ -93,7 +87,7 @@ class CallForegroundService : Service() {
.build()
val builder = NotificationCompat.Builder(this, CHANNEL_ID)
.setSmallIcon(smallIcon)
.setSmallIcon(NotificationSmallIcon.resId(this))
.setOngoing(true)
.setOnlyAlertOnce(true)
.setCategory(Notification.CATEGORY_CALL)
@@ -127,6 +127,7 @@ import ru.fromchat.api.ApiClient
import ru.fromchat.api.calls.CallStore
import ru.fromchat.api.calls.LiveKitConnectSession
import ru.fromchat.api.local.db.store.ProfileCache
import ru.fromchat.notifications.NotificationSmallIcon
import ru.fromchat.call_status_connecting
import ru.fromchat.call_status_reconnecting
import ru.fromchat.call_status_reconnecting_with_detail
@@ -1293,15 +1294,10 @@ private fun CallInlineControlBar(
)
nm.createNotificationChannel(ch)
}
val smallIcon = try {
context.packageManager.getApplicationInfo(context.packageName, 0).icon
} catch (_: Exception) {
R.drawable.stat_sys_upload
}
return NotificationCompat.Builder(context, SCREEN_SHARE_CHANNEL_ID)
.setContentTitle(updatedTitle)
.setContentText(updatedText)
.setSmallIcon(smallIcon)
.setSmallIcon(NotificationSmallIcon.resId(context))
.setOngoing(true)
.build()
}
@@ -8,6 +8,7 @@ import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.launch
import kotlinx.coroutines.FlowPreview
import ru.fromchat.Logger
import ru.fromchat.api.local.WebSocketManager
import ru.fromchat.api.local.cache.CacheContext
import ru.fromchat.api.local.db.store.ConnectionStateStore
@@ -86,14 +87,56 @@ object ChatListSync {
private suspend fun refreshPublicChatPreviewFromLatest() {
runCatching {
val response = ApiClient.getMessages(limit = 1)
val latest = response.messages.maxByOrNull { message ->
val cached = MessageRepository.loadPublicMessages()
val maxCachedId = cached.asSequence().map { it.id }.filter { it > 0 }.maxOrNull() ?: 0
val latestResponse = ApiClient.getMessages(limit = 1)
val latest = latestResponse.messages.maxByOrNull { message ->
parseMessageTimestampMillis(message.timestamp) ?: Long.MIN_VALUE
} ?: return@runCatching
val cachedIds = cached.asSequence().map { it.id }.filter { it > 0 }.toHashSet()
val holeBelowLatest =
latest.id > 0 &&
maxCachedId > 0 &&
latest.id > maxCachedId + 1
val latestMissingWithPriorCache =
latest.id > 0 &&
latest.id !in cachedIds &&
maxCachedId > 0 &&
latest.id > maxCachedId
if (holeBelowLatest || latestMissingWithPriorCache) {
// Preview-only upsert would leave first+last holes; pull a page and merge.
Logger.i(
"ChatListSync",
"Public preview gap: latestId=${latest.id} maxCachedId=$maxCachedId " +
"cachedCount=${cached.size} — fetching page to fill",
)
val page = ApiClient.getMessages(limit = 50)
val networkMessages = page.messages
if (networkMessages.isEmpty()) {
MessageRepository.upsertPublicMessage(latest)
return@runCatching
}
ProfileCache.mergePreviewFromPublicMessages(networkMessages)
val networkIds = networkMessages.map { it.id }.toSet()
val minNetworkId = networkMessages.minOf { it.id }
val maxNetworkId = networkMessages.maxOf { it.id }
val older = cached.filter { it.id > 0 && it.id !in networkIds && it.id < minNetworkId }
val ahead = cached.filter { it.id > 0 && it.id !in networkIds && it.id > maxNetworkId }
val merged = (networkMessages + older + ahead).distinctBy { it.id }
Logger.i(
"ChatListSync",
"Public gap fill: network=${networkMessages.size} older=${older.size} " +
"ahead=${ahead.size} merged=${merged.size} — replaceAll=true",
)
MessageRepository.replacePublicMessages(merged, replaceAll = true)
} else {
// Upsert only — does not wipe older cached messages.
MessageRepository.upsertPublicMessage(latest)
}
}
}
private fun handleWebSocketMessage(message: WebSocketMessage) {
when (message.type) {
@@ -51,13 +51,26 @@ object ProfileUpdateSync {
val data = message.data ?: return
val updates = runCatching {
ApiClient.json.decodeFromJsonElement(WebSocketUpdatesData.serializer(), data)
}.getOrNull() ?: return
}.getOrNull() ?: run {
Logger.w("ProfileUpdateSync", "updates batch decode failed")
return
}
val profileUpdates = updates.updates.count { it.type == "profileUpdate" }
if (profileUpdates > 0) {
Logger.d(
"ProfileUpdateSync",
"updates batch seq=${updates.seq} profileUpdateCount=$profileUpdates",
)
}
updates.updates.forEach { update ->
handleWebSocketMessage(WebSocketMessage(type = update.type, data = update.data))
}
}
"profileUpdate" -> {
val payload = message.data ?: return
val payload = message.data ?: run {
Logger.w("ProfileUpdateSync", "profileUpdate missing data")
return
}
onProfileUpdatePayload(payload)
}
}
@@ -73,9 +86,18 @@ object ProfileUpdateSync {
Logger.d(
"ProfileUpdateSync",
"profileUpdate id=${profile.id} username='${profile.username}' " +
"bio='${profile.bio?.take(48)}'",
"deleted=${profile.deleted} suspended=${profile.suspended} " +
"bio='${profile.bio?.take(48)}' revisionBefore=${ProfileCache.revision.value}",
)
val hadCached = ProfileCache.get(profile.id)
ProfileCache.applyServerProfile(profile, force = true)
val after = ProfileCache.get(profile.id)
Logger.d(
"ProfileUpdateSync",
"profileUpdate applied id=${profile.id} " +
"wasDeleted=${hadCached?.deleted} nowDeleted=${after?.deleted} " +
"revisionAfter=${ProfileCache.revision.value}",
)
UserStatusStore.update(profile.id, profile.online, profile.lastSeen)
if (ApiClient.user?.id == profile.id) {
@@ -83,6 +105,19 @@ object ProfileUpdateSync {
}
runCatching { MessageRepository.patchDmConversationPeerProfile(profile.id) }
.onFailure {
Logger.w(
"ProfileUpdateSync",
"patchDmConversationPeerProfile failed id=${profile.id}: ${it.message}",
it,
)
}
.onSuccess {
Logger.d(
"ProfileUpdateSync",
"patchDmConversationPeerProfile done id=${profile.id}",
)
}
}
private fun parseProfileUpdate(data: JsonElement): UserProfile? {
@@ -58,13 +58,24 @@ object UpdateSyncManager {
/**
* Apply a live or replayed updates envelope, then advance cursor and ack the server.
*
* Ack is fire-and-forget: [WebSocketManager.request] must not be awaited from the WS
* receive loop (it would deadlock the ack response cannot be read while this call blocks).
*/
suspend fun onUpdatesEnvelope(jsonTree: JsonElement) {
applyMutex.withLock {
val seq = UpdatesBatchApplier.applyEnvelope(jsonTree) ?: return@withLock
Logger.d("UpdateSync", "onUpdatesEnvelope begin")
val seq = UpdatesBatchApplier.applyEnvelope(jsonTree) ?: run {
Logger.w("UpdateSync", "onUpdatesEnvelope apply returned null")
return@withLock
}
Logger.d(
"UpdateSync",
"onUpdatesEnvelope applied seq=$seq lastSeq=${_lastSeq.value}",
)
if (seq > _lastSeq.value) {
persistLastSeq(seq)
sendAck(seq)
sendAckFireAndForget(seq)
}
}
}
@@ -90,6 +101,10 @@ object UpdateSyncManager {
/**
* Catch up from [lastSeq]: chunked getUpdates, or tooLong history rebuild.
* Does not advance the cursor until apply/rebuild succeeds.
*
* Loops until [GetUpdatesResponse.hasMore] is false. On repeated getUpdates failures
* (e.g. prior ack-deadlock timeouts), falls back to a full history rebuild so the UI
* is not left with first+last holes filled only by slow incremental envelopes.
*/
suspend fun runGapDetectionIfNeeded() {
if (gapDetectionInProgress) {
@@ -108,16 +123,38 @@ object UpdateSyncManager {
try {
var rounds = 0
var consecutiveFailures = 0
while (rounds < 100) {
rounds++
val startSeq = _lastSeq.value
Logger.i("UpdateSyncManager", "Gap detection from lastSeq=$startSeq (round=$rounds)")
val response = requestGetUpdates(token, startSeq) ?: break
val response = requestGetUpdates(token, startSeq)
if (response == null) {
consecutiveFailures++
Logger.w(
"UpdateSyncManager",
"getUpdates returned null (timeout/disconnect) " +
"failures=$consecutiveFailures lastSeq=$startSeq",
)
if (consecutiveFailures >= 2) {
Logger.w(
"UpdateSyncManager",
"Gap catch-up stalled — rebuilding from history",
)
rebuildStateFromHistory()
break
}
continue
}
consecutiveFailures = 0
val gapHint = (response.lastSeq - startSeq).coerceAtLeast(response.missedCount)
Logger.i(
"UpdateSyncManager",
"Gap detection result: status=${response.status}, lastSeq=${response.lastSeq}, " +
"missed=${response.missedCount}, hasMore=${response.hasMore}",
"missed=${response.missedCount}, hasMore=${response.hasMore}, " +
"gapHint=$gapHint clientSeq=$startSeq",
)
updateMissedCount(response.missedCount)
@@ -126,18 +163,43 @@ object UpdateSyncManager {
val ok = rebuildStateFromHistory()
if (ok) {
persistLastSeq(response.lastSeq)
sendAck(response.lastSeq)
sendAckFireAndForget(response.lastSeq)
} else {
Logger.w("UpdateSyncManager", "History rebuild failed; leaving lastSeq=$startSeq")
}
break
}
"ok" -> {
// Envelopes for this chunk are applied on the receive path before this
// response is delivered; advance cursor here only when there was nothing to apply.
if (response.lastSeq > _lastSeq.value && response.missedCount == 0) {
persistLastSeq(response.lastSeq)
sendAck(response.lastSeq)
sendAckFireAndForget(response.lastSeq)
}
if (!response.hasMore) {
Logger.i(
"UpdateSyncManager",
"Gap catch-up complete after $rounds round(s) lastSeq=${_lastSeq.value}",
)
break
}
if (_lastSeq.value <= startSeq && response.missedCount > 0) {
// Chunk was announced but cursor did not advance — avoid tight spin.
Logger.w(
"UpdateSyncManager",
"Gap chunk did not advance cursor " +
"(start=$startSeq now=${_lastSeq.value} missed=${response.missedCount})",
)
consecutiveFailures++
if (consecutiveFailures >= 2) {
val ok = rebuildStateFromHistory()
if (ok) {
persistLastSeq(response.lastSeq)
sendAckFireAndForget(response.lastSeq)
}
break
}
}
if (!response.hasMore) break
}
else -> {
Logger.w("UpdateSyncManager", "Unknown getUpdates status=${response.status}")
@@ -165,7 +227,7 @@ object UpdateSyncManager {
GetUpdatesRequest(lastSeq = lastSeq),
),
)
val response = WebSocketManager.request(requestMessage)
val response = WebSocketManager.request(requestMessage, timeoutMs = 30_000)
val data = response?.data ?: return null
return runCatching {
ApiClient.json.decodeFromJsonElement(GetUpdatesResponse.serializer(), data)
@@ -174,11 +236,15 @@ object UpdateSyncManager {
}.getOrNull()
}
private suspend fun sendAck(seq: Int) {
/**
* Send ack without waiting for a response. Must not use [WebSocketManager.request] from
* the receive/apply path that deadlocks the incoming frame loop for ~10s per envelope.
*/
private suspend fun sendAckFireAndForget(seq: Int) {
val token = ApiClient.token ?: return
if (seq <= 0) return
runCatching {
WebSocketManager.request(
WebSocketManager.send(
WebSocketMessage(
type = "ackUpdates",
credentials = WebSocketCredentials(scheme = "Bearer", credentials = token),
@@ -188,6 +254,7 @@ object UpdateSyncManager {
),
),
)
Logger.d("UpdateSync", "ackUpdates sent (fire-and-forget) seq=$seq")
}.onFailure {
Logger.w("UpdateSyncManager", "ackUpdates failed for seq=$seq: ${it.message}", it)
}
@@ -236,18 +303,26 @@ object UpdateSyncManager {
val ordered = collected.values.sortedBy {
parseMessageTimestampMillis(it.timestamp) ?: 0L
}
MessageRepository.replacePublicMessages(ordered)
MessageCacheStore.clearPublicMessages()
Logger.i(
"UpdateSync",
"rebuildPublicHistory messages=${ordered.size} — replaceAll=true",
)
MessageRepository.replacePublicMessages(ordered, replaceAll = true)
}
private suspend fun rebuildDmHistories() {
val conversations = MessageRepository.loadCachedDmConversations()
Logger.i("UpdateSync", "rebuildDmHistories conversations=${conversations.size}")
for (conversation in conversations) {
val otherId = conversation.otherUserId
MessageCacheStore.clearDmMessages(otherId)
var beforeId: Int? = null
var pageCount = 0
repeat(MAX_HISTORY_PAGES) {
val page = ApiClient.getDmHistory(otherId, limit = HISTORY_PAGE_SIZE, beforeId = beforeId)
if (page.messages.isEmpty()) return@repeat
pageCount++
for (envelope in page.messages) {
val element = ApiClient.json.encodeToJsonElement(DmEnvelope.serializer(), envelope)
DmInboundMessageProcessor.processNew(element)
@@ -256,6 +331,7 @@ object UpdateSyncManager {
if (page.messages.size < HISTORY_PAGE_SIZE) return@repeat
beforeId = oldest.id
}
Logger.d("UpdateSync", "rebuildDmHistories otherUserId=$otherId pages=$pageCount")
}
}
}
@@ -10,6 +10,7 @@ import kotlinx.coroutines.flow.mapLatest
import kotlinx.coroutines.flow.merge
import kotlinx.coroutines.withContext
import kotlinx.serialization.json.Json
import ru.fromchat.Logger
import ru.fromchat.api.ApiClient
import ru.fromchat.api.local.messages.ChatListPreviewPendingIndicator
import ru.fromchat.api.local.messages.ChatListPreviewState
@@ -167,8 +168,12 @@ object MessageCacheStore {
}
}
suspend fun replacePublicMessages(messages: List<Message>) {
suspend fun replacePublicMessages(messages: List<Message>, replaceAll: Boolean = false) {
val convId = conversationIdForPublic()
Logger.d(
"MessageCache",
"replacePublicMessages count=${messages.size} replaceAll=$replaceAll convId=$convId",
)
val resolved = messages.map { it.resolvePublicAttachmentLayout() }
ProfileCache.mergePreviewFromPublicMessages(resolved)
val pending = loadPendingMessages(convId)
@@ -181,10 +186,11 @@ object MessageCacheStore {
withContext(Dispatchers.Default) {
purgeSupersededPendingRows(iid, convId, before, merged)
}
replaceMessages(convId, merged)
replaceMessages(convId, merged, replaceAll = replaceAll)
}
suspend fun clearPublicMessages() {
Logger.d("MessageCache", "clearPublicMessages")
clearConversationMessages(conversationIdForPublic())
}
@@ -192,11 +198,17 @@ object MessageCacheStore {
loadMessages(conversationIdForDm(otherUserId))
suspend fun clearDmMessages(otherUserId: Int) {
Logger.d("MessageCache", "clearDmMessages otherUserId=$otherUserId")
clearConversationMessages(conversationIdForDm(otherUserId))
}
suspend fun replaceDmMessages(otherUserId: Int, messages: List<Message>) {
suspend fun replaceDmMessages(otherUserId: Int, messages: List<Message>, replaceAll: Boolean = false) {
val convId = conversationIdForDm(otherUserId)
Logger.d(
"MessageCache",
"replaceDmMessages otherUserId=$otherUserId count=${messages.size} " +
"replaceAll=$replaceAll convId=$convId",
)
val pending = loadPendingMessages(convId)
val stillPending = filterStillPendingForReplace(convId, pending, messages)
val before = messages + stillPending
@@ -208,7 +220,7 @@ object MessageCacheStore {
withContext(Dispatchers.Default) {
purgeSupersededPendingRows(iid, convId, before, hydrated)
}
replaceMessages(convId, hydrated)
replaceMessages(convId, hydrated, replaceAll = replaceAll)
pruneEmptyConversations()
}
@@ -246,6 +258,11 @@ object MessageCacheStore {
suspend fun upsertPublicMessage(message: Message) {
val resolved = message.resolvePublicAttachmentLayout()
Logger.d(
"MessageCache",
"upsertPublicMessage id=${resolved.id} userId=${resolved.user_id} " +
"clientId=${resolved.client_message_id}",
)
ProfileCache.mergePreviewFromPublicMessage(resolved)
upsertSingle(conversationIdForPublic(), resolved)
}
@@ -279,25 +296,37 @@ object MessageCacheStore {
}
suspend fun upsertDmMessage(otherUserId: Int, message: Message) {
Logger.d(
"MessageCache",
"upsertDmMessage otherUserId=$otherUserId id=${message.id} " +
"userId=${message.user_id} clientId=${message.client_message_id}",
)
ensureDmConversationRow(otherUserId)
upsertSingle(conversationIdForDm(otherUserId), message)
syncDmConversationPreviewFromCache(otherUserId)
}
suspend fun deletePublicMessageByClientMessageId(clientMessageId: String) {
Logger.d("MessageCache", "deletePublicByClientId clientId=$clientMessageId")
deleteByClientMessageId(conversationIdForPublic(), clientMessageId)
}
suspend fun deleteDmMessageByClientMessageId(otherUserId: Int, clientMessageId: String) {
Logger.d(
"MessageCache",
"deleteDmByClientId otherUserId=$otherUserId clientId=$clientMessageId",
)
deleteByClientMessageId(conversationIdForDm(otherUserId), clientMessageId)
}
suspend fun deleteDmMessageById(otherUserId: Int, messageId: Int) {
Logger.d("MessageCache", "deleteDmById otherUserId=$otherUserId messageId=$messageId")
deleteMessageById(conversationIdForDm(otherUserId), messageId)
syncDmConversationPreviewFromCache(otherUserId)
}
suspend fun deletePublicMessageById(messageId: Int) {
Logger.d("MessageCache", "deletePublicById messageId=$messageId")
deleteMessageById(conversationIdForPublic(), messageId)
}
@@ -440,6 +469,10 @@ object MessageCacheStore {
suspend fun markMessageDeleted(conversationId: String, messageId: Int) {
val iid = instanceId()
Logger.d(
"MessageCache",
"markMessageDeleted (soft) convId=$conversationId messageId=$messageId",
)
withContext(Dispatchers.Default) {
db.messageDatabaseQueries.markMessageDeleted(
instanceId = iid,
@@ -611,10 +644,43 @@ object MessageCacheStore {
withContext(Dispatchers.Default) {
val existing = db.messageDatabaseQueries
.selectConversationById(iid, convId)
.executeAsOneOrNull() ?: return@withContext
val label = resolveDmConversationDisplayLabel(otherUserId, null)
if (label.isEmpty()) return@withContext
if (label == existing.displayName) return@withContext
.executeAsOneOrNull() ?: run {
Logger.d(
"MessageCache",
"patchDmPeerProfile noConversation otherUserId=$otherUserId",
)
return@withContext
}
val profile = ProfileCache.get(otherUserId)
val isDeleted = profile?.deleted == true ||
profile?.username?.startsWith("#deleted") == true
val label = if (isDeleted) {
""
} else {
resolveDmConversationDisplayLabel(otherUserId, null)
}
if (!isDeleted && label.isEmpty()) {
Logger.d(
"MessageCache",
"patchDmPeerProfile skipEmptyLabel otherUserId=$otherUserId " +
"deleted=${profile?.deleted}",
)
return@withContext
}
if (label == existing.displayName) {
Logger.d(
"MessageCache",
"patchDmPeerProfile unchanged otherUserId=$otherUserId " +
"deleted=$isDeleted labelEmpty=${label.isEmpty()}",
)
return@withContext
}
Logger.d(
"MessageCache",
"patchDmPeerProfile otherUserId=$otherUserId deleted=$isDeleted " +
"oldLabelEmpty=${existing.displayName.isNullOrBlank()} " +
"newLabelEmpty=${label.isEmpty()}",
)
db.messageDatabaseQueries.upsertConversation(
instanceId = iid,
id = existing.id,
@@ -947,6 +1013,7 @@ object MessageCacheStore {
private suspend fun clearConversationMessages(conversationId: String) {
val iid = instanceId()
Logger.d("MessageCache", "clearConversationMessages convId=$conversationId")
withContext(Dispatchers.Default) {
db.messageDatabaseQueries.deleteMessagesForConversation(iid, conversationId)
}
@@ -954,6 +1021,10 @@ object MessageCacheStore {
private suspend fun deleteByClientMessageId(conversationId: String, clientMessageId: String) {
val iid = instanceId()
Logger.d(
"MessageCache",
"deleteByClientMessageId convId=$conversationId clientId=$clientMessageId",
)
withContext(Dispatchers.Default) {
db.messageDatabaseQueries.deleteMessageByClientMessageId(iid, conversationId, clientMessageId)
}
@@ -961,6 +1032,12 @@ object MessageCacheStore {
private suspend fun deleteMessageById(conversationId: String, messageId: Int) {
val iid = instanceId()
val beforeCount = withContext(Dispatchers.Default) {
db.messageDatabaseQueries
.selectMessagesByConversation(iid, conversationId)
.executeAsList()
.size
}
withContext(Dispatchers.Default) {
db.messageDatabaseQueries.deleteMessageById(
instanceId = iid,
@@ -968,6 +1045,17 @@ object MessageCacheStore {
id = messageId.toLong(),
)
}
val afterCount = withContext(Dispatchers.Default) {
db.messageDatabaseQueries
.selectMessagesByConversation(iid, conversationId)
.executeAsList()
.size
}
Logger.d(
"MessageCache",
"deleteMessageById convId=$conversationId messageId=$messageId " +
"rowsBefore=$beforeCount rowsAfter=$afterCount removed=${beforeCount - afterCount}",
)
}
private suspend fun upsertSingle(conversationId: String, msg: Message) {
@@ -1249,9 +1337,11 @@ object MessageCacheStore {
val profile = ProfileCache.get(uid)
val usernameResolved = when {
self != null && uid == self.id -> self.username
else -> profile?.username?.takeIf { it.isNotBlank() }
?: profile?.displayName?.takeIf { it.isNotBlank() }
?: ""
else -> profile?.username?.takeIf { it.isNotBlank() }.orEmpty()
}
val displayNameResolved = when {
self != null && uid == self.id -> self.displayName?.trim()?.takeIf { it.isNotEmpty() }
else -> profile?.displayName?.trim()?.takeIf { it.isNotEmpty() }
}
val pictureResolved = when {
self != null && uid == self.id -> self.profile_picture
@@ -1266,6 +1356,7 @@ object MessageCacheStore {
is_read = isRead != 0L,
is_edited = isEdited != 0L,
username = usernameResolved,
displayName = displayNameResolved,
profile_picture = pictureResolved,
verified = profile?.verified,
verificationStatus = profile?.verificationStatus,
@@ -1326,6 +1417,7 @@ object MessageCacheStore {
}
suspend fun clearAll() {
Logger.d("MessageCache", "clearAll")
withContext(Dispatchers.Default) {
db.messageDatabaseQueries.purgeAllCache()
}
@@ -1334,24 +1426,49 @@ object MessageCacheStore {
private fun validatedOrEmpty(conversationId: String, messages: List<Message>): List<Message> {
val self = ApiClient.user?.id
if (!CacheValidator.isConversationCacheCoherent(conversationId, messages, self)) {
Logger.w(
"MessageCache",
"validatedOrEmpty incoherent→empty convId=$conversationId count=${messages.size}",
)
return emptyList()
}
return CacheValidator.filterMessages(conversationId, messages, self)
}
private suspend fun replaceMessages(conversationId: String, messages: List<Message>) {
private suspend fun replaceMessages(
conversationId: String,
messages: List<Message>,
replaceAll: Boolean = false,
) {
val self = ApiClient.user?.id
if (!CacheValidator.isConversationCacheCoherent(conversationId, messages, self)) {
Logger.w(
"MessageCache",
"replaceMessages incoherent→clear convId=$conversationId " +
"count=${messages.size} replaceAll=$replaceAll",
)
clearConversationMessages(conversationId)
return
}
val validated = CacheValidator.filterMessages(conversationId, messages, self)
val iid = instanceId()
val beforeCount = withContext(Dispatchers.Default) {
db.messageDatabaseQueries
.selectMessagesByConversation(iid, conversationId)
.executeAsList()
.size
}
Logger.d(
"MessageCache",
"replaceMessages convId=$conversationId replaceAll=$replaceAll " +
"incoming=${messages.size} validated=${validated.size} rowsBefore=$beforeCount",
)
withContext(Dispatchers.Default) {
val existingReplyToIds = db.messageDatabaseQueries
.selectMessagesByConversation(iid, conversationId)
.executeAsList()
.associate { it.id.toInt() to it.replyToId }
if (replaceAll) {
db.messageDatabaseQueries.transaction {
db.messageDatabaseQueries.deleteMessagesForConversation(iid, conversationId)
validated.forEach { msg: Message ->
@@ -1367,11 +1484,28 @@ object MessageCacheStore {
replyToId = resolveReplyToIdForPersistence(msg, existingReplyToIds[msg.id]),
clientMessageId = msg.client_message_id,
deletedFlag = 0L,
sendStatus = if (msg.id < 0) "pending" else "sent"
sendStatus = if (msg.id < 0) "pending" else "sent",
)
}
}
} else {
// Merge into existing rows — partial UI snapshots must not wipe full history.
validated.forEach { msg: Message ->
upsertSingle(conversationId, msg)
}
}
}
val afterCount = withContext(Dispatchers.Default) {
db.messageDatabaseQueries
.selectMessagesByConversation(iid, conversationId)
.executeAsList()
.size
}
Logger.d(
"MessageCache",
"replaceMessages done convId=$conversationId replaceAll=$replaceAll " +
"rowsAfter=$afterCount",
)
dmOtherUserIdFromConversationId(conversationId)?.let {
syncDmConversationPreviewFromCache(it)
pruneEmptyConversations()
@@ -56,8 +56,13 @@ object MessageRepository {
fun observeActiveDmConversations(): Flow<List<CachedConversation>> =
MessageCacheStore.observeActiveDmConversations(activeInstance())
suspend fun replacePublicMessages(messages: List<Message>) =
MessageCacheStore.replacePublicMessages(messages)
suspend fun replacePublicMessages(messages: List<Message>, replaceAll: Boolean = false) {
ru.fromchat.Logger.d(
"MessageRepo",
"replacePublicMessages count=${messages.size} replaceAll=$replaceAll",
)
MessageCacheStore.replacePublicMessages(messages, replaceAll = replaceAll)
}
suspend fun upsertPublicMessage(message: Message) = MessageCacheStore.upsertPublicMessage(message)
@@ -67,20 +72,32 @@ object MessageRepository {
suspend fun deletePublicMessageByClientMessageId(clientMessageId: String) =
MessageCacheStore.deletePublicMessageByClientMessageId(clientMessageId)
suspend fun markMessageDeleted(conversationId: String, messageId: Int) =
suspend fun deletePublicMessageById(messageId: Int) {
ru.fromchat.Logger.d("MessageRepo", "deletePublicMessageById messageId=$messageId")
MessageCacheStore.deletePublicMessageById(messageId)
}
suspend fun markMessageDeleted(conversationId: String, messageId: Int) {
ru.fromchat.Logger.d(
"MessageRepo",
"markMessageDeleted convId=$conversationId messageId=$messageId",
)
MessageCacheStore.markMessageDeleted(conversationId, messageId)
}
suspend fun markPublicMessageDeleted(messageId: Int) =
markMessageDeleted(conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID), messageId)
suspend fun deletePublicMessageById(messageId: Int) =
MessageCacheStore.deletePublicMessageById(messageId)
suspend fun loadDmMessages(otherUserId: Int): List<Message> =
MessageCacheStore.loadDmMessages(otherUserId)
suspend fun replaceDmMessages(otherUserId: Int, messages: List<Message>) =
MessageCacheStore.replaceDmMessages(otherUserId, messages)
suspend fun replaceDmMessages(otherUserId: Int, messages: List<Message>, replaceAll: Boolean = false) {
ru.fromchat.Logger.d(
"MessageRepo",
"replaceDmMessages otherUserId=$otherUserId count=${messages.size} replaceAll=$replaceAll",
)
MessageCacheStore.replaceDmMessages(otherUserId, messages, replaceAll = replaceAll)
}
suspend fun upsertDmMessage(otherUserId: Int, message: Message) =
MessageCacheStore.upsertDmMessage(otherUserId, message)
@@ -91,8 +108,13 @@ object MessageRepository {
suspend fun deleteDmMessageByClientMessageId(otherUserId: Int, clientMessageId: String) =
MessageCacheStore.deleteDmMessageByClientMessageId(otherUserId, clientMessageId)
suspend fun deleteDmMessageById(otherUserId: Int, messageId: Int) =
suspend fun deleteDmMessageById(otherUserId: Int, messageId: Int) {
ru.fromchat.Logger.d(
"MessageRepo",
"deleteDmMessageById otherUserId=$otherUserId messageId=$messageId",
)
MessageCacheStore.deleteDmMessageById(otherUserId, messageId)
}
suspend fun replaceDmConversations(
conversations: List<DmConversation>,
@@ -12,6 +12,7 @@ import kotlinx.coroutines.flow.map
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import ru.fromchat.Logger
import ru.fromchat.api.ApiClient
import ru.fromchat.api.schema.messages.Message
import ru.fromchat.api.schema.messages.dm.DmConversationUser
@@ -71,19 +72,41 @@ object ProfileCache {
/** Skip force=false network refetch when a full profile was fetched within this window. */
const val FULL_PROFILE_TTL_MS: Long = 5 * 60 * 1000L
private fun bumpRevision() {
_revision.value++
private fun bumpRevision(reason: String) {
val next = _revision.value + 1
_revision.value = next
Logger.d("ProfileCache", "revision=$next reason=$reason size=${profiles.size}")
}
private fun profileSummary(profile: UserProfile): String =
"id=${profile.id} user='${profile.username}' deleted=${profile.deleted} " +
"suspended=${profile.suspended} preview=${profile.isClientPreviewOnly} " +
"verified=${profile.verified} vStatus=${profile.verificationStatus}"
fun get(userId: Int): UserProfile? = profiles[userId]
/** True when a full non-preview profile was fetched recently enough to skip refetch. */
@OptIn(ExperimentalTime::class)
fun hasFreshFullProfile(userId: Int, maxAgeMs: Long = FULL_PROFILE_TTL_MS): Boolean {
val profile = get(userId) ?: return false
if (profile.isClientPreviewOnly) return false
val fetchedAt = fullProfileFetchedAtMs[userId] ?: return false
return (Clock.System.now().toEpochMilliseconds() - fetchedAt) <= maxAgeMs
val profile = get(userId) ?: run {
Logger.d("ProfileCache", "hasFreshFullProfile id=$userId miss")
return false
}
if (profile.isClientPreviewOnly) {
Logger.d("ProfileCache", "hasFreshFullProfile id=$userId stale=previewOnly")
return false
}
val fetchedAt = fullProfileFetchedAtMs[userId] ?: run {
Logger.d("ProfileCache", "hasFreshFullProfile id=$userId stale=noFetchTs")
return false
}
val ageMs = Clock.System.now().toEpochMilliseconds() - fetchedAt
val fresh = ageMs <= maxAgeMs
Logger.d(
"ProfileCache",
"hasFreshFullProfile id=$userId fresh=$fresh ageMs=$ageMs maxAgeMs=$maxAgeMs",
)
return fresh
}
/** Emits whenever this user's cached profile changes (including bio). */
@@ -110,7 +133,13 @@ object ProfileCache {
) {
if (id <= 0) return
val existing = get(id)
if (existing != null && !existing.isClientPreviewOnly) return
if (existing != null && !existing.isClientPreviewOnly) {
Logger.d(
"ProfileCache",
"mergePreview skipFullExists id=$id deleted=${existing.deleted}",
)
return
}
val incomingUsername = username?.trim()?.takeIf { it.isNotEmpty() }
?: existing?.username?.trim()?.takeIf { it.isNotEmpty() }
@@ -120,11 +149,27 @@ object ProfileCache {
} else {
displayName?.trim()?.takeIf { it.isNotEmpty() }
?: existing?.displayName?.takeIf { it.isNotBlank() }
?: incomingUsername
}
if (!isDeleted && incomingUsername.isNullOrEmpty() && incomingDisplayName.isNullOrBlank()) return
if (!isDeleted && incomingUsername.isNullOrEmpty() && incomingDisplayName.isNullOrBlank()) {
Logger.d("ProfileCache", "mergePreview skipEmptyIdentity id=$id")
return
}
if (incomingUsername.isNullOrEmpty() || incomingDisplayName.isNullOrBlank()) {
Logger.d(
"ProfileCache",
"mergePreview missingIdentity id=$id " +
"hasUsername=${!incomingUsername.isNullOrEmpty()} " +
"hasDisplayName=${!incomingDisplayName.isNullOrBlank()}",
)
}
Logger.d(
"ProfileCache",
"mergePreview id=$id deleted=$isDeleted hadExisting=${existing != null} " +
"user='${incomingUsername.orEmpty()}' display='${incomingDisplayName.orEmpty()}'",
)
put(
UserProfile(
id = id,
@@ -159,29 +204,49 @@ object ProfileCache {
val hasIdentity =
profile.username.trim().isNotEmpty() || !profile.displayName.isNullOrBlank()
if (!hasIdentity) {
Logger.d("ProfileCache", "put removeEmptyPreview id=${profile.id}")
remove(profile.id)
return
}
}
val cur = profiles
val existing = cur[profile.id]
val deletedChanged = existing?.deleted != profile.deleted
val suspendedChanged = existing?.suspended != profile.suspended
val verificationChanged = existing?.verified != profile.verified ||
existing?.verificationStatus != profile.verificationStatus
if (
existing != null &&
!existing.isClientPreviewOnly &&
existing.bio != profile.bio
) {
ru.fromchat.Logger.d(
Logger.d(
"ProfileCache",
"put overwrite id=${profile.id} bio '${existing.bio?.take(48)}' -> " +
"'${profile.bio?.take(48)}' preview=${profile.isClientPreviewOnly}",
)
}
if (deletedChanged || suspendedChanged || verificationChanged || existing == null) {
Logger.d(
"ProfileCache",
"put ${profileSummary(profile)} hadExisting=${existing != null} " +
"deletedChanged=$deletedChanged suspendedChanged=$suspendedChanged " +
"verificationChanged=$verificationChanged",
)
}
profiles = cur + (profile.id to profile)
bumpRevision()
bumpRevision("put:${profile.id}")
val instanceId = loadedInstanceId
if (instanceId.isNotEmpty()) {
ioScope.launch {
runCatching { ProfileCacheStore.put(instanceId, profile) }
.onFailure {
Logger.w(
"ProfileCache",
"persist put failed id=${profile.id}: ${it.message}",
it,
)
}
}
}
}
@@ -199,27 +264,56 @@ object ProfileCache {
if (!force) {
val existing = get(profile.id)
if (existing != null && !existing.isClientPreviewOnly) {
val lifecycleMismatch =
existing.deleted != normalized.deleted ||
existing.suspended != normalized.suspended ||
isDeletedPlaceholderUsername(existing.username) !=
isDeletedPlaceholderUsername(normalized.username)
if (lifecycleMismatch) {
Logger.w(
"ProfileCache",
"applyServerProfile force=false lifecycleMismatch " +
"cachedDeleted=${existing.deleted} incomingDeleted=${normalized.deleted} " +
"cachedSuspended=${existing.suspended} " +
"incomingSuspended=${normalized.suspended} " +
"cachedUser='${existing.username}' incomingUser='${normalized.username}' " +
"— applying full server profile",
)
put(normalized)
fullProfileFetchedAtMs = fullProfileFetchedAtMs + (profile.id to nowMs)
return
}
val patched = existing.copy(
verified = normalized.verified ?: existing.verified,
verificationStatus = normalized.verificationStatus
?: existing.verificationStatus,
)
val verificationChanged = patched.verified != existing.verified ||
patched.verificationStatus != existing.verificationStatus
if (patched != existing) put(patched)
if (existing.bio != normalized.bio) {
ru.fromchat.Logger.d(
Logger.d(
"ProfileCache",
"applyServerProfile skipped stale HTTP id=${profile.id} " +
"cachedBio='${existing.bio?.take(48)}' httpBio='${normalized.bio?.take(48)}'",
"cachedBio='${existing.bio?.take(48)}' httpBio='${normalized.bio?.take(48)}' " +
"deleted=${existing.deleted}",
)
} else {
Logger.d(
"ProfileCache",
"applyServerProfile keepCached force=false id=${profile.id} " +
"deleted=${existing.deleted} verificationChanged=$verificationChanged",
)
}
// Refresh TTL so force=false callers stop refetching.
if (!verificationChanged) {
fullProfileFetchedAtMs = fullProfileFetchedAtMs + (profile.id to nowMs)
}
return
}
}
ru.fromchat.Logger.d(
Logger.d(
"ProfileCache",
"applyServerProfile applied force=$force id=${profile.id} " +
"applyServerProfile applied force=$force ${profileSummary(normalized)} " +
"bio='${normalized.bio?.take(48)}'",
)
put(normalized)
@@ -228,9 +322,13 @@ object ProfileCache {
fun remove(userId: Int) {
val cur = profiles
if (userId !in cur) return
if (userId !in cur) {
Logger.d("ProfileCache", "remove miss id=$userId")
return
}
Logger.d("ProfileCache", "remove id=$userId wasDeleted=${cur[userId]?.deleted}")
profiles = cur - userId
bumpRevision()
bumpRevision("remove:$userId")
val instanceId = loadedInstanceId
if (instanceId.isNotEmpty()) {
ioScope.launch {
@@ -251,20 +349,40 @@ object ProfileCache {
if (user.id <= 0) return
val incomingUsername = user.username.trim()
if (incomingUsername.isEmpty()) return
if (incomingUsername.isEmpty()) {
Logger.d("ProfileCache", "mergeFromDmUser skipEmptyUsername id=${user.id}")
return
}
val isDeleted = user.deleted == true || isDeletedPlaceholderUsername(incomingUsername)
val incomingDisplayName = if (isDeleted) {
null
} else {
user.displayName?.trim()?.takeIf { it.isNotEmpty() } ?: incomingUsername
user.displayName?.trim()?.takeIf { it.isNotEmpty() }
}
if (!isDeleted && incomingDisplayName.isNullOrBlank()) {
Logger.d(
"ProfileCache",
"mergeFromDmUser missingDisplayName id=${user.id} user='$incomingUsername'",
)
}
val existing = get(user.id)
Logger.d(
"ProfileCache",
"mergeFromDmUser id=${user.id} deleted=$isDeleted " +
"incomingDeleted=${user.deleted} hadFull=${existing != null && existing.isClientPreviewOnly != true} " +
"user='$incomingUsername' display='${incomingDisplayName.orEmpty()}'",
)
if (existing != null && !existing.isClientPreviewOnly) {
val patched = existing.copy(
username = incomingUsername,
displayName = if (isDeleted) null else incomingDisplayName ?: existing.displayName,
displayName = if (isDeleted) {
null
} else {
incomingDisplayName ?: existing.displayName
},
profilePicture = if (isDeleted) {
null
} else {
@@ -278,7 +396,14 @@ object ProfileCache {
suspensionReason = user.suspensionReason ?: existing.suspensionReason,
deleted = isDeleted,
)
if (patched != existing) put(patched)
if (patched != existing) {
Logger.d(
"ProfileCache",
"mergeFromDmUser patchFull id=${user.id} " +
"deleted ${existing.deleted}${patched.deleted}",
)
put(patched)
}
return
}
@@ -286,8 +411,11 @@ object ProfileCache {
UserProfile(
id = user.id,
username = incomingUsername,
displayName = if (isDeleted) null else existing?.displayName?.takeIf { it.isNotBlank() }
?: incomingDisplayName,
displayName = if (isDeleted) {
null
} else {
incomingDisplayName ?: existing?.displayName?.takeIf { it.isNotBlank() }
},
profilePicture = if (isDeleted) null else user.profile_picture?.takeIf { it.isNotBlank() }
?: existing?.profilePicture,
bio = existing?.bio,
@@ -336,9 +464,18 @@ object ProfileCache {
}
val uname = message.username.trim().ifBlank { existing?.username?.trim().orEmpty() }
if (uname.isBlank()) return
val incomingDisplay = message.displayName?.trim()?.takeIf { it.isNotEmpty() }
?: existing?.displayName?.takeIf { it.isNotBlank() }
if (uname.isBlank() && incomingDisplay.isNullOrBlank()) return
if (uname.isBlank() || incomingDisplay.isNullOrBlank()) {
Logger.d(
"ProfileCache",
"mergePreviewFromPublicMessage missingIdentity id=$uid " +
"hasUsername=${uname.isNotBlank()} hasDisplayName=${!incomingDisplay.isNullOrBlank()}",
)
}
val isDeleted = isDeletedPlaceholderUsername(uname) || existing?.deleted == true
val display = if (isDeleted) null else existing?.displayName?.takeIf { it.isNotBlank() } ?: uname
val display = if (isDeleted) null else incomingDisplay
val pic = if (isDeleted) null else message.profile_picture?.takeIf { it.isNotBlank() }
?: existing?.profilePicture
@@ -379,6 +516,8 @@ object ProfileCache {
val user = ApiClient.user
return message.copy(
username = message.username.trim().ifBlank { user?.username.orEmpty() },
displayName = message.displayName?.trim()?.takeIf { it.isNotEmpty() }
?: user?.displayName?.trim()?.takeIf { it.isNotEmpty() },
profile_picture = message.profile_picture?.takeIf { it.isNotBlank() }
?: user?.profile_picture,
reply_to = enrichedReply,
@@ -389,6 +528,8 @@ object ProfileCache {
username = message.username.trim().ifBlank {
profile?.visibleUsername(self).orEmpty()
},
displayName = message.displayName?.trim()?.takeIf { it.isNotEmpty() }
?: profile?.displayName?.trim()?.takeIf { it.isNotEmpty() },
profile_picture = message.profile_picture?.takeIf { it.isNotBlank() }
?: profile?.profilePicture,
verified = message.verified ?: profile?.verified,
@@ -413,7 +554,7 @@ object ProfileCache {
}
fullProfileFetchedAtMs = emptyMap()
pruneUnusableClientPreviewsLocked()
bumpRevision()
bumpRevision("onActiveInstanceChanged:$instanceId")
}
}
}
@@ -427,6 +568,7 @@ object ProfileCache {
} else {
emptyMap()
}
val before = profiles.size
if (profiles.isEmpty()) {
profiles = diskProfiles
} else {
@@ -442,8 +584,14 @@ object ProfileCache {
}
profiles = merged
}
val deletedCount = profiles.values.count { it.deleted == true }
Logger.d(
"ProfileCache",
"hydrateFromDisk instanceId=$instanceId before=$before " +
"disk=${diskProfiles.size} after=${profiles.size} deletedCount=$deletedCount",
)
pruneUnusableClientPreviewsLocked()
bumpRevision()
bumpRevision("hydrateFromDisk")
}
}
@@ -455,6 +603,7 @@ object ProfileCache {
p.displayName.isNullOrBlank()
}.keys
if (toRemove.isEmpty()) return
Logger.d("ProfileCache", "pruneUnusablePreviews ids=$toRemove")
var cur = profiles
for (id in toRemove) {
cur = cur - id
@@ -464,10 +613,11 @@ object ProfileCache {
suspend fun clear() {
persistMutex.withLock {
Logger.d("ProfileCache", "clear sizeWas=${profiles.size}")
profiles = emptyMap()
fullProfileFetchedAtMs = emptyMap()
loadedInstanceId = ""
bumpRevision()
bumpRevision("clear")
}
}
}
@@ -8,7 +8,6 @@ import ru.fromchat.api.crypto.decryptEnvelope
import ru.fromchat.api.local.db.parseDmMessageContent
import ru.fromchat.api.local.db.store.MessageRepository
import ru.fromchat.api.local.db.store.ProfileCache
import ru.fromchat.api.local.db.store.visibleDisplayName
import ru.fromchat.api.local.messages.ActiveDmChatTracker
import ru.fromchat.api.schema.messages.Message
import ru.fromchat.api.schema.messages.dm.DmEnvelope
@@ -65,16 +64,31 @@ object DmInboundMessageProcessor {
suspend fun processDeleted(element: JsonElement) {
val data = runCatching {
ApiClient.json.decodeFromJsonElement(DmDeletedData.serializer(), element)
}.getOrNull() ?: return
}.getOrNull() ?: run {
ru.fromchat.Logger.w("DmInbox", "processDeleted decode failed")
return
}
val currentUserId = ApiClient.user?.id ?: return
if (data.senderId != currentUserId && data.recipientId != currentUserId) return
if (data.senderId != currentUserId && data.recipientId != currentUserId) {
ru.fromchat.Logger.d(
"DmInbox",
"processDeleted skipNotParticipant messageId=${data.id} " +
"senderId=${data.senderId} recipientId=${data.recipientId} self=$currentUserId",
)
return
}
val otherUserId = when (currentUserId) {
data.senderId -> data.recipientId
else -> data.senderId
} ?: return
ru.fromchat.Logger.i(
"DmInbox",
"processDeleted messageId=${data.id} otherUserId=$otherUserId " +
"senderId=${data.senderId} recipientId=${data.recipientId}",
)
withContext(Dispatchers.Default) {
MessageRepository.deleteDmMessageById(otherUserId, data.id)
}
@@ -132,19 +146,31 @@ object DmInboundMessageProcessor {
otherUserId: Int,
): Message {
val dec = parseDmMessageContent(plaintext)
val senderUsername = envelope.senderUsername?.trim()?.takeIf { it.isNotEmpty() }
val senderDisplayName = envelope.senderDisplayName?.trim()?.takeIf { it.isNotEmpty() }
if (envelope.senderId != currentUserId) {
envelope.senderUsername?.trim()?.takeIf { it.isNotEmpty() }?.let { senderName ->
ProfileCache.mergePreview(id = envelope.senderId, username = senderName)
if (senderUsername != null || senderDisplayName != null) {
ProfileCache.mergePreview(
id = envelope.senderId,
username = senderUsername,
displayName = senderDisplayName,
)
}
}
val senderProfile = ProfileCache.get(envelope.senderId)
val username = if (envelope.senderId == currentUserId) {
"You"
ApiClient.user?.username.orEmpty()
} else {
senderProfile?.visibleDisplayName(currentUserId)?.takeIf { it.isNotBlank() }
?: envelope.senderUsername?.takeIf { it.isNotBlank() }
senderUsername
?: senderProfile?.username?.trim()?.takeIf { it.isNotEmpty() }
?: ""
}
val displayName = if (envelope.senderId == currentUserId) {
ApiClient.user?.displayName?.trim()?.takeIf { it.isNotEmpty() }
} else {
senderDisplayName
?: senderProfile?.displayName?.trim()?.takeIf { it.isNotEmpty() }
}
return Message(
id = envelope.id,
user_id = envelope.senderId,
@@ -153,6 +179,7 @@ object DmInboundMessageProcessor {
is_read = envelope.senderId == currentUserId,
is_edited = false,
username = username,
displayName = displayName,
profile_picture = null,
verified = null,
reply_to = null,
@@ -36,6 +36,7 @@ object DmInboxCoordinator {
}
"dmDeleted" -> message.data?.let { element ->
scope.launch {
ru.fromchat.Logger.d("DmInbox", "handleMessage dmDeleted")
DmInboundMessageProcessor.processDeleted(element)
DmConversationListNotifier.notifyChanged()
}
@@ -3,6 +3,7 @@ package ru.fromchat.api.local.messages
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import kotlinx.serialization.json.JsonElement
import ru.fromchat.Logger
import ru.fromchat.api.ApiClient
import ru.fromchat.api.local.db.store.MessageRepository
import ru.fromchat.api.local.db.store.ProfileCache
@@ -15,7 +16,15 @@ import ru.fromchat.api.schema.websocket.types.MessageDeletedData
*/
object PublicInboxCoordinator {
suspend fun processNew(element: JsonElement) = withContext(Dispatchers.Default) {
val message = decodeMessage(element) ?: return@withContext
val message = decodeMessage(element) ?: run {
Logger.w("PublicInbox", "processNew decode failed")
return@withContext
}
Logger.d(
"PublicInbox",
"processNew id=${message.id} userId=${message.user_id} " +
"clientId=${message.client_message_id}",
)
ProfileCache.mergePreviewFromPublicMessage(message)
val clientId = message.client_message_id?.trim().orEmpty()
val currentUserId = ApiClient.user?.id
@@ -49,7 +58,11 @@ object PublicInboxCoordinator {
}
suspend fun processEdited(element: JsonElement) = withContext(Dispatchers.Default) {
val edited = decodeMessage(element) ?: return@withContext
val edited = decodeMessage(element) ?: run {
Logger.w("PublicInbox", "processEdited decode failed")
return@withContext
}
Logger.d("PublicInbox", "processEdited id=${edited.id} userId=${edited.user_id}")
ProfileCache.mergePreviewFromPublicMessage(edited)
val existing = MessageRepository.loadPublicMessages()
val merged = existing.map { current ->
@@ -62,14 +75,23 @@ object PublicInboxCoordinator {
if (merged.none { it.id == edited.id }) {
MessageRepository.upsertPublicMessage(edited)
} else {
MessageRepository.replacePublicMessages(merged)
MessageRepository.upsertPublicMessage(
merged.first { it.id == edited.id },
)
}
}
suspend fun processDeleted(element: JsonElement) = withContext(Dispatchers.Default) {
val deleted = runCatching {
ApiClient.json.decodeFromJsonElement(MessageDeletedData.serializer(), element)
}.getOrNull() ?: return@withContext
}.getOrNull() ?: run {
Logger.w("PublicInbox", "processDeleted decode failed element=${element.toString().take(120)}")
return@withContext
}
Logger.i(
"PublicInbox",
"processDeleted messageId=${deleted.message_id} — hard-deleting from cache",
)
MessageRepository.deletePublicMessageById(deleted.message_id)
}
@@ -18,8 +18,16 @@ object UpdatesBatchApplier {
suspend fun applyEnvelope(data: kotlinx.serialization.json.JsonElement): Int? = mutex.withLock {
val envelope = runCatching {
ApiClient.json.decodeFromJsonElement(WebSocketUpdatesData.serializer(), data)
}.getOrNull() ?: return@withLock null
}.getOrNull() ?: run {
Logger.w("UpdateSync", "applyEnvelope decode failed: ${data.toString().take(160)}")
return@withLock null
}
val types = envelope.updates.map { it.type }
Logger.d(
"UpdateSync",
"applyEnvelope seq=${envelope.seq} count=${envelope.updates.size} types=$types",
)
for (update in envelope.updates) {
applyOne(WebSocketMessage(type = update.type, data = update.data))
}
@@ -34,8 +42,14 @@ object UpdatesBatchApplier {
}
"newMessage" -> message.data?.let { PublicInboxCoordinator.processNew(it) }
"messageEdited" -> message.data?.let { PublicInboxCoordinator.processEdited(it) }
"messageDeleted" -> message.data?.let { PublicInboxCoordinator.processDeleted(it) }
"dmNew", "dmDeleted", "dmEdited" -> DmInboxCoordinator.handleMessage(message)
"messageDeleted" -> {
Logger.d("UpdateSync", "applyOne messageDeleted")
message.data?.let { PublicInboxCoordinator.processDeleted(it) }
}
"dmNew", "dmDeleted", "dmEdited" -> {
Logger.d("UpdateSync", "applyOne ${message.type}")
DmInboxCoordinator.handleMessage(message)
}
else -> Unit
}
}
@@ -17,6 +17,7 @@ data class Message(
val is_read: Boolean,
val is_edited: Boolean,
val username: String,
@SerialName("display_name") val displayName: String? = null,
val profile_picture: String? = null,
val verified: Boolean? = null,
@SerialName("verification_status") val verificationStatus: VerificationStatus? = null,
@@ -1,9 +1,11 @@
package ru.fromchat.api.schema.messages
import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable
@Serializable
data class ReactionUser(
val id: Int,
val username: String
val username: String,
@SerialName("display_name") val displayName: String? = null,
)
@@ -9,6 +9,7 @@ data class DmEnvelope(
val senderId: Int,
val recipientId: Int,
@SerialName("sender_username") val senderUsername: String? = null,
@SerialName("sender_display_name") val senderDisplayName: String? = null,
@SerialName("iv_b64") val ivB64: String,
@SerialName("ciphertext_b64") val ciphertextB64: String,
@SerialName("wrapped_mek_b64") val wrappedMekB64: String? = null,
@@ -1,5 +1,6 @@
package ru.fromchat.api.schema.websocket.types
import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable
@Serializable
@@ -9,5 +10,6 @@ data class ReactionUpdateData(
val action: String,
val user_id: Int,
val username: String,
@SerialName("display_name") val displayName: String? = null,
val reactions: List<ReactionData>
)
@@ -94,11 +94,26 @@ abstract class ChatPanel(
addMessageMutex.withLock {
batchStateUpdates {
updateState { current ->
val panelSnap = panelMessagesForDbMerge()
val merged = mergeDatabaseMessagesWithPanelState(
panelMessagesForDbMerge(),
panelSnap,
messages,
)
val withReplies = attachPublicReplyReferences(merged)
if (current.messages.size != withReplies.size ||
current.messages.map { it.id }.toSet() != withReplies.map { it.id }.toSet()
) {
val panelIds = panelSnap.map { it.id }.toSet()
val mergedIds = withReplies.map { it.id }.toSet()
val dbIds = messages.map { it.id }.toSet()
Logger.d(
"ChatPanel",
"syncMessagesFromDatabase panel=${panelSnap.size} db=${messages.size} " +
"merged=${withReplies.size} " +
"panelOnlyIds=${(panelIds - mergedIds).take(8)} " +
"dbOnlyIds=${(dbIds - mergedIds).take(8)}",
)
}
if (current.messages == withReplies) current
else current.copy(messages = withReplies)
}
@@ -692,7 +707,8 @@ abstract class ChatPanel(
timestamp = nowMessageTimestampIso(),
is_read = false,
is_edited = false,
username = "You",
username = ApiClient.user?.username.orEmpty(),
displayName = ApiClient.user?.displayName,
client_message_id = tempId,
reply_to = resolvedReply,
replyToId = resolvedReply?.id ?: replyToId?.takeIf { it > 0 },
@@ -949,7 +949,8 @@ fun ChatScreen(
timestamp = nowMessageTimestampIso(),
is_read = false,
is_edited = false,
username = "You",
username = ApiClient.user?.username.orEmpty(),
displayName = ApiClient.user?.displayName,
profile_picture = null,
verified = null,
reply_to = replyTo,
@@ -33,6 +33,7 @@ import androidx.compose.material3.TopAppBar
import androidx.compose.material3.TopAppBarDefaults
import androidx.compose.runtime.Composable
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.collectAsState
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
@@ -105,7 +106,7 @@ fun ChatTopBarInner(
peerIsDeleted(
userId = userId,
currentUserId = ApiClient.user?.id,
username = titleAvatar?.displayName ?: title,
username = ProfileCache.get(userId)?.username,
)
} == true
Row(
@@ -203,7 +204,10 @@ fun ChatTopBarInner(
overflow = TextOverflow.Ellipsis,
)
profileUserId?.let { userId ->
val status = resolveVerificationStatus(userId)
val profileCacheRevision by ProfileCache.revision.collectAsState()
val status = remember(userId, profileCacheRevision) {
resolveVerificationStatus(userId)
}
if (status != null) {
Spacer(modifier = Modifier.width(4.dp))
StatusBadge(
@@ -18,12 +18,13 @@ import androidx.compose.foundation.layout.WindowInsets
import androidx.compose.foundation.layout.aspectRatio
import androidx.compose.foundation.layout.fillMaxSize
import androidx.compose.foundation.layout.fillMaxWidth
import androidx.compose.foundation.layout.navigationBarsPadding
import androidx.compose.foundation.layout.offset
import androidx.compose.foundation.layout.padding
import androidx.compose.foundation.layout.safeDrawing
import androidx.compose.foundation.layout.size
import androidx.compose.foundation.layout.statusBarsPadding
import androidx.compose.foundation.layout.width
import androidx.compose.foundation.layout.windowInsetsPadding
import androidx.compose.material.icons.Icons
import androidx.compose.material.icons.automirrored.rounded.ArrowBack
import androidx.compose.material.icons.automirrored.rounded.Reply
@@ -710,25 +711,26 @@ fun ImageFullscreenPreview(
}
// Top bar: back, display name + date/time, 3-dot menu
Box(
modifier = Modifier
.align(Alignment.TopStart)
.fillMaxWidth()
.windowInsetsPadding(WindowInsets.safeDrawing),
) {
AnimatedVisibility(
visible = effectiveMenusVisible,
enter = androidx.compose.animation.fadeIn(),
exit = androidx.compose.animation.fadeOut(),
modifier = Modifier.fillMaxWidth(),
modifier = Modifier
.align(Alignment.TopStart)
.fillMaxWidth(),
) {
Column(
modifier = Modifier
.fillMaxWidth()
.background(Color.Black.copy(alpha = MENU_BG_ALPHA))
.statusBarsPadding(),
) {
Row(
modifier = Modifier
.fillMaxWidth()
.background(Color.Black.copy(alpha = MENU_BG_ALPHA))
.padding(horizontal = 8.dp, vertical = 12.dp),
verticalAlignment = Alignment.CenterVertically,
horizontalArrangement = Arrangement.SpaceBetween
horizontalArrangement = Arrangement.SpaceBetween,
) {
IconButton(onClick = { dismissRequested = true }) {
Icon(
@@ -820,33 +822,32 @@ fun ImageFullscreenPreview(
}
// Bottom: message text
Box(
modifier = Modifier
.align(Alignment.BottomStart)
.fillMaxWidth()
.windowInsetsPadding(WindowInsets.safeDrawing),
) {
AnimatedVisibility(
visible = effectiveMenusVisible && message.content.isNotBlank(),
enter = androidx.compose.animation.fadeIn(),
exit = androidx.compose.animation.fadeOut(),
modifier = Modifier.fillMaxWidth(),
modifier = Modifier
.align(Alignment.BottomStart)
.fillMaxWidth(),
) {
if (message.content.isNotBlank()) {
Box(
Column(
modifier = Modifier
.fillMaxWidth()
.background(Color.Black.copy(alpha = MENU_BG_ALPHA))
.navigationBarsPadding(),
) {
Box(
modifier = Modifier
.fillMaxWidth()
.padding(16.dp),
) {
Text(
text = message.content,
style = MaterialTheme.typography.bodyMedium,
color = Color.White
color = Color.White,
)
}
}
}
}
}
}
@@ -15,7 +15,8 @@ import ru.fromchat.ui.profile.isRedactedPeerAccount
import ru.fromchat.ui.profile.peerIsDeleted
/**
* Resolves [Message.username] for display: localized «Вы», deleted user label, or server-provided name.
* Resolves the sender label shown in message bubbles: localized «Вы», deleted user label,
* cached/server display name, or login username only as a last resort.
*/
@Composable
fun messageDisplayUsername(message: Message, currentUserId: Int?): String {
@@ -30,12 +31,14 @@ fun messageDisplayUsername(message: Message, currentUserId: Int?): String {
if (isDeletedAccountUsername(message.username)) {
return deletedUserDisplayNameForUi()
}
val cachedUsername = ProfileCache.get(message.user_id)?.visibleDisplayName(currentUserId)
if (cachedUsername != null) return cachedUsername
ProfileCache.get(message.user_id)?.visibleDisplayName(currentUserId)
?.takeIf { it.isNotBlank() }
?.let { return it }
message.displayName?.trim()?.takeIf { it.isNotEmpty() }?.let { return it }
if (message.username.equals("deleted", ignoreCase = true)) {
return deletedUserDisplayNameForUi()
}
return message.username
return message.username.trim()
}
fun messageSenderProfilePicture(
@@ -75,5 +78,5 @@ fun messageSenderAvatarLabel(
if (currentUserId != null && message.user_id == currentUserId) {
return ApiClient.user?.displayName?.trim()?.takeIf { it.isNotBlank() }.orEmpty()
}
return message.username.trim()
return message.displayName?.trim()?.takeIf { it.isNotEmpty() }.orEmpty()
}
@@ -35,6 +35,7 @@ import androidx.compose.material3.Icon
import androidx.compose.material3.MaterialTheme
import androidx.compose.runtime.Composable
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.collectAsState
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
@@ -229,11 +230,14 @@ fun MessageItem(
val sendFailedLabel = stringResource(Res.string.message_send_failed)
val replyPhotoLabel = stringResource(Res.string.message_reply_photo)
val displayUsername = messageDisplayUsername(message, currentUserId)
val profileCacheRevision by ProfileCache.revision.collectAsState()
val senderProfile = ProfileCache.get(message.user_id)
val avatarPictureUrl = senderProfile?.profilePicture?.takeIf { it.isNotBlank() }
?: message.profile_picture
val avatarDisplayName = messageSenderAvatarLabel(message, currentUserId)
val senderVerificationStatus = resolveVerificationStatus(message.user_id, message)
val senderVerificationStatus = remember(message.user_id, message, profileCacheRevision) {
resolveVerificationStatus(message.user_id, message)
}
val isDeletedSender = messageSenderIsDeleted(message, currentUserId)
val replyRef = message.reply_to
@@ -159,6 +159,13 @@ class DmPanel(
scope.launch(Dispatchers.Default) {
val cached = ProfileCache.get(otherUserId)
val displayName = cached?.displayNameText(ApiClient.user?.id).orEmpty()
Logger.d(
"DmPanel",
"applyCachedPeerProfile otherUserId=$otherUserId " +
"deleted=${cached?.deleted} suspended=${cached?.suspended} " +
"revision=${ProfileCache.revision.value} " +
"titleBlank=${displayName.isBlank()}",
)
cached?.let { UserStatusStore.update(it.id, it.online, it.lastSeen) }
withContext(Dispatchers.Main) {
if (displayName.isNotBlank()) {
@@ -236,17 +243,17 @@ class DmPanel(
// Read cache first. Do not setLoading(true) before this: that forced a 1-frame spinner
// when the chat screen re-entered composition (e.g. pop back from profile).
val cached = runCatching { MessageCacheStore.loadDmMessages(otherUserId) }.getOrDefault(emptyList())
if (cached.isNotEmpty()) {
val hadCachedMessages = cached.isNotEmpty()
if (hadCachedMessages) {
batchStateUpdates {
clearMessages()
addMessages(cached)
setLoading(false)
}
messagesLoaded = true
return
} else {
setLoading(true)
}
setLoading(true)
try {
OutgoingMessageCoordinator.pruneStaleAttachmentOutboxForInstance(
CacheContext.requireActiveInstanceId(),
@@ -289,7 +296,7 @@ class DmPanel(
// Persist the most recent DM messages for offline use.
val mergedForCache = _state.messages
MessageCacheStore.replaceDmMessages(otherUserId, mergedForCache)
MessageCacheStore.replaceDmMessages(otherUserId, mergedForCache, replaceAll = true)
messagesLoaded = true
} else {
val error = historyResult.exceptionOrNull()
@@ -299,6 +306,8 @@ class DmPanel(
clearMessages()
setHasMoreMessages(false)
messagesLoaded = true
} else if (hadCachedMessages) {
messagesLoaded = true
}
}
} finally {
@@ -510,8 +519,6 @@ class DmPanel(
MessageCacheStore.confirmDmMessage(otherUserId, cid, mergedForPersistence)
OutgoingMessageCoordinator.clearAttachmentOutboxAfterAck(cid)
}
val snapshot = _state.messages
MessageCacheStore.replaceDmMessages(otherUserId, snapshot)
}
}
@@ -533,31 +540,46 @@ class DmPanel(
}
val outcome = decryptDmEnvelopeForUi(envelope)
val dec = parseDmMessageContent(outcome.plaintext)
updateMessage(envelope.id) {
it.copy(
val editedForCache = (previous ?: createMessage(envelope, outcome.plaintext, outcome.isCorrupted)).copy(
content = dec.text,
is_edited = true,
fileThumbnails = dec.fileThumbnails ?: it.fileThumbnails,
fileAspectRatios = dec.fileAspectRatios ?: it.fileAspectRatios,
fileSizes = dec.fileSizes ?: it.fileSizes,
fileDimensions = dec.fileDimensions ?: it.fileDimensions,
fileThumbnails = dec.fileThumbnails ?: previous?.fileThumbnails,
fileAspectRatios = dec.fileAspectRatios ?: previous?.fileAspectRatios,
fileSizes = dec.fileSizes ?: previous?.fileSizes,
fileDimensions = dec.fileDimensions ?: previous?.fileDimensions,
isContentCorrupted = outcome.isCorrupted,
dmEnvelope = envelope,
reply_to = it.reply_to,
reply_to = previous?.reply_to,
)
updateMessage(envelope.id) {
editedForCache.copy(reply_to = it.reply_to)
}
// Persist edit to cache
MessageCacheStore.replaceDmMessages(otherUserId, _state.messages)
MessageCacheStore.upsertDmMessage(otherUserId, editedForCache)
}
}
private fun createMessage(envelope: DmEnvelope, plaintext: String, isContentCorrupted: Boolean): Message {
val dec = parseDmMessageContent(plaintext)
val username = if (envelope.senderId == currentUserId) {
"You"
val senderUsername = if (envelope.senderId == currentUserId) {
ApiClient.user?.username.orEmpty()
} else {
otherDisplayName
envelope.senderUsername?.trim()?.takeIf { it.isNotEmpty() }
?: ProfileCache.get(envelope.senderId)?.username?.trim().orEmpty()
}
val senderDisplayName = if (envelope.senderId == currentUserId) {
ApiClient.user?.displayName?.trim()?.takeIf { it.isNotEmpty() }
} else {
envelope.senderDisplayName?.trim()?.takeIf { it.isNotEmpty() }
?: otherDisplayName.takeIf { it.isNotBlank() }
?: ProfileCache.get(envelope.senderId)?.displayName?.trim()?.takeIf { it.isNotEmpty() }
}
if (envelope.senderId != currentUserId) {
ProfileCache.mergePreview(
id = envelope.senderId,
username = senderUsername.takeIf { it.isNotEmpty() },
displayName = senderDisplayName,
)
}
return Message(
id = envelope.id,
@@ -569,7 +591,8 @@ class DmPanel(
else -> ActiveDmChatTracker.isActive(otherUserId)
},
is_edited = false,
username = username,
username = senderUsername,
displayName = senderDisplayName,
profile_picture = null,
verified = null,
reply_to = null,
@@ -621,12 +644,27 @@ class DmPanel(
private fun processDeletedEnvelope(element: JsonElement) {
val data = runCatching {
json.decodeFromJsonElement(DmDeletedData.serializer(), element)
}.getOrNull() ?: return
}.getOrNull() ?: run {
Logger.w("DmPanel", "processDeletedEnvelope decode failed")
return
}
val involvesPeer =
data.senderId == otherUserId ||
data.recipientId == otherUserId ||
data.senderId == currentUserId
if (!involvesPeer) return
if (!involvesPeer) {
Logger.d(
"DmPanel",
"processDeletedEnvelope skip messageId=${data.id} " +
"senderId=${data.senderId} recipientId=${data.recipientId} peer=$otherUserId",
)
return
}
Logger.i(
"DmPanel",
"processDeletedEnvelope messageId=${data.id} peer=$otherUserId " +
"uiBefore=${_state.messages.size} inUi=${_state.messages.any { it.id == data.id }}",
)
scope.launch(Dispatchers.Default) {
val clientId = _state.messages.find { it.id == data.id }?.client_message_id
DownloadedFileRegistry.invalidateForMessage(data.id)
@@ -639,6 +677,10 @@ class DmPanel(
}
deleteMessageImmediately(data.id)
MessageRepository.deleteDmMessageById(otherUserId, data.id)
Logger.d(
"DmPanel",
"processDeletedEnvelope done messageId=${data.id} uiAfter=${_state.messages.size}",
)
}
}
@@ -4,6 +4,8 @@ import io.ktor.client.plugins.ClientRequestException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import kotlinx.serialization.json.jsonObject
import kotlinx.serialization.json.jsonPrimitive
@@ -50,7 +52,7 @@ class PublicChatPanel(
scope = scope
) {
private val typingHandler = PublicChatTypingHandler(scope)
private var networkHistoryLoaded = false
private val loadMessagesMutex = Mutex()
/**
* Whether replacing the list would change **structure or message body** (content / edited).
@@ -377,9 +379,8 @@ class PublicChatPanel(
}
override suspend fun loadMessages() {
loadMessagesMutex.withLock {
hydrateMessagesFromLocalCache()
if (networkHistoryLoaded) return
networkHistoryLoaded = true
val cached = _state.messages
if (cached.isEmpty()) {
@@ -388,7 +389,13 @@ class PublicChatPanel(
}
}
// Refresh from network; this may be fast or slow, but runs entirely off main.
// Always refresh from network on open. A retained panel used to skip this after the
// first visit (networkHistoryLoaded), so offline bursts left first+last holes until
// slow WS catch-up filled them.
Logger.d(
"PublicChatPanel",
"loadMessages: network refresh cachedCount=${cached.size}",
)
val responseResult = withContext(Dispatchers.Default) {
runCatching { ApiClient.getMessages(limit = 50) }
}
@@ -400,12 +407,18 @@ class PublicChatPanel(
val optimisticSnapshot = snapshotPendingOptimisticMessages()
val pendingStr = debugPendingKeys().takeIf { it.isNotBlank() } ?: "(none)"
val optIds = optimisticSnapshot.mapNotNull { it.client_message_id }.ifEmpty { listOf<String>() }
val loadMsg = "loadMessages: pendingKeys=$pendingStr optimisticSnapshot=$optIds stateCount=${_state.messages.size}"
Logger.d("PublicChatPanel", loadMsg)
Logger.d(
"PublicChatPanel",
"loadMessages: pendingKeys=$pendingStr optimisticSnapshot=$optIds " +
"stateCount=${_state.messages.size} networkCount=${networkMessages.size}",
)
var mergedForCache: List<Message>? = null
withContext(Dispatchers.Main) {
val shown = snapshotUiMessagesForNetworkMerge()
Logger.d("PublicChatPanel", "loadMessages: snapshotUiMessagesForNetworkMerge size=${shown.size}")
Logger.d(
"PublicChatPanel",
"loadMessages: snapshotUiMessagesForNetworkMerge size=${shown.size}",
)
if (shown.isNotEmpty() && !publicHistoryDiffersForUi(shown, networkMessages)) {
Logger.d("PublicChatPanel", "Network history matches UI; skip clear/re-add")
val withSenders = mergePublicSenderFieldsFromNetwork(shown, networkMessages)
@@ -425,7 +438,11 @@ class PublicChatPanel(
addMessages(
ProfileCache.enrichPublicMessagesForDisplay(merged),
)
Logger.d("PublicChatPanel", "loadMessages: after addMessages mergedSize=${merged.size} restoring optimistic count=${optimisticSnapshot.size}")
Logger.d(
"PublicChatPanel",
"loadMessages: after addMessages mergedSize=${merged.size} " +
"restoring optimistic count=${optimisticSnapshot.size}",
)
restorePendingOptimisticMessages(optimisticSnapshot)
setHasMoreMessages(false) // TODO: Implement has_more from API
setLoading(false)
@@ -439,8 +456,11 @@ class PublicChatPanel(
withContext(Dispatchers.Default) {
val toPersist = mergedForCache
?: mergeNetworkHistoryWithShown(panelMessagesForDbMerge(), networkMessages)
Logger.d("PublicChatPanel", "loadMessages: persisting to cache messages=${toPersist.size}")
MessageCacheStore.replacePublicMessages(toPersist)
Logger.d(
"PublicChatPanel",
"loadMessages: persisting to cache messages=${toPersist.size} replaceAll=true",
)
MessageCacheStore.replacePublicMessages(toPersist, replaceAll = true)
}
} else if (responseResult.isFailure) {
val cause = responseResult.exceptionOrNull()
@@ -452,23 +472,26 @@ class PublicChatPanel(
if (_state.hasMoreMessages) setHasMoreMessages(false)
}
} else if (cached.isEmpty()) {
// Nothing to show at all; hide spinner so the user is not stuck.
withContext(Dispatchers.Main) {
if (_state.isLoading) setLoading(false)
if (_state.hasMoreMessages) setHasMoreMessages(false)
}
} else {
// We already displayed cached messages; just mark pagination state.
withContext(Dispatchers.Main) {
if (_state.hasMoreMessages) setHasMoreMessages(false)
if (_state.isLoading) setLoading(false)
}
}
} else if (cached.isEmpty()) {
// Nothing to show at all; hide spinner so the user is not stuck.
withContext(Dispatchers.Main) {
if (_state.isLoading) setLoading(false)
if (_state.hasMoreMessages) setHasMoreMessages(false)
}
} else {
withContext(Dispatchers.Main) {
if (_state.isLoading) setLoading(false)
}
}
}
}
@@ -494,7 +517,7 @@ class PublicChatPanel(
)
}
withContext(Dispatchers.Default) {
MessageCacheStore.replacePublicMessages(_state.messages)
MessageCacheStore.replacePublicMessages(_state.messages, replaceAll = true)
}
}
setHasMoreMessages(false) // TODO: Implement has_more from API
@@ -529,29 +552,47 @@ class PublicChatPanel(
val data = updateMessage.data ?: return
val editedMsg = json.decodeFromJsonElement(Message.serializer(), data)
DecryptedImageCache.invalidateForMessage(editedMsg.id)
updateMessage(editedMsg.id) { existing ->
editedMsg.copy(reply_to = editedMsg.reply_to ?: existing.reply_to)
val existing = _state.messages.find { it.id == editedMsg.id }
val persisted = editedMsg.copy(reply_to = editedMsg.reply_to ?: existing?.reply_to)
updateMessage(editedMsg.id) { current ->
persisted.copy(reply_to = persisted.reply_to ?: current.reply_to)
}
withContext(Dispatchers.Default) {
MessageCacheStore.replacePublicMessages(_state.messages)
MessageCacheStore.upsertPublicMessage(persisted.resolvePublicAttachmentLayout())
}
}
"messageDeleted" -> {
val data = updateMessage.data ?: return
val deletedData = json.decodeFromJsonElement(MessageDeletedData.serializer(), data)
Logger.i(
"PublicChatPanel",
"messageDeleted messageId=${deletedData.message_id} " +
"uiBefore=${_state.messages.size} inUi=${_state.messages.any { it.id == deletedData.message_id }}",
)
DecryptedImageCache.invalidateForMessage(deletedData.message_id)
removeMessage(deletedData.message_id)
clearReplyReferencesTo(deletedData.message_id)
withContext(Dispatchers.Default) {
MessageRepository.deletePublicMessageById(deletedData.message_id)
}
Logger.d(
"PublicChatPanel",
"messageDeleted done messageId=${deletedData.message_id} " +
"uiAfter=${_state.messages.size}",
)
}
"reactionUpdate" -> {
val data = updateMessage.data ?: return
val reactionUpdate = json.decodeFromJsonElement(ReactionUpdateData.serializer(), data)
val existing = _state.messages.find { it.id == reactionUpdate.message_id }
if (existing != null) {
val updated = existing.copy(reactions = reactionUpdate.reactions)
handleReactionUpdate(reactionUpdate)
withContext(Dispatchers.Default) {
MessageCacheStore.replacePublicMessages(_state.messages)
MessageCacheStore.upsertPublicMessage(updated.resolvePublicAttachmentLayout())
}
} else {
handleReactionUpdate(reactionUpdate)
}
}
"typing" -> {
@@ -620,6 +661,7 @@ class PublicChatPanel(
cancelQueuedMessage(message)
return
}
Logger.d("PublicChatPanel", "handleDeleteMessage messageId=$messageId")
beginMessageDissolve(message)
withContext(Dispatchers.Default) {
MessageRepository.deletePublicMessageById(messageId)
@@ -72,13 +72,28 @@ internal fun mergeDatabaseMessagesWithPanelState(
val mergedClientIds = mergedDb.mapNotNull { it.client_message_id?.trim()?.takeIf { id -> id.isNotEmpty() } }.toSet()
val mergedIds = mergedDb.map { it.id }.toSet()
// Keep in-flight panel optimistics even when the DB Flow emission already stripped them.
// Confirmed (id > 0) rows missing from DB are deletes — do not resurrect them from panel state.
val droppedConfirmed = panelMessages.filter { panel ->
panel.id > 0 && panel.id !in mergedIds
}
if (droppedConfirmed.isNotEmpty()) {
ru.fromchat.Logger.d(
"MessageCache",
"mergeDbPanel dropConfirmedDeletes count=${droppedConfirmed.size} " +
"ids=${droppedConfirmed.map { it.id }.take(12)} " +
"panelSize=${panelMessages.size} dbSize=${dbMessages.size}",
)
}
val extraPanel = panelMessages.filter { panel ->
val cid = panel.client_message_id?.trim()?.takeIf { it.isNotEmpty() }
when {
panel.id < 0 && cid != null && cid !in mergedClientIds -> true
panel.id > 0 && panel.id !in mergedIds && (cid.isNullOrEmpty() || cid !in mergedClientIds) -> true
else -> false
panel.id < 0 && cid != null && cid !in mergedClientIds
}
if (extraPanel.isNotEmpty()) {
ru.fromchat.Logger.d(
"MessageCache",
"mergeDbPanel keepOptimistic count=${extraPanel.size} " +
"ids=${extraPanel.map { it.id }}",
)
}
return dedupeMessagesByClientId(
@@ -45,6 +45,7 @@ import androidx.compose.material3.Surface
import androidx.compose.material3.ripple
import androidx.compose.runtime.Composable
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.collectAsState
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
@@ -130,9 +131,13 @@ internal fun ChatListHeadlineWithBadge(
title: String,
userId: Int,
) {
val profileCacheRevision by ProfileCache.revision.collectAsState()
val verificationStatus = remember(userId, profileCacheRevision) {
resolveVerificationStatus(userId)
}
DisplayName(
displayName = title,
verificationStatus = resolveVerificationStatus(userId),
verificationStatus = verificationStatus,
textStyle = MaterialTheme.typography.bodyLarge,
)
}
@@ -928,7 +933,7 @@ internal fun DmConversationRowContent(
currentUserId = currentUserId,
deleted = cached?.deleted,
suspended = cached?.suspended,
username = cached?.username ?: conversation.displayName.takeIf { it.isNotBlank() },
username = cached?.username,
)
val avatarUrl = if (isPeerDeleted) null else cached?.profilePicture
val peerTitle = when {
@@ -10,9 +10,12 @@ import androidx.compose.foundation.layout.Box
import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.PaddingValues
import androidx.compose.foundation.layout.Spacer
import androidx.compose.foundation.layout.WindowInsets
import androidx.compose.foundation.layout.consumeWindowInsets
import androidx.compose.foundation.layout.fillMaxSize
import androidx.compose.foundation.layout.fillMaxWidth
import androidx.compose.foundation.layout.height
import androidx.compose.foundation.layout.navigationBars
import androidx.compose.foundation.layout.navigationBarsPadding
import androidx.compose.foundation.layout.padding
import androidx.compose.foundation.layout.size
@@ -422,6 +425,8 @@ fun DevicesScreen(onBack: () -> Unit) {
}
Scaffold(
modifier = Modifier.fillMaxSize(),
contentWindowInsets = WindowInsets.navigationBars,
snackbarHost = { FromChatSnackbarHost(hostState = snackbarHostState) },
topBar = {
TopAppBar(
@@ -479,10 +484,15 @@ fun DevicesScreen(onBack: () -> Unit) {
LazyColumn(
modifier = Modifier
.hazeSource(hazeState)
.padding()
.padding(start = 16.dp, end = 16.dp, top = 8.dp, bottom = 24.dp),
contentPadding = innerPadding
.fillMaxSize()
.consumeWindowInsets(innerPadding)
.hazeSource(hazeState),
contentPadding = PaddingValues(
start = 16.dp,
end = 16.dp,
top = innerPadding.calculateTopPadding() + 8.dp,
bottom = innerPadding.calculateBottomPadding() + 24.dp,
),
) {
item {
Column(Modifier.fillMaxWidth()) {
@@ -569,9 +579,8 @@ fun DevicesScreen(onBack: () -> Unit) {
}
}
item {
// Initial load only — background poll must not add/remove list height (overscroll jump).
if (refreshing && devices.isEmpty()) {
item {
Box(
modifier = Modifier
.fillMaxWidth()