diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/ApiClient.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/ApiClient.kt index 137976e..91d71b4 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/ApiClient.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/ApiClient.kt @@ -854,10 +854,9 @@ object ApiClient { } /** - * Fetch encrypted file bytes. Path examples: - * - absolute URL like "http://..." -> returned as-is - * - "/uploads/files/encrypted/xxx.jpg" -> returned as /uploads/... - * - "/api/uploads/..." -> strip legacy `/api` prefix, then combine with apiBaseUrl + * Resolve an attachment path to a full URL. + * Absolute `http(s)://…` URLs are returned as-is; relative paths are joined to [ServerConfig.apiBaseUrl]. + * Leading `/api` is stripped so paths match the backend (legacy payloads still used `/api/uploads/...`). */ fun encryptedFileUrl(path: String): String { if (path.startsWith("http")) return path @@ -1590,6 +1589,9 @@ object ApiClient { suspend fun deleteMessage(messageId: Int) { if (_suspensionState.value.isSuspended) return + runCatching { + http.delete("${ServerConfig.apiBaseUrl}/delete_message/$messageId") + } WebSocketManager.send( WebSocketMessage( type = "deleteMessage", @@ -1729,7 +1731,7 @@ object ApiClient { .body() } - suspend fun sendLiveKitInvite(toUserId: Int, roomName: String, serverUrl: String) { + suspend fun sendLiveKitInvite(toUserId: Int, roomName: String) { if (_suspensionState.value.isSuspended) return WebSocketManager.send( WebSocketMessage( @@ -1742,7 +1744,6 @@ object ApiClient { CallSignalingLiveKitPayload( toUserId = toUserId, roomName = roomName, - serverUrl = serverUrl, ), ), ), diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/calls/CallStore.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/calls/CallStore.kt index 7e33fd7..2a65d67 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/calls/CallStore.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/calls/CallStore.kt @@ -33,7 +33,6 @@ sealed class CallUiState { val fromUserId: Int, val fromUsername: String, val roomName: String, - val serverUrl: String, ) : CallUiState() data class InCall( @@ -101,9 +100,8 @@ object CallStore { } return } - val serverUrl = obj["serverUrl"]?.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 (!ServerConfig.callsEnabled) { Logger.d(TAG, "call_signaling invite ignored (calls disabled in server config)") @@ -116,7 +114,6 @@ object CallStore { fromUserId = fromUserId, fromUsername = fromUsername, roomName = roomName, - serverUrl = serverUrl, ) } @@ -132,13 +129,10 @@ object CallStore { runCatching { withContext(Dispatchers.Default) { val tok = ApiClient.fetchLiveKitToken(peerUserId, null) - // LiveKit WS endpoint is exposed on the same host as the server config, - // using the configured calls port. - val signalUrl = ServerConfig.liveKitWsUrl() - ApiClient.sendLiveKitInvite(peerUserId, tok.roomName, signalUrl) + ApiClient.sendLiveKitInvite(peerUserId, tok.roomName) val label = peerLabel(peerUserId) LiveKitConnectSession( - serverUrl = signalUrl, + serverUrl = ServerConfig.liveKitWsUrl(), token = tok.token, peerUserId = peerUserId, peerDisplayName = label, diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/DmAttachmentOutboxHandler.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/DmAttachmentOutboxHandler.kt index 19841ae..669ad1e 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/DmAttachmentOutboxHandler.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/DmAttachmentOutboxHandler.kt @@ -60,7 +60,9 @@ object DmAttachmentOutboxHandler { .executeAsOneOrNull() != null 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) throw kotlinx.coroutines.CancellationException("Upload cancelled") } @@ -72,6 +74,8 @@ object DmAttachmentOutboxHandler { val payload = json.decodeFromString(row.payloadJson) val clientMessageId = payload.clientMessageId.trim() 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 val conversationId = row.conversationId.ifBlank { @@ -168,6 +172,13 @@ object DmAttachmentOutboxHandler { serverUploadId = serverUploadId, ) } + if ( + OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId) || + !outboxRowExists(instanceId, clientMessageId) + ) { + OutgoingMessageCoordinator.abortDmServerUploadIfNeeded(serverUploadId[0]) + return@runCatching true + } ensureStillQueued(instanceId, clientMessageId) clearPrepared(instanceId, clientMessageId) @@ -183,6 +194,7 @@ object DmAttachmentOutboxHandler { bytesUploaded = row.bytesUploaded, ) } + OutgoingMessageCoordinator.clearOutboundCancelled(clientMessageId) AttachmentUploadNotifier.emit( AttachmentUploadProgress.Success(clientMessageId), messageLabel = payload.plaintext, @@ -221,6 +233,10 @@ object DmAttachmentOutboxHandler { ), messageLabel = payload.plaintext, ) + if (failureKey == UPLOAD_ERROR_FILE_TOO_LARGE || error.isOutboundPermanentFailure()) { + OutgoingMessageCoordinator.markOutboundPaused(clientMessageId) + return true + } false } } diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/OutgoingMessageCoordinator.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/OutgoingMessageCoordinator.kt index 3211d84..6a6990b 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/OutgoingMessageCoordinator.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/OutgoingMessageCoordinator.kt @@ -2,12 +2,16 @@ package ru.fromchat.api.local.send import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext +import kotlin.coroutines.coroutineContext import kotlinx.serialization.json.Json import ru.fromchat.api.ApiClient import ru.fromchat.api.local.workers.AttachmentUploadNotifier @@ -40,6 +44,35 @@ object OutgoingMessageCoordinator { private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true } private val drainMutex = Mutex() private val drainScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + private val retryJobs = mutableMapOf() + /** Client message ids cancelled by the user; in-flight sends must not confirm locally. */ + private val cancelledClientIds = MutableStateFlow>(emptySet()) + /** Permanent failures that must not auto-retry until [retryOutboundMessage] / [retryDmAttachmentUpload]. */ + private val pausedClientIds = MutableStateFlow>(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. */ fun onTransportReady() { @@ -47,9 +80,18 @@ object OutgoingMessageCoordinator { } 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) - 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 conversationId = conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID) + markOutboundActive(clientMessageId) withContext(Dispatchers.Default) { MessageRepository.upsertPublicMessage(optimisticMessage) Logger.d("OutgoingMessageCoordinator", "enqueuePublicMessage: clientId=${clientMessageId.take(12)} contentLen=${content.length}") @@ -186,6 +229,7 @@ object OutgoingMessageCoordinator { ) { val instanceId = CacheContext.requireActiveInstanceId() val conversationId = conversationIdForDm(recipientId) + markOutboundActive(clientMessageId) withContext(Dispatchers.Default) { MessageRepository.upsertDmMessage(recipientId, optimisticMessage) val outboundPlaintext = buildDmOutboundPlaintext(plaintext, replyToId) @@ -226,6 +270,7 @@ object OutgoingMessageCoordinator { ) { val instanceId = CacheContext.requireActiveInstanceId() val conversationId = conversationIdForDm(recipientId) + markOutboundActive(clientMessageId) AttachmentUploadNotifier.emit( AttachmentUploadProgress.Pending(clientMessageId, filename), messageLabel = plaintext, @@ -275,6 +320,7 @@ object OutgoingMessageCoordinator { ) { val instanceId = CacheContext.requireActiveInstanceId() val conversationId = conversationIdForGroup(GENERAL_PUBLIC_GROUP_ID) + markOutboundActive(clientMessageId) AttachmentUploadNotifier.emit( AttachmentUploadProgress.Pending(clientMessageId, filename), messageLabel = content, @@ -358,6 +404,8 @@ object OutgoingMessageCoordinator { if (cid.isEmpty()) return val instanceId = CacheContext.activeInstanceId.value.trim() if (instanceId.isEmpty()) return + clearOutboundPaused(cid) + clearOutboundCancelled(cid) drainScope.launch { withContext(Dispatchers.Default) { MessageCacheStore.clearSendFailed(conversationId, cid) @@ -373,6 +421,8 @@ object OutgoingMessageCoordinator { if (cid.isEmpty()) return val instanceId = CacheContext.activeInstanceId.value.trim() if (instanceId.isEmpty()) return + clearOutboundPaused(cid) + clearOutboundCancelled(cid) AttachmentMediaLog.upload("retry_requested", "job" to cid) kickOutboxDrain(instanceId) } @@ -382,6 +432,8 @@ object OutgoingMessageCoordinator { val cid = clientMessageId.trim() if (cid.isEmpty()) return val instanceId = CacheContext.requireActiveInstanceId() + cancelledClientIds.update { it + cid } + clearOutboundPaused(cid) AttachmentMediaLog.upload("cancel_requested", "job" to cid, "conv" to conversationId) withContext(Dispatchers.Default) { val row = MessageDatabaseProvider.database.messageDatabaseQueries @@ -404,7 +456,14 @@ object OutgoingMessageCoordinator { AttachmentUploadNotifier.emit( 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). */ diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/PublicAttachmentOutboxHandler.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/PublicAttachmentOutboxHandler.kt index 9b08c1c..33d3b88 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/PublicAttachmentOutboxHandler.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/local/send/PublicAttachmentOutboxHandler.kt @@ -21,7 +21,10 @@ import ru.fromchat.api.local.db.store.MessageDatabaseProvider import ru.fromchat.api.local.messages.optimisticMessageIdForClientMessageId import ru.fromchat.api.local.workers.AttachmentUploadNotifier 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.schema.messages.publicchat.resolvePublicAttachmentLayout import ru.fromchat.api.schema.messages.publicchat.upload.PublicUploadCompleteResponse @@ -29,6 +32,7 @@ import ru.fromchat.db.Outbox import ru.fromchat.ui.chat.isImageFilename private const val DEFAULT_CHUNK_SIZE = 262_144 +private const val MAX_ATTACHMENT_OUTBOX_ATTEMPTS = 5L object PublicAttachmentOutboxHandler { private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true } @@ -39,7 +43,9 @@ object PublicAttachmentOutboxHandler { .executeAsOneOrNull() != null 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) throw kotlinx.coroutines.CancellationException("Upload cancelled") } @@ -51,6 +57,8 @@ object PublicAttachmentOutboxHandler { val payload = json.decodeFromString(row.payloadJson) val clientMessageId = payload.clientMessageId.trim() if (clientMessageId.isEmpty()) return true + if (OutgoingMessageCoordinator.isOutboundCancelled(clientMessageId)) return true + if (OutgoingMessageCoordinator.isOutboundPaused(clientMessageId)) return true if (!outboxRowExists(instanceId, clientMessageId)) return true val conversationId = row.conversationId @@ -130,6 +138,21 @@ object PublicAttachmentOutboxHandler { clientMessageId = clientMessageId, 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( confirmed = confirmed.copy(client_message_id = clientMessageId), payload = stagedPayload, @@ -171,6 +194,7 @@ object PublicAttachmentOutboxHandler { bytesUploaded = row.bytesUploaded, ) } + OutgoingMessageCoordinator.clearOutboundCancelled(clientMessageId) AttachmentUploadNotifier.emit( AttachmentUploadProgress.Success(clientMessageId), messageLabel = payload.content, @@ -221,18 +245,81 @@ object PublicAttachmentOutboxHandler { ), messageLabel = payload.content, ) - } else { - // Transient (network / parse / 5xx): keep uploading UI and retry — do not flash failed. + OutgoingMessageCoordinator.markOutboundPaused(clientMessageId) + 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( "outbox_retryable", "job" to clientMessageId.take(12), "err" to (error.message ?: error::class.simpleName), ) } + if (scheduleRetryOrStop(instanceId, row, conversationId, payload.content)) { + return true + } 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( instanceId: String, row: Outbox, diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/CallSignalingLiveKitPayload.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/CallSignalingLiveKitPayload.kt index 2b44dee..5cf46a2 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/CallSignalingLiveKitPayload.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/CallSignalingLiveKitPayload.kt @@ -6,5 +6,4 @@ import kotlinx.serialization.Serializable data class CallSignalingLiveKitPayload( val toUserId: Int, val roomName: String, - val serverUrl: String, -) \ No newline at end of file +) diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/LiveKitTokenResponse.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/LiveKitTokenResponse.kt index 2d0ad3d..56761c8 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/LiveKitTokenResponse.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/api/schema/calls/LiveKitTokenResponse.kt @@ -5,7 +5,6 @@ import kotlinx.serialization.Serializable @Serializable data class LiveKitTokenResponse( - @SerialName("server_url") val serverUrl: String, val token: String, @SerialName("room_name") val roomName: String, -) \ No newline at end of file +) diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/legal/Markdown.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/legal/Markdown.kt index ef04bee..82cf4b1 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/legal/Markdown.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/legal/Markdown.kt @@ -87,7 +87,7 @@ private fun AnnotatedString.linkAt(offset: Int): AnnotatedString.Range 0f } ?: 1f - val (width, height) = remember(ratio, maxWidth, maxHeight) { - computeAttachmentTileSize(ratio, maxWidth, maxHeight) + val (cappedW, cappedH) = 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. */ diff --git a/app/shared/src/commonMain/kotlin/ru/fromchat/ui/main/settings/server/ServerConfigScreen.kt b/app/shared/src/commonMain/kotlin/ru/fromchat/ui/main/settings/server/ServerConfigScreen.kt index f4527da..5cbb106 100644 --- a/app/shared/src/commonMain/kotlin/ru/fromchat/ui/main/settings/server/ServerConfigScreen.kt +++ b/app/shared/src/commonMain/kotlin/ru/fromchat/ui/main/settings/server/ServerConfigScreen.kt @@ -533,7 +533,7 @@ fun ServerConfigScreen() { overflow = TextOverflow.Ellipsis ) }, - placeholder = { Text("8301") }, + placeholder = { Text("8300") }, modifier = Modifier .fillMaxWidth() .trackLazyListFocus(