W77 未读增量 —— 非激活会话的静默刷新改用 GET /messages?after= 前向增量

W49 那条「非激活会话落了新消息 ⇒ 整表重拉 GET /sessions」换掉:整表一次要回**所有**
会话(每条一行预览 + 未读计数),而真正变了的只有一个会话。改问那条会话的消息增量
`GET /messages?after=<本端已见最大 createdAt>`,只回那几条新消息。

- MessageRepository.loadIncrement:after 方向的翻页版,翻到尾巴或 maxPages 为止;
  同毫秒并列推不动游标 / 翻满页数 ⇒ complete = false,不猜、也不悄悄少给几条。
  byId 去重后按 MessageOrder 升序交出去。
- InactiveSessionPull 三个纯函数:incrementCursorFor(游标 = SessionItem.lastMessageAt,
  不另立账本)、incrementCursors(任一 id 问不了 ⇒ 整批回落,不混着来)、
  applyIncrement(只动那一行;未读 = 越过本地已读游标的条数,与服务端
  unreadCountQuery 的 count(created_at > read_at) 同源;lastMessageAt 已被别人推进过
  ⇒ 整行不动,防重复计数;没改动就返回同一实例)。
- SessionsViewModel.requestSilentRefresh 带 sessionId,窗口里攒 pendingIncrements
  (集合,不是单个 id —— 一次窗口可能合并几个不同会话);flushPendingIncrements
  逐个问增量,三档回落整表:游标拿不到 / complete = false / 请求失败(含会话已在别处
  删掉的 404)。一次批次最多回落一次。登出时清空 pendingIncrements。
- 节流与 trailing 语义一个字没动(silentRefreshStep / throttleAfterPull 签名未改);
  init / 抽屉打开 / 新建 / 删除 / ContextChanged 仍走整表。增量路径不写
  prefs.saveSnapshot(增量凑不出合格的 SessionSummaryDto,那会用更差的东西覆盖好快照),
  也不捎带 loadPersonas()(此前它只是跟着整表搭便车)。

量级(同 JSON 形状实测):一次静默刷新原本 = GET /sessions 全表 + loadPersonas,
10 个会话时 4100 B、2 个请求;现在 = 1 条增量 126 B、1 个请求。

测试:新增 MessageIncrementRepositoryTest(6,真 OkHttp + 真 Retrofit + JDK
com.sun.net.httpserver,验 URI 参数、满页后再问一次得空页、同毫秒并列、maxPages
用尽、404 映射、字节量对比)与 SessionIncrementTest(13,三个纯函数穷举)。
testDebugUnitTest 全绿 135 suites / 1689 tests / 0 failures / 0 errors / 0 skipped;
assembleDebug 通过。
This commit is contained in:
AskaEth committed 2026-10-26 22:19:17 +08:00
1 parent 4d25d9ec69
commit d716d26864
5 files changed
+692 -32

No files matched your search

