feat(chat): 加 REST 发送路径 —— 长文与图片(含 clientMsgId 幂等与归一)

客户端此前**只有 WS** 发送,而 WS 帧硬上限 4096 字节、**超限是直接断连**(不是报个错),
所以长文发不出去、图完全发不了。v0.56 给 REST 流式补齐了明确响应面与 clientMsgId,
本次照它接入。

- data/api:ChatApi + ChatDto(窄请求体,不回环整条会话)+ ChatRepository
  —— **刻意不走 `apiCall`**:它把 4xx/5xx 一律折进 `Result.failure` 并丢掉状态码,
  那样 `202`(已排队)与 `409`(队满)的差别会在那一刻消失;改为映射成 ChatAck 密封结果
  (Accepted / Queued / QueueFull / Duplicate / Unavailable / Rejected)
- domain/chat/ChatSendRouting:纯函数路由 —— **UTF-8 字节数 ≤ 3584 且无图 → WS,否则 REST**。
  阈值按**字节**不按字符(中文 1 字 3 字节);预留 512 字节覆盖帧壳。短文本仍走 WS 是刻意的:
  那条路**绕开有损总线**,而超 WS 上限的代价是**断连**而非报错,两边风险不对称
- domain/chat/ChatImageBudget + data/image/ChatImagePreparer:解码 → 等比缩放 → 重编码为
  JPEG/PNG data URI;超预算自动降尺寸重来;**失败返回 null,绝不退回原图 base64**
- clientMsgId:每次发送 `"local-" + UUID`(**随机不自增**——服务端判重窗口是**整个会话历史
  且无过期清理**,进程重启后复用旧值会让那条消息永远发不出去);REST 与 WS 两路都带
- 归一:带 clientMsgId 的消息**只按 id 匹配**(文本不同也命中;匹配不到就什么都不吃,
  **绝不回落到文本**——那会吃掉一条无关气泡);不带键的老行才回落「第一条同文本」
- ack 分支:200 已受理(等 message_persisted)/ 202 已排队(出「已排队 N 条」)/
  **409 既未受理也未落库 → 回滚气泡并把文本与图片还回输入框**(不能当「已排队」,
  那会让气泡永远挂着「发送中…」)/ duplicate 不是错误(收气泡;本地没有该 messageId 就补拉一页)/
  500、503 回滚 / 400 image_too_large 换短文案
- 接下行帧 `chat_duplicate`;图片一律走 REST(`WsOutbound.Chat.image` 仍无生产者)

测试:269 全绿(218 → +51)。含真实抓包的四段响应(200 / 202 / duplicate / 400 image_too_large,
且服务端文案里的 6990572 与本端 SERVER_MAX_CHARS 做成对断言)、路由阈值边界
(1194 汉字 = 3582 字节走 WS、1195 = 3585 走 REST,并直接断言阈值处真实
`WsOutbound.Chat(...)` 序列化后 ≤ 4096 字节)、归一规则、六种 ack 映射。

真机(真后端 127.0.0.1:10942):**图片全链路通过** —— 选图 → 重编码(源 7.8 KB PNG →
1 KB JPEG,magic ffd8ffe0)→ `POST /api/v1/chat`(1558 字节体)→
`200 {"data":{"stream":true}}` → 气泡由「发送中…」归一为「来自 安卓」→ 昔涟回复**准确描述了那张图**。
另 WS 短文本路径未回归(无 POST,且 clientMsgId 被 v0.56 原样回填)。

未验证:`409 chat_busy` 与 `500` 非真实抓包(要填满主人后端的待处理队列、等于真跑 5 个回合,
判断不值得;形状与文案逐字取自后端源码,测试注释里已写明);UI 上的 202/409 未端到端跑过;
降尺寸「减半重来」那步未在设备上跑过(测试图 64×64 只走到重编码)。
This commit is contained in:
AskaEth committed 2026-10-04 16:57:22 +08:00
1 parent 4c14028c70
commit e642b0c18c
19 files changed
+1615 -35

