mirror of
https://github.com/fromchat-messenger/app.git
synced 2026-09-22 19:15:05 +03:00
@@ -854,10 +854,9 @@ object ApiClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Fetch encrypted file bytes. Path examples:
|
* Resolve an attachment path to a full URL.
|
||||||
* - absolute URL like "http://..." -> returned as-is
|
* Absolute `http(s)://…` URLs are returned as-is; relative paths are joined to [ServerConfig.apiBaseUrl].
|
||||||
* - "/uploads/files/encrypted/xxx.jpg" -> returned as <apiBaseUrl>/uploads/...
|
* Leading `/api` is stripped so paths match the backend (legacy payloads still used `/api/uploads/...`).
|
||||||
* - "/api/uploads/..." -> strip legacy `/api` prefix, then combine with apiBaseUrl
|
|
||||||
*/
|
*/
|
||||||
fun encryptedFileUrl(path: String): String {
|
fun encryptedFileUrl(path: String): String {
|
||||||
if (path.startsWith("http")) return path
|
if (path.startsWith("http")) return path
|
||||||
@@ -1590,6 +1589,9 @@ object ApiClient {
|
|||||||
|
|
||||||
suspend fun deleteMessage(messageId: Int) {
|
suspend fun deleteMessage(messageId: Int) {
|
||||||
if (_suspensionState.value.isSuspended) return
|
if (_suspensionState.value.isSuspended) return
|
||||||
|
runCatching {
|
||||||
|
http.delete("${ServerConfig.apiBaseUrl}/delete_message/$messageId")
|
||||||
|
}
|
||||||
WebSocketManager.send(
|
WebSocketManager.send(
|
||||||
WebSocketMessage(
|
WebSocketMessage(
|
||||||
type = "deleteMessage",
|
type = "deleteMessage",
|
||||||
@@ -1729,7 +1731,7 @@ object ApiClient {
|
|||||||
.body()
|
.body()
|
||||||
}
|
}
|
||||||
|
|
||||||
suspend fun sendLiveKitInvite(toUserId: Int, roomName: String, serverUrl: String) {
|
suspend fun sendLiveKitInvite(toUserId: Int, roomName: String) {
|
||||||
if (_suspensionState.value.isSuspended) return
|
if (_suspensionState.value.isSuspended) return
|
||||||
WebSocketManager.send(
|
WebSocketManager.send(
|
||||||
WebSocketMessage(
|
WebSocketMessage(
|
||||||
@@ -1742,7 +1744,6 @@ object ApiClient {
|
|||||||
CallSignalingLiveKitPayload(
|
CallSignalingLiveKitPayload(
|
||||||
toUserId = toUserId,
|
toUserId = toUserId,
|
||||||
roomName = roomName,
|
roomName = roomName,
|
||||||
serverUrl = serverUrl,
|
|
||||||
),
|
),
|
||||||
),
|
),
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -33,7 +33,6 @@ sealed class CallUiState {
|
|||||||
val fromUserId: Int,
|
val fromUserId: Int,
|
||||||
val fromUsername: String,
|
val fromUsername: String,
|
||||||
val roomName: String,
|
val roomName: String,
|
||||||
val serverUrl: String,
|
|
||||||
) : CallUiState()
|
) : CallUiState()
|
||||||
|
|
||||||
data class InCall(
|
data class InCall(
|
||||||
@@ -101,9 +100,8 @@ object CallStore {
|
|||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
val serverUrl = obj["serverUrl"]?.jsonPrimitive?.contentOrNull?.takeIf { it.isNotBlank() }
|
|
||||||
val roomName = obj["roomName"]?.jsonPrimitive?.contentOrNull?.takeIf { it.isNotBlank() }
|
val roomName = obj["roomName"]?.jsonPrimitive?.contentOrNull?.takeIf { it.isNotBlank() }
|
||||||
if (serverUrl == null || roomName == null || fromUserId == null) return
|
if (roomName == null || fromUserId == null) return
|
||||||
if (fromUserId == currentId) return
|
if (fromUserId == currentId) return
|
||||||
if (!ServerConfig.callsEnabled) {
|
if (!ServerConfig.callsEnabled) {
|
||||||
Logger.d(TAG, "call_signaling invite ignored (calls disabled in server config)")
|
Logger.d(TAG, "call_signaling invite ignored (calls disabled in server config)")
|
||||||
@@ -116,7 +114,6 @@ object CallStore {
|
|||||||
fromUserId = fromUserId,
|
fromUserId = fromUserId,
|
||||||
fromUsername = fromUsername,
|
fromUsername = fromUsername,
|
||||||
roomName = roomName,
|
roomName = roomName,
|
||||||
serverUrl = serverUrl,
|
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -132,13 +129,10 @@ object CallStore {
|
|||||||
runCatching {
|
runCatching {
|
||||||
withContext(Dispatchers.Default) {
|
withContext(Dispatchers.Default) {
|
||||||
val tok = ApiClient.fetchLiveKitToken(peerUserId, null)
|
val tok = ApiClient.fetchLiveKitToken(peerUserId, null)
|
||||||
// LiveKit WS endpoint is exposed on the same host as the server config,
|
ApiClient.sendLiveKitInvite(peerUserId, tok.roomName)
|
||||||
// using the configured calls port.
|
|
||||||
val signalUrl = ServerConfig.liveKitWsUrl()
|
|
||||||
ApiClient.sendLiveKitInvite(peerUserId, tok.roomName, signalUrl)
|
|
||||||
val label = peerLabel(peerUserId)
|
val label = peerLabel(peerUserId)
|
||||||
LiveKitConnectSession(
|
LiveKitConnectSession(
|
||||||
serverUrl = signalUrl,
|
serverUrl = ServerConfig.liveKitWsUrl(),
|
||||||
token = tok.token,
|
token = tok.token,
|
||||||
peerUserId = peerUserId,
|
peerUserId = peerUserId,
|
||||||
peerDisplayName = label,
|
peerDisplayName = label,
|
||||||
|
|||||||
+17
-1
@@ -60,7 +60,9 @@ object DmAttachmentOutboxHandler {
|
|||||||
.executeAsOneOrNull() != null
|
.executeAsOneOrNull() != null
|
||||||
|
|
||||||
private fun ensureStillQueued(instanceId: String, clientMessageId: String) {
|
private fun ensureStillQueued(instanceId: String, clientMessageId: String) {
|
||||||
if (!outboxRowExists(instanceId, clientMessageId)) {
|
if (OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId) ||
|
||||||
|
!outboxRowExists(instanceId, clientMessageId)
|
||||||
|
) {
|
||||||
AttachmentMediaLog.upload("cancelled_in_flight", "job" to clientMessageId)
|
AttachmentMediaLog.upload("cancelled_in_flight", "job" to clientMessageId)
|
||||||
throw kotlinx.coroutines.CancellationException("Upload cancelled")
|
throw kotlinx.coroutines.CancellationException("Upload cancelled")
|
||||||
}
|
}
|
||||||
@@ -72,6 +74,8 @@ object DmAttachmentOutboxHandler {
|
|||||||
val payload = json.decodeFromString<DmAttachmentOutboxPayload>(row.payloadJson)
|
val payload = json.decodeFromString<DmAttachmentOutboxPayload>(row.payloadJson)
|
||||||
val clientMessageId = payload.clientMessageId.trim()
|
val clientMessageId = payload.clientMessageId.trim()
|
||||||
if (clientMessageId.isEmpty() || payload.recipientId <= 0) return true
|
if (clientMessageId.isEmpty() || payload.recipientId <= 0) return true
|
||||||
|
if (OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId)) return true
|
||||||
|
if (OutgoingMessageCoordinator.isOutboundPaused(clientMessageId)) return true
|
||||||
if (!outboxRowExists(instanceId, clientMessageId)) return true
|
if (!outboxRowExists(instanceId, clientMessageId)) return true
|
||||||
|
|
||||||
val conversationId = row.conversationId.ifBlank {
|
val conversationId = row.conversationId.ifBlank {
|
||||||
@@ -168,6 +172,13 @@ object DmAttachmentOutboxHandler {
|
|||||||
serverUploadId = serverUploadId,
|
serverUploadId = serverUploadId,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
if (
|
||||||
|
OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId) ||
|
||||||
|
!outboxRowExists(instanceId, clientMessageId)
|
||||||
|
) {
|
||||||
|
OutgoingMessageCoordinator.abortDmServerUploadIfNeeded(serverUploadId[0])
|
||||||
|
return@runCatching true
|
||||||
|
}
|
||||||
ensureStillQueued(instanceId, clientMessageId)
|
ensureStillQueued(instanceId, clientMessageId)
|
||||||
|
|
||||||
clearPrepared(instanceId, clientMessageId)
|
clearPrepared(instanceId, clientMessageId)
|
||||||
@@ -183,6 +194,7 @@ object DmAttachmentOutboxHandler {
|
|||||||
bytesUploaded = row.bytesUploaded,
|
bytesUploaded = row.bytesUploaded,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
OutgoingMessageCoordinator.clearOutboundCancelled(clientMessageId)
|
||||||
AttachmentUploadNotifier.emit(
|
AttachmentUploadNotifier.emit(
|
||||||
AttachmentUploadProgress.Success(clientMessageId),
|
AttachmentUploadProgress.Success(clientMessageId),
|
||||||
messageLabel = payload.plaintext,
|
messageLabel = payload.plaintext,
|
||||||
@@ -221,6 +233,10 @@ object DmAttachmentOutboxHandler {
|
|||||||
),
|
),
|
||||||
messageLabel = payload.plaintext,
|
messageLabel = payload.plaintext,
|
||||||
)
|
)
|
||||||
|
if (failureKey == UPLOAD_ERROR_FILE_TOO_LARGE || error.isOutboundPermanentFailure()) {
|
||||||
|
OutgoingMessageCoordinator.markOutboundPaused(clientMessageId)
|
||||||
|
return true
|
||||||
|
}
|
||||||
false
|
false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+62
-3
@@ -2,12 +2,16 @@ package ru.fromchat.api.local.send
|
|||||||
|
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
|
import kotlinx.coroutines.Job
|
||||||
import kotlinx.coroutines.SupervisorJob
|
import kotlinx.coroutines.SupervisorJob
|
||||||
import kotlinx.coroutines.delay
|
import kotlinx.coroutines.delay
|
||||||
|
import kotlinx.coroutines.flow.MutableStateFlow
|
||||||
|
import kotlinx.coroutines.flow.update
|
||||||
import kotlinx.coroutines.launch
|
import kotlinx.coroutines.launch
|
||||||
import kotlinx.coroutines.sync.Mutex
|
import kotlinx.coroutines.sync.Mutex
|
||||||
import kotlinx.coroutines.sync.withLock
|
import kotlinx.coroutines.sync.withLock
|
||||||
import kotlinx.coroutines.withContext
|
import kotlinx.coroutines.withContext
|
||||||
|
import kotlin.coroutines.coroutineContext
|
||||||
import kotlinx.serialization.json.Json
|
import kotlinx.serialization.json.Json
|
||||||
import ru.fromchat.api.ApiClient
|
import ru.fromchat.api.ApiClient
|
||||||
import ru.fromchat.api.local.workers.AttachmentUploadNotifier
|
import ru.fromchat.api.local.workers.AttachmentUploadNotifier
|
||||||
@@ -40,6 +44,35 @@ object OutgoingMessageCoordinator {
|
|||||||
private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true }
|
private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true }
|
||||||
private val drainMutex = Mutex()
|
private val drainMutex = Mutex()
|
||||||
private val drainScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
private val drainScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||||
|
private val retryJobs = mutableMapOf<String, Job>()
|
||||||
|
/** Client message ids cancelled by the user; in-flight sends must not confirm locally. */
|
||||||
|
private val cancelledClientIds = MutableStateFlow<Set<String>>(emptySet())
|
||||||
|
/** Permanent failures that must not auto-retry until [retryOutboundMessage] / [retryDmAttachmentUpload]. */
|
||||||
|
private val pausedClientIds = MutableStateFlow<Set<String>>(emptySet())
|
||||||
|
|
||||||
|
fun isOutboundCancelled(clientMessageId: String): Boolean =
|
||||||
|
clientMessageId.trim() in cancelledClientIds.value
|
||||||
|
|
||||||
|
fun isOutboundPaused(clientMessageId: String): Boolean =
|
||||||
|
clientMessageId.trim() in pausedClientIds.value
|
||||||
|
|
||||||
|
fun markOutboundPaused(clientMessageId: String) {
|
||||||
|
val cid = clientMessageId.trim()
|
||||||
|
if (cid.isEmpty()) return
|
||||||
|
pausedClientIds.update { it + cid }
|
||||||
|
}
|
||||||
|
|
||||||
|
fun clearOutboundPaused(clientMessageId: String) {
|
||||||
|
val cid = clientMessageId.trim()
|
||||||
|
if (cid.isEmpty()) return
|
||||||
|
pausedClientIds.update { it - cid }
|
||||||
|
}
|
||||||
|
|
||||||
|
fun clearOutboundCancelled(clientMessageId: String) {
|
||||||
|
val cid = clientMessageId.trim()
|
||||||
|
if (cid.isEmpty()) return
|
||||||
|
cancelledClientIds.update { it - cid }
|
||||||
|
}
|
||||||
|
|
||||||
/** Called when network or WebSocket transport is ready; drains pending outbox rows. */
|
/** Called when network or WebSocket transport is ready; drains pending outbox rows. */
|
||||||
fun onTransportReady() {
|
fun onTransportReady() {
|
||||||
@@ -47,9 +80,18 @@ object OutgoingMessageCoordinator {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private fun scheduleOutboxRetry(instanceId: String) {
|
private fun scheduleOutboxRetry(instanceId: String) {
|
||||||
drainScope.launch {
|
val id = instanceId.trim()
|
||||||
|
if (id.isEmpty()) return
|
||||||
|
retryJobs[id]?.cancel()
|
||||||
|
retryJobs[id] = drainScope.launch {
|
||||||
delay(3_000)
|
delay(3_000)
|
||||||
drainOutboxForInstance(instanceId)
|
try {
|
||||||
|
drainOutboxForInstance(id)
|
||||||
|
} finally {
|
||||||
|
if (retryJobs[id] == coroutineContext[Job]) {
|
||||||
|
retryJobs.remove(id)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -157,6 +199,7 @@ object OutgoingMessageCoordinator {
|
|||||||
) {
|
) {
|
||||||
val instanceId = CacheContext.requireActiveInstanceId()
|
val instanceId = CacheContext.requireActiveInstanceId()
|
||||||
val conversationId = conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID)
|
val conversationId = conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID)
|
||||||
|
markOutboundActive(clientMessageId)
|
||||||
withContext(Dispatchers.Default) {
|
withContext(Dispatchers.Default) {
|
||||||
MessageRepository.upsertPublicMessage(optimisticMessage)
|
MessageRepository.upsertPublicMessage(optimisticMessage)
|
||||||
Logger.d("OutgoingMessageCoordinator", "enqueuePublicMessage: clientId=${clientMessageId.take(12)} contentLen=${content.length}")
|
Logger.d("OutgoingMessageCoordinator", "enqueuePublicMessage: clientId=${clientMessageId.take(12)} contentLen=${content.length}")
|
||||||
@@ -186,6 +229,7 @@ object OutgoingMessageCoordinator {
|
|||||||
) {
|
) {
|
||||||
val instanceId = CacheContext.requireActiveInstanceId()
|
val instanceId = CacheContext.requireActiveInstanceId()
|
||||||
val conversationId = conversationIdForDm(recipientId)
|
val conversationId = conversationIdForDm(recipientId)
|
||||||
|
markOutboundActive(clientMessageId)
|
||||||
withContext(Dispatchers.Default) {
|
withContext(Dispatchers.Default) {
|
||||||
MessageRepository.upsertDmMessage(recipientId, optimisticMessage)
|
MessageRepository.upsertDmMessage(recipientId, optimisticMessage)
|
||||||
val outboundPlaintext = buildDmOutboundPlaintext(plaintext, replyToId)
|
val outboundPlaintext = buildDmOutboundPlaintext(plaintext, replyToId)
|
||||||
@@ -226,6 +270,7 @@ object OutgoingMessageCoordinator {
|
|||||||
) {
|
) {
|
||||||
val instanceId = CacheContext.requireActiveInstanceId()
|
val instanceId = CacheContext.requireActiveInstanceId()
|
||||||
val conversationId = conversationIdForDm(recipientId)
|
val conversationId = conversationIdForDm(recipientId)
|
||||||
|
markOutboundActive(clientMessageId)
|
||||||
AttachmentUploadNotifier.emit(
|
AttachmentUploadNotifier.emit(
|
||||||
AttachmentUploadProgress.Pending(clientMessageId, filename),
|
AttachmentUploadProgress.Pending(clientMessageId, filename),
|
||||||
messageLabel = plaintext,
|
messageLabel = plaintext,
|
||||||
@@ -275,6 +320,7 @@ object OutgoingMessageCoordinator {
|
|||||||
) {
|
) {
|
||||||
val instanceId = CacheContext.requireActiveInstanceId()
|
val instanceId = CacheContext.requireActiveInstanceId()
|
||||||
val conversationId = conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID)
|
val conversationId = conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID)
|
||||||
|
markOutboundActive(clientMessageId)
|
||||||
AttachmentUploadNotifier.emit(
|
AttachmentUploadNotifier.emit(
|
||||||
AttachmentUploadProgress.Pending(clientMessageId, filename),
|
AttachmentUploadProgress.Pending(clientMessageId, filename),
|
||||||
messageLabel = content,
|
messageLabel = content,
|
||||||
@@ -358,6 +404,8 @@ object OutgoingMessageCoordinator {
|
|||||||
if (cid.isEmpty()) return
|
if (cid.isEmpty()) return
|
||||||
val instanceId = CacheContext.activeInstanceId.value.trim()
|
val instanceId = CacheContext.activeInstanceId.value.trim()
|
||||||
if (instanceId.isEmpty()) return
|
if (instanceId.isEmpty()) return
|
||||||
|
clearOutboundPaused(cid)
|
||||||
|
clearOutboundCancelled(cid)
|
||||||
drainScope.launch {
|
drainScope.launch {
|
||||||
withContext(Dispatchers.Default) {
|
withContext(Dispatchers.Default) {
|
||||||
MessageCacheStore.clearSendFailed(conversationId, cid)
|
MessageCacheStore.clearSendFailed(conversationId, cid)
|
||||||
@@ -373,6 +421,8 @@ object OutgoingMessageCoordinator {
|
|||||||
if (cid.isEmpty()) return
|
if (cid.isEmpty()) return
|
||||||
val instanceId = CacheContext.activeInstanceId.value.trim()
|
val instanceId = CacheContext.activeInstanceId.value.trim()
|
||||||
if (instanceId.isEmpty()) return
|
if (instanceId.isEmpty()) return
|
||||||
|
clearOutboundPaused(cid)
|
||||||
|
clearOutboundCancelled(cid)
|
||||||
AttachmentMediaLog.upload("retry_requested", "job" to cid)
|
AttachmentMediaLog.upload("retry_requested", "job" to cid)
|
||||||
kickOutboxDrain(instanceId)
|
kickOutboxDrain(instanceId)
|
||||||
}
|
}
|
||||||
@@ -382,6 +432,8 @@ object OutgoingMessageCoordinator {
|
|||||||
val cid = clientMessageId.trim()
|
val cid = clientMessageId.trim()
|
||||||
if (cid.isEmpty()) return
|
if (cid.isEmpty()) return
|
||||||
val instanceId = CacheContext.requireActiveInstanceId()
|
val instanceId = CacheContext.requireActiveInstanceId()
|
||||||
|
cancelledClientIds.update { it + cid }
|
||||||
|
clearOutboundPaused(cid)
|
||||||
AttachmentMediaLog.upload("cancel_requested", "job" to cid, "conv" to conversationId)
|
AttachmentMediaLog.upload("cancel_requested", "job" to cid, "conv" to conversationId)
|
||||||
withContext(Dispatchers.Default) {
|
withContext(Dispatchers.Default) {
|
||||||
val row = MessageDatabaseProvider.database.messageDatabaseQueries
|
val row = MessageDatabaseProvider.database.messageDatabaseQueries
|
||||||
@@ -404,7 +456,14 @@ object OutgoingMessageCoordinator {
|
|||||||
AttachmentUploadNotifier.emit(
|
AttachmentUploadNotifier.emit(
|
||||||
AttachmentUploadProgress.Failed(cid, "Cancelled"),
|
AttachmentUploadProgress.Failed(cid, "Cancelled"),
|
||||||
)
|
)
|
||||||
kickOutboxDrain(instanceId)
|
// Do not kick drain — a concurrent in-flight send must observe cancel via outbox/flags only.
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun markOutboundActive(clientMessageId: String) {
|
||||||
|
val cid = clientMessageId.trim()
|
||||||
|
if (cid.isEmpty()) return
|
||||||
|
clearOutboundCancelled(cid)
|
||||||
|
clearOutboundPaused(cid)
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Drains pending outbox rows for the active instance (shared by workers and iOS). */
|
/** Drains pending outbox rows for the active instance (shared by workers and iOS). */
|
||||||
|
|||||||
+91
-4
@@ -21,7 +21,10 @@ import ru.fromchat.api.local.db.store.MessageDatabaseProvider
|
|||||||
import ru.fromchat.api.local.messages.optimisticMessageIdForClientMessageId
|
import ru.fromchat.api.local.messages.optimisticMessageIdForClientMessageId
|
||||||
import ru.fromchat.api.local.workers.AttachmentUploadNotifier
|
import ru.fromchat.api.local.workers.AttachmentUploadNotifier
|
||||||
import ru.fromchat.api.local.workers.AttachmentUploadProgress
|
import ru.fromchat.api.local.workers.AttachmentUploadProgress
|
||||||
import ru.fromchat.api.local.db.isPlaceholderAttachmentAspectRatio
|
import ru.fromchat.api.local.send.isOutboundPermanentFailure
|
||||||
|
import ru.fromchat.api.local.send.isOutboundTransientFailure
|
||||||
|
import ru.fromchat.api.local.send.outboundFailureErrorKey
|
||||||
|
import ru.fromchat.api.local.send.SEND_ERROR_FAILED
|
||||||
import ru.fromchat.api.local.db.isPlaceholderAttachmentDimensions
|
import ru.fromchat.api.local.db.isPlaceholderAttachmentDimensions
|
||||||
import ru.fromchat.api.schema.messages.publicchat.resolvePublicAttachmentLayout
|
import ru.fromchat.api.schema.messages.publicchat.resolvePublicAttachmentLayout
|
||||||
import ru.fromchat.api.schema.messages.publicchat.upload.PublicUploadCompleteResponse
|
import ru.fromchat.api.schema.messages.publicchat.upload.PublicUploadCompleteResponse
|
||||||
@@ -29,6 +32,7 @@ import ru.fromchat.db.Outbox
|
|||||||
import ru.fromchat.ui.chat.isImageFilename
|
import ru.fromchat.ui.chat.isImageFilename
|
||||||
|
|
||||||
private const val DEFAULT_CHUNK_SIZE = 262_144
|
private const val DEFAULT_CHUNK_SIZE = 262_144
|
||||||
|
private const val MAX_ATTACHMENT_OUTBOX_ATTEMPTS = 5L
|
||||||
|
|
||||||
object PublicAttachmentOutboxHandler {
|
object PublicAttachmentOutboxHandler {
|
||||||
private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true }
|
private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true }
|
||||||
@@ -39,7 +43,9 @@ object PublicAttachmentOutboxHandler {
|
|||||||
.executeAsOneOrNull() != null
|
.executeAsOneOrNull() != null
|
||||||
|
|
||||||
private fun ensureStillQueued(instanceId: String, clientMessageId: String) {
|
private fun ensureStillQueued(instanceId: String, clientMessageId: String) {
|
||||||
if (!outboxRowExists(instanceId, clientMessageId)) {
|
if (OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId) ||
|
||||||
|
!outboxRowExists(instanceId, clientMessageId)
|
||||||
|
) {
|
||||||
AttachmentMediaLog.upload("cancelled_in_flight", "job" to clientMessageId)
|
AttachmentMediaLog.upload("cancelled_in_flight", "job" to clientMessageId)
|
||||||
throw kotlinx.coroutines.CancellationException("Upload cancelled")
|
throw kotlinx.coroutines.CancellationException("Upload cancelled")
|
||||||
}
|
}
|
||||||
@@ -51,6 +57,8 @@ object PublicAttachmentOutboxHandler {
|
|||||||
val payload = json.decodeFromString<PublicAttachmentOutboxPayload>(row.payloadJson)
|
val payload = json.decodeFromString<PublicAttachmentOutboxPayload>(row.payloadJson)
|
||||||
val clientMessageId = payload.clientMessageId.trim()
|
val clientMessageId = payload.clientMessageId.trim()
|
||||||
if (clientMessageId.isEmpty()) return true
|
if (clientMessageId.isEmpty()) return true
|
||||||
|
if (OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId)) return true
|
||||||
|
if (OutgoingMessageCoordinator.isOutboundPaused(clientMessageId)) return true
|
||||||
if (!outboxRowExists(instanceId, clientMessageId)) return true
|
if (!outboxRowExists(instanceId, clientMessageId)) return true
|
||||||
|
|
||||||
val conversationId = row.conversationId
|
val conversationId = row.conversationId
|
||||||
@@ -130,6 +138,21 @@ object PublicAttachmentOutboxHandler {
|
|||||||
clientMessageId = clientMessageId,
|
clientMessageId = clientMessageId,
|
||||||
uploadedFileIds = listOf(completed.fileId),
|
uploadedFileIds = listOf(completed.fileId),
|
||||||
)
|
)
|
||||||
|
if (
|
||||||
|
OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId) ||
|
||||||
|
!outboxRowExists(instanceId, clientMessageId)
|
||||||
|
) {
|
||||||
|
AttachmentMediaLog.send(
|
||||||
|
"outbox_send_aborted_after_cancel",
|
||||||
|
"job" to clientMessageId.take(12),
|
||||||
|
"realId" to confirmed.id,
|
||||||
|
)
|
||||||
|
if (confirmed.id > 0) {
|
||||||
|
runCatching { ApiClient.deleteMessage(confirmed.id) }
|
||||||
|
}
|
||||||
|
OutgoingMessageCoordinator.abortPublicServerUploadIfNeeded(completed.fileId)
|
||||||
|
return@runCatching true
|
||||||
|
}
|
||||||
val resolvedConfirmed = mergeConfirmedPublicAttachment(
|
val resolvedConfirmed = mergeConfirmedPublicAttachment(
|
||||||
confirmed = confirmed.copy(client_message_id = clientMessageId),
|
confirmed = confirmed.copy(client_message_id = clientMessageId),
|
||||||
payload = stagedPayload,
|
payload = stagedPayload,
|
||||||
@@ -171,6 +194,7 @@ object PublicAttachmentOutboxHandler {
|
|||||||
bytesUploaded = row.bytesUploaded,
|
bytesUploaded = row.bytesUploaded,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
OutgoingMessageCoordinator.clearOutboundCancelled(clientMessageId)
|
||||||
AttachmentUploadNotifier.emit(
|
AttachmentUploadNotifier.emit(
|
||||||
AttachmentUploadProgress.Success(clientMessageId),
|
AttachmentUploadProgress.Success(clientMessageId),
|
||||||
messageLabel = payload.content,
|
messageLabel = payload.content,
|
||||||
@@ -221,18 +245,81 @@ object PublicAttachmentOutboxHandler {
|
|||||||
),
|
),
|
||||||
messageLabel = payload.content,
|
messageLabel = payload.content,
|
||||||
)
|
)
|
||||||
} else {
|
OutgoingMessageCoordinator.markOutboundPaused(clientMessageId)
|
||||||
// Transient (network / parse / 5xx): keep uploading UI and retry — do not flash failed.
|
return true
|
||||||
|
}
|
||||||
|
if (error.isOutboundPermanentFailure()) {
|
||||||
|
val errorKey = outboundFailureErrorKey(error)
|
||||||
|
AttachmentUploadNotifier.emit(
|
||||||
|
AttachmentUploadProgress.Failed(
|
||||||
|
jobId = clientMessageId,
|
||||||
|
error = errorKey,
|
||||||
|
),
|
||||||
|
messageLabel = payload.content,
|
||||||
|
)
|
||||||
|
withContext(Dispatchers.Default) {
|
||||||
|
MessageCacheStore.markSendFailed(conversationId, clientMessageId)
|
||||||
|
}
|
||||||
|
OutgoingMessageCoordinator.markOutboundPaused(clientMessageId)
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
if (error.isOutboundTransientFailure()) {
|
||||||
AttachmentMediaLog.send(
|
AttachmentMediaLog.send(
|
||||||
"outbox_retryable",
|
"outbox_retryable",
|
||||||
"job" to clientMessageId.take(12),
|
"job" to clientMessageId.take(12),
|
||||||
"err" to (error.message ?: error::class.simpleName),
|
"err" to (error.message ?: error::class.simpleName),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
if (scheduleRetryOrStop(instanceId, row, conversationId, payload.content)) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
false
|
false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private suspend fun scheduleRetryOrStop(
|
||||||
|
instanceId: String,
|
||||||
|
row: Outbox,
|
||||||
|
conversationId: String,
|
||||||
|
messageLabel: String,
|
||||||
|
): Boolean {
|
||||||
|
if (!outboxRowExists(instanceId, row.clientMessageId)) return true
|
||||||
|
val nextRetry = row.retryCount + 1L
|
||||||
|
if (nextRetry >= MAX_ATTACHMENT_OUTBOX_ATTEMPTS) {
|
||||||
|
AttachmentMediaLog.send(
|
||||||
|
"outbox_give_up",
|
||||||
|
"job" to row.clientMessageId.take(12),
|
||||||
|
"attempts" to nextRetry,
|
||||||
|
)
|
||||||
|
AttachmentUploadNotifier.emit(
|
||||||
|
AttachmentUploadProgress.Failed(
|
||||||
|
jobId = row.clientMessageId,
|
||||||
|
error = SEND_ERROR_FAILED,
|
||||||
|
),
|
||||||
|
messageLabel = messageLabel,
|
||||||
|
)
|
||||||
|
withContext(Dispatchers.Default) {
|
||||||
|
MessageCacheStore.markSendFailed(conversationId, row.clientMessageId)
|
||||||
|
}
|
||||||
|
OutgoingMessageCoordinator.markOutboundPaused(row.clientMessageId)
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
withContext(Dispatchers.Default) {
|
||||||
|
if (!outboxRowExists(instanceId, row.clientMessageId)) return@withContext
|
||||||
|
MessageDatabaseProvider.database.messageDatabaseQueries.upsertOutbox(
|
||||||
|
instanceId = instanceId,
|
||||||
|
clientMessageId = row.clientMessageId,
|
||||||
|
conversationId = row.conversationId,
|
||||||
|
kind = row.kind,
|
||||||
|
payloadJson = row.payloadJson,
|
||||||
|
retryCount = nextRetry,
|
||||||
|
nextAttemptAt = row.nextAttemptAt,
|
||||||
|
bytesUploaded = row.bytesUploaded,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
private suspend fun sendResumable(
|
private suspend fun sendResumable(
|
||||||
instanceId: String,
|
instanceId: String,
|
||||||
row: Outbox,
|
row: Outbox,
|
||||||
|
|||||||
-1
@@ -6,5 +6,4 @@ import kotlinx.serialization.Serializable
|
|||||||
data class CallSignalingLiveKitPayload(
|
data class CallSignalingLiveKitPayload(
|
||||||
val toUserId: Int,
|
val toUserId: Int,
|
||||||
val roomName: String,
|
val roomName: String,
|
||||||
val serverUrl: String,
|
|
||||||
)
|
)
|
||||||
@@ -5,7 +5,6 @@ import kotlinx.serialization.Serializable
|
|||||||
|
|
||||||
@Serializable
|
@Serializable
|
||||||
data class LiveKitTokenResponse(
|
data class LiveKitTokenResponse(
|
||||||
@SerialName("server_url") val serverUrl: String,
|
|
||||||
val token: String,
|
val token: String,
|
||||||
@SerialName("room_name") val roomName: String,
|
@SerialName("room_name") val roomName: String,
|
||||||
)
|
)
|
||||||
@@ -87,7 +87,7 @@ private fun AnnotatedString.linkAt(offset: Int): AnnotatedString.Range<LinkAnnot
|
|||||||
}
|
}
|
||||||
|
|
||||||
private val LEGAL_STATIC_LINK_RE = Regex(
|
private val LEGAL_STATIC_LINK_RE = Regex(
|
||||||
pattern = """(?:^|/)?(?:api/)?static/(TERMS|PRIVACY)\.md$""",
|
pattern = """(?:^|/)?static/(TERMS|PRIVACY)\.md$""",
|
||||||
option = RegexOption.IGNORE_CASE,
|
option = RegexOption.IGNORE_CASE,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -155,6 +155,8 @@ fun AttachmentPreview(
|
|||||||
hasSameAuthorAbove = false,
|
hasSameAuthorAbove = false,
|
||||||
hasSameAuthorBelow = false,
|
hasSameAuthorBelow = false,
|
||||||
),
|
),
|
||||||
|
/** When true, grow the tile to the bubble width (e.g. caption text is wider). */
|
||||||
|
expandToBubbleWidth: Boolean = false,
|
||||||
modifier: Modifier = Modifier,
|
modifier: Modifier = Modifier,
|
||||||
) {
|
) {
|
||||||
val isImage = when {
|
val isImage = when {
|
||||||
@@ -221,9 +223,10 @@ fun AttachmentPreview(
|
|||||||
Modifier
|
Modifier
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
// Explicit px size from dp max + aspect — do not wrap to thumb intrinsics
|
.attachmentTileLayout(
|
||||||
// (IntrinsicSize.Max bubbles otherwise shrink to ~80px ≈ 3× too small).
|
aspectRatio = effectiveAspect,
|
||||||
.attachmentTileLayout(aspectRatio = effectiveAspect)
|
expandToBubbleWidth = expandToBubbleWidth,
|
||||||
|
)
|
||||||
.clip(attachmentImageCornerShape(isAuthor, messageGroup))
|
.clip(attachmentImageCornerShape(isAuthor, messageGroup))
|
||||||
.then(
|
.then(
|
||||||
if (onImageBounds != null && showImageTile) {
|
if (onImageBounds != null && showImageTile) {
|
||||||
@@ -243,7 +246,7 @@ fun AttachmentPreview(
|
|||||||
Modifier
|
Modifier
|
||||||
}
|
}
|
||||||
),
|
),
|
||||||
contentAlignment = Alignment.Center
|
contentAlignment = Alignment.Center,
|
||||||
) {
|
) {
|
||||||
Box(
|
Box(
|
||||||
modifier = Modifier
|
modifier = Modifier
|
||||||
|
|||||||
@@ -76,9 +76,9 @@ fun Avatar(
|
|||||||
val fullUrl = if (profilePictureUrl.startsWith("http")) {
|
val fullUrl = if (profilePictureUrl.startsWith("http")) {
|
||||||
profilePictureUrl
|
profilePictureUrl
|
||||||
} else {
|
} else {
|
||||||
val path = profilePictureUrl
|
val path =
|
||||||
.removePrefix("/api")
|
if (profilePictureUrl.startsWith("/")) profilePictureUrl
|
||||||
.let { if (it.startsWith("/")) it else "/$it" }
|
else "/$profilePictureUrl"
|
||||||
"${ServerConfig.apiBaseUrl}$path"
|
"${ServerConfig.apiBaseUrl}$path"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -514,6 +514,9 @@ fun MessageItem(
|
|||||||
pendingIsImage ||
|
pendingIsImage ||
|
||||||
message.files?.firstOrNull()?.let { isImageFilename(it.name) } == true
|
message.files?.firstOrNull()?.let { isImageFilename(it.name) } == true
|
||||||
)
|
)
|
||||||
|
val hasImageCaption =
|
||||||
|
message.content.isNotBlank() &&
|
||||||
|
!isFilenameOnlyMessageCaption(message)
|
||||||
|
|
||||||
val bubbleBodyGestures =
|
val bubbleBodyGestures =
|
||||||
if (isContextMenuOpen) Modifier
|
if (isContextMenuOpen) Modifier
|
||||||
@@ -812,6 +815,7 @@ fun MessageItem(
|
|||||||
messageLabel = message.content,
|
messageLabel = message.content,
|
||||||
onCancelUpload = onCancelUpload,
|
onCancelUpload = onCancelUpload,
|
||||||
messageGroup = group,
|
messageGroup = group,
|
||||||
|
expandToBubbleWidth = hasImageCaption,
|
||||||
modifier = if (firstContentIsImage) {
|
modifier = if (firstContentIsImage) {
|
||||||
Modifier.padding(all = 2.dp)
|
Modifier.padding(all = 2.dp)
|
||||||
} else {
|
} else {
|
||||||
@@ -941,6 +945,8 @@ fun MessageItem(
|
|||||||
messageLabel = message.content,
|
messageLabel = message.content,
|
||||||
onCancelUpload = onCancelUpload,
|
onCancelUpload = onCancelUpload,
|
||||||
messageGroup = group,
|
messageGroup = group,
|
||||||
|
expandToBubbleWidth =
|
||||||
|
isImage && hasImageCaption,
|
||||||
modifier = if (
|
modifier = if (
|
||||||
isFirstImage &&
|
isFirstImage &&
|
||||||
firstContentIsImage &&
|
firstContentIsImage &&
|
||||||
|
|||||||
+36
-7
@@ -1,12 +1,12 @@
|
|||||||
package ru.fromchat.ui.chat.utils
|
package ru.fromchat.ui.chat.utils
|
||||||
|
|
||||||
import androidx.compose.foundation.layout.requiredSize
|
|
||||||
import androidx.compose.foundation.shape.RoundedCornerShape
|
import androidx.compose.foundation.shape.RoundedCornerShape
|
||||||
import androidx.compose.runtime.Composable
|
|
||||||
import androidx.compose.runtime.remember
|
|
||||||
import androidx.compose.ui.Modifier
|
import androidx.compose.ui.Modifier
|
||||||
|
import androidx.compose.ui.layout.layout
|
||||||
|
import androidx.compose.ui.unit.Constraints
|
||||||
import androidx.compose.ui.unit.Dp
|
import androidx.compose.ui.unit.Dp
|
||||||
import androidx.compose.ui.unit.dp
|
import androidx.compose.ui.unit.dp
|
||||||
|
import kotlin.math.roundToInt
|
||||||
import ru.fromchat.api.local.cache.DecryptedImageCache
|
import ru.fromchat.api.local.cache.DecryptedImageCache
|
||||||
import ru.fromchat.api.local.db.aspectRatioFromDimensionPair
|
import ru.fromchat.api.local.db.aspectRatioFromDimensionPair
|
||||||
import ru.fromchat.api.local.db.isPlaceholderAttachmentAspectRatio
|
import ru.fromchat.api.local.db.isPlaceholderAttachmentAspectRatio
|
||||||
@@ -69,17 +69,46 @@ internal fun computeAttachmentTileSize(
|
|||||||
return width to height
|
return width to height
|
||||||
}
|
}
|
||||||
|
|
||||||
@Composable
|
/**
|
||||||
|
* Fixed or bubble-filling attachment tile.
|
||||||
|
*
|
||||||
|
* Default size is capped at [maxWidth]×[maxHeight]. When [expandToBubbleWidth] is true and the
|
||||||
|
* parent (typically [IntrinsicSize.Max] bubble) offers a wider exact width, the tile grows to
|
||||||
|
* that width with proportional height and no height cap.
|
||||||
|
*
|
||||||
|
* Uses a plain [layout] modifier — not BoxWithConstraints — so it stays valid inside
|
||||||
|
* IntrinsicSize parents.
|
||||||
|
*/
|
||||||
internal fun Modifier.attachmentTileLayout(
|
internal fun Modifier.attachmentTileLayout(
|
||||||
aspectRatio: Float?,
|
aspectRatio: Float?,
|
||||||
maxWidth: Dp = ATTACHMENT_TILE_MAX_WIDTH,
|
maxWidth: Dp = ATTACHMENT_TILE_MAX_WIDTH,
|
||||||
maxHeight: Dp = ATTACHMENT_TILE_MAX_HEIGHT,
|
maxHeight: Dp = ATTACHMENT_TILE_MAX_HEIGHT,
|
||||||
|
expandToBubbleWidth: Boolean = false,
|
||||||
): Modifier {
|
): Modifier {
|
||||||
val ratio = aspectRatio?.takeIf { it.isFinite() && it > 0f } ?: 1f
|
val ratio = aspectRatio?.takeIf { it.isFinite() && it > 0f } ?: 1f
|
||||||
val (width, height) = remember(ratio, maxWidth, maxHeight) {
|
val (cappedW, cappedH) = computeAttachmentTileSize(ratio, maxWidth, maxHeight)
|
||||||
computeAttachmentTileSize(ratio, maxWidth, maxHeight)
|
return this.layout { measurable, constraints ->
|
||||||
|
val cappedWpx = cappedW.roundToPx().coerceAtLeast(1)
|
||||||
|
val cappedHpx = cappedH.roundToPx().coerceAtLeast(1)
|
||||||
|
val boundedMax = constraints.maxWidth
|
||||||
|
val width: Int
|
||||||
|
val height: Int
|
||||||
|
if (
|
||||||
|
expandToBubbleWidth &&
|
||||||
|
boundedMax != Constraints.Infinity &&
|
||||||
|
boundedMax > cappedWpx
|
||||||
|
) {
|
||||||
|
width = boundedMax
|
||||||
|
height = (width / ratio).roundToInt().coerceAtLeast(1)
|
||||||
|
} else {
|
||||||
|
width = cappedWpx
|
||||||
|
height = cappedHpx
|
||||||
|
}
|
||||||
|
val placeable = measurable.measure(Constraints.fixed(width, height))
|
||||||
|
layout(width, height) {
|
||||||
|
placeable.placeRelative(0, 0)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return this.requiredSize(width, height)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Inner clip: top corners follow bubble minus inset; bottom corners lightly rounded. */
|
/** Inner clip: top corners follow bubble minus inset; bottom corners lightly rounded. */
|
||||||
|
|||||||
+1
-1
@@ -533,7 +533,7 @@ fun ServerConfigScreen() {
|
|||||||
overflow = TextOverflow.Ellipsis
|
overflow = TextOverflow.Ellipsis
|
||||||
)
|
)
|
||||||
},
|
},
|
||||||
placeholder = { Text("8301") },
|
placeholder = { Text("8300") },
|
||||||
modifier = Modifier
|
modifier = Modifier
|
||||||
.fillMaxWidth()
|
.fillMaxWidth()
|
||||||
.trackLazyListFocus(
|
.trackLazyListFocus(
|
||||||
|
|||||||
Reference in New Issue
Block a user