Merge branch 'feat/w77-unread-incremental-after-cursor' into android

This commit is contained in:
AskaEth committed 2026-10-26 23:40:17 +08:00
commit 4a9e22ea77
5 files changed
+799 -34

No files matched your search

@@ -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]。
*/
@@ -87,7 +88,9 @@ class MessageRepository(
* `createdAt`([PagedMessages.newestCreatedAt]),不是最旧的。
*
* 同毫秒并列且条数超过 `limit` 时,纯 `createdAt` 游标推不动(契约 §A5 明说的已知边界)——
* 调用方按 `id` 去重、并在游标没前进时停下来,不在这里兜。
* 落在页边界上的那一毫秒里剩下的行**再也追不回来**,因为服务端只给严格 `>`。调用方按 `id`
* 去重,并且**不能把满页之后的一页(空页、不满页都一样)当作追到了尾巴**:要翻页又要把边界
* 说清楚,用 [loadIncrement],不要自己拿这一页凑。
*/
suspend fun loadAfter(
sessionId: String,
@@ -100,6 +103,87 @@ class MessageRepository(
.map { raw -> toPagedMessages(raw, pageSize) }
}
/**
* **增量追尾**(W77):从 [after] 起把某会话的新消息一直翻到尾巴,最多 [maxPages] 页。
*
* 与裸的 [loadAfter] 的差别只有一件:**它负责把「还没翻完」这件事如实说出来**。
*
* ## 什么时候只能说 `complete = false`
* 游标是本页**最新**一条的 `createdAt`,而服务端只给 `created_at > after`(**严格**大于:
* `list_after.go` 的 `ListAfter` 与 `devtools/fake-backend` 两处一致)。于是**落在页边界上被
* 劈开的那一毫秒**里的行永远追不回来——它们的 `createdAt` 恰好等于游标,再问一次也问不到
* (契约 §A5 的已知边界:纯 `createdAt` 游标没有毫秒内的次序)。
*
* 要紧的是**切点**:只有当服务端的 `LIMIT` 在某一毫秒中间切过一刀,才会留下追不回来的尾巴。
* 客户端看不见那一刀落在哪儿,只知道自己**拿过满页**就等于「发生过一次切点」。所以这里只有一条
* 不变式,下面几档都是它的推论:
*
* > `complete = true` **当且仅当整个追尾从未拿过满页**。
*
* - **第一页就是空页 ⇒ `true`**:一条新消息都没有,没有可少给的;
* - **第一页就不满 ⇒ `true`**:`LIMIT` 没切到任何东西,`> cursor` 的行全在这一页里;
* - **拿过整页之后,下一页是空页 ⇒ `false`**:那次空页分不出「干净的尾巴」与「空页底下还压着
* 同一毫秒的几条」;
* - **拿过整页之后,下一页不满 ⇒ 也是 `false`**:不满只说明「`> cursor` 的行就这么多了」,对
* 切点那一毫秒还有没有行**一无所知**。例如积压 `m1@500 m2@500 m3@500 m4@600`、页宽 2:
* 第一页给 `m1 m2`(切点正好劈开 500 那一毫秒),第二页 `after=500` 只给 `m4@600`、不满页,
* 而 `m3@500` 既不 `> 500`、也就永远拿不到;
* - **翻满 [maxPages] ⇒ `false`**。
*
* 代价:积压一到页宽(默认 50 条)就回落整表。这与「翻满 [maxPages] ⇒ 回落」同类——那时增量
* 已经取回 ≥50 行,多问一次整表是边际成本,也正合 [MAX_INCREMENT_PAGES] 那句「积压多到增量不再
* 便宜」。主路径(几条新消息、第一页就不满)不受影响。
*
* 这里**不猜**、也不悄悄少给几条:说不完整,调用方还能去问唯一权威。
*
* 不传 `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
// 只要翻到过一整页,就说明服务端的 LIMIT 切过一刀,那一刀可能劈开了一毫秒(见 KDoc)。
// 此后无论下一页是空页还是不满页,都说不出「追到了尾巴」。
var chasedFullPage = false
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)
// 空页有两种含义,客户端分不出来:第一页就空 = 真的没有新的;拿过一整页之后空 = 可能还有
// 同一毫秒的几条卡在游标上(服务端只给严格大于)。后者只能说「没追完」。
if (paged.messages.isEmpty()) {
return@withContext done(cursor, complete = !chasedFullPage)
}
// 去重只是防御:严格 `>` 下相邻两页不可能重叠(下一页只要 > 本页最新那条)。
paged.messages.forEach { byId[it.id] = it }
val newest = paged.newestCreatedAt ?: return@withContext done(cursor, complete = false)
// 不满页只说明「> cursor 的行就这么多了」,对切点那一毫秒一无所知 —— 与空页档同一个洞,
// 同样是拿过整页之后就不作数。
if (!paged.hasMore) return@withContext done(newest, complete = !chasedFullPage)
cursor = newest
chasedFullPage = true
}
done(cursor, complete = false)
}
/**
* 编辑单条消息(D3,契约 §1 行 6a):`PATCH /api/v1/messages/{id}`,成功返回**服务端更新后的
* 完整消息 DTO**。
@@ -152,6 +236,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 +283,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<HistoryMessageDto>,
val cursor: Long,
val complete: Boolean,
)
@@ -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,117 @@ 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<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]`)筛一遍。
* **过滤口径**与服务端同源:`session_read.go` 的 `unreadCountQuery` 是 `count(created_at > read_at)`,
* 而 `read_at` 是**账号级单一游标**,可能比本端已知的最新消息还新——用户在这台机器上点开过它
* ([SessionsViewModel.select] 推的就是「现在」),随后别的设备又发了消息。那一刻
* `(lastMessageAt, readAt]` 之间的消息**服务端算已读**,一律加进角标就会多出来一个红点。所以只数
* `createdAt > readAt` 的那几条。
*
* 已有的 [SessionItem.unreadCount] 直接保留相加是安全的:它来自服务端,而这次增量里的每一条都严格新于
* [from](服务端的 `after` 就是 `created_at > ?`),两者没有交集,不会重复计数。
*
* ## 已知边界:同源的只是**过滤口径**,基数仍是上次整表的旧值
* 相加的底数 [SessionItem.unreadCount] 来自上一次整表刷新,而 `read_at` 是账号级的、会被别的设备改写。
* 断线期间他端把某个会话读掉时,那条 `session_read` 帧本地**永久收不到**(断线不补帧),于是底数偏大,
* 这一行会一直多算那几条。这是**已接受的边界**,不是漏网:红点本来就只承诺「大概有几条新的」,漂移
* 到下一次整表刷新(打开抽屉 / 初始化 / 任何一次回落)就自愈。
*
* 这里**不**去检测它(例如「[readAt] > [from] ⇒ 底数可疑 ⇒ 回落整表」):本端刚点开过这个会话时
* `readAt` 本来就被推到「现在」、必然大于 `lastMessageAt`,那是最常见的一档;拿它当异常信号会让增量
* 路径在正常使用下持续回落整表,而真正的「他端已读」它一样抓不全(断线时帧就丢了)。抓不准的闸门
* 不如不装。
*
* ## 并发的那一份真相
* [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
}
@@ -28,7 +28,8 @@ data class SessionItem(
val lastMessage: String?,
/**
* 最后一条消息的时间戳(毫秒);会话还没有消息时为 null。
* 只用于已读游标比对(见 [SessionItem.withReadCursor]),不直接显示。
* 两处用:已读游标比对(见 [SessionItem.withReadCursor]),以及增量追尾的 `after` 游标
* (见 `incrementCursorFor`,由 `applyIncrement` 推进到这次增量的 `cursor`)。不直接显示。
*/
val lastMessageAt: Long? = null,
/** 未读条数(服务端 `unreadCount`,D4);`0` = 无未读、不显示角标。 */
@@ -82,9 +83,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=<ms>`(前向增量)而不是整表 `GET /sessions`——W77 换的,理由与回落
* 规则都写在 [flushPendingIncrements] 上。整表仍是**唯一权威**,只是不再需要为一条新消息付它的价钱。
*
* ## WS 连接:全 App 共享一条,而**本类是它唯一的主人**
* `session_read` / `message_persisted` 都在 WS 上,这个连接就是**进程内共享**的那一条
@@ -136,6 +140,17 @@ class SessionsViewModel(private val container: AppContainer) : ViewModel() {
*/
private var trailingRefreshJob: Job? = null
/**
* 窗口里攒下的「要问增量的会话 id」(W77)。
*
* 为什么是一个**集合**而不是一个 id:节流窗口把同一段时间里的落库广播合并成一次拉取,而那几个
* 会话**未必是同一个**(主会话说一句、另一个会话也说一句)。只记住最后一条的话,前面那个会话的
* 红点就永远不会出现——那正是 W49 修过的同一类病,只是换了个成因。
*
* 只在 [requestSilentRefresh] / [flushPendingIncrements] 两处写;登出时清空(它属于上一个身份)。
*/
private val pendingIncrements = mutableSetOf<String>()
/**
* 「谁在跑」的账本(S9):会话 id → 本端见过的、还没收到收尾帧的在途路 id。
*
@@ -201,6 +216,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 +334,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 +395,75 @@ 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` —— 整个追尾里拿过整页(服务端的 `LIMIT` 切过一刀,那一刀
* 可能劈开了同一毫秒,而服务端只给严格 `>`,被劈开的尾巴再也追不回来;此后的空页与不满页
* 都说不清),或翻满了 `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 +684,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 +725,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 +738,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 的同一档一样「看到即已读」,免得抽屉为她刚说的话亮一个红点。
@@ -0,0 +1,234 @@
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 `an empty first page is a clean tail`() = runBlocking {
backend.enqueue(200, """{"data":[]}""")
val increment = repository().loadIncrement("web", after = 300).getOrThrow()
// 第一页就空 ⇒ 一条新消息都没有,没有可少给的:这是真的追到尾巴,游标原地不动。
assertEquals(1, backend.uris.size)
assertTrue(backend.uris.single(), backend.uris.single().contains("after=300"))
assertTrue(increment.complete)
assertEquals(300L, increment.cursor)
assertTrue(increment.messages.isEmpty())
}
@Test
fun `an empty page after a full page is not proof of a clean tail`() = 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()
// 满页 ⇒ 多问一次,第二次拿到空页。但客户端**分不出**这次空页是「干净的尾巴」还是
// 「空页底下还压着同一毫秒的几条」:服务端只给严格 `> cursor`,被页边界劈开的那一毫秒
// 再也追不回来。分不出来就只能说「没追完」,把「要不要回落整表」交回调用方。
assertEquals(2, backend.uris.size)
assertTrue(backend.uris[1], backend.uris[1].contains("after=200"))
assertFalse(increment.complete)
assertEquals(200L, increment.cursor)
assertEquals(listOf("m1", "m2"), increment.messages.map { it.id })
}
@Test
fun `a short page after a full page is not proof of a clean tail either`() = runBlocking {
// 页宽 2、积压 m1@500 / m2@500 / m3@500 / m4@600。第一页给 m1、m2 —— LIMIT 恰好把 500 那一
// 毫秒劈开;第二页 after=500 只能给 > 500 的,于是 [m4@600]、不满页。不满只证明「> 500 的行
// 就这一条」,对切点那一毫秒还有没有行一无所知:m3@500 既不 > 500,也就永远拿不到。
// 空页档与不满页档是同一个洞的两个出口,都要说「没追完」。
backend.enqueue(200, """{"data":[${message("m1", 500)}, ${message("m2", 500)}]}""")
backend.enqueue(200, """{"data":[${message("m4", 600)}]}""")
val increment = repository().loadIncrement("web", after = 0, pageSize = 2).getOrThrow()
assertEquals(2, backend.uris.size)
assertTrue(backend.uris[1], backend.uris[1].contains("after=500"))
assertFalse(increment.complete)
assertEquals(600L, increment.cursor)
assertEquals(listOf("m1", "m2", "m4"), increment.messages.map { it.id })
}
@Test
fun `a full page stuck on one millisecond is reported as incomplete`() = runBlocking {
// 真后端的形状:m1 / m2 / m3 三条落在同一毫秒,页宽 2。第一页给 m1、m2,第二页问
// `after=500` —— 严格 `>` 下后端只会回空页,m3@500 永远拿不到(这里不入队 m3:队列里
// 那份空响应就是真后端在 `after=500` 时会给的答案)。
backend.enqueue(200, """{"data":[${message("m1", 500)}, ${message("m2", 500)}]}""")
backend.enqueue(200, """{"data":[]}""")
val increment = repository().loadIncrement("web", after = 100, pageSize = 2).getOrThrow()
assertFalse(increment.complete)
// 游标停在那一毫秒(推不动就不装作推得动)。
assertEquals(500L, increment.cursor)
assertEquals(listOf("m1", "m2"), increment.messages.map { it.id })
// 线上证据:第二次真的带上了 `after=500`,即「m3 拿不到」的那一问。
assertEquals(2, backend.uris.size)
assertTrue(backend.uris[1], backend.uris[1].contains("after=500"))
}
@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,212 @@
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` 是账号级单一游标,这台机器点开过它(游标推到「现在」)之后别的设备又发了消息,
// 那段区间服务端算已读 —— 加进角标就多出一个红点,所以先按 readAt 筛。
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 `a read elsewhere while offline over-counts until the next whole table`() {
// 已接受的边界(见 `applyIncrement` 的 KDoc),钉在这里是为了它不会被当成「已修好」:
// 底数 `unreadCount` 来自上一次整表刷新,而 `read_at` 是账号级的。断线期间他端把这个会话
// 读掉了,那条 `session_read` 帧本地**永久收不到**(断线不补帧)⇒ 本端 readAt 还是 null,
// 整批都进了角标(4 + 3 = 7),而服务端此刻算 0。漂移到下一次整表刷新自愈,这里不改行为。
val sessions = listOf(item(id = "web", lastMessageAt = 100, unreadCount = 4))
val updated = sessions.applyIncrement(
"web",
from = 100,
readAt = null,
increment = increment(listOf(message("m1", 200), message("m2", 300), message("m3", 400))),
)
assertEquals(7, 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)
}
}