fix(import): normalize jsonl timestamps for incremental import

This commit is contained in:
Yui authored and digua committed 2026-06-09 09:54:50 +08:00
1 parent 44cc22817b
commit d70ab042ce
2 files changed
+104 -4

No files matched your search

@@ -0,0 +1,91 @@
import assert from 'node:assert/strict'
import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import test from 'node:test'
import { CHAT_DB_SCHEMA } from '@openchatlab/core'
import { openBetterSqliteDatabase } from '../better-sqlite3-adapter'
import { analyzeIncrementalImport, incrementalImport, type IncrementalImportDeps } from './incremental-importer'
const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node')
function makeTempDir(): string {
const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()
return fs.mkdtempSync(path.join(baseDir, 'chatlab-incremental-import-'))
}
function writeChatLabJsonl(filePath: string): void {
const lines = [
{
_type: 'header',
chatlab: { version: '0.0.2', exportedAt: 1780330900 },
meta: { name: 'CipherTalk Export', platform: 'wechat', type: 'private' },
},
{
_type: 'member',
platformId: 'wxid_alice',
accountName: 'Alice',
},
{
_type: 'message',
sender: 'wxid_alice',
accountName: 'Alice',
timestamp: '1780330832',
type: 0,
content: 'hello from CipherTalk',
},
]
fs.writeFileSync(filePath, `${lines.map((line) => JSON.stringify(line)).join('\n')}\n`, 'utf8')
}
function seedSessionDb(dbPath: string): void {
const db = openBetterSqliteDatabase(dbPath, { nativeBinding })
db.exec(CHAT_DB_SCHEMA)
db.prepare(
`INSERT INTO meta (name, platform, type, imported_at, schema_version)
VALUES (?, ?, ?, ?, ?)`
).run('Existing Session', 'wechat', 'private', 1780330000, 6)
db.close()
}
function createDeps(dbPath: string): IncrementalImportDeps {
return {
openDatabase: (_sessionId, readonly = false) => openBetterSqliteDatabase(dbPath, { readonly, nativeBinding }),
onProgress: () => {},
}
}
test('imports ChatLab JSONL messages with numeric string timestamps consistently with analysis', async (t) => {
const tempDir = makeTempDir()
t.after(() => fs.rmSync(tempDir, { recursive: true, force: true }))
const dbPath = path.join(tempDir, 'session.db')
const filePath = path.join(tempDir, 'cipher-talk.jsonl')
seedSessionDb(dbPath)
writeChatLabJsonl(filePath)
const deps = createDeps(dbPath)
const analysis = await analyzeIncrementalImport('session', filePath, deps)
assert.deepEqual(analysis, {
newMessageCount: 1,
duplicateCount: 0,
totalInFile: 1,
})
const result = await incrementalImport('session', filePath, deps)
assert.equal(result.success, true)
assert.equal(result.newMessageCount, 1)
assert.equal(result.batch?.writtenCount, 1)
assert.equal(result.batch?.errorCount, 0)
const db = openBetterSqliteDatabase(dbPath, { readonly: true, nativeBinding })
const row = db.prepare('SELECT ts, content FROM message').get() as { ts: number; content: string } | undefined
db.close()
assert.deepEqual(row, {
ts: 1780330832,
content: 'hello from CipherTalk',
})
})
@@ -113,6 +113,11 @@ function isDuplicate(
return false
}
function normalizeTimestamp(timestamp: unknown): number | null {
const value = typeof timestamp === 'string' && timestamp.trim() !== '' ? Number(timestamp) : timestamp
return typeof value === 'number' && value > 0 && Number.isFinite(value) ? value : null
}
// ==================== Analyze (dry-run) ====================
export async function analyzeIncrementalImport(
@@ -148,7 +153,10 @@ export async function analyzeIncrementalImport(
onMessageBatch: (batch) => {
for (const msg of batch) {
totalInFile++
if (isDuplicate(msg, existingPlatformMsgIds, existingKeys)) {
const timestamp = normalizeTimestamp(msg.timestamp)
if (timestamp === null) continue
if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys)) {
duplicateCount++
} else {
newMessageCount++
@@ -297,12 +305,13 @@ export async function incrementalImport(
trackError(processedCount, 'MISSING_TIMESTAMP', 'timestamp field is missing')
continue
}
if (typeof msg.timestamp !== 'number' || msg.timestamp <= 0 || !isFinite(msg.timestamp)) {
const timestamp = normalizeTimestamp(msg.timestamp)
if (timestamp === null) {
trackError(processedCount, 'INVALID_TIMESTAMP', `timestamp value: ${msg.timestamp}`)
continue
}
if (isDuplicate(msg, existingPlatformMsgIds, existingKeys)) {
if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys)) {
duplicateCount++
continue
}
@@ -328,7 +337,7 @@ export async function incrementalImport(
memberId,
msg.senderAccountName || null,
msg.senderGroupNickname || null,
msg.timestamp,
timestamp,
msg.type,
msg.content || null,
msg.replyToMessageId || null,