mirror of
https://github.com/hellodigua/ChatLab.git
synced 2026-10-26 02:04:37 +08:00
feat(node-runtime): extract IncrementalImporter from Electron worker
Platform-agnostic incremental import with dual-path dedup (platformMessageId + content hash), member upsert, meta patching, FTS sync, and injectable dependencies (IncrementalImportDeps).
This commit is contained in:
digua committed
1 parent
bf91c4b7df
commit
2e564e8063
3 files changed
+456
-2
No files matched your search
@@ -0,0 +1,436 @@
|
||||
/**
|
||||
* Platform-agnostic incremental importer.
|
||||
*
|
||||
* Extracted from electron/main/worker/import/incrementalImport.ts.
|
||||
* Appends new messages to an existing session, using dual-path dedup:
|
||||
* platformMessageId (preferred) + content hash (fallback).
|
||||
*
|
||||
* Callers provide DatabaseAdapter + progress callback via dependency injection.
|
||||
*/
|
||||
|
||||
import type { DatabaseAdapter } from '@openchatlab/core'
|
||||
import { generateMessageKey } from '@openchatlab/core'
|
||||
import { streamParseFile, detectFormat, type ParseProgress } from '@openchatlab/parser'
|
||||
import { insertFtsEntries, hasFtsTable } from '../fts'
|
||||
import type { ImportProgressCallback } from './streaming-importer'
|
||||
|
||||
// ==================== Public interfaces ====================
|
||||
|
||||
export interface ImportOptions {
|
||||
metaUpdateMode?: 'patch' | 'none'
|
||||
memberUpdateMode?: 'upsert' | 'none'
|
||||
}
|
||||
|
||||
export interface IncrementalAnalyzeResult {
|
||||
newMessageCount: number
|
||||
duplicateCount: number
|
||||
totalInFile: number
|
||||
error?: string
|
||||
}
|
||||
|
||||
interface ErrorSample {
|
||||
index: number
|
||||
reason: string
|
||||
detail: string
|
||||
}
|
||||
|
||||
export interface IncrementalImportResult {
|
||||
success: boolean
|
||||
newMessageCount: number
|
||||
error?: string
|
||||
batch?: {
|
||||
receivedCount: number
|
||||
writtenCount: number
|
||||
duplicateCount: number
|
||||
errorCount: number
|
||||
errorReasonCounts: Record<string, number>
|
||||
errorSample: ErrorSample[]
|
||||
}
|
||||
session?: {
|
||||
totalCount: number
|
||||
memberCount: number
|
||||
firstTimestamp: number
|
||||
lastTimestamp: number
|
||||
}
|
||||
updates?: {
|
||||
metaUpdated: boolean
|
||||
membersAdded: number
|
||||
membersUpdated: number
|
||||
}
|
||||
}
|
||||
|
||||
export interface IncrementalImportDeps {
|
||||
/** Open existing session DB for read-only access (analyze) or read-write (import). */
|
||||
openDatabase(sessionId: string, readonly?: boolean): DatabaseAdapter
|
||||
onProgress: ImportProgressCallback
|
||||
/** Optional hook after incremental import (e.g. update overview cache). */
|
||||
postImportHook?: (db: DatabaseAdapter, sessionId: string) => void
|
||||
}
|
||||
|
||||
// ==================== Internal helpers ====================
|
||||
|
||||
function loadExistingDedup(db: DatabaseAdapter): {
|
||||
existingPlatformMsgIds: Set<string>
|
||||
existingKeys: Set<string>
|
||||
} {
|
||||
const existingPlatformMsgIds = new Set<string>()
|
||||
const existingKeys = new Set<string>()
|
||||
|
||||
const pmidRows = db
|
||||
.prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL')
|
||||
.all() as Array<{ platform_message_id: string }>
|
||||
for (const row of pmidRows) {
|
||||
existingPlatformMsgIds.add(row.platform_message_id)
|
||||
}
|
||||
|
||||
const hashRows = db
|
||||
.prepare(
|
||||
`SELECT ts, m.platform_id as sender_platform_id, content
|
||||
FROM message msg
|
||||
JOIN member m ON msg.sender_id = m.id`
|
||||
)
|
||||
.all() as Array<{ ts: number; sender_platform_id: string; content: string | null }>
|
||||
for (const row of hashRows) {
|
||||
existingKeys.add(generateMessageKey(row.ts, row.sender_platform_id, row.content))
|
||||
}
|
||||
|
||||
return { existingPlatformMsgIds, existingKeys }
|
||||
}
|
||||
|
||||
function isDuplicate(
|
||||
msg: { platformMessageId?: string; timestamp: number; senderPlatformId: string; content: string | null },
|
||||
existingPlatformMsgIds: Set<string>,
|
||||
existingKeys: Set<string>
|
||||
): boolean {
|
||||
if (msg.platformMessageId) {
|
||||
if (existingPlatformMsgIds.has(msg.platformMessageId)) return true
|
||||
existingPlatformMsgIds.add(msg.platformMessageId)
|
||||
return false
|
||||
}
|
||||
const key = generateMessageKey(msg.timestamp, msg.senderPlatformId, msg.content)
|
||||
if (existingKeys.has(key)) return true
|
||||
existingKeys.add(key)
|
||||
return false
|
||||
}
|
||||
|
||||
// ==================== Analyze (dry-run) ====================
|
||||
|
||||
export async function analyzeIncrementalImport(
|
||||
sessionId: string,
|
||||
filePath: string,
|
||||
deps: IncrementalImportDeps
|
||||
): Promise<IncrementalAnalyzeResult> {
|
||||
const formatFeature = detectFormat(filePath)
|
||||
if (!formatFeature) {
|
||||
return { error: 'error.unrecognized_format', newMessageCount: 0, duplicateCount: 0, totalInFile: 0 }
|
||||
}
|
||||
|
||||
let db: DatabaseAdapter
|
||||
try {
|
||||
db = deps.openDatabase(sessionId, true)
|
||||
} catch {
|
||||
return { error: 'error.session_not_found', newMessageCount: 0, duplicateCount: 0, totalInFile: 0 }
|
||||
}
|
||||
|
||||
const { existingPlatformMsgIds, existingKeys } = loadExistingDedup(db)
|
||||
db.close()
|
||||
|
||||
let totalInFile = 0
|
||||
let newMessageCount = 0
|
||||
let duplicateCount = 0
|
||||
|
||||
await streamParseFile(filePath, {
|
||||
onMeta: () => {},
|
||||
onMembers: () => {},
|
||||
onProgress: (progress: ParseProgress) => {
|
||||
deps.onProgress(progress)
|
||||
},
|
||||
onMessageBatch: (batch) => {
|
||||
for (const msg of batch) {
|
||||
totalInFile++
|
||||
if (isDuplicate(msg, existingPlatformMsgIds, existingKeys)) {
|
||||
duplicateCount++
|
||||
} else {
|
||||
newMessageCount++
|
||||
}
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
return { newMessageCount, duplicateCount, totalInFile }
|
||||
}
|
||||
|
||||
// ==================== Execute incremental import ====================
|
||||
|
||||
export async function incrementalImport(
|
||||
sessionId: string,
|
||||
filePath: string,
|
||||
deps: IncrementalImportDeps,
|
||||
options?: ImportOptions
|
||||
): Promise<IncrementalImportResult> {
|
||||
const formatFeature = detectFormat(filePath)
|
||||
if (!formatFeature) {
|
||||
return { success: false, newMessageCount: 0, error: 'error.unrecognized_format' }
|
||||
}
|
||||
|
||||
let db: DatabaseAdapter
|
||||
try {
|
||||
db = deps.openDatabase(sessionId, false)
|
||||
} catch {
|
||||
return { success: false, newMessageCount: 0, error: 'error.session_not_found' }
|
||||
}
|
||||
|
||||
const metaUpdateMode = options?.metaUpdateMode ?? 'patch'
|
||||
const memberUpdateMode = options?.memberUpdateMode ?? 'upsert'
|
||||
|
||||
try {
|
||||
const { existingPlatformMsgIds, existingKeys } = loadExistingDedup(db)
|
||||
|
||||
const memberIdMap = new Map<string, number>()
|
||||
const existingMembers = db.prepare('SELECT id, platform_id FROM member').all() as Array<{
|
||||
id: number
|
||||
platform_id: string
|
||||
}>
|
||||
for (const m of existingMembers) {
|
||||
memberIdMap.set(m.platform_id, m.id)
|
||||
}
|
||||
|
||||
const upsertMember = db.prepare(`
|
||||
INSERT INTO member (platform_id, account_name, group_nickname, avatar, roles)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(platform_id) DO UPDATE SET
|
||||
account_name = COALESCE(NULLIF(excluded.account_name, ''), account_name),
|
||||
group_nickname = COALESCE(NULLIF(excluded.group_nickname, ''), group_nickname),
|
||||
avatar = COALESCE(NULLIF(excluded.avatar, ''), avatar),
|
||||
roles = CASE WHEN excluded.roles != '[]' THEN excluded.roles ELSE roles END
|
||||
`)
|
||||
|
||||
const insertMemberMinimal = db.prepare(`
|
||||
INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar)
|
||||
VALUES (?, ?, ?, ?)
|
||||
`)
|
||||
|
||||
const getMemberId = db.prepare('SELECT id FROM member WHERE platform_id = ?')
|
||||
|
||||
const insertMessage = db.prepare(`
|
||||
INSERT INTO message (sender_id, sender_account_name, sender_group_nickname, ts, type, content, reply_to_message_id, platform_message_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`)
|
||||
|
||||
const updateMeta = db.prepare(`
|
||||
UPDATE meta SET
|
||||
name = COALESCE(NULLIF(?, ''), name),
|
||||
group_id = COALESCE(NULLIF(?, ''), group_id),
|
||||
group_avatar = COALESCE(NULLIF(?, ''), group_avatar),
|
||||
owner_id = COALESCE(NULLIF(?, ''), owner_id),
|
||||
imported_at = ?
|
||||
`)
|
||||
|
||||
db.exec('BEGIN TRANSACTION')
|
||||
|
||||
let newMessageCount = 0
|
||||
let duplicateCount = 0
|
||||
let processedCount = 0
|
||||
let metaUpdated = false
|
||||
let membersAdded = 0
|
||||
let membersUpdated = 0
|
||||
let errorCount = 0
|
||||
const errorReasonCounts: Record<string, number> = {}
|
||||
const errorSamples: ErrorSample[] = []
|
||||
const MAX_ERROR_SAMPLES = 5
|
||||
const BATCH_SIZE = 5000
|
||||
|
||||
function trackError(index: number, reason: string, detail: string) {
|
||||
errorCount++
|
||||
errorReasonCounts[reason] = (errorReasonCounts[reason] || 0) + 1
|
||||
if (errorSamples.length < MAX_ERROR_SAMPLES) {
|
||||
errorSamples.push({ index, reason, detail })
|
||||
}
|
||||
}
|
||||
|
||||
const newFtsEntries: Array<{ id: number; content: string | null }> = []
|
||||
|
||||
await streamParseFile(filePath, {
|
||||
onMeta: (meta) => {
|
||||
if (metaUpdateMode === 'none') return
|
||||
updateMeta.run(
|
||||
meta.name || '',
|
||||
meta.groupId || '',
|
||||
meta.groupAvatar || '',
|
||||
meta.ownerId || '',
|
||||
Math.floor(Date.now() / 1000)
|
||||
)
|
||||
metaUpdated = true
|
||||
},
|
||||
onMembers: (members) => {
|
||||
if (memberUpdateMode === 'none') return
|
||||
for (const m of members) {
|
||||
const existed = memberIdMap.has(m.platformId)
|
||||
upsertMember.run(
|
||||
m.platformId,
|
||||
m.accountName || null,
|
||||
m.groupNickname || null,
|
||||
m.avatar || null,
|
||||
m.roles ? JSON.stringify(m.roles) : '[]'
|
||||
)
|
||||
if (!existed) {
|
||||
const row = getMemberId.get(m.platformId) as { id: number } | undefined
|
||||
if (row) memberIdMap.set(m.platformId, row.id)
|
||||
membersAdded++
|
||||
} else {
|
||||
membersUpdated++
|
||||
}
|
||||
}
|
||||
},
|
||||
onProgress: (progress: ParseProgress) => {
|
||||
deps.onProgress(progress)
|
||||
},
|
||||
onMessageBatch: (batch) => {
|
||||
for (const msg of batch) {
|
||||
processedCount++
|
||||
|
||||
if (!msg.senderPlatformId) {
|
||||
trackError(processedCount, 'MISSING_SENDER', 'sender field is empty')
|
||||
continue
|
||||
}
|
||||
if (msg.timestamp === undefined || msg.timestamp === null) {
|
||||
trackError(processedCount, 'MISSING_TIMESTAMP', 'timestamp field is missing')
|
||||
continue
|
||||
}
|
||||
if (typeof msg.timestamp !== 'number' || msg.timestamp <= 0 || !isFinite(msg.timestamp)) {
|
||||
trackError(processedCount, 'INVALID_TIMESTAMP', `timestamp value: ${msg.timestamp}`)
|
||||
continue
|
||||
}
|
||||
|
||||
if (isDuplicate(msg, existingPlatformMsgIds, existingKeys)) {
|
||||
duplicateCount++
|
||||
continue
|
||||
}
|
||||
|
||||
let memberId = memberIdMap.get(msg.senderPlatformId)
|
||||
if (!memberId) {
|
||||
insertMemberMinimal.run(
|
||||
msg.senderPlatformId,
|
||||
msg.senderAccountName || null,
|
||||
msg.senderGroupNickname || null,
|
||||
null
|
||||
)
|
||||
const row = getMemberId.get(msg.senderPlatformId) as { id: number } | undefined
|
||||
if (row) {
|
||||
memberId = row.id
|
||||
memberIdMap.set(msg.senderPlatformId, memberId)
|
||||
membersAdded++
|
||||
}
|
||||
}
|
||||
if (!memberId) continue
|
||||
|
||||
const msgResult = insertMessage.run(
|
||||
memberId,
|
||||
msg.senderAccountName || null,
|
||||
msg.senderGroupNickname || null,
|
||||
msg.timestamp,
|
||||
msg.type,
|
||||
msg.content || null,
|
||||
msg.replyToMessageId || null,
|
||||
msg.platformMessageId || null
|
||||
)
|
||||
|
||||
newFtsEntries.push({
|
||||
id: Number((msgResult as any).lastInsertRowid ?? 0),
|
||||
content: msg.content || null,
|
||||
})
|
||||
newMessageCount++
|
||||
}
|
||||
|
||||
if (processedCount % BATCH_SIZE === 0) {
|
||||
deps.onProgress({
|
||||
stage: 'saving',
|
||||
bytesRead: 0,
|
||||
totalBytes: 0,
|
||||
messagesProcessed: processedCount,
|
||||
percentage: 50,
|
||||
message: `Processed ${processedCount}, added ${newMessageCount}`,
|
||||
})
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
db.exec('COMMIT')
|
||||
|
||||
if (!metaUpdated) {
|
||||
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 */
|
||||
}
|
||||
}
|
||||
|
||||
const sessionStats = db
|
||||
.prepare(
|
||||
`SELECT
|
||||
COUNT(*) as totalCount,
|
||||
MIN(ts) as firstTimestamp,
|
||||
MAX(ts) as lastTimestamp
|
||||
FROM message`
|
||||
)
|
||||
.get() as { totalCount: number; firstTimestamp: number; lastTimestamp: number }
|
||||
const memberCountRow = db.prepare('SELECT COUNT(*) as count FROM member').get() as { count: number }
|
||||
|
||||
// Post-import hook (e.g. overview cache)
|
||||
try {
|
||||
deps.postImportHook?.(db, sessionId)
|
||||
} catch {
|
||||
/* non-fatal */
|
||||
}
|
||||
|
||||
db.close()
|
||||
|
||||
deps.onProgress({
|
||||
stage: 'done',
|
||||
bytesRead: 0,
|
||||
totalBytes: 0,
|
||||
messagesProcessed: processedCount,
|
||||
percentage: 100,
|
||||
message: `Import complete, added ${newMessageCount} messages`,
|
||||
})
|
||||
|
||||
return {
|
||||
success: true,
|
||||
newMessageCount,
|
||||
batch: {
|
||||
receivedCount: processedCount,
|
||||
writtenCount: newMessageCount,
|
||||
duplicateCount,
|
||||
errorCount,
|
||||
errorReasonCounts,
|
||||
errorSample: errorSamples,
|
||||
},
|
||||
session: {
|
||||
totalCount: sessionStats.totalCount,
|
||||
memberCount: memberCountRow.count,
|
||||
firstTimestamp: sessionStats.firstTimestamp,
|
||||
lastTimestamp: sessionStats.lastTimestamp,
|
||||
},
|
||||
updates: {
|
||||
metaUpdated,
|
||||
membersAdded,
|
||||
membersUpdated,
|
||||
},
|
||||
}
|
||||
} catch (error) {
|
||||
try {
|
||||
db.exec('ROLLBACK')
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
db.close()
|
||||
|
||||
console.error('[IncrementalImport] Error:', error)
|
||||
return { success: false, newMessageCount: 0, error: String(error) }
|
||||
}
|
||||
}
|
||||
@@ -24,3 +24,10 @@ export type {
|
||||
StreamParseFileInfoResult,
|
||||
StreamParseFileInfoDeps,
|
||||
} from './streaming-importer'
|
||||
export { analyzeIncrementalImport, incrementalImport } from './incremental-importer'
|
||||
export type {
|
||||
ImportOptions,
|
||||
IncrementalAnalyzeResult,
|
||||
IncrementalImportResult,
|
||||
IncrementalImportDeps,
|
||||
} from './incremental-importer'
|
||||
@@ -7,8 +7,15 @@
|
||||
|
||||
export { BetterSqliteAdapter, openBetterSqliteDatabase } from './better-sqlite3-adapter'
|
||||
|
||||
// Import data writing + perf logging + streaming importer
|
||||
export { writeParseResultToDb, streamingImport, analyzeNewImport, streamParseFileInfo } from './import'
|
||||
// Import data writing + perf logging + streaming importer + incremental importer
|
||||
export {
|
||||
writeParseResultToDb,
|
||||
streamingImport,
|
||||
analyzeNewImport,
|
||||
streamParseFileInfo,
|
||||
analyzeIncrementalImport,
|
||||
incrementalImport,
|
||||
} from './import'
|
||||
export type {
|
||||
ImportMeta,
|
||||
WriteParseResultStats,
|
||||
@@ -21,6 +28,10 @@ export type {
|
||||
AnalyzeNewImportResult,
|
||||
StreamParseFileInfoResult,
|
||||
StreamParseFileInfoDeps,
|
||||
ImportOptions,
|
||||
IncrementalAnalyzeResult,
|
||||
IncrementalImportResult,
|
||||
IncrementalImportDeps,
|
||||
} from './import'
|
||||
export {
|
||||
LogLevel,
|
||||
|
||||
Reference in new issue
Block a user