feat(chat): 接上 A5 正向增量 after,并用 404 session_not_found 把「会话已删」如实说出来
改了什么:
- SessionsApi.listMessages 加可选 after(契约 §1 行 6)。before / q / after 三者互斥是服务端的
硬约束,本端三个调用点各只给一个;after 为 null 时 Retrofit 整个省掉这个参数 ⇒ 既有三条路径
的线上请求逐字节不变。
- MessageRepository 加 loadAfter:同一个端点,服务端按【升序】返回(与 before 的倒序相反),
仍统一转成升序的 PagedMessages。顺手去掉 toPagedMessages 参数名里那个已经不对的 newestFirst。
- ChatViewModel:
· 新增 newestKnownAt(从状态现算的 after 游标)与 absorb(只并消息与乐观气泡,不碰
loading / loadingOlder / hasMore / error —— 一次静默补拉不该把上拉位置和错误横幅掀掉)。
· 掉线那一刻记下游标、重新连上时用它做一次正向增量。记录时机必须在「掉线」而不是「重连后」:
重连的 history 帧会把最近 50 条灌进来、游标随即被推到「现在」,那时再取就取不到那个洞了。
「再拉一页最新」永远补不回中间断掉的那一段,这正是 after 存在的理由(02/A5 的欠账)。
· ensureMessageKnown 先用增量(只要新来的那几条),还认不出来才退回整页重拉。
· 循环推进抽成纯函数 loadForwardPages:游标必须严格前进(同毫秒并列且超过一页时推不动,
契约 §A5 明写的已知边界)+ 轮数上限 20,两道防死循环的闸都由单测钉住。
- 404 session_not_found(契约 v0.63):首屏 / 上拉 / 增量三条读路径都判,非主会话就置
ChatUiState.sessionGone;ChatScreen 据此显示一句实话并回调 ChatHomeScreen 重拉会话列表
(refresh 内部会把「指向已删会话」的 activeSessionId 收敛回主会话)。
主会话不置位:那说明部署本身出了问题,该显示的是普通错误横幅,而不是一句假的「已切回主会话」。
为什么:
主人那次「图片凭空消失、别的端也看不到」的真相,就是会话在别处被删掉之后,客户端把它渲染成了
一条空会话 —— 让用户对着空白猜。现在有了机器可读的 404,就能如实说一句并把人带回主会话。
另一半是重连补洞:一直是「再拉一页最新」在顶着,断线期间超过 50 条必丢中间那一段。
验证:assembleDebug + testDebugUnitTest 全绿,846 测试 / 0 失败 / 0 错误。
较基线 831 新增 15 条(MessageAfterRepositoryTest 5 条走真 OkHttp 断言线上 URI;
ChatRecoveryTest 10 条钉住 404 判据与增量循环的两道闸)。
设备侧验证(点开看到那句实话、断线补洞)另行做,不在本次提交的断言范围内。
This commit is contained in:
1 parent
a84d473e2f
commit
28c314262e
7 files changed
+549
-24
No files matched your search
@@ -81,7 +81,14 @@ interface SessionsApi {
|
||||
* 支持搜索之前**逐字节相同**(Retrofit 会整个省掉 null 的查询参数)。非空时服务端按
|
||||
* 大小写不敏感的 ASCII LIKE 匹配 `content`(`%` / `_` / `\` 已转义为字面量),
|
||||
* 仍按时间倒序、`before` / `limit` 同分页语义,搜索**严格限定在 `{id}` 会话内**。
|
||||
* **`q` 与 `after` 互斥**(同传 `400`)——本方法本来就没有 `after` 参数,天然不冲突。
|
||||
* **`q` 与 `after` 互斥**(同传 `400`)。
|
||||
* @param after **正向增量**(A5,契约 §1 行 6):返回 `created_at > after` 的消息,**按时间
|
||||
* **升序**(与 `before` 的倒序**相反**)——用来补「WS 重连只回放最近 50 条」漏掉的那一段。
|
||||
* 游标是**本端已经看到的最后一条 `createdAt`**;服务端配 `limit` 从最旧的一条开始给,所以
|
||||
* 用「本页最后一条的 `createdAt`」反复推进就能一直追到当前。
|
||||
* **与 `before` / `q` 互斥**(同传 `400`)——三个调用点各自只给一个
|
||||
* (`loadPage` 只给 `before`、`search` 只给 `q`、`loadAfter` 只给 `after`),天然不冲突。
|
||||
* 同毫秒并列超过 `limit` 时纯 `createdAt` 游标会跨页重读,调用方按 `id` 去重兜底。
|
||||
*/
|
||||
@GET("api/v1/sessions/{id}/messages")
|
||||
suspend fun listMessages(
|
||||
@@ -90,6 +97,7 @@ interface SessionsApi {
|
||||
@Query("before") before: Long? = null,
|
||||
@Query("platform") platform: String? = null,
|
||||
@Query("q") query: String? = null,
|
||||
@Query("after") after: Long? = null,
|
||||
): Response<ApiEnvelope<List<HistoryMessageDto>>>
|
||||
|
||||
/**
|
||||
|
||||
@@ -11,8 +11,14 @@ import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
|
||||
/**
|
||||
* 会话消息层:分页 / 搜索**读取**,以及单条消息的**编辑 / 删除**(D3)。
|
||||
* 服务端按倒序返回分页结果,这里统一转成升序的 [PagedMessages]。去重 / 合并请用 [mergeMessages]。
|
||||
* 会话消息层:分页 / 搜索 / **正向增量**读取,以及单条消息的**编辑 / 删除**(D3)。
|
||||
*
|
||||
* 三条读取出口共用同一个端点,只是给的查询参数不同:
|
||||
* - [loadPage] / [loadLatest]:`before` 往回翻(服务端**倒序**返回);
|
||||
* - [search]:`q` 会话内搜索(也是倒序);
|
||||
* - [loadAfter]:`after` 往前追(服务端**升序**返回,契约 §1 行 6)。
|
||||
*
|
||||
* 三者都统一转成**升序**的 [PagedMessages] 再交出去。去重 / 合并请用 [mergeMessages]。
|
||||
*/
|
||||
class MessageRepository(
|
||||
private val api: SessionsApi,
|
||||
@@ -69,6 +75,31 @@ class MessageRepository(
|
||||
.map { raw -> toPagedMessages(raw, pageSize) }
|
||||
}
|
||||
|
||||
/**
|
||||
* **正向增量**(A5,契约 §1 行 6):`GET /messages?after=<ms>`,返回 `created_at > after` 的
|
||||
* 消息,**升序**(与 [loadPage] 的倒序相反)。
|
||||
*
|
||||
* 它存在的唯一理由是补断线的洞:WS 重连只回放主会话**最近 50 条**,断线期间超过 50 条的
|
||||
* 消息用「再拉一页最新」是补不回来的——最新那一页会直接跳过中间那段。[after] 就是那个游标:
|
||||
* 传本端已看到的最后一条 `createdAt`,服务端从**最旧的一条开始**给。
|
||||
*
|
||||
* 分页推进的规矩(与 [loadPage] 不同,别照抄):下一页的游标是本页**最新**一条的
|
||||
* `createdAt`([PagedMessages.newestCreatedAt]),不是最旧的。
|
||||
*
|
||||
* 同毫秒并列且条数超过 `limit` 时,纯 `createdAt` 游标推不动(契约 §A5 明说的已知边界)——
|
||||
* 调用方按 `id` 去重、并在游标没前进时停下来,不在这里兜。
|
||||
*/
|
||||
suspend fun loadAfter(
|
||||
sessionId: String,
|
||||
after: Long,
|
||||
limit: Int = DEFAULT_PAGE_SIZE,
|
||||
platform: String? = null,
|
||||
): Result<PagedMessages> = withContext(dispatcher) {
|
||||
val pageSize = limit.coerceIn(1, MAX_PAGE_SIZE)
|
||||
apiCall { api.listMessages(sessionId, pageSize, before = null, platform = platform, after = after) }
|
||||
.map { raw -> toPagedMessages(raw, pageSize) }
|
||||
}
|
||||
|
||||
/**
|
||||
* 编辑单条消息(D3,契约 §1 行 6a):`PATCH /api/v1/messages/{id}`,成功返回**服务端更新后的
|
||||
* 完整消息 DTO**。
|
||||
@@ -140,14 +171,21 @@ class MessageRepository(
|
||||
*/
|
||||
fun normalizeSearchQuery(raw: String): String? = raw.trim().takeIf { it.isNotEmpty() }
|
||||
|
||||
/**
|
||||
* 原始响应 → [PagedMessages]:**无论服务端给的是倒序(`before` / `q`)还是升序(`after`)**,
|
||||
* 一律按 [MessageOrder] 排成升序。
|
||||
*
|
||||
* [hasMore] 两向同义:拿到满页 ⇒ 该方向**可能还有**(`before` 方向是更旧的、`after` 方向是更新的)——
|
||||
* 它是保守判断,具体能不能继续由调用方的游标决定。
|
||||
*/
|
||||
private fun toPagedMessages(
|
||||
rawNewestFirst: List<HistoryMessageDto>,
|
||||
raw: List<HistoryMessageDto>,
|
||||
pageSize: Int,
|
||||
): PagedMessages {
|
||||
val ascending = rawNewestFirst.sortedWith(MessageOrder)
|
||||
val ascending = raw.sortedWith(MessageOrder)
|
||||
return PagedMessages(
|
||||
messages = ascending,
|
||||
hasMore = rawNewestFirst.size >= pageSize,
|
||||
hasMore = raw.size >= pageSize,
|
||||
oldestCreatedAt = ascending.firstOrNull()?.createdAt,
|
||||
newestCreatedAt = ascending.lastOrNull()?.createdAt,
|
||||
)
|
||||
|
||||
@@ -75,6 +75,10 @@ fun ChatHomeScreen(
|
||||
sessionsViewModel.refresh(silent = true)
|
||||
},
|
||||
onLogout = onLogout,
|
||||
// 当前会话在别处被删了(`404 session_not_found`)⇒ 重拉列表:`refresh` 里面会把
|
||||
// 「指向已删会话」的 activeSessionId 收敛回主会话(`resolveActiveSession`),
|
||||
// 抽屉里那条也跟着消失。所以这里不需要自己判「该切到哪儿」。
|
||||
onSessionGone = { sessionsViewModel.refresh(silent = true) },
|
||||
)
|
||||
},
|
||||
drawerContent = {
|
||||
|
||||
@@ -139,6 +139,7 @@ fun ChatScreen(
|
||||
sessionId: String,
|
||||
onOpenDrawer: () -> Unit = {},
|
||||
onLogout: () -> Unit = {},
|
||||
onSessionGone: () -> Unit = {},
|
||||
) {
|
||||
val viewModel: ChatViewModel = viewModel(
|
||||
key = sessionId,
|
||||
@@ -147,6 +148,13 @@ fun ChatScreen(
|
||||
val state by viewModel.state.collectAsState()
|
||||
val listState = rememberLazyListState()
|
||||
|
||||
// 这个会话在服务端已经没了(`404 session_not_found`,契约 v0.63):交给上层把用户带回主会话。
|
||||
// 只有**非主会话**会置这个位(主会话回 404 说明部署出了问题,那时该显示的是普通错误横幅),
|
||||
// 所以这里不会出现「已经主会话还往上跳一次」的回环。
|
||||
LaunchedEffect(state.sessionGone) {
|
||||
if (state.sessionGone) onSessionGone()
|
||||
}
|
||||
|
||||
// —— 会话内搜索(D2)——
|
||||
// 搜索态活在**聊天页内部**、自带一个小 VM:它不改消息窗、不发消息、不碰 WS,与另一个并行
|
||||
// 模块正在改动的 ChatViewModel 毫无关系,所以刻意不并进那个 VM。
|
||||
@@ -485,6 +493,12 @@ fun ChatScreen(
|
||||
state.error?.let { message ->
|
||||
Banner(text = message, color = MiuixTheme.colorScheme.error)
|
||||
}
|
||||
if (state.sessionGone) {
|
||||
Banner(
|
||||
text = "这个会话在服务端已经不存在了(可能已在别处删除),正在切回主会话",
|
||||
color = MiuixTheme.colorScheme.error,
|
||||
)
|
||||
}
|
||||
state.shareNotice?.let { message ->
|
||||
Banner(text = message, color = MiuixTheme.colorScheme.onSurfaceVariantSummary)
|
||||
}
|
||||
|
||||
@@ -23,6 +23,7 @@ import top.yeij.cyrene.data.image.PreparedImage
|
||||
import top.yeij.cyrene.data.repository.ChatAck
|
||||
import top.yeij.cyrene.data.repository.MessageEditDraft
|
||||
import top.yeij.cyrene.data.repository.MessageRepository
|
||||
import top.yeij.cyrene.data.repository.PagedMessages
|
||||
import top.yeij.cyrene.data.repository.applyMessageEdit
|
||||
import top.yeij.cyrene.data.repository.canDeleteMessage
|
||||
import top.yeij.cyrene.data.repository.canEditMessage
|
||||
@@ -465,6 +466,17 @@ data class ChatUiState(
|
||||
* 动作总是针对**一条**消息,两个弹层同时在飞只会让「失败该说给谁听」变得含糊。
|
||||
*/
|
||||
val messageAction: MessageActionState? = null,
|
||||
/**
|
||||
* **这个会话在服务端已经不存在了**(`404 session_not_found`,契约 v0.63)。
|
||||
*
|
||||
* 与 [error] 分开:它不是「这一次请求失败了」,而是「**这个会话本身没了**」——常见于别的端
|
||||
* 把它删掉了(主人的真实遭遇:正看着的临时会话在别处被清理,客户端这边看着就像一条空会话,
|
||||
* 消息「凭空消失」,让人以为是丢数据)。界面据此**如实说一句**并把用户带回主会话。
|
||||
*
|
||||
* **主会话([MAIN_SESSION_ID])不会置它**:主会话若回 404,说明部署本身出了问题,
|
||||
* 那时候该显示的是普通的错误横幅,而不是一句「已切回主会话」(那会是一句假话)。
|
||||
*/
|
||||
val sessionGone: Boolean = false,
|
||||
) {
|
||||
/** 渲染顺序:已落库 → 乐观发送中 → 流式中(升序,UI 再反向喂给 reverseLayout)。 */
|
||||
val items: List<ChatDisplayItem>
|
||||
@@ -530,6 +542,15 @@ class ChatViewModel(
|
||||
private var oldestLoadedAt: Long? = null
|
||||
private var streamParser: StreamingContentParser? = null
|
||||
|
||||
/**
|
||||
* 掉线那一刻的 `after` 游标(契约 §1 行 6):为 null = 当时没有需要补的洞。
|
||||
*
|
||||
* 存在理由:WS 重连只回放主会话**最近 50 条**,而重连后本端会因为这一帧回放把游标直接推到
|
||||
* 「现在」——不**在掉线那一刻**先记下来,那个洞就永远补不上了。所以记的时机是「离开 Connected」,
|
||||
* 用的时机是「重新 Connected」。
|
||||
*/
|
||||
private var gapCursor: Long? = null
|
||||
|
||||
/**
|
||||
* 「没发出去的那几条」的记忆(**多槽**,见 [RetryKeyMemory]):**回滚**时按内容指纹记下那条用的
|
||||
* 幂等键,**下一次发送**指纹相等就复用那个键。复用时**只消费命中的那一条**,别的失败记录原样
|
||||
@@ -576,18 +597,43 @@ class ChatViewModel(
|
||||
}
|
||||
.onFailure { e ->
|
||||
_state.update { it.copy(loading = false, error = e.userMessage()) }
|
||||
markSessionGoneIfNeeded(e)
|
||||
}
|
||||
}
|
||||
|
||||
val ws = container.createChatSocket()
|
||||
socket = ws
|
||||
viewModelScope.launch {
|
||||
ws.state.collect { connection -> _state.update { it.copy(connection = connection) } }
|
||||
}
|
||||
viewModelScope.launch { ws.state.collect(::onConnectionChanged) }
|
||||
viewModelScope.launch { ws.events.collect(::onEvent) }
|
||||
ws.connect()
|
||||
}
|
||||
|
||||
/**
|
||||
* 连接状态变化:除了照抄给界面,还要顺手维护「断线期间漏了哪一段」这个游标。
|
||||
*
|
||||
* - **离开 Connected 那一刻**记下 [gapCursor](= 当时本端已有最新一条的 `createdAt`);
|
||||
* - **重新 Connected 那一刻**用它做一次 `after` 正向增量。
|
||||
*
|
||||
* 记录时机必须是「掉线」而不是「重连后」:重连的 `history` 帧会把最近 50 条灌进来,本端游标
|
||||
* 随即被推到「现在」,那时候再取游标就取不到那个洞了。这也正是 `after`(契约 §1 行 6)
|
||||
* 存在的理由——「再拉一页最新」永远补不回中间断掉的那一段。
|
||||
*
|
||||
* 冷启动(Idle → Connected)不动:[gapCursor] 是 null,首屏 REST 本来就覆盖了。
|
||||
*/
|
||||
private fun onConnectionChanged(connection: WsState) {
|
||||
val previous = _state.value.connection
|
||||
_state.update { it.copy(connection = connection) }
|
||||
if (connection == previous) return
|
||||
|
||||
if (previous == WsState.Connected) gapCursor = newestKnownAt
|
||||
if (connection == WsState.Connected) {
|
||||
val cursor = gapCursor
|
||||
gapCursor = null
|
||||
val repo = repository
|
||||
if (cursor != null && repo != null) viewModelScope.launch { catchUp(repo, cursor) }
|
||||
}
|
||||
}
|
||||
|
||||
fun onDraftChange(text: String) = _state.update { it.copy(draft = text) }
|
||||
|
||||
/** 选好待发图片(本地 `content://` URI);传 null 或空白串即清除。 */
|
||||
@@ -1129,33 +1175,77 @@ class ChatViewModel(
|
||||
}
|
||||
|
||||
/**
|
||||
* 本地不认识这条 `messageId` 时补拉最新一页(认得的就什么也不做,省一次往返)。
|
||||
* 「本端已看到的最新一条」= `after` 游标(契约 §1 行 6)。
|
||||
*
|
||||
* **从状态里现算**,而不是单独维护一个字段:消息从 REST 分页、WS 回放、`message_persisted`、
|
||||
* 编辑回执好几条路进来,单独维护的游标只要有一条路忘了更新,它就会静默落后——而落后表现为
|
||||
* 「补拉时把已经看过的几百条又拉一遍」。现算的代价只是一次 `maxOf`,且只在补拉时才走到。
|
||||
*/
|
||||
private val newestKnownAt: Long?
|
||||
get() = _state.value.messages.maxOfOrNull { it.createdAt }
|
||||
|
||||
/**
|
||||
* 本地不认识这条 `messageId` 时补拉(认得的就什么也不做,省一次往返)。
|
||||
*
|
||||
* 为什么要补:复用同一个 `clientMsgId` 重试 / 收编时,正常路径本端早就收到过
|
||||
* `message_persisted`(第一次发的时候),但弱网 / 重连丢帧会留下「气泡没了、消息也没了」
|
||||
* 的空档——补一次增量拉取才是真的「对齐到服务端真相」,而不是让气泡凭空消失。
|
||||
*
|
||||
* 先用 `after` 做**增量**(便宜:只要新来的那几条),还认不出来才退回整页重拉——兜住
|
||||
* 「服务端把这条排在了本端游标更早的位置」这种增量拿不到的情形(`after` 只往前看)。
|
||||
*/
|
||||
private fun ensureMessageKnown(messageId: String) {
|
||||
if (_state.value.messages.any { it.id == messageId }) return
|
||||
refreshLatest()
|
||||
}
|
||||
|
||||
/** 静默补拉最新一页(成功才并进来;失败不改状态——这只是一次对齐尝试)。 */
|
||||
private fun refreshLatest() {
|
||||
val repo = repository ?: return
|
||||
viewModelScope.launch {
|
||||
repo.loadLatest(sessionId).onSuccess { page ->
|
||||
page.oldestCreatedAt?.let { oldest -> oldestLoadedAt = minOf(oldestLoadedAt ?: oldest, oldest) }
|
||||
_state.update {
|
||||
it.copy(
|
||||
messages = mergeMessages(it.messages, page.messages),
|
||||
pending = reconcilePending(it.pending, page.messages),
|
||||
)
|
||||
}
|
||||
}
|
||||
newestKnownAt?.let { catchUp(repo, it) }
|
||||
if (_state.value.messages.any { it.id == messageId }) return@launch
|
||||
repo.loadLatest(sessionId).onSuccess(::absorb)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 [from] 起**正向**追到当前(契约 §1 行 6 的 `after` 正向增量)。
|
||||
*
|
||||
* 推进的规矩与两道闸都在 [loadForwardPages] 里(那是纯编排,可单测);这里只负责把取回来的
|
||||
* 每一页并进状态,外加把失败交给 [markSessionGoneIfNeeded]——失败**静默**(除了「会话没了」):
|
||||
* 这只是一次对齐尝试,弹一条错反而比丢帧更吵。
|
||||
*/
|
||||
private suspend fun catchUp(repo: MessageRepository, from: Long) {
|
||||
var failure: Throwable? = null
|
||||
val pages = loadForwardPages(from) { cursor ->
|
||||
repo.loadAfter(sessionId, cursor).onFailure { failure = it }
|
||||
}
|
||||
pages.forEach(::absorb)
|
||||
failure?.let(::markSessionGoneIfNeeded)
|
||||
}
|
||||
|
||||
/**
|
||||
* 把一页**服务端权威**消息并进状态:按 `id` 去重([mergeMessages])并收掉对应的乐观气泡。
|
||||
*
|
||||
* 刻意**不碰** `loading` / `loadingOlder` / `hasMore` / `error`:那几项是「首屏」与「上拉分页」
|
||||
* 的状态,一次静默补拉去改它们,会把用户的上拉位置和错误横幅一起掀掉。
|
||||
*/
|
||||
private fun absorb(page: PagedMessages) {
|
||||
page.oldestCreatedAt?.let { oldest -> oldestLoadedAt = minOf(oldestLoadedAt ?: oldest, oldest) }
|
||||
_state.update {
|
||||
it.copy(
|
||||
messages = mergeMessages(it.messages, page.messages),
|
||||
pending = reconcilePending(it.pending, page.messages),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 现行会话在服务端已经没了(`404 session_not_found`)就如实置位(见 [ChatUiState.sessionGone])。
|
||||
*
|
||||
* 主会话**不置**:那说明部署出了问题,该显示的是普通错误横幅。判据见 [isSessionGone]。
|
||||
*/
|
||||
private fun markSessionGoneIfNeeded(e: Throwable) {
|
||||
if (!e.isSessionGone() || sessionId == MAIN_SESSION_ID) return
|
||||
_state.update { it.copy(sessionGone = true) }
|
||||
}
|
||||
|
||||
/**
|
||||
* 一批**服务端权威消息**到达时,把它们对应的乐观气泡一并收掉(按 `clientMsgId`)。
|
||||
*
|
||||
@@ -1472,6 +1562,7 @@ class ChatViewModel(
|
||||
}
|
||||
.onFailure { e ->
|
||||
_state.update { it.copy(loadingOlder = false, error = e.userMessage()) }
|
||||
markSessionGoneIfNeeded(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1488,6 +1579,18 @@ class ChatViewModel(
|
||||
private fun Throwable.userMessage(): String =
|
||||
(this as? CyreneException)?.error?.userMessage ?: "加载失败,请稍后重试"
|
||||
|
||||
/**
|
||||
* 这个失败是不是「**当前会话在服务端已经不存在了**」(契约 v0.63:`404 session_not_found`)。
|
||||
*
|
||||
* 判据以**服务端给的码**为准([SESSION_NOT_FOUND_CODE]),并兜住「服务端没给 error 信封」的
|
||||
* 情况——那时代码是空串,但 404 仍然只有这一个含义。这就是主人那次「图片凭空消失」的答案:
|
||||
* 那个会话在别处被删了,客户端当时把它渲染成了一条**空会话**。
|
||||
*/
|
||||
internal fun Throwable.isSessionGone(): Boolean {
|
||||
val error = (this as? CyreneException)?.error as? CyreneError.Server ?: return false
|
||||
return error.code == SESSION_NOT_FOUND_CODE || error.httpStatus == 404
|
||||
}
|
||||
|
||||
/**
|
||||
* 上传失败的文案(纯函数,可单测)。
|
||||
*
|
||||
@@ -1510,6 +1613,58 @@ fun uploadFailureMessage(e: Throwable): String {
|
||||
/** 思维链在流式结束后继续保留的时长。 */
|
||||
private const val THINKING_LINGER_MILLIS = 10 * 60 * 1000L
|
||||
|
||||
/**
|
||||
* 一次正向补拉最多追几轮(每轮一页,见 [loadForwardPages])。
|
||||
*
|
||||
* 20 × 50 = 1000 条:比任何一次真实的断线都长,又短到不会让界面卡在连续请求里。追不完的洞
|
||||
* 留给下一次(下一次补拉或下一条新消息都会把游标往前带)。
|
||||
*/
|
||||
internal const val MAX_CATCH_UP_ROUNDS = 20
|
||||
|
||||
/**
|
||||
* 正向增量循环(**纯编排,可单测**):从 [from] 起反复用本页**最新**一条的 `createdAt` 当下一轮的
|
||||
* `after`,直到某一页不满一页、或游标推进不动。
|
||||
*
|
||||
* 游标为什么取**最新**那条:`after` 是**升序**返回(契约 §1 行 6),服务端配 `limit` 是从最旧的
|
||||
* 一端给的——取最旧那条当游标就等于原地不动。
|
||||
*
|
||||
* 两道**防死循环**的闸,缺一不可:
|
||||
* 1. **游标必须严格前进**:同毫秒并列且条数超过 `limit` 时 `createdAt` 游标推不动(契约 §A5
|
||||
* 明写的已知边界),不检查就会拿同一批消息无限请求;
|
||||
* 2. **轮数上限** [maxRounds]:再长的洞留给下一次。
|
||||
*
|
||||
* 失败即停(把已经取到的页交出去):这是尽力而为的对齐,不是必须成功的事务。
|
||||
* 返回**按序**取回的每一页,合并是调用方的事——所以这里没有状态、没有协程作用域。
|
||||
*/
|
||||
internal suspend fun loadForwardPages(
|
||||
from: Long,
|
||||
maxRounds: Int = MAX_CATCH_UP_ROUNDS,
|
||||
load: suspend (Long) -> Result<PagedMessages>,
|
||||
): List<PagedMessages> {
|
||||
val pages = mutableListOf<PagedMessages>()
|
||||
var cursor = from
|
||||
var rounds = 0
|
||||
while (rounds++ < maxRounds) {
|
||||
val page = load(cursor).getOrElse { return pages }
|
||||
if (page.messages.isEmpty()) return pages
|
||||
pages += page
|
||||
val next = page.messages.maxOf { it.createdAt }
|
||||
if (next <= cursor) return pages
|
||||
cursor = next
|
||||
if (!page.hasMore) return pages
|
||||
}
|
||||
return pages
|
||||
}
|
||||
|
||||
/**
|
||||
* 「这个会话已经不在服务端」的错误码(契约 v0.63)。
|
||||
*
|
||||
* `/messages` 上的 404 只可能是这一件事(消息端点 `/messages/{id}` 走的是另一条路径,
|
||||
* `PATCH /sessions/{id}` 用的是通用码 `not_found`),所以按码判**并**兜一个状态码就够——
|
||||
* 不把 404 当万能的「会话没了」,免得将来别的 404 也被读成「会话已删」。
|
||||
*/
|
||||
private const val SESSION_NOT_FOUND_CODE = "session_not_found"
|
||||
|
||||
/** 分享提示在界面上停留的时长(它描述的是一次刚发生的动作,不是持续状态)。 */
|
||||
private const val SHARE_NOTICE_MILLIS = 8 * 1000L
|
||||
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
package top.yeij.cyrene.data.repository
|
||||
|
||||
import com.sun.net.httpserver.HttpExchange
|
||||
import com.sun.net.httpserver.HttpServer
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
import top.yeij.cyrene.core.error.CyreneError
|
||||
import top.yeij.cyrene.data.api.ApiClientFactory
|
||||
import top.yeij.cyrene.data.api.CyreneException
|
||||
import top.yeij.cyrene.data.api.SessionsApi
|
||||
import java.net.InetSocketAddress
|
||||
import java.util.concurrent.CopyOnWriteArrayList
|
||||
import java.util.concurrent.Executors
|
||||
|
||||
/**
|
||||
* 正向增量(A5)的**数据层**:`after` 有没有真的上线上、以及服务端的**升序**口径有没有被本端
|
||||
* 如实接住。
|
||||
*
|
||||
* 与 `MessageSearchRepositoryTest` 同一套做法(真 OkHttp + 真 Retrofit + JDK 自带的
|
||||
* `com.sun.net.httpserver`,不引新依赖),断言落在**收到的原始 URI** 上:`before` / `q` / `after`
|
||||
* 三者互斥是**服务端**的硬约束(同传回 `400`),所以「一次请求里只出现其中一个」必须由 URI 证据
|
||||
* 来钉,不能只看返回值。
|
||||
*/
|
||||
class MessageAfterRepositoryTest {
|
||||
|
||||
private val backend = FakeMessagesBackend()
|
||||
|
||||
@After
|
||||
fun tearDown() {
|
||||
backend.close()
|
||||
}
|
||||
|
||||
private fun repository() = MessageRepository(
|
||||
ApiClientFactory.create(baseUrl = backend.baseUrl, service = SessionsApi::class.java),
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `the forward cursor is the only paging parameter on the wire`() = runBlocking {
|
||||
backend.reply(200, """{"data":[${message("m1", 100)}]}""")
|
||||
|
||||
repository().loadAfter("web", after = 100).getOrThrow()
|
||||
|
||||
val uri = backend.uris.single()
|
||||
assertTrue(uri, uri.contains("after=100"))
|
||||
assertTrue(uri, uri.contains("limit=50"))
|
||||
// 互斥的那两个一个都不许出现:服务端同传就回 400,而这条路径恰恰不需要它们。
|
||||
assertFalse(uri, uri.contains("before="))
|
||||
assertFalse(uri, uri.contains("q="))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ascending server order is preserved and the newest cursor is the max`() = runBlocking {
|
||||
// `after` 的契约是**升序**(与 `before` 的倒序相反),照抄这个形状喂进来。
|
||||
backend.reply(
|
||||
200,
|
||||
"""{"data":[${message("m1", 100)}, ${message("m2", 200)}, ${message("m3", 300)}]}""",
|
||||
)
|
||||
|
||||
val page = repository().loadAfter("web", after = 100, limit = 50).getOrThrow()
|
||||
|
||||
assertEquals(listOf("m1", "m2", "m3"), page.messages.map { it.id })
|
||||
assertEquals(100L, page.oldestCreatedAt)
|
||||
// 下一页的游标是**最新**那条(`after` 方向与 `before` 方向相反,规则也相反)。
|
||||
assertEquals(300L, page.newestCreatedAt)
|
||||
assertFalse(page.hasMore)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a full page says there may be more ahead`() = runBlocking {
|
||||
backend.reply(200, """{"data":[${message("m1", 100)}, ${message("m2", 200)}]}""")
|
||||
|
||||
val page = repository().loadAfter("web", after = 0, limit = 2).getOrThrow()
|
||||
|
||||
assertTrue(page.hasMore)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the page size is clamped to the server ceiling`() = runBlocking {
|
||||
backend.reply(200, """{"data":[]}""")
|
||||
|
||||
repository().loadAfter("web", after = 0, limit = 999).getOrThrow()
|
||||
|
||||
assertTrue(backend.uris.single(), backend.uris.single().contains("limit=200"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a deleted session surfaces as session_not_found with 404`() = runBlocking {
|
||||
backend.reply(
|
||||
404,
|
||||
"""{"error":{"code":"session_not_found","message":"session \"web:gone\" not found"}}""",
|
||||
)
|
||||
|
||||
val error = repository().loadAfter("web:gone", after = 0).exceptionOrNull() as CyreneException
|
||||
|
||||
val server = error.error as CyreneError.Server
|
||||
assertEquals("session_not_found", server.code)
|
||||
assertEquals(404, server.httpStatus)
|
||||
}
|
||||
|
||||
private fun message(id: String, createdAt: Long): String =
|
||||
"""{"id":"$id","sessionId":"web","role":"user","content":"内容","createdAt":$createdAt}"""
|
||||
|
||||
/** 只回一份固定响应、但把每次请求的原始 URI 记下来的假后端。 */
|
||||
private inner class FakeMessagesBackend : AutoCloseable {
|
||||
|
||||
private val server = HttpServer.create(InetSocketAddress("127.0.0.1", 0), 0).apply {
|
||||
executor = Executors.newFixedThreadPool(2)
|
||||
createContext("/", ::handle)
|
||||
start()
|
||||
}
|
||||
|
||||
val baseUrl: String get() = "http://127.0.0.1:${server.address.port}/"
|
||||
|
||||
val uris = CopyOnWriteArrayList<String>()
|
||||
|
||||
@Volatile
|
||||
private var status: Int = 200
|
||||
|
||||
@Volatile
|
||||
private var body: String = """{"data":[]}"""
|
||||
|
||||
fun reply(status: Int, body: String) {
|
||||
this.status = status
|
||||
this.body = body
|
||||
}
|
||||
|
||||
private fun handle(exchange: HttpExchange) {
|
||||
uris += exchange.requestURI.toString()
|
||||
val bytes = body.toByteArray()
|
||||
exchange.responseHeaders.add("Content-Type", "application/json")
|
||||
exchange.sendResponseHeaders(status, bytes.size.toLong())
|
||||
exchange.responseBody.use { it.write(bytes) }
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
server.stop(0)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
package top.yeij.cyrene.ui.chat
|
||||
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
import top.yeij.cyrene.core.error.CyreneError
|
||||
import top.yeij.cyrene.data.api.CyreneException
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
import top.yeij.cyrene.data.repository.PagedMessages
|
||||
|
||||
/**
|
||||
* 断线恢复的两块纯逻辑:**「会话没了」怎么判**(契约 v0.63 的 `404 session_not_found`)
|
||||
* 与 **正向增量循环什么时候停**(契约 §1 行 6 的 `after`)。
|
||||
*
|
||||
* 两者都是本次新增的判据,所以都从 `ChatViewModel` 里拎出来做成可单测的纯函数:
|
||||
* 真要跑起那个 VM 得搭一整套 `AppContainer`,而这里要证的只是两句话。
|
||||
*/
|
||||
class ChatRecoveryTest {
|
||||
|
||||
// —— 一、「这个会话在服务端已经不存在了」——
|
||||
|
||||
@Test
|
||||
fun `the v0 63 code marks the session as gone`() {
|
||||
val error = CyreneException(
|
||||
CyreneError.Server(
|
||||
code = "session_not_found",
|
||||
message = """session "web:gone" not found""",
|
||||
httpStatus = 404,
|
||||
),
|
||||
)
|
||||
|
||||
assertTrue(error.isSessionGone())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a 404 without an error envelope still counts`() {
|
||||
// 老部署 / 代理兜底时可能没有 `error` 信封(code 是空串):`/messages` 上的 404 只有
|
||||
// 一个含义,不能因为少一个字段就把它读成「空会话」——那正是主人那次「图凭空消失」。
|
||||
assertTrue(CyreneException(CyreneError.Server("", "", 404)).isSessionGone())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the code wins over the status`() {
|
||||
// 码是机器可读的那一份,状态码只是兜底:有码就按码判。
|
||||
assertTrue(CyreneException(CyreneError.Server("session_not_found", "", 410)).isSessionGone())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `other failures are not a deleted session`() {
|
||||
assertFalse(CyreneException(CyreneError.Server("bad_request", "invalid limit", 400)).isSessionGone())
|
||||
assertFalse(CyreneException(CyreneError.Server("internal", "boom", 500)).isSessionGone())
|
||||
assertFalse(CyreneException(CyreneError.Server("insufficient_quota", "", 402)).isSessionGone())
|
||||
assertFalse(CyreneException(CyreneError.Unauthorized("", "")).isSessionGone())
|
||||
assertFalse(CyreneException(CyreneError.Local(CyreneError.LocalKind.Timeout)).isSessionGone())
|
||||
assertFalse(RuntimeException("boom").isSessionGone())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `any 404 from the messages endpoint counts as gone`() {
|
||||
// `/messages` 上的 404 只有「会话不存在」这一个来源(消息端点 `/messages/{id}` 走的是
|
||||
// 另一条路径),所以码对不上也照判:宁可如实说「这个会话没了」,也不要把它渲染成
|
||||
// 一条空会话——那正是主人那次「图片凭空消失」。
|
||||
assertTrue(CyreneException(CyreneError.Server("not_found", "会话不存在", 404)).isSessionGone())
|
||||
}
|
||||
|
||||
// —— 二、正向增量循环 ——
|
||||
|
||||
@Test
|
||||
fun `it keeps asking with the newest cursor until a short page`() = runBlocking {
|
||||
val cursors = mutableListOf<Long>()
|
||||
val pages = listOf(
|
||||
page(listOf("m1" to 100L, "m2" to 200L), hasMore = true),
|
||||
page(listOf("m3" to 300L), hasMore = false),
|
||||
)
|
||||
|
||||
val taken = loadForwardPages(from = 50L) { cursor ->
|
||||
cursors += cursor
|
||||
Result.success(pages[cursors.size - 1])
|
||||
}
|
||||
|
||||
// 游标取的是**最新**那条(升序口径),不是最旧的。
|
||||
assertEquals(listOf(50L, 200L), cursors)
|
||||
assertEquals(2, taken.size)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `an empty page ends the walk`() = runBlocking {
|
||||
var calls = 0
|
||||
|
||||
val taken = loadForwardPages(from = 100L) {
|
||||
calls++
|
||||
Result.success(page(emptyList(), hasMore = false))
|
||||
}
|
||||
|
||||
assertEquals(1, calls)
|
||||
assertTrue(taken.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a cursor that cannot advance stops instead of looping forever`() = runBlocking {
|
||||
// 同毫秒并列且超过一页:契约 §A5 明写的已知边界——纯 createdAt 游标推不动。
|
||||
// 不做这道闸,这里会拿同一批消息无限请求下去。
|
||||
var calls = 0
|
||||
|
||||
val taken = loadForwardPages(from = 100L) {
|
||||
calls++
|
||||
Result.success(page(listOf("m1" to 100L, "m2" to 100L), hasMore = true))
|
||||
}
|
||||
|
||||
assertEquals(1, calls)
|
||||
assertEquals(1, taken.size)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `the round cap bounds one catch-up`() = runBlocking {
|
||||
var calls = 0
|
||||
|
||||
val taken = loadForwardPages(from = 0L, maxRounds = 3) { cursor ->
|
||||
calls++
|
||||
Result.success(page(listOf("m$calls" to cursor + 1), hasMore = true))
|
||||
}
|
||||
|
||||
assertEquals(3, calls)
|
||||
assertEquals(3, taken.size)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `a failure keeps what was already fetched and stops`() = runBlocking {
|
||||
var calls = 0
|
||||
|
||||
val taken = loadForwardPages(from = 0L) {
|
||||
calls++
|
||||
if (calls == 1) {
|
||||
Result.success(page(listOf("m1" to 100L), hasMore = true))
|
||||
} else {
|
||||
Result.failure(CyreneException(CyreneError.Local(CyreneError.LocalKind.Timeout)))
|
||||
}
|
||||
}
|
||||
|
||||
assertEquals(2, calls)
|
||||
assertEquals(listOf("m1"), taken.single().messages.map { it.id })
|
||||
}
|
||||
|
||||
private fun page(entries: List<Pair<String, Long>>, hasMore: Boolean): PagedMessages {
|
||||
val messages = entries.map { (id, createdAt) ->
|
||||
HistoryMessageDto(
|
||||
id = id,
|
||||
sessionId = "web",
|
||||
role = "user",
|
||||
content = "",
|
||||
createdAt = createdAt,
|
||||
)
|
||||
}
|
||||
return PagedMessages(
|
||||
messages = messages,
|
||||
hasMore = hasMore,
|
||||
oldestCreatedAt = messages.minOfOrNull { it.createdAt },
|
||||
newestCreatedAt = messages.maxOfOrNull { it.createdAt },
|
||||
)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user