diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/repository/MessageRepository.kt b/app/src/main/kotlin/top/yeij/cyrene/data/repository/MessageRepository.kt index b14cc22..4334439 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/repository/MessageRepository.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/repository/MessageRepository.kt @@ -16,7 +16,8 @@ import kotlinx.coroutines.withContext * 三条读取出口共用同一个端点，只是给的查询参数不同： * - [loadPage] / [loadLatest]：`before` 往回翻（服务端**倒序**返回）； * - [search]：`q` 会话内搜索（也是倒序）； - * - [loadAfter]：`after` 往前追（服务端**升序**返回，契约 §1 行 6）。 + * - [loadAfter]：`after` 往前追（服务端**升序**返回，契约 §1 行 6）；[loadIncrement] 是它在 + * `after` 方向上的翻页版（一直追到尾巴，追不到就如实说 `complete = false`）。 * * 三者都统一转成**升序**的 [PagedMessages] 再交出去。去重 / 合并请用 [mergeMessages]。 */ @@ -100,6 +101,54 @@ class MessageRepository( .map { raw -> toPagedMessages(raw, pageSize) } } + /** + * **增量追尾**（W77）：从 [after] 起把某会话的新消息一直翻到尾巴，最多 [maxPages] 页。 + * + * 与裸的 [loadAfter] 的差别只有一件：**它负责把「还没翻完」这件事如实说出来**。`after` 方向的 + * 游标是本页**最新**一条的 `createdAt`，同毫秒并列超过一页时纯 `createdAt` 游标推不动 + * （契约 §A5 明说的已知边界），再翻也只会拿到同一批行——那种情况返回 `complete = false`， + * 由调用方决定怎么办（未读面上是**回落到整表**），这里**不猜**、也不悄悄少给几条。 + * + * 不传 `platform`：未读面上的口径是服务端 `count(created_at > read_at)`（`unreadCountQuery`， + * 不带平台过滤），按平台过滤会让本地数出来的条数**与服务端的 `unreadCount` 不同源**。 + * + * @param after 本端**已见**的最大 `createdAt`（毫秒）；服务端返回严格新于它的消息，升序。 + * @param maxPages 最多翻几页（默认 [MAX_INCREMENT_PAGES]）。用尽即 `complete = false`： + * 一次静默刷新不该为了补一个角标翻十几页。 + */ + suspend fun loadIncrement( + sessionId: String, + after: Long, + pageSize: Int = DEFAULT_PAGE_SIZE, + maxPages: Int = MAX_INCREMENT_PAGES, + platform: String? = null, + ): Result = withContext(dispatcher) { + val size = pageSize.coerceIn(1, MAX_PAGE_SIZE) + val byId = LinkedHashMap() + var cursor = after + + fun done(at: Long, complete: Boolean) = Result.success( + MessageIncrement(messages = byId.values.sortedWith(MessageOrder), cursor = at, complete = complete), + ) + + var page = 0 + while (page < maxPages.coerceAtLeast(1)) { + page++ + val raw = apiCall { api.listMessages(sessionId, size, before = null, platform = platform, after = cursor) } + .getOrElse { return@withContext Result.failure(it) } + val paged = toPagedMessages(raw, size) + // 空页 = 真的追到了尾巴。`hasMore` 只是「拿到满页」的保守判断，条数正好是页宽的整数倍时 + // 会多问一次，那一次拿到的就是这一页空的。 + if (paged.messages.isEmpty()) return@withContext done(cursor, complete = true) + paged.messages.forEach { byId[it.id] = it } + val newest = paged.newestCreatedAt ?: return@withContext done(cursor, complete = false) + if (!paged.hasMore) return@withContext done(newest, complete = true) + if (newest <= cursor) return@withContext done(newest, complete = false) + cursor = newest + } + done(cursor, complete = false) + } + /** * 编辑单条消息（D3，契约 §1 行 6a）：`PATCH /api/v1/messages/{id}`，成功返回**服务端更新后的 * 完整消息 DTO**。 @@ -152,6 +201,15 @@ class MessageRepository( /** 搜索结果一页的条数：命中项要一条条读，比历史分页小一些更好翻。 */ const val DEFAULT_SEARCH_PAGE_SIZE = 30 + /** + * [loadIncrement] 一次最多翻几页（500 条）。 + * + * 这个上限只为一件事存在：一次**后台**的顺手对齐不该翻十几页。越过它就说明积压已经多到 + * 「增量」不再便宜，那时整表（`GET /sessions` 一次拿回权威的 `unreadCount` 与预览） + * 反而更划算——调用方据此回落，见 `SessionsViewModel.flushPendingIncrements`。 + */ + const val MAX_INCREMENT_PAGES = 10 + private val EMPTY_PAGE = PagedMessages( messages = emptyList(), hasMore = false, @@ -190,3 +248,18 @@ private fun toPagedMessages( newestCreatedAt = ascending.lastOrNull()?.createdAt, ) } + +/** + * [MessageRepository.loadIncrement] 的结论：升序增量消息 + 推进后的游标 + 有没有真的追到尾巴。 + * + * [complete] 为 `false` 时 [messages] 仍是**已拿到的全部**（不丢），但调用方不能把它当成 + * 「这之后就没有了」——未读面上那种情况要回落到整表，见 `SessionsViewModel.flushPendingIncrements`。 + * + * 放在这里而不是 `PagedMessages.kt`：`PagedMessages` 是纯分页形状（一页的元数据）， + * 这个类型是 [MessageRepository] 翻页循环的产物，跟着它的生产者走。 + */ +data class MessageIncrement( + val messages: List, + val cursor: Long, + val complete: Boolean, +) diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/InactiveSessionPull.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/InactiveSessionPull.kt index 343021d..1f95dbb 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/InactiveSessionPull.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/InactiveSessionPull.kt @@ -1,5 +1,7 @@ package top.yeij.cyrene.ui.sessions +import top.yeij.cyrene.data.repository.MessageIncrement + /** * 一帧 `message_persisted` 属于哪条路（[persistedRoute] 的取值）。 * @@ -18,11 +20,15 @@ internal enum class PersistedRoute { /** * **别的**会话落了新消息：它攒未读是正确行为（红点该亮），而本端**数不出这个数** - * （`message_persisted` 帧不带 `unreadCount`）⇒ 节流地静默重拉一次列表，问服务端要那个数。 + * （`message_persisted` 帧不带 `unreadCount`）⇒ 节流地问服务端要。 * * 这正是 W49 修的那一幕：**App 停在主会话时，次级会话来了新消息，抽屉红点不出现**。 * 此前帧在这一档直接 `return`，而列表更新的触发点只有抽屉打开 / 新建 / 删除 / init 四处， - * 于是红点必须靠用户开一次抽屉（那次静默重拉）才补上。 + * 于是红点必须靠用户开一次抽屉才补上。 + * + * 「要」的是什么，W77 换过一次：原来是整表 `GET /sessions`，现在是**这条会话的消息增量** + * `GET /messages?after=`（见 [incrementCursors] / [applyIncrement]）；整表退回成「增量给不出 + * 可信答案」时的回落。判据本身不受影响——这一档该不该被窗口挡住，与问的是哪种口径无关。 */ InactiveSession, } @@ -190,3 +196,105 @@ internal fun silentRefreshStep( */ internal fun throttleAfterPull(nowMillis: Long): SilentRefreshThrottle = SilentRefreshThrottle(lastPullAt = nowMillis) + +// --------------------------------------------------------------------------------------------- +// W77：非激活会话的静默刷新，从「整表重拉 GET /sessions」换成「前向增量 GET /messages?after=」 +// +// 后端明确批了这一换（回执二十八第 3 条：`?after=` 毫秒 `createdAt` 早就在册）。换掉的理由是**量**： +// 整表一次要回**所有**会话（每条一行预览 + 未读计数），而这里只需要回**真的变了的那一个**会话的新消息。 +// 下面三个纯函数就是这一换的全部判据，放在这里而不是 VM 里，理由与 [silentRefreshStep] 一样： +// 纯函数才穷举得了、才不用在 JVM 上摆一台假服务器。 +// --------------------------------------------------------------------------------------------- + +/** + * 某个会话现在该从哪个 `createdAt` 起问增量（纯函数）；`null` = **这一行问不了**。 + * + * 游标就取 [SessionItem.lastMessageAt]——最近一次权威 `GET /sessions` 给的、本端已知最新一条的 + * `createdAt`——**不另立账本**： + * - 服务端的 `after` 是 `created_at > ?`（**严格**大于），而 [SessionItem.lastMessageAt] 正是 + * 「已经见过的最新一条」⇒ 两边语义天然对齐，多立一个账本只会多一个会漂移的真相源； + * - 增量自己会把 [SessionItem.lastMessageAt] 推到新游标（[applyIncrement]），于是账本自洽地滚动， + * 不需要第二处推进。 + * + * 两种「问不了」都落到 `null`，它们其实是同一件事——**本端不知道这条会话看到哪儿了**： + * - 行不在列表里（会话在别处新建、或本端那份列表还没拉到它）⇒ 没有游标可用； + * - 行在但 `lastMessageAt` 为 `null`（本端认为它一条消息都没有，却收到了它的落库广播）⇒ + * 手里那份列表已经陈旧到不可信。 + * + * 两种都由调用方回落到整表（唯一权威），见 `SessionsViewModel.flushPendingIncrements`。 + */ +internal fun incrementCursorFor(item: SessionItem?): Long? = item?.lastMessageAt + +/** + * 一批待问增量的会话 id → 各自的 `after` 游标（纯函数）。 + * + * `null` = **这批里有一个问不了**（[incrementCursorFor] 给了 `null`）⇒ 调用方**整批**回落整表。 + * 不混着来是刻意的：一半靠增量、一半靠整表，得到的那份列表没有任何一处能说得清「它是什么时刻的」， + * 而未读面上最不能出的就是这种「说不清」。 + * + * 返回 `emptyMap` = 这批本来就是空的（没有事要做），**不是**失败。 + */ +internal fun incrementCursors( + sessionIds: Collection, + sessions: List, +): Map? { + if (sessionIds.isEmpty()) return emptyMap() + val byId = sessions.associateBy { it.id } + val cursors = LinkedHashMap(sessionIds.size) + for (id in sessionIds) { + cursors[id] = incrementCursorFor(byId[id]) ?: return null + } + return cursors +} + +/** + * 把一次增量折进列表（纯函数）：**只动那一行**，其余原样返回。 + * + * ## 未读数怎么算：先过已读游标，再加 + * 增量条数**不是**直接往角标上加的，要先按本地已知的已读游标 [readAt]（`readCursors[sessionId]`）筛一遍。 + * 服务端的口径是 `count(created_at > read_at)`（`session_read.go` 的 `unreadCountQuery`），而 `read_at` + * 是**账号级单一游标**，可能比本端已知的最新消息还新——用户在这台机器上点开过它（[SessionsViewModel.select] + * 推的就是「现在」），随后别的设备又发了消息。那一刻 `(lastMessageAt, readAt]` 之间的消息**服务端算已读**， + * 一律加进角标就会多出来一个红点。所以只数 `createdAt > readAt` 的那几条。 + * + * 已有的 [SessionItem.unreadCount] 直接保留相加是安全的：它来自服务端，而这次增量里的每一条都严格新于 + * [from]（服务端的 `after` 就是 `created_at > ?`），两者没有交集，不会重复计数。 + * + * ## 并发的那一份真相 + * [from] 不是「这次用了什么游标」的装饰：行上的 [SessionItem.lastMessageAt] **已经不等于** [from]， + * 就说明在这次请求飞在天上的时候，另一份更权威的结果（一次整表刷新、或另一次增量）已经推进过这一行。 + * 那些条数已经被别人数过一遍了，再加一次就是重复计数 ⇒ 整行不动。 + * + * 一行都没被改动时返回**同一个列表实例**（与 `applyBusyFrame` 同一习惯）：这条路径跑在后台， + * 没有必要为了「什么都没变」让整个抽屉重组一遍。 + * + * @param from 这次请求用的 `after` 游标。 + * @param readAt 本端已知的已读游标；`null` = 不知道（当作 `0`，即「一条都还没读过」）。 + * @param increment 升序的增量（`MessageRepository.loadIncrement` 的产物）。 + */ +internal fun List.applyIncrement( + sessionId: String, + from: Long, + readAt: Long?, + increment: MessageIncrement, +): List { + if (increment.messages.isEmpty()) return this + val floor = readAt ?: 0L + val unseen = increment.messages.count { it.createdAt > floor } + // 升序 ⇒ 最后一条就是最新的一条（`loadIncrement` 保证这一点）。 + val preview = increment.messages.last().content.toPreviewText().takeIf { it.isNotBlank() } + var touched = false + val updated = map { item -> + if (item.id != sessionId || item.lastMessageAt != from) { + item + } else { + touched = true + item.copy( + unreadCount = item.unreadCount + unseen, + lastMessage = preview, + lastMessageAt = increment.cursor, + ) + } + } + return if (touched) updated else this +} diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/SessionsViewModel.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/SessionsViewModel.kt index 5742cf4..80a3f66 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/SessionsViewModel.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/sessions/SessionsViewModel.kt @@ -82,9 +82,12 @@ data class SessionsUiState( * * **非激活会话攒未读**（别人发来、别的设备发来）本端**不自己数**——`message_persisted` 帧不带 * `unreadCount`，逐条累加会与编辑 / 删除 / 别端操作漂移。本端做的是**自己去问一次服务端**： - * 收到非激活会话的落库广播就**节流地静默重拉**（[requestSilentRefresh]，窗口 - * [SILENT_REFRESH_WINDOW_MILLIS]），于是抽屉的红点**不需要用户先开一次抽屉**就会出现 - * （S6，登记见 `devdocs/05-开发进度与交接.md`）。 + * 收到非激活会话的落库广播就**节流地问一次那条会话的消息增量**（[requestSilentRefresh] → + * [flushPendingIncrements]，窗口 [SILENT_REFRESH_WINDOW_MILLIS]），于是抽屉的红点**不需要用户先开一次 + * 抽屉**就会出现（S6，登记见 `devdocs/05-开发进度与交接.md`）。 + * + * 取数走的是 `GET /messages?after=`（前向增量）而不是整表 `GET /sessions`——W77 换的，理由与回落 + * 规则都写在 [flushPendingIncrements] 上。整表仍是**唯一权威**，只是不再需要为一条新消息付它的价钱。 * * ## WS 连接：全 App 共享一条，而**本类是它唯一的主人** * `session_read` / `message_persisted` 都在 WS 上，这个连接就是**进程内共享**的那一条 @@ -136,6 +139,17 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { */ private var trailingRefreshJob: Job? = null + /** + * 窗口里攒下的「要问增量的会话 id」（W77）。 + * + * 为什么是一个**集合**而不是一个 id：节流窗口把同一段时间里的落库广播合并成一次拉取，而那几个 + * 会话**未必是同一个**（主会话说一句、另一个会话也说一句）。只记住最后一条的话，前面那个会话的 + * 红点就永远不会出现——那正是 W49 修过的同一类病，只是换了个成因。 + * + * 只在 [requestSilentRefresh] / [flushPendingIncrements] 两处写；登出时清空（它属于上一个身份）。 + */ + private val pendingIncrements = mutableSetOf() + /** * 「谁在跑」的账本（S9）：会话 id → 本端见过的、还没收到收尾帧的在途路 id。 * @@ -201,6 +215,9 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { trailingRefreshJob?.cancel() trailingRefreshJob = null silentRefreshThrottle = SilentRefreshThrottle() + // 攒下的会话 id 同理：它们是上一个身份的会话，重登后那些 id 要么不存在、 + // 要么属于别人；留着只会让重登后的第一次窗口拿旧 id 去问增量。 + pendingIncrements.clear() pulledForIdentity = null busyTurns = emptyMap() _state.value = SessionsUiState() @@ -316,42 +333,54 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { } /** - * 节流地静默重拉一次列表 —— W49 修的「停在主会话时，次级会话来了新消息，抽屉红点不出现， + * 节流地静默对齐一次 —— W49 修的「停在主会话时，次级会话来了新消息，抽屉红点不出现， * 必须开一次抽屉才补上」（S6）。 * * 列表此前只在抽屉打开 / 新建 / 删除 / init 四处更新，而 `message_persisted` 只处理 * 「正在看的那个会话」—— 别的会话攒未读**没有任何人会去发现**。 * - * ## 为什么是「重拉」而不是「本地把未读数加一」 - * `message_persisted` 帧**不带 `unreadCount`**：本地没有逐条时间线，自己数出来的数会与编辑 / + * ## 为什么是「问服务端」而不是「本地把未读数加一」 + * `message_persisted` 帧**不带 `unreadCount`**：本端没有逐条时间线，靠帧自己数出来的数会与编辑 / * 删除 / 别端操作漂移（这正是 [onMessagePersisted] 里那条纪律的由来）。所以这里**一个数都不造**， - * 只是把「服务端才是唯一真源」这件事提前做一次 —— 走的是**与抽屉打开时那次完全相同**的路径 - * （[refresh]，含已读游标遮罩），只是不需要用户先开一次抽屉。 + * 只是把「服务端才是唯一真源」这件事提前做一次。 + * + * ## W77：问的是**那一条会话的增量**，不再是整表 + * 曾经的落点是 `refresh(silent = true)`（整表 `GET /sessions`），一次要为**所有**会话付一行预览 + + * 未读计数的价钱，而真正变了的只有一个会话。现在落到 [flushPendingIncrements]： + * `GET /messages?after=<已见最大 createdAt>`，只回那几条新消息。整表仍是唯一权威，但只在 + * 「增量给不出可信答案」时才请它出场（回落规则写在 [flushPendingIncrements] 上）。 * * ## 三条纪律 * 1. **节流**（判据 [silentRefreshStep]，窗口 [SILENT_REFRESH_WINDOW_MILLIS]）：一次连发只发一次 * 请求；同时保证**最后一条之后还会再拉一次**（trailing），红点不会停在旧值上。 - * 2. **静默**：`silent = true`，不闪「同步中…」——用户根本没开抽屉，不该有任何东西在动。 - * 失败时做什么**完全沿用既有行为**（挂 [SessionsUiState.error] 那一行，只有打开抽屉才看得见； - * 以及「网络异常，显示 x 分钟前的结果」那一句），本方法不另加提示：一次后台的顺手对齐失败， - * 不该比用户自己打开抽屉时的那次刷新更吵。 - * 3. **不抢跑**：不改 `activeSessionId`、不推进已读游标、不碰 [select]。[refresh] 内部唯一会动 - * `activeSessionId` 的地方是 [resolveActiveSession]（当前会话被别处删掉时收敛回主会话）， - * 那是既有的、正确的收敛，不是重置。 + * 2. **静默**：不闪「同步中…」——用户根本没开抽屉，不该有任何东西在动。失败时做什么**完全沿用既有 + * 行为**（挂 [SessionsUiState.error] 那一行，只有打开抽屉才看得见；以及「网络异常，显示 x 分钟前 + * 的结果」那一句），本方法不另加提示：一次后台的顺手对齐失败，不该比用户自己打开抽屉时的那次 + * 刷新更吵。 + * 3. **不抢跑**：不改 `activeSessionId`、不推进已读游标、不碰 [select]。回落那条路（[refresh]） + * 内部唯一会动 `activeSessionId` 的地方是 [resolveActiveSession]（当前会话被别处删掉时收敛回 + * 主会话），那是既有的、正确的收敛，不是重置。 * * @param cause 这次为什么要拉；决定它走不走窗口（见 [SilentRefreshCause]）。 + * @param sessionId 这一帧说的是**哪个**会话；`null` = 没有具体会话（[SilentRefreshCause.ContextChanged]）， + * 那一档只能走整表。窗口里攒下的 id 由 [pendingIncrements] 记着，因为一次窗口可能合并**几个不同**会话。 */ - private fun requestSilentRefresh(cause: SilentRefreshCause) { + private fun requestSilentRefresh(cause: SilentRefreshCause, sessionId: String? = null) { val step = silentRefreshStep(cause, System.currentTimeMillis(), silentRefreshThrottle) silentRefreshThrottle = step.throttle + // 没有具体会话时不留 id：那一次整表会把所有会话都对上，攒着的 id 一个也不缺。 + if (sessionId != null) pendingIncrements += sessionId when (step.action) { // 该拉了。先把**等在半路**的那次补拉撤掉：它要做的就是同一件事，留着会多拉一次。 // （只有 ContextChanged 压过窗口时才可能真的取消到一个活着的补拉任务——窗口外那一档 // 走到这里时 trailingPending 必为 false，补拉早跑完了。） + // + // 撤掉补拉**不等于**可以丢掉它攒下的 id：[pendingIncrements] 是独立的一份账本， + // 这一趟会把它们一并做掉。 SilentRefreshAction.PullNow -> { trailingRefreshJob?.cancel() trailingRefreshJob = null - refresh(silent = true) + flushPendingIncrements(fullTable = sessionId == null) } // 窗口内：排一次补拉。它是「最后一条之后还会再拉一次」的**唯一**保证——没有它，一串 @@ -365,15 +394,74 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { trailingRefreshJob = viewModelScope.launch { delay(step.delayMillis) silentRefreshThrottle = throttleAfterPull(System.currentTimeMillis()) - refresh(silent = true) + flushPendingIncrements(fullTable = false) } } // 窗口内、补拉已经排上：那一次本来就是把窗口里所有成因合起来做掉，这里不必再排。 + // （但这一帧的会话 id 已经在上面记进 [pendingIncrements] 了 —— 补拉不会漏掉它。） SilentRefreshAction.Ignore -> Unit } } + /** + * 把 [pendingIncrements] 里攒下的会话**逐个问一次消息增量**，问不了就整批回落整表（W77）。 + * + * 这就是 W49 那条「非激活会话落了新消息 ⇒ 整表重拉 `GET /sessions`」的替代品：整表一次要回 + * **所有**会话（每条一行预览 + 未读计数），而这里只回**真的变了的那一个**会话的新消息。 + * + * ## 回落的三条理由（都落到同一条路：整表） + * 1. **问不了增量**：[incrementCursors] 给 `null` —— 行不在列表里，或本端不知道它看到哪儿了； + * 2. **追不到尾**：服务端说有更多却推不动（`complete = false`：同毫秒并列超过一页、或翻满了 + * `MessageRepository.MAX_INCREMENT_PAGES`）—— 那时增量只覆盖了积压的一部分，拿它当「全部」 + * 就是把红点做小； + * 3. **请求失败**：含会话已在别处被删掉的 `404` —— 那一档要靠整表把这一行从列表里去掉。 + * + * 三条都是「增量给不出可信答案」⇒ 去问唯一权威。**一次批次最多回落一次**：整批共用这一个失败 + * 出口，否则 N 个会话同时失败就是 N 次整表。 + * + * ## 刻意不做的一件事：不写快照 + * [refresh] 成功时会 `prefs.saveSnapshot`，这里**不写**。快照的类型是 `List` + * （整表响应），而增量只回了消息——拿它去拼一份 DTO 列表，缺的字段（`type` / `createdAt` / + * `pinned` / `personaId`）只能靠猜，那等于用一个更差的东西覆盖掉一份好快照。代价是冷启动第一帧 + * 可能显示上一次整表时的未读与预览，紧接着 `init` 里那次 `refresh()` 就会把它修正。 + * + * ## 也不顺带刷角色目录 + * [refresh] 会捎带 `loadPersonas()`，这条路径不会。角色目录的刷新点是 init / 新建 / 删除 / 抽屉 + * 打开（那几个都是用户动作或一次性事件），「别的会话来了条消息」从来不是它的刷新时机——此前它只是 + * 跟着整表搭了便车。W77 把便车取消，角色目录少掉的是几次没人要的请求。 + * + * @param fullTable `true` = 这一趟没有具体会话（或已判明必须整表），直接走 [refresh]。 + */ + private fun flushPendingIncrements(fullTable: Boolean) { + val targets = pendingIncrements.toList() + pendingIncrements.clear() + if (fullTable) { + refresh(silent = true) + return + } + val cursors = incrementCursors(targets, _state.value.sessions) + if (cursors == null) { + refresh(silent = true) + return + } + // 空批次 = 没有事要做（不是失败）：不回落、也不发请求。 + if (cursors.isEmpty()) return + viewModelScope.launch { + for ((sessionId, from) in cursors) { + val increment = container.messageRepository.loadIncrement(sessionId, after = from).getOrNull() + if (increment == null || !increment.complete) { + refresh(silent = true) + return@launch + } + val readAt = readCursors[sessionId] + _state.update { + it.copy(sessions = it.sessions.applyIncrement(sessionId, from, readAt, increment)) + } + } + } + } + /** * 切换 / 进入会话。 * @@ -594,23 +682,24 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { * 两半共用同一条纪律：**本端哪里都不替服务端数未读**（逐条累加会与编辑 / 删除 / 别端操作漂移）。 * 区别只在「那它靠什么显示」： * - [PersistedRoute.ActiveSession] ⇒ 就地清零 + 推进游标（这段一个字没改）； - * - [PersistedRoute.InactiveSession] ⇒ **节流地静默重拉**（[requestSilentRefresh]）， - * 用服务端下发的 `unreadCount` 说话。这正是 W49 修的那一幕：此前这一档直接 `return`，于是 - * 「App 停在主会话、次级会话来了新消息」没有任何人会去发现，红点要等用户打开一次抽屉（那次 - * 静默重拉）才补上。 + * - [PersistedRoute.InactiveSession] ⇒ **节流地问一次这条会话的消息增量**（[requestSilentRefresh] + * 带上 `sessionId` → [flushPendingIncrements] → `MessageRepository.loadIncrement`），未读由 + * 「增量里越过已读游标的条数」说话（W77；W49 那一档原本整表重拉 `GET /sessions`）。 + * 这正是 W49 修的那一幕：此前这一档直接 `return`，于是「App 停在主会话、次级会话来了新消息」 + * 没有任何人会去发现，红点要等用户打开一次抽屉才补上。 * * 两档的**载荷要求不同**，是刻意的：推游标要用消息自己的 `createdAt`（缺了就只能什么都不做）， - * 而「重拉一次列表」不需要载荷里的任何一个字段 —— **帧壳本身就是全部证据**（服务端只为「某条 + * 而「问一次增量」不需要载荷里的任何一个字段 —— **帧壳本身就是全部证据**（服务端只为「某条 * 消息已在某会话落库」才推这一帧）。所以非激活那一档**不检查 `event.message`**：真遇上载荷解不出的 * 坏帧（契约漂移的典型症状），宁可多问一次服务端，也不要让红点又停在旧值上。 */ private fun onMessagePersisted(event: WsEvent.MessagePersisted) { val sessionId = event.sessionId?.takeIf { it.isNotBlank() } ?: return when (persistedRoute(sessionId, _state.value.activeSessionId)) { - // 别的会话：本端数不出它的未读（帧不带 `unreadCount`）⇒ 节流地静默重拉一次列表， - // 红点于是在用户**没开抽屉**的时候也会出现。这一帧到此为止。 + // 别的会话：本端数不出它的未读（帧不带 `unreadCount`）⇒ 节流地问一次**这条会话**的消息 + // 增量，红点于是在用户**没开抽屉**的时候也会出现。这一帧到此为止。 PersistedRoute.InactiveSession -> - requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage) + requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage, sessionId) // 屏幕上就是这个会话：就地清零 + 推进游标，**不发任何请求去拉列表**。 // 这里用 `when` 穷举而不是「不满足上面那条就走这段」：路由将来多一档时， @@ -634,9 +723,9 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { * 判据也共用 [persistedRoute]（只是会话 id 由这一帧自己钉：它不带 `sessionId`）。 * * 为什么必须在这里补一句：服务端的主动推送 Web 路由只发这一帧，**不补** `message_persisted` - * （那条只有 IM 路由会广播），所以 W49 那条「非激活会话落新消息 ⇒ 节流静默重拉」根本不会被 + * （那条只有 IM 路由会广播），所以 W49 那条「非激活会话落新消息 ⇒ 节流刷新」根本不会被 * 触发——抽屉里主会话的红点就停在她开口之前，直到用户自己开一次抽屉。这里复用**同一条**路 - * （[requestSilentRefresh] + 既有的 [SilentRefreshCause.InactiveSessionMessage]）， + * （[requestSilentRefresh] + 既有的 [SilentRefreshCause.InactiveSessionMessage]，会话 id 取主会话）， * 不新写取数路径，也不动节流判据本身。 * * `id` 缺席 = 服务端那一刻**没有落库行** ⇒ 历史里没有这一行，未读面上没有任何东西要更新 @@ -647,7 +736,7 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() { val sessionId = SessionSummaryDto.MAIN_SESSION_ID when (persistedRoute(sessionId, _state.value.activeSessionId)) { PersistedRoute.InactiveSession -> - requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage) + requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage, sessionId) // 屏幕上就是主会话：她这句话已经显示在聊天流里（ChatViewModel 那侧并入）⇒ 与 // message_persisted 的同一档一样「看到即已读」，免得抽屉为她刚说的话亮一个红点。 diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/repository/MessageIncrementRepositoryTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/repository/MessageIncrementRepositoryTest.kt new file mode 100644 index 0000000..1b08ede --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/repository/MessageIncrementRepositoryTest.kt @@ -0,0 +1,197 @@ +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.ConcurrentLinkedQueue +import java.util.concurrent.CopyOnWriteArrayList +import java.util.concurrent.Executors + +/** + * 增量追尾（W77 `MessageRepository.loadIncrement`）的**数据层**。 + * + * 与 `MessageAfterRepositoryTest` 同一套做法（真 OkHttp + 真 Retrofit + JDK 自带的 + * `com.sun.net.httpserver`，不引新依赖），只是假后端从「一份固定响应」换成**响应队列**： + * 追尾是一次循环，翻页语义只有在「第一页满、第二页空」这种**按次序**的响应下才验得出来。 + * + * 断言分两类，都不看实现细节： + * - **线上证据**：每次请求的原始 URI（`after=` 只能有一个值，且随游标前进）、请求次数； + * - **结论**：[MessageIncrement.complete] 与 `cursor` —— 这两项是调用方决定「要不要回落整表」的 + * 唯一依据，错了就是红点做小或做多。 + */ +class MessageIncrementRepositoryTest { + + private val backend = FakeMessagesBackend() + + @After + fun tearDown() { + backend.close() + } + + private fun repository() = MessageRepository( + ApiClientFactory.create(baseUrl = backend.baseUrl, service = SessionsApi::class.java), + ) + + @Test + fun `a short page is the tail and only the forward cursor goes on the wire`() = runBlocking { + backend.enqueue(200, """{"data":[${message("m1", 100)}, ${message("m2", 200)}]}""") + + val increment = repository().loadIncrement("web", after = 100).getOrThrow() + + val uri = backend.uris.single() + assertTrue(uri, uri.contains("after=100")) + assertTrue(uri, uri.contains("limit=50")) + assertFalse(uri, uri.contains("before=")) + assertFalse(uri, uri.contains("q=")) + // 不满一页 ⇒ 真的追到尾巴了，游标停在最新一条。 + assertTrue(increment.complete) + assertEquals(200L, increment.cursor) + assertEquals(listOf("m1", "m2"), increment.messages.map { it.id }) + } + + @Test + fun `a full page is followed by one more request and the empty page ends the chase`() = runBlocking { + backend.enqueue(200, """{"data":[${message("m1", 100)}, ${message("m2", 200)}]}""") + backend.enqueue(200, """{"data":[]}""") + + val increment = repository().loadIncrement("web", after = 0, pageSize = 2).getOrThrow() + + // 满页 ⇒ 多问一次；那一次拿到空页，于是「追到了」而不是「翻不动了」。 + assertEquals(2, backend.uris.size) + assertTrue(backend.uris[1], backend.uris[1].contains("after=200")) + assertTrue(increment.complete) + assertEquals(200L, increment.cursor) + assertEquals(listOf("m1", "m2"), increment.messages.map { it.id }) + } + + @Test + fun `a full page stuck on one millisecond is reported as incomplete`() = runBlocking { + // 同毫秒并列且超过页宽：纯 `createdAt` 游标推不动，服务端只会把同一批行再给一次。 + val sameMillisecond = """{"data":[${message("m1", 500)}, ${message("m2", 500)}]}""" + backend.enqueue(200, sameMillisecond) + backend.enqueue(200, sameMillisecond) + + val increment = repository().loadIncrement("web", after = 100, pageSize = 2).getOrThrow() + + assertFalse(increment.complete) + // 游标没有越过那一毫秒（推不动就不装作推得动）。 + assertEquals(500L, increment.cursor) + // 同一批行被问了两次，按 id 去重后仍是两条 —— 不能因为翻了两页就把条数翻倍。 + assertEquals(listOf("m1", "m2"), increment.messages.map { it.id }) + } + + @Test + fun `the page budget stops the chase instead of chasing forever`() = runBlocking { + // 永远满页：预算只有 1 页时必须停下来并如实说「没追完」。 + backend.enqueue(200, """{"data":[${message("m1", 100)}, ${message("m2", 200)}]}""") + + val increment = repository() + .loadIncrement("web", after = 0, pageSize = 2, maxPages = 1) + .getOrThrow() + + assertEquals(1, backend.uris.size) + assertFalse(increment.complete) + assertEquals(200L, increment.cursor) + } + + @Test + fun `a session deleted elsewhere surfaces as session_not_found with 404`() = runBlocking { + backend.enqueue( + 404, + """{"error":{"code":"session_not_found","message":"session \"web:gone\" not found"}}""", + ) + + val error = repository().loadIncrement("web:gone", after = 0).exceptionOrNull() as CyreneException + + val server = error.error as CyreneError.Server + assertEquals("session_not_found", server.code) + assertEquals(404, server.httpStatus) + } + + @Test + fun `one increment costs one request and k messages instead of the whole table`() = runBlocking { + // 10 个会话的整表长什么样（每条一行预览 + 未读计数）：这是被换掉的那条路。 + val wholeTable = """{"data":[${sessions(10)}]}""" + backend.enqueue(200, """{"data":[${message("new", 9_000)}]}""") + + val increment = repository().loadIncrement("web", after = 8_000).getOrThrow() + + assertEquals(1, backend.uris.size) + assertEquals(1, increment.messages.size) + assertTrue(increment.complete) + // 量级证据：增量回的字节数远小于整表（这里只有一条新消息 vs 十行会话预览）。 + val incrementBytes = backend.responseBytes.single() + assertTrue( + "increment=$incrementBytes wholeTable=${wholeTable.toByteArray().size}", + incrementBytes * 4 < wholeTable.toByteArray().size, + ) + } + + /** 一条 `GET /messages` 形状的消息（字段与 `HistoryMessageDto` 对齐，只留断言用得到的）。 */ + private fun message(id: String, createdAt: Long): String = + """{"id":"$id","sessionId":"web","role":"user","content":"内容","createdAt":$createdAt}""" + + /** 一段 `GET /sessions` 形状的会话行（约 60 rune 预览 + 未读计数），只用于量级对比。 */ + private fun sessions(count: Int): String = (1..count).joinToString(",") { i -> + val preview = "这是第 $i 个会话的最后一条消息预览，长度大致与真实数据同量级，用来量字节数。" + """{"id":"s$i","type":"secondary","title":"会话 $i","createdAt":${i * 1000},""" + + """"pinned":false,"personaId":null,"unreadCount":$i,""" + + """"lastMessage":{"id":"p$i","sessionId":"s$i","role":"assistant",""" + + """"content":"$preview","createdAt":${i * 1000}}}""" + } + + /** + * 按**入队顺序**逐次回响应的假后端：追尾是一次循环，固定响应验不出翻页。 + * + * 队列见底时重复最后一份 —— 这样「永远满页」那种用例只需要入队一次。 + */ + 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}/" + + /** 每次请求的原始 URI（含 query），按到达顺序。 */ + val uris = CopyOnWriteArrayList() + + /** 每次响应体（UTF-8）的字节数，按到达顺序 —— 量化对比用。 */ + val responseBytes = CopyOnWriteArrayList() + + private val queue = ConcurrentLinkedQueue>() + + private val last = java.util.concurrent.atomic.AtomicReference(200 to """{"data":[]}""") + + fun enqueue(status: Int, body: String) { + queue += status to body + } + + private fun handle(exchange: HttpExchange) { + uris += exchange.requestURI.toString() + val (status, body) = queue.poll() ?: last.get() + last.set(status to body) + val bytes = body.toByteArray() + responseBytes += bytes.size + 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) + } + } +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/ui/sessions/SessionIncrementTest.kt b/app/src/test/kotlin/top/yeij/cyrene/ui/sessions/SessionIncrementTest.kt new file mode 100644 index 0000000..9e19314 --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/ui/sessions/SessionIncrementTest.kt @@ -0,0 +1,193 @@ +package top.yeij.cyrene.ui.sessions + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Test +import top.yeij.cyrene.data.api.dto.HistoryMessageDto +import top.yeij.cyrene.data.repository.MessageIncrement + +/** + * W77 的判据层：非激活会话的静默刷新从「整表重拉 `GET /sessions`」换成「`GET /messages?after=`」 + * 之后，**什么时候问得了增量、问回来的数怎么折进列表**。 + * + * 与 `InactiveSessionPullTest` 同一套理由：这三件都是纯函数，穷举得了，不用在 JVM 上摆一台假服务器 + * （那条路由 `MessageIncrementRepositoryTest` 从线上证据那一侧验）。VM 里剩下的只有「拼装」—— + * 取游标、逐个发请求、回落 —— 每一步的分支判据都在下面。 + */ +class SessionIncrementTest { + + private fun item( + id: String = "web", + lastMessageAt: Long? = null, + unreadCount: Int = 0, + lastMessage: String? = null, + ) = SessionItem( + id = id, + title = "标题", + isMain = id == "web", + pinned = false, + lastMessage = lastMessage, + lastMessageAt = lastMessageAt, + unreadCount = unreadCount, + ) + + private fun message(id: String, createdAt: Long, content: String = "内容") = HistoryMessageDto( + id = id, + sessionId = "web", + role = "assistant", + content = content, + createdAt = createdAt, + ) + + private fun increment( + messages: List, + cursor: Long = messages.lastOrNull()?.createdAt ?: 0L, + complete: Boolean = true, + ) = MessageIncrement(messages = messages, cursor = cursor, complete = complete) + + // —— 游标从哪来 —— + + @Test + fun `no row means no cursor`() { + // 会话在别处新建 / 本端那份列表还没拉到它：不知道它看到哪儿了，只能回落整表。 + assertNull(incrementCursorFor(null)) + } + + @Test + fun `a row without a last message is not a cursor either`() { + // 本端认为它一条消息都没有，却收到了它的落库广播 ⇒ 手里那份列表已经陈旧到不可信。 + assertNull(incrementCursorFor(item(lastMessageAt = null))) + } + + @Test + fun `the cursor is the last message time already on the row`() { + // 不另立账本：`after` 是严格大于，而 `lastMessageAt` 正是「已见的最新一条」。 + assertEquals(500L, incrementCursorFor(item(lastMessageAt = 500))) + } + + // —— 一批 id → 一批游标 —— + + @Test + fun `an empty batch is nothing to do, not a failure`() { + assertEquals(emptyMap(), incrementCursors(emptyList(), listOf(item(lastMessageAt = 1)))) + } + + @Test + fun `every id in the batch gets its own cursor`() { + val sessions = listOf(item(id = "web", lastMessageAt = 100), item(id = "s2", lastMessageAt = 200)) + + assertEquals(mapOf("web" to 100L, "s2" to 200L), incrementCursors(listOf("web", "s2"), sessions)) + } + + @Test + fun `an unknown id rolls the whole batch back to the table`() { + // 一半靠增量、一半靠整表得到的列表说不清是哪个时刻的 —— 未读面上最不能出这种「说不清」。 + val sessions = listOf(item(id = "web", lastMessageAt = 100)) + + assertNull(incrementCursors(listOf("web", "gone"), sessions)) + } + + @Test + fun `one row without a cursor rolls the whole batch back too`() { + val sessions = listOf(item(id = "web", lastMessageAt = 100), item(id = "s2", lastMessageAt = null)) + + assertNull(incrementCursors(listOf("web", "s2"), sessions)) + } + + // —— 折进列表 —— + + @Test + fun `an empty increment changes nothing and keeps the same instance`() { + val sessions = listOf(item(id = "web", lastMessageAt = 100)) + + assertSame(sessions, sessions.applyIncrement("web", from = 100, readAt = null, increment = increment(emptyList()))) + } + + @Test + fun `only the target row moves and the cursor advances to the increment tail`() { + val sessions = listOf( + item(id = "web", lastMessageAt = 100, unreadCount = 2, lastMessage = "旧预览"), + item(id = "s2", lastMessageAt = 900, unreadCount = 7, lastMessage = "别人的预览"), + ) + + val updated = sessions.applyIncrement( + "web", + from = 100, + readAt = null, + increment = increment(listOf(message("m1", 200), message("m2", 300, "她轻轻说"))), + ) + + val web = updated.first { it.id == "web" } + // 已有的 unreadCount 直接相加是安全的：增量的每一条都严格新于 from，两者没有交集。 + assertEquals(4, web.unreadCount) + assertEquals(300L, web.lastMessageAt) + // 预览取最新一条，且剥掉 `` 这类标记（与整表那条路同一个 `toPreviewText`）。 + assertEquals("她 轻轻 说", web.lastMessage) + // 别的行一个字节都不动。 + assertEquals(sessions[1], updated.first { it.id == "s2" }) + } + + @Test + fun `messages at or before the read cursor are already read on the server`() { + // 服务端口径是 `count(created_at > read_at)`，而 `read_at` 是账号级单一游标： + // 这台机器点开过它（游标推到「现在」）之后别的设备又发了消息，那段区间服务端算已读。 + val sessions = listOf(item(id = "web", lastMessageAt = 100)) + + val updated = sessions.applyIncrement( + "web", + from = 100, + readAt = 250, + increment = increment(listOf(message("m1", 200), message("m2", 250), message("m3", 300))), + ) + + // 只有 300 那一条越过游标。 + assertEquals(1, updated.single().unreadCount) + } + + @Test + fun `an unknown read cursor counts everything in the increment`() { + val sessions = listOf(item(id = "web", lastMessageAt = 100)) + + val updated = sessions.applyIncrement( + "web", + from = 100, + readAt = null, + increment = increment(listOf(message("m1", 200), message("m2", 300))), + ) + + assertEquals(2, updated.single().unreadCount) + } + + @Test + fun `a row already advanced by a more authoritative result is left alone`() { + // 请求飞在天上时，一次整表刷新（或另一次增量）已经推进过这一行 ⇒ 那些条数已经被数过一遍了， + // 再加一次就是重复计数。 + val sessions = listOf(item(id = "web", lastMessageAt = 500, unreadCount = 3)) + + val updated = sessions.applyIncrement( + "web", + from = 100, + readAt = null, + increment = increment(listOf(message("m1", 200))), + ) + + assertSame(sessions, updated) + } + + @Test + fun `a blank newest message clears the preview the same way the table would`() { + // 与 `toSessionItem`（整表那条路）同一个口径：内容折叠后为空 ⇒ 预览为 null，不留一句空话。 + val sessions = listOf(item(id = "web", lastMessageAt = 100, lastMessage = "旧预览")) + + val updated = sessions.applyIncrement( + "web", + from = 100, + readAt = null, + increment = increment(listOf(message("m1", 200, " "))), + ) + + assertNull(updated.single().lastMessage) + assertEquals(200L, updated.single().lastMessageAt) + } +}