From d70ab042ced5c815c298bfd6ca17c6f03f4ed863 Mon Sep 17 00:00:00 2001 From: Yui Date: Mon, 8 Jun 2026 13:18:55 +0900 Subject: [PATCH] fix(import): normalize jsonl timestamps for incremental import --- .../src/import/incremental-importer.test.ts | 91 +++++++++++++++++++ .../src/import/incremental-importer.ts | 17 +++- 2 files changed, 104 insertions(+), 4 deletions(-) create mode 100644 packages/node-runtime/src/import/incremental-importer.test.ts diff --git a/packages/node-runtime/src/import/incremental-importer.test.ts b/packages/node-runtime/src/import/incremental-importer.test.ts new file mode 100644 index 00000000..ab76ec87 --- /dev/null +++ b/packages/node-runtime/src/import/incremental-importer.test.ts @@ -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', + }) +}) diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index adec4e88..12ebaa86 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -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,