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.

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 {

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<string, unknown> {
+  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<string, number>()
   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: {