No files matched your search

@@ -0,0 +1,23 @@
package top.yeij.cyrene.data.api
import top.yeij.cyrene.data.api.dto.ChatAckDto
import top.yeij.cyrene.data.api.dto.ChatRequest
import retrofit2.Response
import retrofit2.http.Body
import retrofit2.http.POST
/**
* 发送一条消息(`POST /api/v1/chat`,api-client §1 行 3)。权限 `owner|admin`。
*
* 与会话只读的 [SessionsApi]、管理面 [AdminApi] 分开:这是**写**路径,且它的失败面比只读端点
* 复杂得多(`202` / `409` / `500` / `503` 都是编排器的真实回执,不是「请求错了」),
* 放在一起会让只读端点的调用方以为错误处理是同一种。
*
* 返回原始 [Response](而不是直接返回信封):状态码本身是语义的一部分——
* `202` 排队 与 `409` 排队满 必须能被调用方分开,折进 `Result` 前就先丢了。
*/
interface ChatApi {
@POST("api/v1/chat")
suspend fun sendChat(@Body body: ChatRequest): Response<ApiEnvelope<ChatAckDto>>
}
@@ -0,0 +1,59 @@
package top.yeij.cyrene.data.api.dto
import kotlinx.serialization.Serializable
/**
* `POST /api/v1/chat` 请求体(契约 v0.56,api-client §1 行 3)。
*
* 刻意只带这一个端点要的字段——**不复用整条会话对象**:请求体越小,「WS 帧 4096 字节上限」
* 的反面(REST 请求体没有长度上限)就越不会被无关字段吃掉,抓包时也一眼能看出这一轮发了什么。
*
* 字段语义:
* - [text] / [image] **至少一项非空**,两项都空服务端回 `400 bad_request`;
* - [image] 是 base64 `data:` URI 或 http(s) URL,**解码后上限 5 MiB**
* (超限 `400 image_too_large`,文案指路 `POST /api/v1/upload`);
* - [clientMsgId] 是客户端自带的幂等键:同一会话内**同一个非空值只承认第一条**,
* 重复到达回 `200 {data:{duplicate:true,messageId,sessionId}}` 且不落库、不起回合;
* - [stream] 必须为 `true`(本端**不用** `stream:false`:它是同步的,服务端没有写超时,
* 客户端 60s 读超时会先超时、而回合还在跑——重发就是第二次对话)。
*
* ⚠️ 本项目 `ApiClientFactory.json` 用 kotlinx 默认的 `encodeDefaults = false`:
* 取值等于默认值的字段**整键省略**。因此**纯图片消息**(`text == ""`)线上不会带 `text` 键——
* 服务端 Go 侧零值同为空串,语义完全等价(已按此实现,见 [ChatDtoRealPayloadTest])。
*/
@Serializable
data class ChatRequest(
val sessionId: String? = null,
val text: String = "",
val image: String? = null,
val deviceId: String? = null,
val mode: String? = null,
val stream: Boolean? = null,
val clientMsgId: String? = null,
)
/**
* `POST /api/v1/chat` 的**受理回执**(`data` 载荷)。v0.56 起响应面不再是「任何情况都 200」,
* 四种判定各有形状,因此这里**全部字段可空**、按组合判读,绝不假设某字段恒在:
*
* | 判定 | HTTP | 载荷 |
* |---|---|---|
* | 已起回合 | 200 | `{stream:true}` |
* | 正忙已续接排队 | 202 | `{stream:true,queued:true,queueDepth:n}` |
* | 幂等键重复 | 200 | `{duplicate:true,messageId,sessionId}` |
*
* 其余三种(`409 chat_busy` / `500 chat_unavailable` / `503 chat_dispatch_failed`)是
* `{error:{code,message}}` 信封,由 [top.yeij.cyrene.data.repository.ChatRepository] 走错误路径分类。
*
* 不开枚举:服务端还有别的判定会在将来加进来,未知组合一律按「已受理但形状不认识」处理,
* 不因为不认识一个布尔而把整轮当失败。
*/
@Serializable
data class ChatAckDto(
val stream: Boolean? = null,
val queued: Boolean? = null,
val queueDepth: Int? = null,
val duplicate: Boolean? = null,
val messageId: String? = null,
val sessionId: String? = null,
)
@@ -20,5 +20,12 @@ data class HistoryMessageDto(
val deviceId: String? = null,
val senderName: String? = null,
val adapterId: String? = null,
/**
* 客户端幂等键(契约 v0.56 §3.13,`omitempty`):带 `clientMsgId` 发出的 user 消息会被
* 服务端**原样回填**,`GET /messages` / WS `history` / `message_persisted` 都能看到。
* 客户端据此把本地乐观气泡对齐到服务端真相——取代「靠文本相等」的脆弱匹配。
* 空 = 这条消息没有键(老客户端发的、IM 镜像进来的、v0.56 之前的历史行)。
*/
val clientMsgId: String? = null,
val createdAt: Long,
)
@@ -0,0 +1,115 @@
package top.yeij.cyrene.data.image
import android.content.Context
import android.graphics.Bitmap
import android.graphics.BitmapFactory
import android.net.Uri
import android.util.Base64
import top.yeij.cyrene.domain.chat.ChatImageBudget
import java.io.ByteArrayOutputStream
/**
* 把用户选的本地图片压成可以内联发送的 `data:` URI(**发送前的必经一步**)。
*
* 为什么要重编码而不是直接读原始字节:服务端把 `image` 字符串**原样落库**,
* 每一页历史与每一次 WS `history` 回放都要重传它(协同文档 G3)。所以这里
* 无论用户选了什么,出口都保证 ≤ [ChatImageBudget.MAX_DECODED_BYTES]:
*
* 1. 先按边界解码拿原尺寸 → 等比缩到最长边 [ChatImageBudget.MAX_DIMENSION_PX];
* 2. 编码(有 alpha 用 PNG,否则用 JPEG);
* 3. 仍超预算 → **尺寸减半重来**(最多 [MAX_ATTEMPTS] 档);
* 4. 出口再按服务端的**字符串长度**上限复核一次(4/3 膨胀 + 前缀余量)。
*
* 任何一步失败(读不到 / 解不出 / 编不出 / 压不到预算内)都返回 `null`——由调用方回滚并报错,
* **绝不**退回「原图 base64」:那既会撞服务端 5 MiB 硬顶,也会把历史背成几 MB。
*
* 预算数学在 [ChatImageBudget](纯函数,可单测);本对象只剩「必须真机才有」的解码与编码。
*/
object ChatImagePreparer {
private const val JPEG_QUALITY = 80
/** 缩不动就减半重来的次数上限(1 MiB 预算下 1280px 起步一般 1–2 次就够)。 */
private const val MAX_ATTEMPTS = 4
/**
* @return 可直接作为 `image` 字段发送的 `data:` URI;失败返回 `null`。
* 必须在后台线程调用(解码 + 缩放 + 编码都是重活)。
*/
fun prepare(context: Context, uri: Uri): String? {
val raw = readBytes(context, uri) ?: return null
val decoded = decodeScaled(raw) ?: return null
return try {
encodeWithinBudget(decoded)
} finally {
decoded.recycle()
}
}
private fun readBytes(context: Context, uri: Uri): ByteArray? =
runCatching { context.contentResolver.openInputStream(uri)?.use { it.readBytes() } }.getOrNull()
/** 按**目标尺寸**两步解码:先读边界算 `inSampleSize`(省内存),再精确缩放到目标。 */
private fun decodeScaled(raw: ByteArray): Bitmap? {
val bounds = BitmapFactory.Options().apply { inJustDecodeBounds = true }
BitmapFactory.decodeByteArray(raw, 0, raw.size, bounds)
if (bounds.outWidth <= 0 || bounds.outHeight <= 0) return null
val target = ChatImageBudget.scaledSize(bounds.outWidth, bounds.outHeight)
val options = BitmapFactory.Options().apply {
inSampleSize = sampleSizeFor(bounds.outWidth, bounds.outHeight, target.width, target.height)
}
val sampled = runCatching { BitmapFactory.decodeByteArray(raw, 0, raw.size, options) }.getOrNull()
?: return null
if (sampled.width == target.width && sampled.height == target.height) return sampled
val scaled = runCatching {
Bitmap.createScaledBitmap(sampled, target.width, target.height, true)
}.getOrNull() ?: return null
if (scaled !== sampled) sampled.recycle()
return scaled
}
/** 在不小于目标尺寸的前提下,取最大的 2 的幂作为采样步长。 */
private fun sampleSizeFor(width: Int, height: Int, targetWidth: Int, targetHeight: Int): Int {
var sample = 1
while (width / (sample * 2) >= targetWidth && height / (sample * 2) >= targetHeight) {
sample *= 2
}
return sample
}
/** 逐档降尺寸,直到编码后字节数落进预算;都落不进返回 `null`。 */
private fun encodeWithinBudget(source: Bitmap): String? {
var width = source.width
var height = source.height
repeat(MAX_ATTEMPTS) { attempt ->
val frame = if (attempt == 0) source else scale(source, width, height) ?: return null
// 有 alpha 的图走 JPEG 会把透明区压成黑块,所以这类图保持 PNG(无损、体积更大)。
val mime = if (frame.hasAlpha()) MIME_PNG else MIME_JPEG
val bytes = compress(frame, mime)
if (frame !== source) frame.recycle()
if (ChatImageBudget.withinTarget(bytes.size)) {
val dataUri = ChatImageBudget.dataUri(mime, Base64.encodeToString(bytes, Base64.NO_WRAP))
return dataUri.takeIf { ChatImageBudget.withinServerLimit(it.length) }
}
width = (width / 2).coerceAtLeast(1)
height = (height / 2).coerceAtLeast(1)
}
return null
}
private fun scale(source: Bitmap, width: Int, height: Int): Bitmap? =
runCatching { Bitmap.createScaledBitmap(source, width, height, true) }.getOrNull()
private fun compress(bitmap: Bitmap, mime: String): ByteArray {
val format = if (mime == MIME_PNG) Bitmap.CompressFormat.PNG else Bitmap.CompressFormat.JPEG
val out = ByteArrayOutputStream()
bitmap.compress(format, JPEG_QUALITY, out)
return out.toByteArray()
}
private const val MIME_JPEG = "image/jpeg"
private const val MIME_PNG = "image/png"
}
@@ -0,0 +1,102 @@
package top.yeij.cyrene.data.repository
import top.yeij.cyrene.core.error.CyreneError
import top.yeij.cyrene.data.api.ApiErrorMapper
import top.yeij.cyrene.data.api.ChatApi
import top.yeij.cyrene.data.api.CyreneException
import top.yeij.cyrene.data.api.dto.ChatAckDto
import top.yeij.cyrene.data.api.dto.ChatRequest
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
/**
* `POST /api/v1/chat` 的受理判定(契约 v0.56)。
*
* 它是一个**密封结果**而不是 `Result.failure`:v0.56 起「会话正忙」与「排队已满」不再是异常,
* 而是编排器的正常回执(修复前 handler 提前 `return 200`,这两种状态**永远不可能出现**)。
* 把它们和网络错误混在同一条失败流里,调用方就只能靠字符串猜——这正是本类型要消除的东西。
*/
sealed interface ChatAck {
/** `200 {stream:true}`:回合已起,内容帧走**已连接的 WS**。 */
data object Accepted : ChatAck
/** `202 {stream:true,queued:true,queueDepth:n}`:会话正忙,这条已续接排队(**已落库**)。 */
data class Queued(val queueDepth: Int) : ChatAck
/** `200 {duplicate:true,messageId,sessionId}`:幂等键重复,服务端**没有**再落库 / 再起回合。 */
data class Duplicate(val messageId: String?, val sessionId: String?) : ChatAck
/** `409 {error:{code:"chat_busy"}}`:待处理队列已满,这条**没有被受理、没有落库**。 */
data object QueueFull : ChatAck
/**
* `500 chat_unavailable` / `503 chat_dispatch_failed`:服务端接不住这一轮。
* [code] 保留原始错误码——两者的区别对排查有用,对用户文案无差别。
*/
data class Unavailable(val code: String, val message: String) : ChatAck
/** 其余明确的拒绝(`400 bad_request` / `400 image_too_large` / 未知码)。 */
data class Rejected(val code: String, val message: String) : ChatAck
}
/**
* 发送一条消息(REST 通道)。
*
* 与别的仓库不同的一点:**非 2xx 里有一半是业务态而不是失败**(见 [ChatAck]),
* 所以这里刻意**不使用** [top.yeij.cyrene.data.api.apiCall]——它把 4xx/5xx 一律折进
* `Result.failure`、且只回 `data` 丢掉状态码,于是 `202` 与 `409` 的差别在折进去的那一刻
* 就没了。信封解析与错误映射仍走 [ApiErrorMapper],口径与其它端点一致。
*
* `Result.failure` 只留给**真正发不出去**的情况:网络 / 超时 / 解析不出来 / 401。
*/
class ChatRepository(
private val api: ChatApi,
private val dispatcher: CoroutineDispatcher = Dispatchers.IO,
) {
suspend fun send(request: ChatRequest): Result<ChatAck> = withContext(dispatcher) {
try {
val response = api.sendChat(request)
val data = response.body()?.data
when {
response.isSuccessful && data != null -> Result.success(ackOf(data))
// 2xx 却没有 data:与 apiCall 同一口径,判解析失败而不是「已受理」。
response.isSuccessful -> Result.failure(parseFailure())
else -> {
val error = ApiErrorMapper.fromHttp(response.code(), response.errorBody()?.string())
val ack = serviceAck(error)
if (ack != null) Result.success(ack) else Result.failure(CyreneException(error))
}
}
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
Result.failure(CyreneException(ApiErrorMapper.fromThrowable(e)))
}
}
private fun ackOf(data: ChatAckDto): ChatAck = when {
data.duplicate == true -> ChatAck.Duplicate(data.messageId, data.sessionId)
data.queued == true -> ChatAck.Queued(data.queueDepth ?: 0)
// `{stream:true}` 以及将来新增的未知组合:受理了、形状不认识 → 按已受理处理。
else -> ChatAck.Accepted
}
/** 非 2xx 里属于**编排器真实回执**的部分;返回 null = 真失败(401 或未知服务端错误)。 */
private fun serviceAck(error: CyreneError): ChatAck? {
if (error is CyreneError.Unauthorized) return null
val code = (error as? CyreneError.Server)?.code.orEmpty()
val message = error.userMessage
return when (code) {
"chat_busy" -> ChatAck.QueueFull
"chat_unavailable", "chat_dispatch_failed" -> ChatAck.Unavailable(code, message)
else -> ChatAck.Rejected(code, message)
}
}
private fun parseFailure() =
CyreneException(CyreneError.Local(CyreneError.LocalKind.Parse))
}
@@ -138,10 +138,22 @@ class ChatSocket(
/**
* 发送 `chat`(api-reference §6.3)。
* @param clientMsgId 幂等键(契约 v0.56 §3.13);重复到达服务端回 `chat_duplicate`,不落库。
* @return 未连接或写失败返回 `false`(本层不做离线队列,由上层标红重试)。
*/
fun sendChat(sessionId: String?, text: String, deviceId: String?): Boolean =
send(WsOutbound.Chat(text = text, sessionId = sessionId, deviceId = deviceId))
fun sendChat(
sessionId: String?,
text: String,
deviceId: String?,
clientMsgId: String? = null,
): Boolean = send(
WsOutbound.Chat(
text = text,
sessionId = sessionId,
deviceId = deviceId,
clientMsgId = clientMsgId,
),
)
/** 发送 `interrupt`(取消在跑 turn + 清空 pending)。 */
fun sendInterrupt(sessionId: String): Boolean = send(WsOutbound.Interrupt(sessionId))
@@ -288,6 +288,27 @@ sealed class WsEvent {
override val type: String get() = "pending_full"
}
/**
* `clientMsgId` 重复到达的明确回执(契约 v0.56 §3.13,纯新增帧)。
*
* 语义 = **这条消息早已被受理过**:服务端**不落库、不起回合、不发任何内容帧**
* (它由编排器的状态通道下发,与 `session_busy` 同一出口,因此帧壳带 `sessionId`、
* 不带 `turnId` / `seq`)。[messageId] 是**既有那条用户消息的 id**,
* 客户端据此把本地乐观气泡对齐到服务端真相。
*
* 广播是发给**所有** owner/admin 连接的(hub 无按会话订阅),没发过该
* [clientMsgId] 的客户端按值自行忽略即可。
*/
data class ChatDuplicate(
override val sessionId: String,
val clientMsgId: String,
val messageId: String? = null,
override val turnId: String? = null,
override val seq: Long? = null,
) : WsEvent() {
override val type: String get() = "chat_duplicate"
}
/** 主会话(`web` / `web:<uuid>`)消息落库后广播。 */
data class MessagePersisted(
val message: HistoryMessageDto?,
@@ -427,6 +448,17 @@ sealed class WsEvent {
.let { SessionInterrupted(resolveSessionId(sessionId, it.sessionId), it.mode, it.dropped, turnId, seq) }
"pending_full" -> decode<PendingFullData>(data)
.let { PendingFull(resolveSessionId(sessionId, it.sessionId), it.mode, it.max, turnId, seq) }
// 幂等回执(v0.56 §3.13):帧壳带 sessionId,data 内也带一份(两者取有值者)。
"chat_duplicate" -> decode<ChatDuplicateData>(data)
.let {
ChatDuplicate(
resolveSessionId(sessionId, it.sessionId),
it.clientMsgId,
it.messageId,
turnId,
seq,
)
}
"message_persisted" -> decode<PersistedData>(data)
.let { MessagePersisted(it.message, turnId, seq, sessionId ?: it.message?.sessionId) }
"channel_persisted" -> decode<PersistedData>(data)
@@ -527,6 +559,14 @@ private data class InterruptedData(val sessionId: String = "", val mode: String
@Serializable
private data class PendingFullData(val sessionId: String = "", val mode: String = "", val max: Int = 0)
/** `chat_duplicate` 载荷(v0.56 §3.13):`{sessionId, clientMsgId, messageId}`。 */
@Serializable
private data class ChatDuplicateData(
val sessionId: String = "",
val clientMsgId: String = "",
val messageId: String? = null,
)
@Serializable
private data class PersistedData(val message: HistoryMessageDto? = null)
@@ -32,12 +32,17 @@ sealed interface WsOutbound {
data()?.let { put("data", it) }
}.toString()
/** `chat`:`{sessionId?, text, image?, deviceId?}`;text+image 至少一项。 */
/** `chat`:`{sessionId?, text, image?, deviceId?, clientMsgId?}`;text+image 至少一项。 */
data class Chat(
val text: String,
val sessionId: String? = null,
val deviceId: String? = null,
val image: String? = null,
/**
* 客户端幂等键(契约 v0.56 §3.13):同一会话内同一个非空值只承认第一条,
* 重复到达服务端回下行帧 `chat_duplicate`、**不落库、不起回合**。省略 = 老行为逐字不变。
*/
val clientMsgId: String? = null,
override val reqId: String? = null,
) : WsOutbound {
override val type: String get() = "chat"
@@ -47,6 +52,7 @@ sealed interface WsOutbound {
put("text", text)
image?.let { put("image", it) }
deviceId?.let { put("deviceId", it) }
clientMsgId?.let { put("clientMsgId", it) }
}
}