From 710b21f2ae2be6c223b75dccf9edbc1952f97204 Mon Sep 17 00:00:00 2001 From: digua Date: Tue, 14 Jul 2026 11:50:19 +0800 Subject: [PATCH] fix(records): order backfilled messages by timestamp Use timestamp and message ID as the stable viewer cursor so historical messages imported later remain in chronological order. Resolve timeline anchors by message time and cover out-of-order IDs with SQLite regressions. --- .../__tests__/message-query-functions.test.ts | 79 +++++++++++++++++++ .../src/query/__tests__/session-index.test.ts | 6 +- .../core/src/query/message-query-functions.ts | 34 +++++--- packages/core/src/query/session-queries.ts | 9 ++- .../common/ChatRecord/MessageList.vue | 2 +- src/services/session-index/fetch.test.ts | 23 ++++-- src/services/session-index/fetch.ts | 3 +- 7 files changed, 132 insertions(+), 24 deletions(-) diff --git a/packages/core/src/query/__tests__/message-query-functions.test.ts b/packages/core/src/query/__tests__/message-query-functions.test.ts index f1dafcba..6a39cd2c 100644 --- a/packages/core/src/query/__tests__/message-query-functions.test.ts +++ b/packages/core/src/query/__tests__/message-query-functions.test.ts @@ -9,6 +9,7 @@ import { describe, it } from 'node:test' import assert from 'node:assert/strict' +import Database from 'better-sqlite3' import type { AsyncSqlExecutor } from '../message-query-functions' import type { FullMessageRow } from '../message-sql' import { @@ -87,6 +88,49 @@ function createAsyncExecutor(store: Map): AsyncSqlExecutor { } } +function createSqliteExecutor(db: Database.Database): AsyncSqlExecutor { + return { + all(sql: string, params: unknown[] = []): Promise { + return Promise.resolve(db.prepare(sql).all(...params) as T[]) + }, + get(sql: string, params: unknown[] = []): Promise { + return Promise.resolve(db.prepare(sql).get(...params) as T | undefined) + }, + } +} + +function createBackfilledMessageDb(): Database.Database { + const db = new Database(':memory:') + db.exec(` + CREATE TABLE member ( + id INTEGER PRIMARY KEY, + platform_id TEXT NOT NULL, + account_name TEXT, + group_nickname TEXT, + aliases TEXT DEFAULT '[]', + avatar TEXT + ); + CREATE TABLE message ( + id INTEGER PRIMARY KEY, + sender_id INTEGER NOT NULL, + ts INTEGER NOT NULL, + type INTEGER NOT NULL, + content TEXT, + reply_to_message_id TEXT, + platform_message_id TEXT + ); + INSERT INTO member (id, platform_id, account_name) VALUES (1, 'alice', 'Alice'); + + -- 较新的消息先入库，随后增量回填较早的消息，因此 id 与时间顺序不同。 + INSERT INTO message (id, sender_id, ts, type, content) VALUES + (1, 1, 300, 0, 'newer-first'), + (2, 1, 400, 0, 'newest-first'), + (3, 1, 100, 0, 'oldest-backfill'), + (4, 1, 200, 0, 'older-backfill'); + `) + return db +} + // ==================== Tests ==================== describe('fetchMessagesBefore', () => { @@ -135,6 +179,41 @@ describe('fetchMessagesAfter', () => { }) }) +describe('timestamp cursor ordering after historical backfill', () => { + it('paginates before and after an anchor by timestamp then id', async () => { + const db = createBackfilledMessageDb() + try { + const executor = createSqliteExecutor(db) + const before = await fetchMessagesBefore(executor, 1, 10) + const after = await fetchMessagesAfter(executor, 1, 10) + + assert.deepEqual( + before.messages.map((message) => message.id), + [3, 4] + ) + assert.deepEqual( + after.messages.map((message) => message.id), + [2] + ) + } finally { + db.close() + } + }) + + it('loads message context in chronological order instead of insertion order', async () => { + const db = createBackfilledMessageDb() + try { + const messages = await fetchMessageContext(createSqliteExecutor(db), 1, 2) + assert.deepEqual( + messages.map((message) => message.id), + [3, 4, 1, 2] + ) + } finally { + db.close() + } + }) +}) + describe('searchMessagesLikeAsync', () => { it('returns total and messages consistently across executors', async () => { const store = new Map([ diff --git a/packages/core/src/query/__tests__/session-index.test.ts b/packages/core/src/query/__tests__/session-index.test.ts index 3de21346..0785e425 100644 --- a/packages/core/src/query/__tests__/session-index.test.ts +++ b/packages/core/src/query/__tests__/session-index.test.ts @@ -234,15 +234,15 @@ describe('getChatSessionList', () => { it('returns sessions with firstMessageId', () => { const db = createSqliteDb() seedMessages(db, [ - { id: 10, ts: 1000 }, - { id: 20, ts: 1100 }, + { id: 10, ts: 1100 }, + { id: 20, ts: 1000 }, { id: 30, ts: 5000 }, ]) generateSessionIndex(db, 2000) const list = getChatSessionList(db) assert.equal(list.length, 2) - assert.equal(list[0].firstMessageId, 10) + assert.equal(list[0].firstMessageId, 20) assert.equal(list[0].messageCount, 2) assert.equal(list[1].firstMessageId, 30) }) diff --git a/packages/core/src/query/message-query-functions.ts b/packages/core/src/query/message-query-functions.ts index f6d91ea7..4fa44602 100644 --- a/packages/core/src/query/message-query-functions.ts +++ b/packages/core/src/query/message-query-functions.ts @@ -74,7 +74,11 @@ export async function fetchMessagesBefore( keywords?: string[] ): Promise { const { clause, params } = filterConditions(filter, senderId, keywords) - const sql = `${FULL_MSG_SELECT} WHERE msg.id < ? ${clause} ORDER BY msg.id DESC LIMIT ?` + const sql = `${FULL_MSG_SELECT} + JOIN message anchor ON anchor.id = ? + WHERE (msg.ts < anchor.ts OR (msg.ts = anchor.ts AND msg.id < anchor.id)) + ${clause} + ORDER BY msg.ts DESC, msg.id DESC LIMIT ?` const rows = await executor.all(sql, [beforeId, ...params, limit + 1]) const hasMore = rows.length > limit const sliced = hasMore ? rows.slice(0, limit) : rows @@ -94,7 +98,11 @@ export async function fetchMessagesAfter( keywords?: string[] ): Promise { const { clause, params } = filterConditions(filter, senderId, keywords) - const sql = `${FULL_MSG_SELECT} WHERE msg.id > ? ${clause} ORDER BY msg.id ASC LIMIT ?` + const sql = `${FULL_MSG_SELECT} + JOIN message anchor ON anchor.id = ? + WHERE (msg.ts > anchor.ts OR (msg.ts = anchor.ts AND msg.id > anchor.id)) + ${clause} + ORDER BY msg.ts ASC, msg.id ASC LIMIT ?` const rows = await executor.all(sql, [afterId, ...params, limit + 1]) const hasMore = rows.length > limit const sliced = hasMore ? rows.slice(0, limit) : rows @@ -149,7 +157,7 @@ export async function searchMessagesLikeAsync( /** * Get surrounding context messages for given message IDs. - * Uses simple id-based ordering (not session-aware). + * Neighbor selection and output use chronological (timestamp, id) ordering. */ export async function fetchMessageContext( executor: AsyncSqlExecutor, @@ -165,15 +173,21 @@ export async function fetchMessageContext( allIds.add(id) if (contextSize > 0) { const before = await executor.all<{ id: number }>( - 'SELECT id FROM message WHERE id < ? ORDER BY id DESC LIMIT ?', + `SELECT msg.id FROM message msg + JOIN message anchor ON anchor.id = ? + WHERE msg.ts < anchor.ts OR (msg.ts = anchor.ts AND msg.id < anchor.id) + ORDER BY msg.ts DESC, msg.id DESC LIMIT ?`, [id, contextSize] ) before.forEach((r) => allIds.add(r.id)) - const after = await executor.all<{ id: number }>('SELECT id FROM message WHERE id > ? ORDER BY id ASC LIMIT ?', [ - id, - contextSize, - ]) + const after = await executor.all<{ id: number }>( + `SELECT msg.id FROM message msg + JOIN message anchor ON anchor.id = ? + WHERE msg.ts > anchor.ts OR (msg.ts = anchor.ts AND msg.id > anchor.id) + ORDER BY msg.ts ASC, msg.id ASC LIMIT ?`, + [id, contextSize] + ) after.forEach((r) => allIds.add(r.id)) } } @@ -182,7 +196,7 @@ export async function fetchMessageContext( if (idList.length === 0) return [] const placeholders = idList.map(() => '?').join(', ') - const sql = `${FULL_MSG_SELECT} WHERE msg.id IN (${placeholders}) ORDER BY msg.id ASC` + const sql = `${FULL_MSG_SELECT} WHERE msg.id IN (${placeholders}) ORDER BY msg.ts ASC, msg.id ASC` const rows = await executor.all(sql, idList) return rows.map(mapMessageRow) } @@ -283,7 +297,7 @@ export async function fetchAllRecentMessages( const countRow = await executor.get<{ total: number }>(countSql, params) const total = countRow?.total ?? 0 - const sql = `${FULL_MSG_SELECT} WHERE 1=1 ${clause} ORDER BY msg.ts DESC LIMIT ?` + const sql = `${FULL_MSG_SELECT} WHERE 1=1 ${clause} ORDER BY msg.ts DESC, msg.id DESC LIMIT ?` const rows = await executor.all(sql, [...params, limit]) return { messages: rows.map(mapMessageRow).reverse(), total } } diff --git a/packages/core/src/query/session-queries.ts b/packages/core/src/query/session-queries.ts index b513ca7e..55df2207 100644 --- a/packages/core/src/query/session-queries.ts +++ b/packages/core/src/query/session-queries.ts @@ -439,7 +439,8 @@ export function getSessionsByTimeRange(db: DatabaseAdapter, startTs: number, end id, start_ts as startTs, end_ts as endTs, message_count as messageCount, summary, (SELECT mc.message_id FROM message_context mc - WHERE mc.segment_id = cs.id ORDER BY mc.message_id LIMIT 1) as firstMessageId + JOIN message first_msg ON first_msg.id = mc.message_id + WHERE mc.segment_id = cs.id ORDER BY first_msg.ts ASC, first_msg.id ASC LIMIT 1) as firstMessageId FROM segment cs WHERE start_ts >= ? AND start_ts <= ? ORDER BY start_ts DESC` @@ -462,7 +463,8 @@ export function getRecentChatSessions(db: DatabaseAdapter, limit: number): ChatS id, start_ts as startTs, end_ts as endTs, message_count as messageCount, summary, (SELECT mc.message_id FROM message_context mc - WHERE mc.segment_id = cs.id ORDER BY mc.message_id LIMIT 1) as firstMessageId + JOIN message first_msg ON first_msg.id = mc.message_id + WHERE mc.segment_id = cs.id ORDER BY first_msg.ts ASC, first_msg.id ASC LIMIT 1) as firstMessageId FROM segment cs ORDER BY start_ts DESC LIMIT ?` @@ -488,7 +490,8 @@ export function getChatSessionList(db: DatabaseAdapter): ChatSessionItem[] { cs.message_count as messageCount, cs.summary, (SELECT mc.message_id FROM message_context mc - WHERE mc.segment_id = cs.id ORDER BY mc.message_id LIMIT 1) as firstMessageId + JOIN message first_msg ON first_msg.id = mc.message_id + WHERE mc.segment_id = cs.id ORDER BY first_msg.ts ASC, first_msg.id ASC LIMIT 1) as firstMessageId FROM segment cs ORDER BY cs.start_ts ASC` ) diff --git a/src/components/common/ChatRecord/MessageList.vue b/src/components/common/ChatRecord/MessageList.vue index 4b14aa32..87eb5d28 100644 --- a/src/components/common/ChatRecord/MessageList.vue +++ b/src/components/common/ChatRecord/MessageList.vue @@ -354,7 +354,7 @@ async function loadMoreAfter() { hasMoreAfter.value = searchOffset.value < result.total } else { - // 普通模式：使用消息 ID 加载 + // 普通模式：服务端会根据末条消息 ID 解析 (timestamp, id) 复合时间游标。 const lastMessage = messages.value[messages.value.length - 1] if (!lastMessage) return diff --git a/src/services/session-index/fetch.test.ts b/src/services/session-index/fetch.test.ts index 55bd64d4..fc9c5d52 100644 --- a/src/services/session-index/fetch.test.ts +++ b/src/services/session-index/fetch.test.ts @@ -29,12 +29,23 @@ function createDb(): Database.Database { segment_id INTEGER, topic_id INTEGER ); + CREATE TABLE message ( + id INTEGER PRIMARY KEY, + ts INTEGER NOT NULL + ); `) const insSeg = db.prepare('INSERT INTO segment (id, start_ts, end_ts, message_count, summary) VALUES (?, ?, ?, ?, ?)') insSeg.run(1, 100, 200, 3, null) insSeg.run(2, 300, 400, 2, 'summary-2') - // 故意打乱插入顺序，验证 firstMessageId 取的是 message_id 最小值 + // 模拟先导入新消息、再回填旧消息，验证首条消息按时间而不是按自增 ID 选择。 + const insMsg = db.prepare('INSERT INTO message (id, ts) VALUES (?, ?)') + insMsg.run(10, 150) + insMsg.run(11, 100) + insMsg.run(12, 200) + insMsg.run(30, 400) + insMsg.run(31, 300) + const insCtx = db.prepare('INSERT INTO message_context (message_id, segment_id, topic_id) VALUES (?, ?, NULL)') insCtx.run(11, 1) insCtx.run(10, 1) @@ -60,9 +71,9 @@ describe('FetchSessionIndexAdapter firstMessageId', () => { const sessions = await new FetchSessionIndexAdapter().getSessions('s') assert.equal(sessions.length, 2) assert.equal(sessions[0].id, 1) - assert.equal(sessions[0].firstMessageId, 10) + assert.equal(sessions[0].firstMessageId, 11) assert.equal(sessions[1].id, 2) - assert.equal(sessions[1].firstMessageId, 30) + assert.equal(sessions[1].firstMessageId, 31) } finally { db.close() } @@ -73,7 +84,7 @@ describe('FetchSessionIndexAdapter firstMessageId', () => { registerSqliteDataAdapter(db) try { const sessions = await new FetchSessionIndexAdapter().getRecent('s', 10) - assert.equal(sessions[0].firstMessageId, 30) + assert.equal(sessions[0].firstMessageId, 31) assert.ok(sessions.every((s) => typeof s.firstMessageId === 'number')) } finally { db.close() @@ -86,8 +97,8 @@ describe('FetchSessionIndexAdapter firstMessageId', () => { try { const sessions = await new FetchSessionIndexAdapter().getByTimeRange('s', 0, 1000) assert.equal(sessions.length, 2) - assert.equal(sessions[0].firstMessageId, 10) - assert.equal(sessions[1].firstMessageId, 30) + assert.equal(sessions[0].firstMessageId, 11) + assert.equal(sessions[1].firstMessageId, 31) } finally { db.close() } diff --git a/src/services/session-index/fetch.ts b/src/services/session-index/fetch.ts index 067d3a78..7648acbf 100644 --- a/src/services/session-index/fetch.ts +++ b/src/services/session-index/fetch.ts @@ -25,7 +25,8 @@ function getDataAdapter(): DataAdapter { // 取会话首条消息 ID，供时间线点击跳转到对应消息位置使用（与 core getChatSessionList 保持一致） const FIRST_MESSAGE_ID_SUBQUERY = `(SELECT mc.message_id FROM message_context mc - WHERE mc.segment_id = segment.id ORDER BY mc.message_id LIMIT 1) as firstMessageId` + JOIN message first_msg ON first_msg.id = mc.message_id + WHERE mc.segment_id = segment.id ORDER BY first_msg.ts ASC, first_msg.id ASC LIMIT 1) as firstMessageId` export class FetchSessionIndexAdapter implements SessionIndexAdapter { async generate(sessionId: string, gapThreshold: number = 1800): Promise {