From db26950d4b334e2a6cc721c67ddca14731ace30d Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 13 Jul 2026 15:25:54 +0800 Subject: [PATCH] fix(import): harden desktop shared imports Use the long-running worker timeout and refresh Desktop caches after successful push imports. Align auto-created member metadata and counts across initial and incremental imports, exclude SYSTEM from member updates, and keep existing-session writes validated and atomic. --- .../main/worker/import/pushImport.test.ts | 49 ++++ apps/desktop/main/worker/import/pushImport.ts | 27 +- apps/desktop/main/worker/workerManager.ts | 4 +- .../main/worker/workerTimeoutPolicy.test.ts | 7 +- .../main/worker/workerTimeoutPolicy.ts | 10 + .../src/services/push-importer.test.ts | 265 ++++++++++++++++++ .../src/services/push-importer.ts | 128 +++++++-- 7 files changed, 458 insertions(+), 32 deletions(-) diff --git a/apps/desktop/main/worker/import/pushImport.test.ts b/apps/desktop/main/worker/import/pushImport.test.ts index a50e55c9..fb236e97 100644 --- a/apps/desktop/main/worker/import/pushImport.test.ts +++ b/apps/desktop/main/worker/import/pushImport.test.ts @@ -3,6 +3,12 @@ import fs from 'node:fs' import os from 'node:os' import path from 'node:path' import test from 'node:test' +import { + CACHE_KEY_MEMBERS, + getCache, + setCache, + type MembersCache, +} from '@openchatlab/node-runtime/src/cache/session-cache' import { initDbDir } from '../core/dbCore' import { pushImport } from './pushImport' @@ -35,3 +41,46 @@ test('Desktop worker push import uses the shared writer and canonical database d assert.equal(fs.existsSync(path.join(dbDir, 'worker-session.db')), true) assert.equal(fs.existsSync(path.join(root, 'worker-session.db')), false) }) + +test('Desktop worker push import refreshes caches after duplicate-only member updates', async (t) => { + const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir() + const root = fs.mkdtempSync(path.join(baseDir, 'chatlab-worker-push-cache-')) + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + + const cacheDir = path.join(root, 'cache') + const queryCacheDir = path.join(cacheDir, 'query') + initDbDir(path.join(root, 'databases'), cacheDir, path.join(root, 'temp'), nativeBinding, path.join(root, 'logs')) + + const payload = { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Worker Push Import', platform: 'wechat', type: 'private' as const }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + platformMessageId: 'msg-1', + sender: 'wxid_alice', + timestamp: 1780330832, + type: 0, + content: 'hello', + }, + ], + } + + const initialOutcome = await pushImport('cache-session', payload) + assert.equal(initialOutcome.ok, true, JSON.stringify(initialOutcome)) + + setCache('cache-session', CACHE_KEY_MEMBERS, { members: { 1: { name: 'Alice', count: 1 } } }, cacheDir) + setCache('cache-session', 'analysis:test', { stale: true }, queryCacheDir) + + const outcome = await pushImport('cache-session', { + ...payload, + members: [{ platformId: 'wxid_alice', accountName: 'Alice', groupNickname: 'Updated Alice' }], + }) + + assert.equal(outcome.ok, true, JSON.stringify(outcome)) + if (!outcome.ok) return + assert.equal(outcome.result.batch.writtenCount, 0) + assert.equal(outcome.result.updates.membersUpdated, 1) + assert.equal(getCache('cache-session', CACHE_KEY_MEMBERS, cacheDir)?.members[1]?.name, 'Updated Alice') + assert.equal(getCache('cache-session', 'analysis:test', queryCacheDir), null) +}) diff --git a/apps/desktop/main/worker/import/pushImport.ts b/apps/desktop/main/worker/import/pushImport.ts index 4fe818e3..748dfb93 100644 --- a/apps/desktop/main/worker/import/pushImport.ts +++ b/apps/desktop/main/worker/import/pushImport.ts @@ -1,19 +1,39 @@ import * as fs from 'node:fs' import * as path from 'node:path' import { BetterSqliteAdapter } from '@openchatlab/node-runtime/src/better-sqlite3-adapter' +import { computeAndSetOverviewCache, deleteSessionCache } from '@openchatlab/node-runtime/src/cache/session-cache' import { executePushImportUnlocked, type PushImportOutcome, type PushImportPayload, } from '@openchatlab/node-runtime/src/services/push-importer' -import { closeDatabase, getDbDir, openRawDatabase } from '../core/dbCore' +import { closeDatabase, getCacheDir, getDbDir, openRawDatabase } from '../core/dbCore' function getDbPath(sessionId: string): string { return path.join(getDbDir(), `${sessionId}.db`) } +function refreshImportCaches(sessionId: string): void { + const cacheDir = getCacheDir() + try { + const rawDb = openRawDatabase(getDbPath(sessionId)) + try { + computeAndSetOverviewCache(new BetterSqliteAdapter(rawDb), sessionId, cacheDir) + } finally { + rawDb.close() + } + } catch (error) { + // Cache refresh is best-effort and must not turn a successful import into a failure. + console.warn('[Worker] pushImport: failed to refresh overview cache', error) + } + + if (cacheDir) { + deleteSessionCache(sessionId, path.join(cacheDir, 'query')) + } +} + export async function pushImport(sessionId: string, payload: PushImportPayload): Promise { - return executePushImportUnlocked( + const outcome = await executePushImportUnlocked( { getDbPath, openDatabase(id, options) { @@ -40,4 +60,7 @@ export async function pushImport(sessionId: string, payload: PushImportPayload): sessionId, payload ) + + if (outcome.ok) refreshImportCaches(sessionId) + return outcome } diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index 14094557..4713eb6a 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -22,7 +22,7 @@ import { withDataDirImportLock, } from '@openchatlab/node-runtime' import type { PushImportOutcome, PushImportPayload } from '@openchatlab/node-runtime' -import { isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' +import { getWorkerRequestTimeoutMs, isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' interface WorkerRequestOptions { timeoutMs?: number @@ -212,7 +212,7 @@ function sendToWorker(type: string, payload: any, options: number | WorkerReq } } - const timeoutMs = typeof options === 'number' ? options : (options.timeoutMs ?? 60000) + const timeoutMs = typeof options === 'number' ? options : (options.timeoutMs ?? getWorkerRequestTimeoutMs(type)) const restartOnTimeout = typeof options === 'number' ? isRestartableReadOnlyRequestType(type) diff --git a/apps/desktop/main/worker/workerTimeoutPolicy.test.ts b/apps/desktop/main/worker/workerTimeoutPolicy.test.ts index 8194245b..0988e344 100644 --- a/apps/desktop/main/worker/workerTimeoutPolicy.test.ts +++ b/apps/desktop/main/worker/workerTimeoutPolicy.test.ts @@ -1,6 +1,6 @@ import { describe, it } from 'node:test' import assert from 'node:assert/strict' -import { isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' +import { getWorkerRequestTimeoutMs, isRestartableReadOnlyRequestType } from './workerTimeoutPolicy' describe('worker timeout policy', () => { it('allows timed-out read-only AI/debug requests to restart the worker', () => { @@ -26,4 +26,9 @@ describe('worker timeout policy', () => { assert.equal(isRestartableReadOnlyRequestType('exportFilterResultToFile'), false) assert.equal(isRestartableReadOnlyRequestType('unknownFutureWorkerRequest'), false) }) + + it('gives push imports the long-running worker timeout', () => { + assert.equal(getWorkerRequestTimeoutMs('pushImport'), 10 * 60 * 1000) + assert.equal(getWorkerRequestTimeoutMs('getSession'), 60 * 1000) + }) }) diff --git a/apps/desktop/main/worker/workerTimeoutPolicy.ts b/apps/desktop/main/worker/workerTimeoutPolicy.ts index 2649366b..0228cfa0 100644 --- a/apps/desktop/main/worker/workerTimeoutPolicy.ts +++ b/apps/desktop/main/worker/workerTimeoutPolicy.ts @@ -42,6 +42,16 @@ const RESTARTABLE_READ_ONLY_REQUEST_TYPES = new Set([ 'getSegmentSummaries', ]) +const DEFAULT_WORKER_REQUEST_TIMEOUT_MS = 60 * 1000 +const LONG_RUNNING_WORKER_REQUEST_TIMEOUT_MS = 10 * 60 * 1000 +const LONG_RUNNING_WORKER_REQUEST_TYPES = new Set(['pushImport']) + export function isRestartableReadOnlyRequestType(type: string): boolean { return RESTARTABLE_READ_ONLY_REQUEST_TYPES.has(type) } + +export function getWorkerRequestTimeoutMs(type: string): number { + return LONG_RUNNING_WORKER_REQUEST_TYPES.has(type) + ? LONG_RUNNING_WORKER_REQUEST_TIMEOUT_MS + : DEFAULT_WORKER_REQUEST_TIMEOUT_MS +} diff --git a/packages/node-runtime/src/services/push-importer.test.ts b/packages/node-runtime/src/services/push-importer.test.ts index 1efed946..73b63f5d 100644 --- a/packages/node-runtime/src/services/push-importer.test.ts +++ b/packages/node-runtime/src/services/push-importer.test.ts @@ -222,6 +222,40 @@ test('deduplicates initial push batch when a no-platform-id message is repeated } }) +test('counts explicit and auto-created members in initial push import results', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const outcome = await pushImport(manager, 'initial-auto-member-count', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Initial Auto Member Count', platform: 'wechat', type: 'group' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }, + { sender: 'wxid_bob', accountName: 'Bob', timestamp: 1780330833, type: 0, content: 'hi' }, + { sender: 'SYSTEM', timestamp: 1780330834, type: 80, content: 'Bob joined the group' }, + ], + }) + + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.equal(outcome.result.session.memberCount, 2) + assert.deepEqual(outcome.result.updates, { + metaUpdated: true, + membersAdded: 2, + membersUpdated: 0, + }) + + const db = manager.openRawSessionDatabase('initial-auto-member-count', { readonly: true }) + try { + const row = db.prepare('SELECT COUNT(*) as count FROM member').get() as { count: number } + assert.equal(row.count, 3) + } finally { + db.close() + } +}) + test('deduplicates incremental push when a later platform-id message matches an existing content hash', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -330,6 +364,55 @@ test('keeps distinct platform messages even when timestamp, sender and content m } }) +test('preserves metadata and counts senders auto-created by incremental push imports', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const initialOutcome = await pushImport(manager, 'incremental-auto-member', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Incremental Auto Member', platform: 'wechat', type: 'group' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [{ sender: 'wxid_alice', timestamp: 1780330832, type: 0, content: 'hello' }], + }) + assert.equal(initialOutcome.ok, true) + + const outcome = await pushImport(manager, 'incremental-auto-member', { + messages: [ + { + sender: 'wxid_bob', + accountName: 'Bob', + groupNickname: 'Product', + timestamp: 1780330833, + type: 0, + content: 'hi', + }, + ], + }) + + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.deepEqual(outcome.result.updates, { + metaUpdated: false, + membersAdded: 1, + membersUpdated: 0, + }) + + const db = manager.openRawSessionDatabase('incremental-auto-member', { readonly: true }) + try { + const member = db + .prepare('SELECT platform_id, account_name, group_nickname FROM member WHERE platform_id = ?') + .get('wxid_bob') as { platform_id: string; account_name: string; group_nickname: string } + assert.deepEqual(member, { + platform_id: 'wxid_bob', + account_name: 'Bob', + group_nickname: 'Product', + }) + } finally { + db.close() + } +}) + test('creates push-import databases only inside the canonical databases directory', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -587,6 +670,144 @@ test('rejects non-string reply targets before applying incremental metadata upda } }) +test('rejects malformed message metadata without partially applying incremental updates', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const initialOutcome = await pushImport(manager, 'invalid-message-metadata', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Original Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice', groupNickname: 'Original Member' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'hello', + }, + ], + }) + assert.equal(initialOutcome.ok, true) + + const malformedPayload = { + meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', groupNickname: 'Updated Member' }], + messages: [ + { + sender: 'wxid_alice', + accountName: { invalid: true }, + timestamp: 1780330833, + type: 0, + content: 'should fail', + }, + ], + } as unknown as PushImportPayload + + const outcome = await pushImport(manager, 'invalid-message-metadata', malformedPayload) + const db = manager.openRawSessionDatabase('invalid-message-metadata', { readonly: true }) + try { + const meta = db.prepare('SELECT name FROM meta').get() as { name: string } + const member = db.prepare('SELECT group_nickname FROM member WHERE platform_id = ?').get('wxid_alice') as { + group_nickname: string + } + const messages = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } + + assert.deepEqual( + { + reason: outcome.ok ? 'ok' : outcome.reason, + metaName: meta.name, + memberName: member.group_nickname, + messageCount: messages.count, + }, + { + reason: 'invalid_payload', + metaName: 'Original Session', + memberName: 'Original Member', + messageCount: 1, + } + ) + } finally { + db.close() + } +}) + +test('rolls back metadata and member updates when incremental message writes fail', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const initialOutcome = await pushImport(manager, 'atomic-message-write', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Original Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice', groupNickname: 'Original Member' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330832, + type: 0, + content: 'hello', + }, + ], + }) + assert.equal(initialOutcome.ok, true) + + const writableDb = manager.openRawSessionDatabase('atomic-message-write', { readonly: false }) + try { + writableDb.exec(` + CREATE TRIGGER fail_incremental_message_insert + BEFORE INSERT ON message + BEGIN + SELECT RAISE(ABORT, 'forced message write failure'); + END + `) + } finally { + writableDb.close() + } + + const outcome = await pushImport(manager, 'atomic-message-write', { + meta: { name: 'Renamed Session', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', groupNickname: 'Updated Member' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 1780330833, + type: 0, + content: 'should fail', + }, + ], + }) + + const db = manager.openRawSessionDatabase('atomic-message-write', { readonly: true }) + try { + const meta = db.prepare('SELECT name FROM meta').get() as { name: string } + const member = db.prepare('SELECT group_nickname FROM member WHERE platform_id = ?').get('wxid_alice') as { + group_nickname: string + } + const messages = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number } + + assert.deepEqual( + { + reason: outcome.ok ? 'ok' : outcome.reason, + metaName: meta.name, + memberName: member.group_nickname, + messageCount: messages.count, + }, + { + reason: 'import_failed', + metaName: 'Original Session', + memberName: 'Original Member', + messageCount: 1, + } + ) + } finally { + db.close() + } +}) + test('treats auto-created SYSTEM sender as a system member during initial import', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -609,6 +830,7 @@ test('treats auto-created SYSTEM sender as a system member during initial import assert.equal(outcome.ok, true) if (!outcome.ok) return assert.equal(outcome.result.session.memberCount, 0) + assert.equal(outcome.result.updates.membersAdded, 0) const db = manager.openRawSessionDatabase('system-sender', { readonly: true }) try { @@ -625,6 +847,44 @@ test('treats auto-created SYSTEM sender as a system member during initial import } }) +test('excludes an auto-created SYSTEM sender from incremental member counts', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const initialOutcome = await pushImport(manager, 'incremental-system-count', { + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'System Count', platform: 'wechat', type: 'group' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [{ sender: 'wxid_alice', accountName: 'Alice', timestamp: 1780330832, type: 0, content: 'hello' }], + }) + assert.equal(initialOutcome.ok, true) + + const outcome = await pushImport(manager, 'incremental-system-count', { + messages: [{ sender: 'SYSTEM', timestamp: 1780330833, type: 80, content: 'Bob joined the group' }], + }) + + assert.equal(outcome.ok, true) + if (!outcome.ok) return + assert.equal(outcome.result.session.memberCount, 1) + assert.deepEqual(outcome.result.updates, { + metaUpdated: false, + membersAdded: 0, + membersUpdated: 0, + }) + + const db = manager.openRawSessionDatabase('incremental-system-count', { readonly: true }) + try { + const members = db.prepare('SELECT platform_id, account_name FROM member ORDER BY platform_id').all() + assert.deepEqual(members, [ + { platform_id: 'SYSTEM', account_name: '系统消息' }, + { platform_id: 'wxid_alice', account_name: 'Alice' }, + ]) + } finally { + db.close() + } +}) + test('keeps SYSTEM member canonical during incremental member upserts', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) @@ -662,6 +922,11 @@ test('keeps SYSTEM member canonical during incremental member upserts', async (t assert.equal(outcome.ok, true) if (!outcome.ok) return assert.equal(outcome.result.session.memberCount, 0) + assert.deepEqual(outcome.result.updates, { + metaUpdated: false, + membersAdded: 0, + membersUpdated: 0, + }) const db = manager.openRawSessionDatabase('system-member-upsert', { readonly: true }) try { diff --git a/packages/node-runtime/src/services/push-importer.ts b/packages/node-runtime/src/services/push-importer.ts index 55dff1ee..63a191ad 100644 --- a/packages/node-runtime/src/services/push-importer.ts +++ b/packages/node-runtime/src/services/push-importer.ts @@ -82,19 +82,48 @@ export interface PushImportExecutionDeps { deleteDatabase(sessionId: string): void } +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function validateOptionalString(value: unknown, field: string): string | null { + return value !== undefined && typeof value !== 'string' ? `${field} must be a string` : null +} + function validatePayload(payload: PushImportPayload, isNew: boolean): string | null { + if (!isRecord(payload)) return 'payload must be an object' + const messages = payload.messages if (!Array.isArray(messages) || messages.length === 0) return 'messages is required and must be a non-empty array' + if (payload.chatlab !== undefined) { + if (!isRecord(payload.chatlab)) return 'chatlab must be an object' + if (typeof payload.chatlab.version !== 'string' || payload.chatlab.version.length === 0) + return 'chatlab.version must be a string' + if (typeof payload.chatlab.exportedAt !== 'number' || !Number.isFinite(payload.chatlab.exportedAt)) + return 'chatlab.exportedAt must be a number' + const generatorError = validateOptionalString(payload.chatlab.generator, 'chatlab.generator') + if (generatorError) return generatorError + } + if (isNew) { if (!payload.chatlab) return 'chatlab is required for new sessions' if (!payload.meta) return 'meta is required for new sessions' - const m = payload.meta - if (!m.name) return 'meta.name is required' - if (!m.platform) return 'meta.platform is required' - if (!m.type) return 'meta.type is required' } + if (payload.meta !== undefined) { + if (!isRecord(payload.meta)) return 'meta must be an object' + const metaFields = ['name', 'platform', 'type', 'groupId', 'groupAvatar', 'ownerId'] as const + for (const field of metaFields) { + const error = validateOptionalString(payload.meta[field], `meta.${field}`) + if (error) return error + } + if (isNew && !payload.meta.name) return 'meta.name is required' + if (isNew && !payload.meta.platform) return 'meta.platform is required' + if (isNew && !payload.meta.type) return 'meta.type is required' + } + + if (payload.options !== undefined && !isRecord(payload.options)) return 'options must be an object' const { metaUpdateMode, memberUpdateMode } = payload.options ?? {} if (metaUpdateMode !== undefined && metaUpdateMode !== 'patch' && metaUpdateMode !== 'none') { return `options.metaUpdateMode must be 'patch' or 'none'` @@ -106,17 +135,36 @@ function validatePayload(payload: PushImportPayload, isNew: boolean): string | n if (payload.members !== undefined && !Array.isArray(payload.members)) return 'members must be an array' if (payload.members) { for (let i = 0; i < payload.members.length; i++) { - if (typeof payload.members[i].platformId !== 'string' || payload.members[i].platformId.length === 0) + const member = payload.members[i] + if (!isRecord(member)) return `members[${i}] must be an object` + if (typeof member.platformId !== 'string' || member.platformId.length === 0) return `members[${i}].platformId must be a string` + for (const field of ['accountName', 'groupNickname', 'avatar'] as const) { + const error = validateOptionalString(member[field], `members[${i}].${field}`) + if (error) return error + } + if (member.roles !== undefined) { + if (!Array.isArray(member.roles)) return `members[${i}].roles must be an array` + for (let j = 0; j < member.roles.length; j++) { + const role = member.roles[j] + if (!isRecord(role) || typeof role.id !== 'string' || role.id.length === 0) + return `members[${i}].roles[${j}].id must be a string` + } + } } } for (let i = 0; i < messages.length; i++) { const msg = messages[i] + if (!isRecord(msg)) return `messages[${i}] must be an object` if (typeof msg.sender !== 'string' || msg.sender.length === 0) return `messages[${i}].sender must be a string` - if (typeof msg.timestamp !== 'number' || msg.timestamp <= 0) + if (typeof msg.timestamp !== 'number' || !Number.isFinite(msg.timestamp) || msg.timestamp <= 0) return `messages[${i}].timestamp must be a positive number` - if (typeof msg.type !== 'number') return `messages[${i}].type must be a number` + if (typeof msg.type !== 'number' || !Number.isFinite(msg.type)) return `messages[${i}].type must be a number` + for (const field of ['accountName', 'groupNickname'] as const) { + const error = validateOptionalString(msg[field], `messages[${i}].${field}`) + if (error) return error + } if (msg.content !== undefined && msg.content !== null && typeof msg.content !== 'string') return `messages[${i}].content must be a string or null` if (msg.platformMessageId !== undefined && typeof msg.platformMessageId !== 'string') @@ -148,6 +196,10 @@ function normalizeSenderGroupNickname(sender: string, groupNickname: string | un return sender === SYSTEM_SENDER_ID ? SYSTEM_MEMBER_NAME : groupNickname } +function isCountedMember(platformId: string): boolean { + return platformId !== SYSTEM_SENDER_ID +} + function writeMessages( db: DatabaseAdapter, messages: PushImportMessage[], @@ -155,6 +207,7 @@ function writeMessages( ): { writtenCount: number duplicateCount: number + membersAdded: number minWrittenTs: number ftsEntries: Array<{ id: number; content: string | null }> } { @@ -163,11 +216,14 @@ function writeMessages( VALUES (?, ?, ?, ?, ?, ?, ?, ?)` ) const getMemberId = db.prepare('SELECT id FROM member WHERE platform_id = ?') - const insertMinimalMember = db.prepare('INSERT OR IGNORE INTO member (platform_id, account_name) VALUES (?, ?)') + const insertMinimalMember = db.prepare( + 'INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname) VALUES (?, ?, ?)' + ) const memberIdCache = new Map() let writtenCount = 0 let duplicateCount = 0 + let membersAdded = 0 let minWrittenTs = Infinity const ftsEntries: Array<{ id: number; content: string | null }> = [] @@ -194,11 +250,16 @@ function writeMessages( let memberId = memberIdCache.get(msg.sender) if (!memberId) { - insertMinimalMember.run(msg.sender, normalizeSenderAccountName(msg.sender, msg.accountName) || null) + const insertResult = insertMinimalMember.run( + msg.sender, + normalizeSenderAccountName(msg.sender, msg.accountName) || null, + normalizeSenderGroupNickname(msg.sender, msg.groupNickname) || null + ) const row = getMemberId.get(msg.sender) as { id: number } | undefined if (row) { memberId = row.id memberIdCache.set(msg.sender, memberId) + if (isCountedMember(msg.sender)) membersAdded += insertResult.changes } } if (!memberId) continue @@ -219,7 +280,7 @@ function writeMessages( } }) - return { writtenCount, duplicateCount, minWrittenTs, ftsEntries } + return { writtenCount, duplicateCount, membersAdded, minWrittenTs, ftsEntries } } function fullImport( @@ -227,7 +288,7 @@ function fullImport( meta: PushImportMeta, members: PushImportMember[], messages: PushImportMessage[] -): { writtenCount: number; duplicateCount: number; membersAdded: number } { +): { writtenCount: number; duplicateCount: number } { db.exec(CHAT_DB_SCHEMA) // Auto-create minimal member entries for senders not listed in members, @@ -300,19 +361,18 @@ function fullImport( /* non-fatal */ } - return { writtenCount: stats.messageCount, duplicateCount, membersAdded: members.length } + return { writtenCount: stats.messageCount, duplicateCount } } -function incrementalImport( - db: DatabaseAdapter, - payload: PushImportPayload -): { +interface IncrementalImportStats { writtenCount: number duplicateCount: number metaUpdated: boolean membersAdded: number membersUpdated: number -} { +} + +function writeIncrementalImport(db: DatabaseAdapter, payload: PushImportPayload): IncrementalImportStats { const metaUpdateMode = payload.options?.metaUpdateMode ?? 'patch' const memberUpdateMode = payload.options?.memberUpdateMode ?? 'upsert' @@ -387,10 +447,10 @@ function incrementalImport( m.roles ? JSON.stringify(m.roles) : '[]' ) if (!existed) { - membersAdded++ + if (isCountedMember(m.platformId)) membersAdded++ const row = getMemberId.get(m.platformId) as { id: number } | undefined if (row) existingMemberIds.add(m.platformId) - } else { + } else if (isCountedMember(m.platformId)) { membersUpdated++ } } @@ -401,7 +461,13 @@ function incrementalImport( const preWriteMaxTs = (db.prepare('SELECT MAX(ts) as max_ts FROM message').get() as { max_ts: number | null })?.max_ts ?? 0 - const { writtenCount, duplicateCount, minWrittenTs, ftsEntries } = writeMessages(db, payload.messages!, dedupState) + const { + writtenCount, + duplicateCount, + membersAdded: autoCreatedMembersAdded, + minWrittenTs, + ftsEntries, + } = writeMessages(db, payload.messages!, dedupState) if (ftsEntries.length > 0) { try { @@ -429,7 +495,17 @@ function incrementalImport( db.prepare('UPDATE meta SET imported_at = ?').run(Math.floor(Date.now() / 1000)) } - return { writtenCount, duplicateCount, metaUpdated, membersAdded, membersUpdated } + return { + writtenCount, + duplicateCount, + metaUpdated, + membersAdded: membersAdded + autoCreatedMembersAdded, + membersUpdated, + } +} + +function incrementalImport(db: DatabaseAdapter, payload: PushImportPayload): IncrementalImportStats { + return db.transaction(() => writeIncrementalImport(db, payload)) } export async function pushImport( @@ -497,13 +573,11 @@ export async function executePushImportUnlocked( if (isNew) { const db = deps.openDatabase(sessionId, { create: true }) try { - const { writtenCount, duplicateCount, membersAdded } = fullImport( - db, - payload.meta!, - payload.members ?? [], - payload.messages! - ) + const { writtenCount, duplicateCount } = fullImport(db, payload.meta!, payload.members ?? [], payload.messages!) const session = queryStats(db) + // Every non-system member in a new database was inserted by this request, + // including senders that were omitted from the submitted members array. + const membersAdded = session.memberCount const outcome: PushImportOutcome = { ok: true, result: {