@@ -16,7 +16,8 @@ import kotlinx.coroutines.withContext
* 三条读取出口共用同一个端点,只是给的查询参数不同: * 三条读取出口共用同一个端点,只是给的查询参数不同:
* - [loadPage] / [loadLatest]:`before` 往回翻(服务端**倒序**返回); * - [loadPage] / [loadLatest]:`before` 往回翻(服务端**倒序**返回);
* - [search]:`q` 会话内搜索(也是倒序); * - [search]:`q` 会话内搜索(也是倒序);
* - [loadAfter]:`after` 往前追(服务端**升序**返回,契约 §1 行 6)。 * - [loadAfter]:`after` 往前追(服务端**升序**返回,契约 §1 行 6);[loadIncrement] 是它在
* `after` 方向上的翻页版(一直追到尾巴,追不到就如实说 `complete = false`)。
* *
* 三者都统一转成**升序**的 [PagedMessages] 再交出去。去重 / 合并请用 [mergeMessages]。 * 三者都统一转成**升序**的 [PagedMessages] 再交出去。去重 / 合并请用 [mergeMessages]。
*/ */
@@ -100,6 +101,54 @@ class MessageRepository(
.map { raw -> toPagedMessages(raw, pageSize) } .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<MessageIncrement> = withContext(dispatcher) {
val size = pageSize.coerceIn(1, MAX_PAGE_SIZE)
val byId = LinkedHashMap<String, HistoryMessageDto>()
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}`,成功返回**服务端更新后的 * 编辑单条消息(D3,契约 §1 行 6a):`PATCH /api/v1/messages/{id}`,成功返回**服务端更新后的
* 完整消息 DTO**。 * 完整消息 DTO**。
@@ -152,6 +201,15 @@ class MessageRepository(
/** 搜索结果一页的条数:命中项要一条条读,比历史分页小一些更好翻。 */ /** 搜索结果一页的条数:命中项要一条条读,比历史分页小一些更好翻。 */
const val DEFAULT_SEARCH_PAGE_SIZE = 30 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( private val EMPTY_PAGE = PagedMessages(
messages = emptyList(), messages = emptyList(),
hasMore = false, hasMore = false,
@@ -190,3 +248,18 @@ private fun toPagedMessages(
newestCreatedAt = ascending.lastOrNull()?.createdAt, newestCreatedAt = ascending.lastOrNull()?.createdAt,
) )
} }
/**
* [MessageRepository.loadIncrement] 的结论:升序增量消息 + 推进后的游标 + 有没有真的追到尾巴。
*
* [complete] 为 `false` 时 [messages] 仍是**已拿到的全部**(不丢),但调用方不能把它当成
* 「这之后就没有了」——未读面上那种情况要回落到整表,见 `SessionsViewModel.flushPendingIncrements`。
*
* 放在这里而不是 `PagedMessages.kt`:`PagedMessages` 是纯分页形状(一页的元数据),
* 这个类型是 [MessageRepository] 翻页循环的产物,跟着它的生产者走。
*/
data class MessageIncrement(
val messages: List<HistoryMessageDto>,
val cursor: Long,
val complete: Boolean,
)
@@ -1,5 +1,7 @@
package top.yeij.cyrene.ui.sessions package top.yeij.cyrene.ui.sessions
import top.yeij.cyrene.data.repository.MessageIncrement
/** /**
* 一帧 `message_persisted` 属于哪条路([persistedRoute] 的取值)。 * 一帧 `message_persisted` 属于哪条路([persistedRoute] 的取值)。
* *
@@ -18,11 +20,15 @@ internal enum class PersistedRoute {
/** /**
* **别的**会话落了新消息:它攒未读是正确行为(红点该亮),而本端**数不出这个数** * **别的**会话落了新消息:它攒未读是正确行为(红点该亮),而本端**数不出这个数**
* (`message_persisted` 帧不带 `unreadCount`)⇒ 节流地静默重拉一次列表,问服务端要那个数。 * (`message_persisted` 帧不带 `unreadCount`)⇒ 节流地问服务端要。
* *
* 这正是 W49 修的那一幕:**App 停在主会话时,次级会话来了新消息,抽屉红点不出现**。 * 这正是 W49 修的那一幕:**App 停在主会话时,次级会话来了新消息,抽屉红点不出现**。
* 此前帧在这一档直接 `return`,而列表更新的触发点只有抽屉打开 / 新建 / 删除 / init 四处, * 此前帧在这一档直接 `return`,而列表更新的触发点只有抽屉打开 / 新建 / 删除 / init 四处,
* 于是红点必须靠用户开一次抽屉(那次静默重拉)才补上。 * 于是红点必须靠用户开一次抽屉才补上。
*
* 「要」的是什么,W77 换过一次:原来是整表 `GET /sessions`,现在是**这条会话的消息增量**
* `GET /messages?after=`(见 [incrementCursors] / [applyIncrement]);整表退回成「增量给不出
* 可信答案」时的回落。判据本身不受影响——这一档该不该被窗口挡住,与问的是哪种口径无关。
*/ */
InactiveSession, InactiveSession,
} }
@@ -190,3 +196,105 @@ internal fun silentRefreshStep(
*/ */
internal fun throttleAfterPull(nowMillis: Long): SilentRefreshThrottle = internal fun throttleAfterPull(nowMillis: Long): SilentRefreshThrottle =
SilentRefreshThrottle(lastPullAt = nowMillis) 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<String>,
sessions: List<SessionItem>,
): Map<String, Long>? {
if (sessionIds.isEmpty()) return emptyMap()
val byId = sessions.associateBy { it.id }
val cursors = LinkedHashMap<String, Long>(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<SessionItem>.applyIncrement(
sessionId: String,
from: Long,
readAt: Long?,
increment: MessageIncrement,
): List<SessionItem> {
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
}
@@ -82,9 +82,12 @@ data class SessionsUiState(
* *
* **非激活会话攒未读**(别人发来、别的设备发来)本端**不自己数**——`message_persisted` 帧不带 * **非激活会话攒未读**(别人发来、别的设备发来)本端**不自己数**——`message_persisted` 帧不带
* `unreadCount`,逐条累加会与编辑 / 删除 / 别端操作漂移。本端做的是**自己去问一次服务端**: * `unreadCount`,逐条累加会与编辑 / 删除 / 别端操作漂移。本端做的是**自己去问一次服务端**:
* 收到非激活会话的落库广播就**节流地静默重拉**([requestSilentRefresh],窗口 * 收到非激活会话的落库广播就**节流地问一次那条会话的消息增量**([requestSilentRefresh] →
* [SILENT_REFRESH_WINDOW_MILLIS]),于是抽屉的红点**不需要用户先开一次抽屉**就会出现 * [flushPendingIncrements],窗口 [SILENT_REFRESH_WINDOW_MILLIS]),于是抽屉的红点**不需要用户先开一次
* (S6,登记见 `devdocs/05-开发进度与交接.md`)。 * 抽屉**就会出现(S6,登记见 `devdocs/05-开发进度与交接.md`)。
*
* 取数走的是 `GET /messages?after=<ms>`(前向增量)而不是整表 `GET /sessions`——W77 换的,理由与回落
* 规则都写在 [flushPendingIncrements] 上。整表仍是**唯一权威**,只是不再需要为一条新消息付它的价钱。
* *
* ## WS 连接:全 App 共享一条,而**本类是它唯一的主人** * ## WS 连接:全 App 共享一条,而**本类是它唯一的主人**
* `session_read` / `message_persisted` 都在 WS 上,这个连接就是**进程内共享**的那一条 * `session_read` / `message_persisted` 都在 WS 上,这个连接就是**进程内共享**的那一条
@@ -136,6 +139,17 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
*/ */
private var trailingRefreshJob: Job? = null private var trailingRefreshJob: Job? = null
/**
* 窗口里攒下的「要问增量的会话 id」(W77)。
*
* 为什么是一个**集合**而不是一个 id:节流窗口把同一段时间里的落库广播合并成一次拉取,而那几个
* 会话**未必是同一个**(主会话说一句、另一个会话也说一句)。只记住最后一条的话,前面那个会话的
* 红点就永远不会出现——那正是 W49 修过的同一类病,只是换了个成因。
*
* 只在 [requestSilentRefresh] / [flushPendingIncrements] 两处写;登出时清空(它属于上一个身份)。
*/
private val pendingIncrements = mutableSetOf<String>()
/** /**
* 「谁在跑」的账本(S9):会话 id → 本端见过的、还没收到收尾帧的在途路 id。 * 「谁在跑」的账本(S9):会话 id → 本端见过的、还没收到收尾帧的在途路 id。
* *
@@ -201,6 +215,9 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
trailingRefreshJob?.cancel() trailingRefreshJob?.cancel()
trailingRefreshJob = null trailingRefreshJob = null
silentRefreshThrottle = SilentRefreshThrottle() silentRefreshThrottle = SilentRefreshThrottle()
// 攒下的会话 id 同理:它们是上一个身份的会话,重登后那些 id 要么不存在、
// 要么属于别人;留着只会让重登后的第一次窗口拿旧 id 去问增量。
pendingIncrements.clear()
pulledForIdentity = null pulledForIdentity = null
busyTurns = emptyMap() busyTurns = emptyMap()
_state.value = SessionsUiState() _state.value = SessionsUiState()
@@ -316,42 +333,54 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
} }
/** /**
* 节流地静默重拉一次列表 —— W49 修的「停在主会话时,次级会话来了新消息,抽屉红点不出现, * 节流地静默对齐一次 —— W49 修的「停在主会话时,次级会话来了新消息,抽屉红点不出现,
* 必须开一次抽屉才补上」(S6)。 * 必须开一次抽屉才补上」(S6)。
* *
* 列表此前只在抽屉打开 / 新建 / 删除 / init 四处更新,而 `message_persisted` 只处理 * 列表此前只在抽屉打开 / 新建 / 删除 / init 四处更新,而 `message_persisted` 只处理
* 「正在看的那个会话」—— 别的会话攒未读**没有任何人会去发现**。 * 「正在看的那个会话」—— 别的会话攒未读**没有任何人会去发现**。
* *
* ## 为什么是「重拉」而不是「本地把未读数加一」 * ## 为什么是「问服务端」而不是「本地把未读数加一」
* `message_persisted` 帧**不带 `unreadCount`**:本地没有逐条时间线,自己数出来的数会与编辑 / * `message_persisted` 帧**不带 `unreadCount`**:本端没有逐条时间线,靠帧自己数出来的数会与编辑 /
* 删除 / 别端操作漂移(这正是 [onMessagePersisted] 里那条纪律的由来)。所以这里**一个数都不造**, * 删除 / 别端操作漂移(这正是 [onMessagePersisted] 里那条纪律的由来)。所以这里**一个数都不造**,
* 只是把「服务端才是唯一真源」这件事提前做一次 —— 走的是**与抽屉打开时那次完全相同**的路径 * 只是把「服务端才是唯一真源」这件事提前做一次。
* ([refresh],含已读游标遮罩),只是不需要用户先开一次抽屉。 *
* ## W77:问的是**那一条会话的增量**,不再是整表
* 曾经的落点是 `refresh(silent = true)`(整表 `GET /sessions`),一次要为**所有**会话付一行预览 +
* 未读计数的价钱,而真正变了的只有一个会话。现在落到 [flushPendingIncrements]:
* `GET /messages?after=<已见最大 createdAt>`,只回那几条新消息。整表仍是唯一权威,但只在
* 「增量给不出可信答案」时才请它出场(回落规则写在 [flushPendingIncrements] 上)。
* *
* ## 三条纪律 * ## 三条纪律
* 1. **节流**(判据 [silentRefreshStep],窗口 [SILENT_REFRESH_WINDOW_MILLIS]):一次连发只发一次 * 1. **节流**(判据 [silentRefreshStep],窗口 [SILENT_REFRESH_WINDOW_MILLIS]):一次连发只发一次
* 请求;同时保证**最后一条之后还会再拉一次**(trailing),红点不会停在旧值上。 * 请求;同时保证**最后一条之后还会再拉一次**(trailing),红点不会停在旧值上。
* 2. **静默**:`silent = true`,不闪「同步中…」——用户根本没开抽屉,不该有任何东西在动。 * 2. **静默**:不闪「同步中…」——用户根本没开抽屉,不该有任何东西在动。失败时做什么**完全沿用既有
* 失败时做什么**完全沿用既有行为**(挂 [SessionsUiState.error] 那一行,只有打开抽屉才看得见; * 行为**(挂 [SessionsUiState.error] 那一行,只有打开抽屉才看得见;以及「网络异常,显示 x 分钟前
* 以及「网络异常,显示 x 分钟前的结果」那一句),本方法不另加提示:一次后台的顺手对齐失败, * 的结果」那一句),本方法不另加提示:一次后台的顺手对齐失败,不该比用户自己打开抽屉时的那次
* 不该比用户自己打开抽屉时的那次刷新更吵。 * 刷新更吵。
* 3. **不抢跑**:不改 `activeSessionId`、不推进已读游标、不碰 [select]。[refresh] 内部唯一会动 * 3. **不抢跑**:不改 `activeSessionId`、不推进已读游标、不碰 [select]。回落那条路([refresh])
* `activeSessionId` 的地方是 [resolveActiveSession](当前会话被别处删掉时收敛回主会话), * 内部唯一会动 `activeSessionId` 的地方是 [resolveActiveSession](当前会话被别处删掉时收敛回
* 那是既有的、正确的收敛,不是重置。 * 主会话),那是既有的、正确的收敛,不是重置。
* *
* @param cause 这次为什么要拉;决定它走不走窗口(见 [SilentRefreshCause])。 * @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) val step = silentRefreshStep(cause, System.currentTimeMillis(), silentRefreshThrottle)
silentRefreshThrottle = step.throttle silentRefreshThrottle = step.throttle
// 没有具体会话时不留 id:那一次整表会把所有会话都对上,攒着的 id 一个也不缺。
if (sessionId != null) pendingIncrements += sessionId
when (step.action) { when (step.action) {
// 该拉了。先把**等在半路**的那次补拉撤掉:它要做的就是同一件事,留着会多拉一次。 // 该拉了。先把**等在半路**的那次补拉撤掉:它要做的就是同一件事,留着会多拉一次。
// (只有 ContextChanged 压过窗口时才可能真的取消到一个活着的补拉任务——窗口外那一档 // (只有 ContextChanged 压过窗口时才可能真的取消到一个活着的补拉任务——窗口外那一档
// 走到这里时 trailingPending 必为 false,补拉早跑完了。) // 走到这里时 trailingPending 必为 false,补拉早跑完了。)
//
// 撤掉补拉**不等于**可以丢掉它攒下的 id:[pendingIncrements] 是独立的一份账本,
// 这一趟会把它们一并做掉。
SilentRefreshAction.PullNow -> { SilentRefreshAction.PullNow -> {
trailingRefreshJob?.cancel() trailingRefreshJob?.cancel()
trailingRefreshJob = null trailingRefreshJob = null
refresh(silent = true) flushPendingIncrements(fullTable = sessionId == null)
} }
// 窗口内:排一次补拉。它是「最后一条之后还会再拉一次」的**唯一**保证——没有它,一串 // 窗口内:排一次补拉。它是「最后一条之后还会再拉一次」的**唯一**保证——没有它,一串
@@ -365,15 +394,74 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
trailingRefreshJob = viewModelScope.launch { trailingRefreshJob = viewModelScope.launch {
delay(step.delayMillis) delay(step.delayMillis)
silentRefreshThrottle = throttleAfterPull(System.currentTimeMillis()) silentRefreshThrottle = throttleAfterPull(System.currentTimeMillis())
refresh(silent = true) flushPendingIncrements(fullTable = false)
} }
} }
// 窗口内、补拉已经排上:那一次本来就是把窗口里所有成因合起来做掉,这里不必再排。 // 窗口内、补拉已经排上:那一次本来就是把窗口里所有成因合起来做掉,这里不必再排。
// (但这一帧的会话 id 已经在上面记进 [pendingIncrements] 了 —— 补拉不会漏掉它。)
SilentRefreshAction.Ignore -> Unit 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<SessionSummaryDto>`
* (整表响应),而增量只回了消息——拿它去拼一份 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.ActiveSession] ⇒ 就地清零 + 推进游标(这段一个字没改);
* - [PersistedRoute.InactiveSession] ⇒ **节流地静默重拉**([requestSilentRefresh]), * - [PersistedRoute.InactiveSession] ⇒ **节流地问一次这条会话的消息增量**([requestSilentRefresh]
* 用服务端下发的 `unreadCount` 说话。这正是 W49 修的那一幕:此前这一档直接 `return`,于是 * 带上 `sessionId` → [flushPendingIncrements] → `MessageRepository.loadIncrement`),未读由
* 「App 停在主会话、次级会话来了新消息」没有任何人会去发现,红点要等用户打开一次抽屉(那次 * 「增量里越过已读游标的条数」说话(W77;W49 那一档原本整表重拉 `GET /sessions`)。
* 静默重拉)才补上。 * 这正是 W49 修的那一幕:此前这一档直接 `return`,于是「App 停在主会话、次级会话来了新消息」
* 没有任何人会去发现,红点要等用户打开一次抽屉才补上。
* *
* 两档的**载荷要求不同**,是刻意的:推游标要用消息自己的 `createdAt`(缺了就只能什么都不做), * 两档的**载荷要求不同**,是刻意的:推游标要用消息自己的 `createdAt`(缺了就只能什么都不做),
* 而「重拉一次列表」不需要载荷里的任何一个字段 —— **帧壳本身就是全部证据**(服务端只为「某条 * 而「问一次增量」不需要载荷里的任何一个字段 —— **帧壳本身就是全部证据**(服务端只为「某条
* 消息已在某会话落库」才推这一帧)。所以非激活那一档**不检查 `event.message`**:真遇上载荷解不出的 * 消息已在某会话落库」才推这一帧)。所以非激活那一档**不检查 `event.message`**:真遇上载荷解不出的
* 坏帧(契约漂移的典型症状),宁可多问一次服务端,也不要让红点又停在旧值上。 * 坏帧(契约漂移的典型症状),宁可多问一次服务端,也不要让红点又停在旧值上。
*/ */
private fun onMessagePersisted(event: WsEvent.MessagePersisted) { private fun onMessagePersisted(event: WsEvent.MessagePersisted) {
val sessionId = event.sessionId?.takeIf { it.isNotBlank() } ?: return val sessionId = event.sessionId?.takeIf { it.isNotBlank() } ?: return
when (persistedRoute(sessionId, _state.value.activeSessionId)) { when (persistedRoute(sessionId, _state.value.activeSessionId)) {
// 别的会话:本端数不出它的未读(帧不带 `unreadCount`)⇒ 节流地静默重拉一次列表, // 别的会话:本端数不出它的未读(帧不带 `unreadCount`)⇒ 节流地问一次**这条会话**的消息
// 红点于是在用户**没开抽屉**的时候也会出现。这一帧到此为止。 // 增量,红点于是在用户**没开抽屉**的时候也会出现。这一帧到此为止。
PersistedRoute.InactiveSession -> PersistedRoute.InactiveSession ->
requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage) requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage, sessionId)
// 屏幕上就是这个会话:就地清零 + 推进游标,**不发任何请求去拉列表**。 // 屏幕上就是这个会话:就地清零 + 推进游标,**不发任何请求去拉列表**。
// 这里用 `when` 穷举而不是「不满足上面那条就走这段」:路由将来多一档时, // 这里用 `when` 穷举而不是「不满足上面那条就走这段」:路由将来多一档时,
@@ -634,9 +723,9 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
* 判据也共用 [persistedRoute](只是会话 id 由这一帧自己钉:它不带 `sessionId`)。 * 判据也共用 [persistedRoute](只是会话 id 由这一帧自己钉:它不带 `sessionId`)。
* *
* 为什么必须在这里补一句:服务端的主动推送 Web 路由只发这一帧,**不补** `message_persisted` * 为什么必须在这里补一句:服务端的主动推送 Web 路由只发这一帧,**不补** `message_persisted`
* (那条只有 IM 路由会广播),所以 W49 那条「非激活会话落新消息 ⇒ 节流静默重拉」根本不会被 * (那条只有 IM 路由会广播),所以 W49 那条「非激活会话落新消息 ⇒ 节流刷新」根本不会被
* 触发——抽屉里主会话的红点就停在她开口之前,直到用户自己开一次抽屉。这里复用**同一条**路 * 触发——抽屉里主会话的红点就停在她开口之前,直到用户自己开一次抽屉。这里复用**同一条**路
* ([requestSilentRefresh] + 既有的 [SilentRefreshCause.InactiveSessionMessage]), * ([requestSilentRefresh] + 既有的 [SilentRefreshCause.InactiveSessionMessage],会话 id 取主会话),
* 不新写取数路径,也不动节流判据本身。 * 不新写取数路径,也不动节流判据本身。
* *
* `id` 缺席 = 服务端那一刻**没有落库行** ⇒ 历史里没有这一行,未读面上没有任何东西要更新 * `id` 缺席 = 服务端那一刻**没有落库行** ⇒ 历史里没有这一行,未读面上没有任何东西要更新
@@ -647,7 +736,7 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
val sessionId = SessionSummaryDto.MAIN_SESSION_ID val sessionId = SessionSummaryDto.MAIN_SESSION_ID
when (persistedRoute(sessionId, _state.value.activeSessionId)) { when (persistedRoute(sessionId, _state.value.activeSessionId)) {
PersistedRoute.InactiveSession -> PersistedRoute.InactiveSession ->
requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage) requestSilentRefresh(SilentRefreshCause.InactiveSessionMessage, sessionId)
// 屏幕上就是主会话:她这句话已经显示在聊天流里(ChatViewModel 那侧并入)⇒ 与 // 屏幕上就是主会话:她这句话已经显示在聊天流里(ChatViewModel 那侧并入)⇒ 与
// message_persisted 的同一档一样「看到即已读」,免得抽屉为她刚说的话亮一个红点。 // message_persisted 的同一档一样「看到即已读」,免得抽屉为她刚说的话亮一个红点。
@@ -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<String>()
/** 每次响应体(UTF-8)的字节数,按到达顺序 —— 量化对比用。 */
val responseBytes = CopyOnWriteArrayList<Int>()
private val queue = ConcurrentLinkedQueue<Pair<Int, String>>()
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)
}
}
}
@@ -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<HistoryMessageDto>,
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<String, Long>(), 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, "她<action>轻轻</action>说"))),
)
val web = updated.first { it.id == "web" }
// 已有的 unreadCount 直接相加是安全的:增量的每一条都严格新于 from,两者没有交集。
assertEquals(4, web.unreadCount)
assertEquals(300L, web.lastMessageAt)
// 预览取最新一条,且剥掉 `<action>` 这类标记(与整表那条路同一个 `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)
}
}