mirror of
https://github.com/hellodigua/ChatLab.git
synced 2026-10-26 02:04:37 +08:00
refactor(search): stop maintaining session FTS
This commit is contained in:
1 parent
50a390b2de
commit
ef4f8de74d
30 files changed
+29
-771
No files matched your search
@@ -1,87 +0,0 @@
|
||||
/**
|
||||
* FTS5 full-text search index management — Electron adapter layer.
|
||||
*
|
||||
* Core FTS operations are delegated to @openchatlab/node-runtime.
|
||||
* This module adds session-level DB connection management.
|
||||
*/
|
||||
|
||||
import type Database from 'better-sqlite3'
|
||||
import { BetterSqliteAdapter } from '@openchatlab/node-runtime'
|
||||
import {
|
||||
hasFtsTable,
|
||||
createFtsTable as coreCreateFtsTable,
|
||||
buildFtsIndex as coreBuildFtsIndex,
|
||||
rebuildFtsIndex as coreRebuildFtsIndex,
|
||||
insertFtsEntries as coreInsertFtsEntries,
|
||||
searchByFts as coreSearchByFts,
|
||||
} from '@openchatlab/node-runtime'
|
||||
import { getDbPath, openRawDatabase, openDatabase } from '../core'
|
||||
|
||||
function openWritableDb(sessionId: string): Database.Database | null {
|
||||
try {
|
||||
return openRawDatabase(getDbPath(sessionId))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export function hasFtsIndex(sessionId: string): boolean {
|
||||
const db = openDatabase(sessionId)
|
||||
if (!db) return false
|
||||
try {
|
||||
return hasFtsTable(new BetterSqliteAdapter(db))
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export function createFtsTable(db: Database.Database): void {
|
||||
coreCreateFtsTable(new BetterSqliteAdapter(db))
|
||||
}
|
||||
|
||||
export function buildFtsIndex(sessionId: string): { indexed: number } {
|
||||
const db = openWritableDb(sessionId)
|
||||
if (!db) return { indexed: 0 }
|
||||
|
||||
try {
|
||||
return coreBuildFtsIndex(new BetterSqliteAdapter(db))
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
export function rebuildFtsIndex(sessionId: string): { indexed: number } {
|
||||
const db = openWritableDb(sessionId)
|
||||
if (!db) return { indexed: 0 }
|
||||
|
||||
try {
|
||||
return coreRebuildFtsIndex(new BetterSqliteAdapter(db))
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
export function insertFtsEntries(sessionId: string, entries: Array<{ id: number; content: string | null }>): void {
|
||||
const db = openWritableDb(sessionId)
|
||||
if (!db) return
|
||||
|
||||
try {
|
||||
coreInsertFtsEntries(new BetterSqliteAdapter(db), entries)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
export function searchByFts(
|
||||
sessionId: string,
|
||||
keywords: string[],
|
||||
limit = 1000,
|
||||
offset = 0
|
||||
): { rowids: number[]; total: number } {
|
||||
if (keywords.length === 0) return { rowids: [], total: 0 }
|
||||
|
||||
const db = openDatabase(sessionId)
|
||||
if (!db) return { rowids: [], total: 0 }
|
||||
|
||||
return coreSearchByFts(new BetterSqliteAdapter(db), keywords, limit, offset)
|
||||
}
|
||||
@@ -96,6 +96,3 @@ export type { ChatSessionItem, SessionMessagesResult } from './session'
|
||||
|
||||
// NLP 查询
|
||||
export { getWordFrequency } from './nlp'
|
||||
|
||||
// FTS 索引管理
|
||||
export { hasFtsIndex, buildFtsIndex, rebuildFtsIndex } from './fts'
|
||||
Vendored
-1
@@ -50,7 +50,6 @@ interface ImportDiagnostics {
|
||||
messageWriteMs: number
|
||||
nicknameHistoryMs: number
|
||||
indexCreationMs: number
|
||||
ftsMs: number
|
||||
checkpointMs: number
|
||||
sessionIndexMs: number
|
||||
postImportHookMs: number
|
||||
|
||||
@@ -120,7 +120,6 @@ export {
|
||||
fetchMessagesBefore,
|
||||
fetchMessagesAfter,
|
||||
searchMessagesLikeAsync,
|
||||
searchMessagesWithFtsAsync,
|
||||
fetchMessageContext,
|
||||
fetchSearchMessageContext,
|
||||
fetchAllRecentMessages,
|
||||
@@ -384,7 +383,6 @@ export {
|
||||
CHAT_DB_TABLES,
|
||||
CHAT_DB_INDEXES,
|
||||
CHAT_DB_SCHEMA,
|
||||
FTS_TABLE_SCHEMA,
|
||||
getSchemaVersion,
|
||||
setSchemaVersion,
|
||||
needsMigration,
|
||||
|
||||
@@ -189,7 +189,6 @@ export {
|
||||
fetchMessagesBefore,
|
||||
fetchMessagesAfter,
|
||||
searchMessagesLikeAsync,
|
||||
searchMessagesWithFtsAsync,
|
||||
fetchMessageContext,
|
||||
fetchSearchMessageContext,
|
||||
fetchAllRecentMessages,
|
||||
|
||||
@@ -109,33 +109,7 @@ export async function fetchMessagesAfter(
|
||||
return { messages: sliced.map(mapMessageRow), hasMore }
|
||||
}
|
||||
|
||||
/**
|
||||
* LIKE-based keyword search with count + pagination.
|
||||
* FTS search is NOT included here — platforms should handle FTS themselves
|
||||
* and fall back to this function when FTS is unavailable.
|
||||
*/
|
||||
/**
|
||||
* FTS5-based message search. The matchQuery must be pre-tokenized for the platform's FTS tokenizer.
|
||||
*/
|
||||
export async function searchMessagesWithFtsAsync(
|
||||
executor: AsyncSqlExecutor,
|
||||
matchQuery: string,
|
||||
filter?: TimeFilter,
|
||||
limit: number = 20,
|
||||
offset: number = 0,
|
||||
senderId?: number
|
||||
): Promise<AsyncMessagesWithTotal> {
|
||||
const { clause, params } = filterConditions(filter, senderId)
|
||||
|
||||
const countSql = `SELECT COUNT(*) as total ${MSG_COUNT_FROM} WHERE msg.id IN (SELECT rowid FROM message_fts WHERE content MATCH ?) ${clause}`
|
||||
const countRow = await executor.get<{ total: number }>(countSql, [matchQuery, ...params])
|
||||
const total = countRow?.total ?? 0
|
||||
|
||||
const sql = `${FULL_MSG_SELECT} WHERE msg.id IN (SELECT rowid FROM message_fts WHERE content MATCH ?) ${clause} ORDER BY msg.ts DESC LIMIT ? OFFSET ?`
|
||||
const rows = await executor.all<FullMessageRow>(sql, [matchQuery, ...params, limit, offset])
|
||||
return { messages: rows.map(mapMessageRow), total }
|
||||
}
|
||||
|
||||
/** LIKE-based keyword search with count and pagination. */
|
||||
export async function searchMessagesLikeAsync(
|
||||
executor: AsyncSqlExecutor,
|
||||
keywords: string[],
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
export { CURRENT_SCHEMA_VERSION, CHAT_DB_TABLES, CHAT_DB_INDEXES, CHAT_DB_SCHEMA, FTS_TABLE_SCHEMA } from './tables'
|
||||
export { CURRENT_SCHEMA_VERSION, CHAT_DB_TABLES, CHAT_DB_INDEXES, CHAT_DB_SCHEMA } from './tables'
|
||||
export { getSchemaVersion, setSchemaVersion, needsMigration, runMigrations } from './migrations'
|
||||
export type { Migration } from './migrations'
|
||||
@@ -96,17 +96,3 @@ export const CHAT_DB_INDEXES = `
|
||||
* For new database creation where deferred indexing is not needed.
|
||||
*/
|
||||
export const CHAT_DB_SCHEMA = CHAT_DB_TABLES + CHAT_DB_INDEXES
|
||||
|
||||
/**
|
||||
* FTS5 全文搜索虚拟表 DDL
|
||||
*
|
||||
* content='' 表示使用外部内容表模式(不存储原始内容,只存储索引)。
|
||||
* 填充需要在导入或迁移时手动执行。
|
||||
*/
|
||||
export const FTS_TABLE_SCHEMA = `
|
||||
CREATE VIRTUAL TABLE IF NOT EXISTS message_fts USING fts5(
|
||||
content,
|
||||
content='',
|
||||
content_rowid=id
|
||||
);
|
||||
`
|
||||
@@ -33,27 +33,4 @@ export function registerSessionIndexRoutes(server: FastifyInstance, ctx: Session
|
||||
sessionIndexService.clearIndex(adapter, request.params.id)
|
||||
return { success: true }
|
||||
})
|
||||
|
||||
server.get<{
|
||||
Params: { id: string }
|
||||
Querystring: { keywords: string; limit?: string; offset?: string }
|
||||
}>('/_web/sessions/:id/search/fts', async (request, reply) => {
|
||||
if (!sessionIndexService.getFtsStatus(adapter, request.params.id)) {
|
||||
return reply.code(400).send({ error: 'FTS index not built for this session' })
|
||||
}
|
||||
const keywords = request.query.keywords.split(/\s+/).filter(Boolean)
|
||||
if (keywords.length === 0) return { rowids: [], total: 0 }
|
||||
const limit = parseInt(request.query.limit || '100', 10)
|
||||
const offset = parseInt(request.query.offset || '0', 10)
|
||||
return sessionIndexService.searchFts(adapter, request.params.id, keywords, limit, offset)
|
||||
})
|
||||
|
||||
server.get<{ Params: { id: string } }>('/_web/sessions/:id/fts/status', async (request) => {
|
||||
return { hasFtsIndex: sessionIndexService.getFtsStatus(adapter, request.params.id) }
|
||||
})
|
||||
|
||||
server.post<{ Params: { id: string } }>('/_web/sessions/:id/fts/rebuild', async (request) => {
|
||||
const result = sessionIndexService.rebuildFts(adapter, request.params.id)
|
||||
return { success: true, indexed: result.indexed }
|
||||
})
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import path from 'node:path'
|
||||
import test from 'node:test'
|
||||
import Database from 'better-sqlite3'
|
||||
import { BetterSqliteAdapter } from '../better-sqlite3-adapter'
|
||||
import { buildFtsIndex, searchByFts } from './index'
|
||||
|
||||
const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
|
||||
|
||||
test('buildFtsIndex indexes sparse eligible rows exactly once across keyset batches', () => {
|
||||
const raw = new Database(':memory:', { nativeBinding })
|
||||
raw.exec(`
|
||||
CREATE TABLE message (
|
||||
id INTEGER PRIMARY KEY,
|
||||
type INTEGER NOT NULL,
|
||||
content TEXT
|
||||
)
|
||||
`)
|
||||
const insert = raw.prepare('INSERT INTO message (id, type, content) VALUES (?, ?, ?)')
|
||||
const expectedRowIds = [-2, 0]
|
||||
insert.run(-2, 0, 'negative marker')
|
||||
insert.run(0, 0, 'zero marker')
|
||||
|
||||
for (let index = 1; index <= 5_001; index++) {
|
||||
const id = index * 2
|
||||
insert.run(id, 0, `marker${index}`)
|
||||
expectedRowIds.push(id)
|
||||
}
|
||||
insert.run(10_003, 1, 'non text')
|
||||
insert.run(10_004, 0, '')
|
||||
insert.run(10_005, 0, null)
|
||||
insert.run(10_006, 0, ' ')
|
||||
|
||||
const db = new BetterSqliteAdapter(raw)
|
||||
const result = buildFtsIndex(db)
|
||||
const actualRowIds = raw
|
||||
.prepare('SELECT rowid FROM message_fts ORDER BY rowid')
|
||||
.all()
|
||||
.map((row) => (row as { rowid: number }).rowid)
|
||||
|
||||
assert.equal(result.indexed, expectedRowIds.length + 1)
|
||||
assert.deepEqual(actualRowIds, expectedRowIds)
|
||||
assert.deepEqual(searchByFts(db, ['marker1']).rowids, [2])
|
||||
assert.deepEqual(searchByFts(db, ['marker5001']).rowids, [10_002])
|
||||
raw.close()
|
||||
})
|
||||
@@ -1,147 +0,0 @@
|
||||
/**
|
||||
* FTS5 full-text search index operations (platform-agnostic).
|
||||
*
|
||||
* Extracted from electron/main/worker/query/fts.ts.
|
||||
* Works with DatabaseAdapter + NLP tokenizer from this package.
|
||||
*/
|
||||
|
||||
import type { DatabaseAdapter } from '@openchatlab/core'
|
||||
import { FTS_TABLE_SCHEMA } from '@openchatlab/core'
|
||||
import { tokenizeForFts, tokenizeQueryForFts } from '../nlp'
|
||||
|
||||
const BATCH_SIZE = 5000
|
||||
|
||||
/**
|
||||
* Check if FTS virtual table exists in the database.
|
||||
*/
|
||||
export function hasFtsTable(db: DatabaseAdapter): boolean {
|
||||
try {
|
||||
const row = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='message_fts'").get()
|
||||
return !!row
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create FTS virtual table if it doesn't exist.
|
||||
*/
|
||||
export function createFtsTable(db: DatabaseAdapter): void {
|
||||
db.exec(FTS_TABLE_SCHEMA)
|
||||
}
|
||||
|
||||
/**
|
||||
* Build FTS index from all text messages in the database.
|
||||
* Processes in batches for memory efficiency.
|
||||
*/
|
||||
export function buildFtsIndex(db: DatabaseAdapter): { indexed: number } {
|
||||
createFtsTable(db)
|
||||
|
||||
const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)')
|
||||
|
||||
let indexed = 0
|
||||
let lastId: number | null = null
|
||||
const selectFirstBatch = db.prepare(
|
||||
`SELECT id, content FROM message
|
||||
WHERE type = 0
|
||||
AND content IS NOT NULL
|
||||
AND content != ''
|
||||
ORDER BY id ASC
|
||||
LIMIT ?`
|
||||
)
|
||||
const selectNextBatch = db.prepare(
|
||||
`SELECT id, content FROM message
|
||||
WHERE type = 0
|
||||
AND content IS NOT NULL
|
||||
AND content != ''
|
||||
AND id > ?
|
||||
ORDER BY id ASC
|
||||
LIMIT ?`
|
||||
)
|
||||
|
||||
while (true) {
|
||||
const rows = (
|
||||
lastId === null ? selectFirstBatch.all(BATCH_SIZE) : selectNextBatch.all(lastId, BATCH_SIZE)
|
||||
) as Array<{ id: number; content: string }>
|
||||
if (rows.length === 0) break
|
||||
|
||||
db.transaction(() => {
|
||||
for (const row of rows) {
|
||||
const tokens = tokenizeForFts(row.content)
|
||||
if (tokens) {
|
||||
insertFts.run(row.id, tokens)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
indexed += rows.length
|
||||
lastId = rows[rows.length - 1].id
|
||||
}
|
||||
|
||||
return { indexed }
|
||||
}
|
||||
|
||||
/**
|
||||
* Rebuild FTS index by dropping and recreating.
|
||||
*/
|
||||
export function rebuildFtsIndex(db: DatabaseAdapter): { indexed: number } {
|
||||
if (hasFtsTable(db)) {
|
||||
db.exec('DROP TABLE message_fts')
|
||||
}
|
||||
return buildFtsIndex(db)
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert FTS entries for a batch of messages.
|
||||
* Used during incremental import to sync new messages.
|
||||
*/
|
||||
export function insertFtsEntries(db: DatabaseAdapter, entries: Array<{ id: number; content: string | null }>): void {
|
||||
if (!hasFtsTable(db)) return
|
||||
|
||||
const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)')
|
||||
|
||||
db.transaction(() => {
|
||||
for (const entry of entries) {
|
||||
if (entry.content) {
|
||||
const tokens = tokenizeForFts(entry.content)
|
||||
if (tokens) {
|
||||
insertFts.run(entry.id, tokens)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Search messages using FTS5, returning matching rowids.
|
||||
*/
|
||||
export function searchByFts(
|
||||
db: DatabaseAdapter,
|
||||
keywords: string[],
|
||||
limit = 1000,
|
||||
offset = 0
|
||||
): { rowids: number[]; total: number } {
|
||||
if (keywords.length === 0) return { rowids: [], total: 0 }
|
||||
|
||||
const matchQuery = tokenizeQueryForFts(keywords)
|
||||
if (!matchQuery) return { rowids: [], total: 0 }
|
||||
|
||||
try {
|
||||
const countRow = db.prepare('SELECT COUNT(*) as total FROM message_fts WHERE content MATCH ?').get(matchQuery) as
|
||||
| { total: number }
|
||||
| undefined
|
||||
const total = countRow?.total ?? 0
|
||||
|
||||
const rows = db
|
||||
.prepare(`SELECT rowid FROM message_fts WHERE content MATCH ? ORDER BY rank LIMIT ? OFFSET ?`)
|
||||
.all(matchQuery, limit, offset) as Array<{ rowid: number }>
|
||||
|
||||
return {
|
||||
rowids: rows.map((r) => r.rowid),
|
||||
total,
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[FTS] Search failed, query:', matchQuery, error)
|
||||
return { rowids: [], total: 0 }
|
||||
}
|
||||
}
|
||||
@@ -69,7 +69,6 @@ function createDeps(options?: {
|
||||
messageWriteMs: 0,
|
||||
nicknameHistoryMs: 0,
|
||||
indexCreationMs: 0,
|
||||
ftsMs: 0,
|
||||
checkpointMs: 0,
|
||||
sessionIndexMs: 0,
|
||||
postImportHookMs: 0,
|
||||
|
||||
@@ -137,12 +137,14 @@ test('imports ChatLab JSONL messages with numeric string timestamps consistently
|
||||
|
||||
const db = openBetterSqliteDatabase(dbPath, { readonly: true, nativeBinding })
|
||||
const row = db.prepare('SELECT ts, content FROM message').get() as { ts: number; content: string } | undefined
|
||||
const ftsTable = db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'message_fts'").get()
|
||||
db.close()
|
||||
|
||||
assert.deepEqual(row, {
|
||||
ts: 1780330832,
|
||||
content: 'hello from CipherTalk',
|
||||
})
|
||||
assert.equal(ftsTable, undefined)
|
||||
})
|
||||
|
||||
test('continues incremental session indexing with the stored gap threshold', async (t) => {
|
||||
|
||||
@@ -11,7 +11,6 @@
|
||||
import type { DatabaseAdapter } from '@openchatlab/core'
|
||||
import { generateSessionIndex, generateIncrementalSessionIndex, getSessionIndexStats } from '@openchatlab/core'
|
||||
import { streamParseFile, detectFormat, getFormatFeatureById, type ParseProgress } from '@openchatlab/parser'
|
||||
import { insertFtsEntries, hasFtsTable } from '../fts'
|
||||
import { generateFallbackMessageKey, registerMessageAndCheckDuplicate } from './message-deduplicator'
|
||||
import { normalizeSystemMemberName, SYSTEM_MEMBER_NAME, type ImportProgressCallback } from './streaming-importer'
|
||||
|
||||
@@ -296,8 +295,6 @@ export async function incrementalImport(
|
||||
}
|
||||
}
|
||||
|
||||
const newFtsEntries: Array<{ id: number; content: string | null }> = []
|
||||
|
||||
await streamParseFile(
|
||||
filePath,
|
||||
{
|
||||
@@ -389,7 +386,7 @@ export async function incrementalImport(
|
||||
}
|
||||
if (!memberId) continue
|
||||
|
||||
const msgResult = insertMessage.run(
|
||||
insertMessage.run(
|
||||
memberId,
|
||||
senderAccountName || null,
|
||||
senderGroupNickname || null,
|
||||
@@ -400,10 +397,6 @@ export async function incrementalImport(
|
||||
msg.platformMessageId || null
|
||||
)
|
||||
|
||||
newFtsEntries.push({
|
||||
id: Number((msgResult as any).lastInsertRowid ?? 0),
|
||||
content: msg.content || null,
|
||||
})
|
||||
if (timestamp < minWrittenTs) minWrittenTs = timestamp
|
||||
newMessageCount++
|
||||
}
|
||||
@@ -429,15 +422,6 @@ export async function incrementalImport(
|
||||
db.prepare('UPDATE meta SET imported_at = ?').run(Math.floor(Date.now() / 1000))
|
||||
}
|
||||
|
||||
// Incremental FTS update
|
||||
if (newFtsEntries.length > 0 && hasFtsTable(db)) {
|
||||
try {
|
||||
insertFtsEntries(db, newFtsEntries)
|
||||
} catch {
|
||||
/* FTS failure is non-fatal */
|
||||
}
|
||||
}
|
||||
|
||||
// Incremental session index (segment / message_context tables).
|
||||
// Use full rebuild for backfill batches: generateIncrementalSessionIndex
|
||||
// compares the first new message with the latest existing segment, so an
|
||||
|
||||
@@ -404,8 +404,9 @@ test('streamParseFileInfo preserves member metadata from a batched ChatLab JSONL
|
||||
assert.equal(row?.avatar, 'data:image/png;base64,AAAA')
|
||||
})
|
||||
|
||||
test('streamingImport updates avatars for members first created from message batches', async () => {
|
||||
test('streamingImport updates avatars without creating a per-session FTS table', async (t) => {
|
||||
const root = makeTempDir()
|
||||
t.after(() => fs.rmSync(root, { recursive: true, force: true }))
|
||||
const manifestPath = writeChunkedShuakamiQqExport(root)
|
||||
const dbPath = path.join(root, 'avatar-test.db')
|
||||
|
||||
@@ -417,10 +418,12 @@ test('streamingImport updates avatars for members first created from message bat
|
||||
const row = db.prepare('SELECT platform_id, avatar FROM member WHERE platform_id = ?').get('10001') as
|
||||
| { platform_id: string; avatar: string | null }
|
||||
| undefined
|
||||
const ftsTable = db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'message_fts'").get()
|
||||
db.close()
|
||||
|
||||
assert.equal(row?.platform_id, '10001')
|
||||
assert.equal(row?.avatar, 'data:image/png;base64,AAAA')
|
||||
assert.equal(ftsTable, undefined)
|
||||
})
|
||||
|
||||
test('streamingImport builds the session index exactly once with the requested gap threshold', async (t) => {
|
||||
@@ -454,7 +457,7 @@ test('streamingImport builds the session index exactly once with the requested g
|
||||
})
|
||||
|
||||
test(
|
||||
'streamingImport persists shuakami/qq-chat-exporter V4 native-first output with deduplication, reply and FTS',
|
||||
'streamingImport persists shuakami/qq-chat-exporter V4 native-first output without a derived FTS table',
|
||||
{
|
||||
skip: !isNativeFormatAvailable('shuakami-qq-exporter') && 'native shuakami/qq-chat-exporter kernel not built',
|
||||
},
|
||||
@@ -499,9 +502,7 @@ test(
|
||||
.get('qq-message-2') as {
|
||||
reply_to_message_id: string | null
|
||||
}
|
||||
const fts = db.prepare("SELECT COUNT(*) AS count FROM message_fts WHERE content MATCH 'searchable'").get() as {
|
||||
count: number
|
||||
}
|
||||
const ftsTable = db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'message_fts'").get()
|
||||
db.close()
|
||||
|
||||
assert.deepEqual(meta, {
|
||||
@@ -512,7 +513,7 @@ test(
|
||||
})
|
||||
assert.deepEqual(counts, { members: 2, messages: 2 })
|
||||
assert.equal(reply.reply_to_message_id, 'qq-message-1')
|
||||
assert.equal(fts.count, 1)
|
||||
assert.equal(ftsTable, undefined)
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
*
|
||||
* Extracted from electron/main/worker/import/streamImport.ts.
|
||||
* Streams parsed data directly into SQLite with batched transactions,
|
||||
* deferred index creation, nickname history tracking, and FTS indexing.
|
||||
* deferred index creation and nickname history tracking.
|
||||
*
|
||||
* Both Electron and Server/CLI use this module via dependency injection:
|
||||
* the caller provides a DatabaseAdapter, progress callback, and optional hooks.
|
||||
@@ -30,7 +30,6 @@ import {
|
||||
} from '@openchatlab/parser'
|
||||
import * as fs from 'fs'
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { buildFtsIndex } from '../fts'
|
||||
import { MessageBatchInserter, type MessageInsertRow } from './message-batch-inserter'
|
||||
import { createMessageDedupState, registerMessageAndCheckDuplicate, type DedupMessage } from './message-deduplicator'
|
||||
|
||||
@@ -64,7 +63,6 @@ export interface ImportStageTimings {
|
||||
messageWriteMs: number
|
||||
nicknameHistoryMs: number
|
||||
indexCreationMs: number
|
||||
ftsMs: number
|
||||
checkpointMs: number
|
||||
sessionIndexMs: number
|
||||
postImportHookMs: number
|
||||
@@ -302,7 +300,6 @@ async function streamImportSingle(
|
||||
messageWriteMs: 0,
|
||||
nicknameHistoryMs: 0,
|
||||
indexCreationMs: 0,
|
||||
ftsMs: 0,
|
||||
checkpointMs: 0,
|
||||
sessionIndexMs: 0,
|
||||
postImportHookMs: 0,
|
||||
@@ -760,18 +757,6 @@ async function streamImportSingle(
|
||||
sampleRss()
|
||||
logger?.perf('Indexes created', totalMessageCount)
|
||||
|
||||
// Build FTS index
|
||||
const ftsStartedAt = now()
|
||||
try {
|
||||
buildFtsIndex(db)
|
||||
logger?.perf('FTS index built', totalMessageCount)
|
||||
} catch (ftsError) {
|
||||
logger?.error('FTS index build failed (non-fatal)', ftsError instanceof Error ? ftsError : undefined)
|
||||
} finally {
|
||||
timings.ftsMs = elapsedMs(ftsStartedAt, now)
|
||||
sampleRss()
|
||||
}
|
||||
|
||||
// Final WAL checkpoint + session index + post-import hook
|
||||
onProgress({
|
||||
stage: 'indexing',
|
||||
@@ -817,7 +802,7 @@ async function streamImportSingle(
|
||||
|
||||
logger?.perfDetail(
|
||||
`[Stages] parser=${timings.parserMs.toFixed(1)}ms | message-write=${timings.messageWriteMs.toFixed(1)}ms | ` +
|
||||
`indexes=${timings.indexCreationMs.toFixed(1)}ms | fts=${timings.ftsMs.toFixed(1)}ms | ` +
|
||||
`indexes=${timings.indexCreationMs.toFixed(1)}ms | ` +
|
||||
`session-index=${timings.sessionIndexMs.toFixed(1)}ms | hook=${timings.postImportHookMs.toFixed(1)}ms`
|
||||
)
|
||||
logger?.perf('Import completed', totalMessageCount)
|
||||
|
||||
@@ -104,9 +104,6 @@ export {
|
||||
logSummary,
|
||||
} from './import'
|
||||
|
||||
// FTS5 full-text search operations
|
||||
export { hasFtsTable, createFtsTable, buildFtsIndex, rebuildFtsIndex, insertFtsEntries, searchByFts } from './fts'
|
||||
|
||||
// AI Logger & Error formatting
|
||||
export { AiLogger, extractErrorInfo, extractErrorStack, formatAIError } from './ai'
|
||||
export type { FormatAIErrorOptions } from './ai'
|
||||
|
||||
@@ -3,19 +3,16 @@ import fs from 'node:fs'
|
||||
import os from 'node:os'
|
||||
import path from 'node:path'
|
||||
import test from 'node:test'
|
||||
import { CHAT_DB_SCHEMA, FTS_TABLE_SCHEMA } from '@openchatlab/core'
|
||||
import { CHAT_DB_SCHEMA } from '@openchatlab/core'
|
||||
import { openBetterSqliteDatabase, type BetterSqliteAdapter } from '../../better-sqlite3-adapter'
|
||||
import { buildFtsIndex } from '../../fts'
|
||||
import { createChatDbMessageSource } from './message-source'
|
||||
import { createChatDbMessageRangeReader } from './message-range-reader'
|
||||
import { createChatDbFtsSearcher, extractFtsKeywords } from './fts-searcher'
|
||||
|
||||
function makeChatDb(): { db: BetterSqliteAdapter; dir: string } {
|
||||
const baseDir = process.env.CHATLAB_TEST_TMPDIR ?? (fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir())
|
||||
const dir = fs.mkdtempSync(path.join(baseDir, 'chatlab-chatdb-'))
|
||||
const db = openBetterSqliteDatabase(path.join(dir, 'chat.db'))
|
||||
db.exec(CHAT_DB_SCHEMA)
|
||||
db.exec(FTS_TABLE_SCHEMA)
|
||||
|
||||
db.exec(`
|
||||
INSERT INTO member (id, platform_id, account_name, group_nickname) VALUES
|
||||
@@ -27,7 +24,6 @@ function makeChatDb(): { db: BetterSqliteAdapter; dir: string } {
|
||||
(3, 1, 1020, 1, NULL),
|
||||
(4, 2, 1030, 0, '排期可以放到下周一');
|
||||
`)
|
||||
buildFtsIndex(db)
|
||||
return { db, dir }
|
||||
}
|
||||
|
||||
@@ -68,32 +64,3 @@ test('range reader returns inclusive id range including non-text messages', () =
|
||||
assert.equal(range[0].ts, 1010 * 1000)
|
||||
db.close()
|
||||
})
|
||||
|
||||
test('fts searcher maps natural-language query to ranked message hits', () => {
|
||||
const { db } = makeChatDb()
|
||||
const fts = createChatDbFtsSearcher(db)
|
||||
|
||||
const hits = fts.search('排期', 10)
|
||||
assert.ok(hits.some((h) => h.id === 1))
|
||||
assert.ok(hits.some((h) => h.id === 4))
|
||||
assert.ok(!hits.some((h) => h.id === 2))
|
||||
// ts 应为毫秒(chat DB ts 为秒,fts-searcher 乘 1000 转换)
|
||||
assert.ok(hits.every((h) => h.ts > 0))
|
||||
db.close()
|
||||
})
|
||||
|
||||
test('fts searcher returns empty for blank query', () => {
|
||||
const { db } = makeChatDb()
|
||||
const fts = createChatDbFtsSearcher(db)
|
||||
assert.deepEqual(fts.search(' ', 10), [])
|
||||
db.close()
|
||||
})
|
||||
|
||||
test('extractFtsKeywords tokenizes and dedupes regardless of segmentation granularity', () => {
|
||||
const keywords = extractFtsKeywords('需求 需求')
|
||||
assert.ok(keywords.length > 0)
|
||||
// 去重:无重复 token
|
||||
assert.equal(new Set(keywords).size, keywords.length)
|
||||
// 重复输入去重后与单次输入 token 集合一致
|
||||
assert.deepEqual(new Set(keywords), new Set(extractFtsKeywords('需求')))
|
||||
})
|
||||
@@ -1,51 +0,0 @@
|
||||
/**
|
||||
* 聊天库 FTS 检索适配器
|
||||
*
|
||||
* 把自然语言 query 用 jieba 分词成关键词,作为多关键词传入 searchByFts
|
||||
* (多关键词以 OR 组合,最大化召回;精度由 dense 路与 RRF 融合保证)。
|
||||
* 返回按 FTS rank 排序的 message_id。
|
||||
*/
|
||||
|
||||
import type { DatabaseAdapter } from '@openchatlab/core'
|
||||
import { searchByFts } from '../../fts'
|
||||
import { getJieba } from '../../nlp/segmenter'
|
||||
|
||||
interface FtsSearcher {
|
||||
search(query: string, topN: number): Array<{ id: number; ts: number }>
|
||||
}
|
||||
|
||||
/** 自然语言 query -> 去重关键词列表 */
|
||||
export function extractFtsKeywords(query: string): string[] {
|
||||
const trimmed = query.trim()
|
||||
if (!trimmed) return []
|
||||
try {
|
||||
const tokens = getJieba()
|
||||
.cut(trimmed, false)
|
||||
.map((t) => t.trim())
|
||||
.filter((t) => t.length > 0)
|
||||
return [...new Set(tokens)]
|
||||
} catch {
|
||||
return [...new Set(trimmed.split(/\s+/).filter((t) => t.length > 0))]
|
||||
}
|
||||
}
|
||||
|
||||
export function createChatDbFtsSearcher(db: DatabaseAdapter): FtsSearcher {
|
||||
return {
|
||||
search(query: string, topN: number): Array<{ id: number; ts: number }> {
|
||||
const keywords = extractFtsKeywords(query)
|
||||
if (keywords.length === 0) return []
|
||||
const { rowids } = searchByFts(db, keywords, topN, 0)
|
||||
if (rowids.length === 0) return []
|
||||
const placeholders = rowids.map(() => '?').join(', ')
|
||||
const tsRows = db.prepare(`SELECT id, ts FROM message WHERE id IN (${placeholders})`).all(...rowids) as Array<{
|
||||
id: number
|
||||
ts: number
|
||||
}>
|
||||
const tsById = new Map(tsRows.map((r) => [r.id, r.ts * 1000]))
|
||||
return rowids.flatMap((id) => {
|
||||
const ts = tsById.get(id)
|
||||
return ts !== undefined ? [{ id, ts }] : []
|
||||
})
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@
|
||||
* 语义索引模块(Phase 1)
|
||||
*
|
||||
* 独立于 dormant 的旧 `ai/rag` 模块,按 chunking-decision-final.md 重新实现:
|
||||
* 向量存储、parent/child chunker、embedding provider、混合召回与后台预热。
|
||||
* 向量存储、parent/child chunker、embedding provider、dense 召回与后台预热。
|
||||
*/
|
||||
|
||||
export { EmbeddingIndexStore } from './store'
|
||||
@@ -23,14 +23,7 @@ export type {
|
||||
EnableParams,
|
||||
ProgressPatch,
|
||||
} from './session-state-store'
|
||||
export type {
|
||||
ChunkRecord,
|
||||
ChunkInsert,
|
||||
ChunkStatus,
|
||||
DenseQueryParams,
|
||||
DenseQueryResult,
|
||||
MessageToChunkParams,
|
||||
} from './types'
|
||||
export type { ChunkRecord, ChunkInsert, ChunkStatus, DenseQueryParams, DenseQueryResult } from './types'
|
||||
export {
|
||||
CHUNKER_VERSION,
|
||||
STRATEGY_ID,
|
||||
@@ -120,8 +113,6 @@ export type {
|
||||
} from './warmup/runner'
|
||||
export { SemanticIndexJobQueue } from './warmup/job-queue'
|
||||
export type { SemanticIndexJob, SemanticIndexJobType, JobContext, JobExecutor } from './warmup/job-queue'
|
||||
export { reciprocalRankFusion } from './retrieval/rrf'
|
||||
export type { RrfResult } from './retrieval/rrf'
|
||||
export { semanticSearch } from './retrieval/semantic-search'
|
||||
export type {
|
||||
SemanticSearchDeps,
|
||||
@@ -131,7 +122,6 @@ export type {
|
||||
} from './retrieval/semantic-search'
|
||||
export { createChatDbMessageSource } from './chat-db/message-source'
|
||||
export { createChatDbMessageRangeReader } from './chat-db/message-range-reader'
|
||||
export { createChatDbFtsSearcher, extractFtsKeywords } from './chat-db/fts-searcher'
|
||||
export { assembleEvidence, formatEvidenceMessages } from './retrieval/evidence'
|
||||
export type {
|
||||
EvidenceMessage,
|
||||
|
||||
@@ -1,52 +0,0 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
import { reciprocalRankFusion } from './rrf'
|
||||
|
||||
test('single list preserves order with descending scores', () => {
|
||||
const result = reciprocalRankFusion([['a', 'b', 'c']], 60)
|
||||
assert.deepEqual(
|
||||
result.map((r) => r.id),
|
||||
['a', 'b', 'c']
|
||||
)
|
||||
assert.ok(result[0].score > result[1].score && result[1].score > result[2].score)
|
||||
assert.ok(Math.abs(result[0].score - 1 / 60) < 1e-9)
|
||||
})
|
||||
|
||||
test('items present in both lists outrank items in one list', () => {
|
||||
// x 在两个列表都靠前;y 仅在 dense 第一;z 仅在 fts 第一
|
||||
const dense = ['y', 'x']
|
||||
const fts = ['z', 'x']
|
||||
const result = reciprocalRankFusion([dense, fts], 60)
|
||||
assert.equal(result[0].id, 'x')
|
||||
})
|
||||
|
||||
test('item appearing in only one list still scored', () => {
|
||||
const result = reciprocalRankFusion([['a'], ['b']], 60)
|
||||
const ids = result.map((r) => r.id)
|
||||
assert.ok(ids.includes('a') && ids.includes('b'))
|
||||
assert.ok(Math.abs(result[0].score - result[1].score) < 1e-9)
|
||||
})
|
||||
|
||||
test('ties break by first-seen order for stability', () => {
|
||||
const result = reciprocalRankFusion([['a', 'b']], 60)
|
||||
// a 与 b 不同分,但验证同分场景:两个独立单元素列表同 rank
|
||||
const tie = reciprocalRankFusion([['a'], ['b']], 60)
|
||||
assert.deepEqual(
|
||||
tie.map((r) => r.id),
|
||||
['a', 'b']
|
||||
)
|
||||
assert.equal(result[0].id, 'a')
|
||||
})
|
||||
|
||||
test('smaller k increases score separation between ranks', () => {
|
||||
const small = reciprocalRankFusion([['a', 'b']], 1)
|
||||
const large = reciprocalRankFusion([['a', 'b']], 1000)
|
||||
const smallGap = small[0].score - small[1].score
|
||||
const largeGap = large[0].score - large[1].score
|
||||
assert.ok(smallGap > largeGap)
|
||||
})
|
||||
|
||||
test('empty input returns empty array', () => {
|
||||
assert.deepEqual(reciprocalRankFusion([], 60), [])
|
||||
assert.deepEqual(reciprocalRankFusion([[], []], 60), [])
|
||||
})
|
||||
@@ -1,35 +0,0 @@
|
||||
/**
|
||||
* Reciprocal Rank Fusion(RRF)
|
||||
*
|
||||
* chunking-decision-final.md 第 10 节:dense 与 FTS 各自独立排序,在 Node 层用 RRF 融合,
|
||||
* 不依赖跨库 join。每个 id 的融合分 = Σ 1 / (k + rank),rank 为该 id 在某个排序列表中的
|
||||
* 位次(0 为最优);id 未出现在某列表时该列表不贡献分数。
|
||||
*
|
||||
* 纯函数,按 id 操作(此处 id 即 chunkId),与具体检索来源解耦。
|
||||
*/
|
||||
|
||||
export interface RrfResult {
|
||||
id: string
|
||||
score: number
|
||||
}
|
||||
|
||||
/**
|
||||
* @param rankedLists 多个已排序 id 列表,每个列表内 index 0 为最相关
|
||||
* @param k RRF 常数,默认 60(业界常用值,降低高位排名的极端权重)
|
||||
*/
|
||||
export function reciprocalRankFusion(rankedLists: string[][], k = 60): RrfResult[] {
|
||||
const scores = new Map<string, number>()
|
||||
const firstSeen = new Map<string, number>()
|
||||
let seq = 0
|
||||
|
||||
for (const list of rankedLists) {
|
||||
list.forEach((id, rank) => {
|
||||
scores.set(id, (scores.get(id) ?? 0) + 1 / (k + rank))
|
||||
if (!firstSeen.has(id)) firstSeen.set(id, seq++)
|
||||
})
|
||||
}
|
||||
|
||||
return [...scores.entries()]
|
||||
.map(([id, score]) => ({ id, score }))
|
||||
.sort((a, b) => b.score - a.score || firstSeen.get(a.id)! - firstSeen.get(b.id)!)
|
||||
}
|
||||
@@ -3,7 +3,6 @@ import fs from 'node:fs'
|
||||
import os from 'node:os'
|
||||
import path from 'node:path'
|
||||
import test from 'node:test'
|
||||
import Database from 'better-sqlite3'
|
||||
import { EmbeddingIndexStore } from './store'
|
||||
import type { ChunkRecord } from './types'
|
||||
|
||||
@@ -102,98 +101,6 @@ test('insertChunk rejects embedding whose length mismatches dim', () => {
|
||||
store.close()
|
||||
})
|
||||
|
||||
test('mapMessageToChunk returns the chunk whose ts range covers the message', () => {
|
||||
const dbPath = makeTempDbPath()
|
||||
const store = new EmbeddingIndexStore(dbPath)
|
||||
|
||||
// ts 用 messageId * 100 映射,三段连续
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c1', startMessageId: 0, endMessageId: 11, startTs: 0, endTs: 1100 }),
|
||||
[1, 0, 0, 0]
|
||||
)
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c2', startMessageId: 12, endMessageId: 23, startTs: 1200, endTs: 2300 }),
|
||||
[0, 1, 0, 0]
|
||||
)
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c3', startMessageId: 24, endMessageId: 35, startTs: 2400, endTs: 3500 }),
|
||||
[0, 0, 1, 0]
|
||||
)
|
||||
|
||||
const params = { dbPathHash: 'dbA', modelId: 'qwen3', strategyId: 'balanced' }
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 15, messageTs: 1500 })?.chunkId, 'c2')
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 0, messageTs: 0 })?.chunkId, 'c1')
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 35, messageTs: 3500 })?.chunkId, 'c3')
|
||||
|
||||
store.close()
|
||||
})
|
||||
|
||||
test('mapMessageToChunk returns null when message ts falls in a gap between chunks', () => {
|
||||
const dbPath = makeTempDbPath()
|
||||
const store = new EmbeddingIndexStore(dbPath)
|
||||
|
||||
// chunk 之间留 ts 空洞
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c1', startMessageId: 0, endMessageId: 11, startTs: 0, endTs: 1100 }),
|
||||
[1, 0, 0, 0]
|
||||
)
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c2', startMessageId: 100, endMessageId: 111, startTs: 10000, endTs: 11100 }),
|
||||
[0, 1, 0, 0]
|
||||
)
|
||||
|
||||
const params = { dbPathHash: 'dbA', modelId: 'qwen3', strategyId: 'balanced' }
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 50, messageTs: 5000 }), null)
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 200, messageTs: 20000 }), null)
|
||||
|
||||
store.close()
|
||||
})
|
||||
|
||||
test('mapMessageToChunk breaks ties when multiple chunks share the same start_ts', () => {
|
||||
const dbPath = makeTempDbPath()
|
||||
const store = new EmbeddingIndexStore(dbPath)
|
||||
|
||||
// 同一秒内两个 chunk:c1 覆盖 id 1-2,c2 覆盖 id 3-4,start_ts 相同
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c1', startMessageId: 1, endMessageId: 2, startTs: 1000, endTs: 1000 }),
|
||||
[1, 0, 0, 0]
|
||||
)
|
||||
store.insertChunk(
|
||||
baseRecord({ chunkId: 'c2', startMessageId: 3, endMessageId: 4, startTs: 1000, endTs: 1000 }),
|
||||
[0, 1, 0, 0]
|
||||
)
|
||||
|
||||
const params = { dbPathHash: 'dbA', modelId: 'qwen3', strategyId: 'balanced' }
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 1, messageTs: 1000 })?.chunkId, 'c1')
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 3, messageTs: 1000 })?.chunkId, 'c2')
|
||||
assert.equal(store.mapMessageToChunk({ ...params, messageId: 4, messageTs: 1000 })?.chunkId, 'c2')
|
||||
|
||||
store.close()
|
||||
})
|
||||
|
||||
test('message-to-chunk lookup uses a timestamp-leading index query plan', () => {
|
||||
const dbPath = makeTempDbPath()
|
||||
const store = new EmbeddingIndexStore(dbPath)
|
||||
store.close()
|
||||
|
||||
const db = new Database(dbPath, { readonly: true })
|
||||
const plan = db
|
||||
.prepare(
|
||||
`EXPLAIN QUERY PLAN
|
||||
SELECT chunk_id FROM chunk_vector_index
|
||||
WHERE db_path_hash = ? AND model_id = ? AND strategy_id = ?
|
||||
AND (start_ts < CAST(? AS INTEGER)
|
||||
OR (start_ts = CAST(? AS INTEGER) AND start_message_id <= CAST(? AS INTEGER)))
|
||||
ORDER BY start_ts DESC, start_message_id DESC LIMIT 1`
|
||||
)
|
||||
.all('dbA', 'qwen3', 'balanced', 1000, 1000, 1) as Array<{ detail: string }>
|
||||
const searchStep = plan.find((step) => step.detail.includes('SEARCH chunk_vector_index'))
|
||||
|
||||
assert.ok(searchStep)
|
||||
assert.match(searchStep.detail, /USING INDEX idx_chunk_ts_range/)
|
||||
db.close()
|
||||
})
|
||||
|
||||
test('data persists across store reopen', () => {
|
||||
const dbPath = makeTempDbPath()
|
||||
const store = new EmbeddingIndexStore(dbPath)
|
||||
|
||||
@@ -5,21 +5,13 @@
|
||||
* - 初始化元数据表与按维度的 vec0 表。
|
||||
* - 写入 chunk 元数据与向量(同一事务,rowid 关联)。
|
||||
* - dense ANN 查询(限定 db_path_hash + model_id 分区)。
|
||||
* - FTS message_id -> chunk 的 O(log n) 范围映射。
|
||||
*
|
||||
* P0-1 硬约定:vec0 的 INTEGER 列与 k 参数绑定必须使用 CAST(? AS INTEGER)。
|
||||
*/
|
||||
|
||||
import Database from 'better-sqlite3'
|
||||
import * as sqliteVec from 'sqlite-vec'
|
||||
import type {
|
||||
ChunkInsert,
|
||||
ChunkRecord,
|
||||
ChunkStatus,
|
||||
DenseQueryParams,
|
||||
DenseQueryResult,
|
||||
MessageToChunkParams,
|
||||
} from './types'
|
||||
import type { ChunkInsert, ChunkRecord, ChunkStatus, DenseQueryParams, DenseQueryResult } from './types'
|
||||
import { EMBEDDING_INDEX_SCHEMA, vecTableName, vecTableSchema } from './schema'
|
||||
|
||||
/** 加载 sqlite-vec 扩展。Electron 打包后可注入自定义实现处理 asar 解包路径。 */
|
||||
@@ -193,36 +185,6 @@ export class EmbeddingIndexStore {
|
||||
return rows.map((row) => ({ chunkId: row.chunk_id, distance: row.distance, record: rowToRecord(row) }))
|
||||
}
|
||||
|
||||
/**
|
||||
* FTS message -> chunk 映射。按 (ts, id) 复合查找:取 (start_ts, start_message_id) <= (messageTs, messageId)
|
||||
* 的最大值,再校验 end 覆盖。
|
||||
* - 用 ts 而非 ID 做主键:回填旧消息后 ID 不再单调
|
||||
* - 用 (ts, id) 复合而非仅 ts:同秒多 chunk 时 ts 相同,需 id 打破 tie
|
||||
*/
|
||||
mapMessageToChunk(params: MessageToChunkParams): ChunkRecord | null {
|
||||
const row = this.db
|
||||
.prepare(
|
||||
`SELECT ${CHUNK_COLUMNS} FROM chunk_vector_index
|
||||
WHERE db_path_hash = ? AND model_id = ? AND strategy_id = ?
|
||||
AND (start_ts < CAST(? AS INTEGER)
|
||||
OR (start_ts = CAST(? AS INTEGER) AND start_message_id <= CAST(? AS INTEGER)))
|
||||
ORDER BY start_ts DESC, start_message_id DESC LIMIT 1`
|
||||
)
|
||||
.get(
|
||||
params.dbPathHash,
|
||||
params.modelId,
|
||||
params.strategyId,
|
||||
params.messageTs,
|
||||
params.messageTs,
|
||||
params.messageId
|
||||
) as ChunkRow | undefined
|
||||
|
||||
if (!row) return null
|
||||
if (row.end_ts < params.messageTs) return null
|
||||
if (row.end_ts === params.messageTs && row.end_message_id < params.messageId) return null
|
||||
return rowToRecord(row)
|
||||
}
|
||||
|
||||
/** 删除某聊天库(全模型)的所有 chunk 元数据与向量,返回删除的元数据行数 */
|
||||
deleteByDbPathHash(dbPathHash: string): number {
|
||||
return this.db.transaction(() => {
|
||||
|
||||
@@ -56,13 +56,3 @@ export interface DenseQueryResult {
|
||||
distance: number
|
||||
record: ChunkRecord
|
||||
}
|
||||
|
||||
/** message_id -> chunk 范围映射参数 */
|
||||
export interface MessageToChunkParams {
|
||||
dbPathHash: string
|
||||
modelId: string
|
||||
strategyId: string
|
||||
messageId: number
|
||||
/** 消息时间戳(毫秒),用于 ts-based 区间映射,与 chunk_vector_index 的 start_ts/end_ts 单位一致 */
|
||||
messageTs: number
|
||||
}
|
||||
@@ -37,15 +37,7 @@ export type {
|
||||
} from './owner-profile-service'
|
||||
|
||||
// Session index service
|
||||
export {
|
||||
generateIndex,
|
||||
generateIncrementalIndex,
|
||||
clearIndex,
|
||||
getFtsStatus,
|
||||
searchFts,
|
||||
rebuildFts,
|
||||
getAllIndexStats,
|
||||
} from './session-index-service'
|
||||
export { generateIndex, generateIncrementalIndex, clearIndex, getAllIndexStats } from './session-index-service'
|
||||
export type { SessionIndexStatusItem } from './session-index-service'
|
||||
|
||||
// Summary service
|
||||
|
||||
@@ -79,6 +79,14 @@ test('analyzes a new push payload with the same optional-account and member sema
|
||||
assert.equal(outcome.result.batch.writtenCount, analysis.result.newMessageCount)
|
||||
assert.equal(outcome.result.batch.duplicateCount, analysis.result.duplicateCount)
|
||||
assert.equal(outcome.result.updates.membersAdded, analysis.result.newMemberCount)
|
||||
|
||||
const db = manager.openRawSessionDatabase('push-analysis', { readonly: true })
|
||||
try {
|
||||
const ftsTable = db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'message_fts'").get()
|
||||
assert.equal(ftsTable, undefined)
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
test('analyzes incremental push deduplication without writing', async (t) => {
|
||||
|
||||
@@ -15,7 +15,6 @@ import {
|
||||
} from '@openchatlab/core'
|
||||
import type { DatabaseAdapter } from '@openchatlab/core'
|
||||
import type { DatabaseManager } from '../database-manager'
|
||||
import { buildFtsIndex, insertFtsEntries } from '../fts'
|
||||
import { writeParseResultToDb } from '../import'
|
||||
import { ImportInProgressError, withDataDirImportLock } from '../import/import-lock'
|
||||
import {
|
||||
@@ -299,7 +298,6 @@ function writeMessages(
|
||||
duplicateCount: number
|
||||
membersAdded: number
|
||||
minWrittenTs: number
|
||||
ftsEntries: Array<{ id: number; content: string | null }>
|
||||
} {
|
||||
const insertMsg = db.prepare(
|
||||
`INSERT INTO message (sender_id, sender_account_name, sender_group_nickname, ts, type, content, reply_to_message_id, platform_message_id)
|
||||
@@ -315,7 +313,6 @@ function writeMessages(
|
||||
let duplicateCount = 0
|
||||
let membersAdded = 0
|
||||
let minWrittenTs = Infinity
|
||||
const ftsEntries: Array<{ id: number; content: string | null }> = []
|
||||
|
||||
const sorted = [...messages].sort((a, b) => a.timestamp - b.timestamp)
|
||||
|
||||
@@ -342,7 +339,7 @@ function writeMessages(
|
||||
}
|
||||
if (!memberId) continue
|
||||
|
||||
const result = insertMsg.run(
|
||||
insertMsg.run(
|
||||
memberId,
|
||||
normalizeSenderAccountName(msg.sender, msg.accountName) || null,
|
||||
normalizeSenderGroupNickname(msg.sender, msg.groupNickname) || null,
|
||||
@@ -352,13 +349,12 @@ function writeMessages(
|
||||
msg.replyToMessageId || null,
|
||||
msg.platformMessageId || null
|
||||
)
|
||||
ftsEntries.push({ id: Number(result.lastInsertRowid), content: msg.content ?? null })
|
||||
if (msg.timestamp < minWrittenTs) minWrittenTs = msg.timestamp
|
||||
writtenCount++
|
||||
}
|
||||
})
|
||||
|
||||
return { writtenCount, duplicateCount, membersAdded, minWrittenTs, ftsEntries }
|
||||
return { writtenCount, duplicateCount, membersAdded, minWrittenTs }
|
||||
}
|
||||
|
||||
function fullImport(
|
||||
@@ -419,8 +415,6 @@ function fullImport(
|
||||
}))
|
||||
)
|
||||
|
||||
buildFtsIndex(db)
|
||||
|
||||
try {
|
||||
generateSessionIndex(db)
|
||||
} catch {
|
||||
@@ -502,17 +496,8 @@ function writeIncrementalImport(db: DatabaseAdapter, payload: PushImportPayload)
|
||||
duplicateCount,
|
||||
membersAdded: autoCreatedMembersAdded,
|
||||
minWrittenTs,
|
||||
ftsEntries,
|
||||
} = writeMessages(db, payload.messages!, dedupState)
|
||||
|
||||
if (ftsEntries.length > 0) {
|
||||
try {
|
||||
insertFtsEntries(db, ftsEntries)
|
||||
} catch {
|
||||
/* non-fatal */
|
||||
}
|
||||
}
|
||||
|
||||
if (writtenCount > 0) {
|
||||
try {
|
||||
const sessionGapThreshold = getSessionIndexStats(db).gapThreshold
|
||||
|
||||
@@ -10,7 +10,6 @@ import {
|
||||
clearSessionIndex as coreClearSessionIndex,
|
||||
getSessionIndexStats,
|
||||
} from '@openchatlab/core'
|
||||
import { hasFtsTable, searchByFts, rebuildFtsIndex } from '../fts'
|
||||
import type { SessionRuntimeAdapter } from './adapters'
|
||||
|
||||
export function generateIndex(adapter: SessionRuntimeAdapter, sessionId: string, gapThreshold: number = 1800): number {
|
||||
@@ -32,27 +31,6 @@ export function clearIndex(adapter: SessionRuntimeAdapter, sessionId: string): v
|
||||
coreClearSessionIndex(db)
|
||||
}
|
||||
|
||||
export function getFtsStatus(adapter: SessionRuntimeAdapter, sessionId: string): boolean {
|
||||
const db = adapter.ensureReadonly(sessionId)
|
||||
return hasFtsTable(db)
|
||||
}
|
||||
|
||||
export function searchFts(
|
||||
adapter: SessionRuntimeAdapter,
|
||||
sessionId: string,
|
||||
keywords: string[],
|
||||
limit: number,
|
||||
offset: number
|
||||
) {
|
||||
const db = adapter.ensureReadonly(sessionId)
|
||||
return searchByFts(db, keywords, limit, offset)
|
||||
}
|
||||
|
||||
export function rebuildFts(adapter: SessionRuntimeAdapter, sessionId: string) {
|
||||
const db = adapter.ensureWritable(sessionId)
|
||||
return rebuildFtsIndex(db)
|
||||
}
|
||||
|
||||
export interface SessionIndexStatusItem {
|
||||
sessionId: string
|
||||
hasIndex: boolean
|
||||
|
||||
@@ -81,7 +81,6 @@ export interface ImportPerformanceDiagnosticsInfo {
|
||||
messageWriteMs: number
|
||||
nicknameHistoryMs: number
|
||||
indexCreationMs: number
|
||||
ftsMs: number
|
||||
checkpointMs: number
|
||||
sessionIndexMs: number
|
||||
postImportHookMs: number
|
||||
|
||||
Reference in new issue
Block a user