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.
This commit is contained in:
digua committed 2026-07-14 11:56:09 +08:00
1 parent 82710e2d5a
commit 710b21f2ae
7 files changed
+132 -24

No files matched your search

@@ -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<string, unknown[]>): AsyncSqlExecutor {
}
}
function createSqliteExecutor(db: Database.Database): AsyncSqlExecutor {
return {
all<T>(sql: string, params: unknown[] = []): Promise<T[]> {
return Promise.resolve(db.prepare(sql).all(...params) as T[])
},
get<T>(sql: string, params: unknown[] = []): Promise<T | undefined> {
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<string, unknown[]>([
@@ -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)
})
@@ -74,7 +74,11 @@ export async function fetchMessagesBefore(
keywords?: string[]
): Promise<AsyncPaginatedMessages> {
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<FullMessageRow>(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<AsyncPaginatedMessages> {
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<FullMessageRow>(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<FullMessageRow>(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<FullMessageRow>(sql, [...params, limit])
return { messages: rows.map(mapMessageRow).reverse(), total }
}
+6 -3
View File
@@ -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`
)
@@ -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
+17 -6
View File
@@ -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()
}
+2 -1
View File
@@ -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<number> {