feat(api): honor idempotency keys on shared imports

docs/cn/standard/chatlab-import.md

@@ -21,7 +21,7 @@ outline: deep
 
 1. **统一入口**：首次导入和增量导入使用同一个端点，调用方无需区分。
 2. **接口最小化**：1 个导入接口 + 2 个查询接口覆盖全部 Push 场景。
-3. **双层幂等**：请求级幂等（Idempotency-Key）+ 记录级去重（platformMessageId / 内容哈希），承诺 **at-least-once + deterministic dedupe**。
+3. **双层幂等**：请求级幂等（Idempotency-Key）+ 记录级去重（platformMessageId / 确定性 fallback key），承诺 **at-least-once + deterministic dedupe**。
 4. **同步优先**：小批量导入同步返回 `200 OK` 和写入结果。
 5. **默认自动更新**：meta 和 members 默认随导入请求自动更新，可通过 `options` 控制。
 
@@ -320,17 +320,19 @@ curl http://127.0.0.1:3110/api/v1/imports/group_abc123 \
 **优先级：**
 
 1. 若消息提供了 `platformMessageId`，以此作为唯一键去重（高精度，推荐）。
-2. 若未提供 `platformMessageId`，退化为内容哈希去重：`sha256(timestamp + '\0' + sender + '\0' + contentTag + '\0' + normalizedContent)`
+2. 若未提供 `platformMessageId`，退化为确定性 fallback key：
+   `timestamp + sender + type + normalizedContent + replyToMessageId`。
 
 | 层次               | 机制                                            | 适用范围                      | 精度     |
 | ------------------ | ----------------------------------------------- | ----------------------------- | -------- |
 | 请求级幂等         | `Idempotency-Key`                               | 同一 HTTP 请求的重试          | 精确     |
 | 消息级去重（主键） | `platformMessageId`                             | 跨批次、跨窗口的同一条消息    | 精确     |
-| 消息级去重（降级） | 内容哈希 `sha256(timestamp + sender + content)` | 无 platformMessageId 时的兜底 | 最大努力 |
+| 消息级去重（降级） | 上述字段生成的确定性 fallback key                 | 无 platformMessageId 时的兜底 | 最大努力 |
 
 ::: warning 注意
 - 同一 `platformMessageId` 的消息不会被重复写入，即使 content 不同（以首次写入为准）
-- 内容哈希去重在"同一人、同一秒、完全相同内容"时判定为重复，存在极小概率误判
+- 两个不同的 `platformMessageId` 即使其他字段完全相同，也会保留为两条消息
+- fallback 去重在"同一人、同一秒、相同类型、相同规范化内容和相同回复目标"时判定为重复，存在极小概率误判
 - **强烈建议**外部数据源提供 `platformMessageId`，这是最可靠的去重依据
 :::
 
@@ -442,6 +444,7 @@ ChatLab 不为调用方维护游标。推荐结构：
 | `BODY_TOO_LARGE` | 413 | JSON body 超过 50MB | 否 |
 | `IMPORT_IN_PROGRESS` | 409 | 当前有其他导入正在执行 | 是 |
 | `IDEMPOTENCY_CONFLICT` | 409 | 相同幂等键但请求体不一致 | 否 |
+| `IDEMPOTENCY_PENDING` | 409 | 相同幂等键的首次请求仍在执行 | 是 |
 | `IMPORT_FAILED` | 500 | 导入过程内部错误 | 是 |
 | `SERVER_ERROR` | 500 | 服务内部错误 | 是 |
 

docs/en/standard/chatlab-import.md

@@ -21,7 +21,7 @@ Both modes share the same underlying import pipeline (deduplication, meta/member
 
 1. **Single endpoint**: Initial and incremental imports use the same endpoint — callers don't need to distinguish.
 2. **Minimal surface**: 1 import endpoint + 2 query endpoints cover all Push scenarios.
-3. **Two-layer idempotency**: Request-level (`Idempotency-Key`) + record-level dedup (`platformMessageId` / content hash), guaranteeing **at-least-once + deterministic dedupe**.
+3. **Two-layer idempotency**: Request-level (`Idempotency-Key`) + record-level dedup (`platformMessageId` / deterministic fallback key), guaranteeing **at-least-once + deterministic dedupe**.
 4. **Synchronous by default**: Small batches return `200 OK` with write results synchronously.
 5. **Auto-update by default**: `meta` and `members` are updated automatically with each import request; controllable via `options`.
 
@@ -320,17 +320,19 @@ Deduplication is scoped to a single session and does not cross sessions.
 **Priority:**
 
 1. If `platformMessageId` is provided, it is used as the unique key (high precision, recommended).
-2. If `platformMessageId` is absent, falls back to content hash: `sha256(timestamp + '\0' + sender + '\0' + contentTag + '\0' + normalizedContent)`
+2. If `platformMessageId` is absent, a deterministic fallback key is generated from
+   `timestamp + sender + type + normalizedContent + replyToMessageId`.
 
 | Layer | Mechanism | Scope | Precision |
 | --- | --- | --- | --- |
 | Request-level idempotency | `Idempotency-Key` | Retries of the same HTTP request | Exact |
 | Message-level dedup (primary) | `platformMessageId` | Same message across batches/windows | Exact |
-| Message-level dedup (fallback) | Content hash | Fallback when `platformMessageId` is absent | Best effort |
+| Message-level dedup (fallback) | Deterministic key from the fields above | Fallback when `platformMessageId` is absent | Best effort |
 
 ::: warning
 - A message with the same `platformMessageId` will not be written again, even if `content` differs (first write wins).
-- Content hash dedup treats "same sender, same second, identical content" as a duplicate; there is a very small false positive rate.
+- Two different `platformMessageId` values are preserved as different messages even when every other field matches.
+- Fallback dedup treats "same sender, same second, same type, normalized content, and reply target" as a duplicate; there is a very small false positive rate.
 - **Strongly recommended**: provide `platformMessageId` — it is the most reliable deduplication key.
 :::
 
@@ -442,6 +444,7 @@ See [ChatLab Format Specification](./chatlab-format.md) for the reserved field s
 | `BODY_TOO_LARGE` | 413 | JSON body exceeds 50MB | No |
 | `IMPORT_IN_PROGRESS` | 409 | Another import is currently running | Yes |
 | `IDEMPOTENCY_CONFLICT` | 409 | Same idempotency key but different request body | No |
+| `IDEMPOTENCY_PENDING` | 409 | The first request with this idempotency key is still running | Yes |
 | `IMPORT_FAILED` | 500 | Internal error during import | Yes |
 | `SERVER_ERROR` | 500 | Internal server error | Yes |
 

packages/http-routes/src/errors.ts

@@ -17,6 +17,8 @@ export enum ApiErrorCode {
   EXPORT_TOO_LARGE = 'EXPORT_TOO_LARGE',
   BODY_TOO_LARGE = 'BODY_TOO_LARGE',
   IMPORT_IN_PROGRESS = 'IMPORT_IN_PROGRESS',
+  IDEMPOTENCY_CONFLICT = 'IDEMPOTENCY_CONFLICT',
+  IDEMPOTENCY_PENDING = 'IDEMPOTENCY_PENDING',
   IMPORT_FAILED = 'IMPORT_FAILED',
   DATA_DIR_INCOMPATIBLE = 'DATA_DIR_INCOMPATIBLE',
   SERVER_ERROR = 'SERVER_ERROR',
@@ -32,6 +34,8 @@ const HTTP_STATUS: Record<ApiErrorCode, number> = {
   [ApiErrorCode.EXPORT_TOO_LARGE]: 400,
   [ApiErrorCode.BODY_TOO_LARGE]: 413,
   [ApiErrorCode.IMPORT_IN_PROGRESS]: 409,
+  [ApiErrorCode.IDEMPOTENCY_CONFLICT]: 409,
+  [ApiErrorCode.IDEMPOTENCY_PENDING]: 409,
   [ApiErrorCode.IMPORT_FAILED]: 500,
   [ApiErrorCode.DATA_DIR_INCOMPATIBLE]: 409,
   [ApiErrorCode.SERVER_ERROR]: 500,
@@ -84,6 +88,17 @@ export function importInProgress(): ApiError {
   return new ApiError(ApiErrorCode.IMPORT_IN_PROGRESS, 'Another import is already in progress')
 }
 
+export function idempotencyConflict(): ApiError {
+  return new ApiError(ApiErrorCode.IDEMPOTENCY_CONFLICT, 'Same Idempotency-Key with different request body hash')
+}
+
+export function idempotencyPending(): ApiError {
+  return new ApiError(
+    ApiErrorCode.IDEMPOTENCY_PENDING,
+    'A request with this Idempotency-Key is still in progress. Please retry later.'
+  )
+}
+
 export function importFailed(message: string): ApiError {
   return new ApiError(ApiErrorCode.IMPORT_FAILED, message)
 }

packages/http-routes/src/routes/imports.test.ts

@@ -0,0 +1,114 @@
+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 Fastify from 'fastify'
+import type { PathProvider } from '@openchatlab/core'
+import { DatabaseManager } from '@openchatlab/node-runtime'
+import type { HttpRouteContext } from '../context'
+import { registerImportRoutes } from './imports'
+
+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-import-route-'))
+}
+
+function createContext(rootDir: string): HttpRouteContext {
+  const pathProvider: PathProvider = {
+    getSystemDir: () => rootDir,
+    getUserDataDir: () => rootDir,
+    getDatabaseDir: () => path.join(rootDir, 'databases'),
+    getVectorDir: () => path.join(rootDir, 'vector'),
+    getAiDataDir: () => path.join(rootDir, 'ai'),
+    getSettingsDir: () => path.join(rootDir, 'settings'),
+    getCacheDir: () => path.join(rootDir, 'cache'),
+    getTempDir: () => path.join(rootDir, 'temp'),
+    getLogsDir: () => path.join(rootDir, 'logs'),
+    getDownloadsDir: () => rootDir,
+  }
+  const dbManager = new DatabaseManager(pathProvider, { nativeBinding, allowMissingRuntimeForTests: true })
+
+  return {
+    dbManager,
+    pathProvider,
+    getVersion: () => '0.0.0-test',
+  } as HttpRouteContext
+}
+
+function createPayload(content: string) {
+  return {
+    chatlab: { version: '0.0.2', exportedAt: 1780330900 },
+    meta: { name: 'Route Import', platform: 'wechat', type: 'private' },
+    members: [{ platformId: 'wxid_alice', accountName: 'Alice' }],
+    messages: [
+      {
+        platformMessageId: `msg-${content}`,
+        sender: 'wxid_alice',
+        timestamp: 1780330832,
+        type: 0,
+        content,
+      },
+    ],
+  }
+}
+
+test('replays the cached response for the same Idempotency-Key and body', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  const app = Fastify()
+  registerImportRoutes(app, createContext(root))
+  await app.ready()
+  t.after(() => app.close())
+
+  const request = {
+    method: 'POST' as const,
+    url: '/api/v1/imports/idempotent-session',
+    headers: { 'content-type': 'application/json', 'idempotency-key': 'batch-1' },
+    payload: createPayload('hello'),
+  }
+  const first = await app.inject(request)
+  const replay = await app.inject(request)
+
+  assert.equal(first.statusCode, 200)
+  assert.equal(replay.statusCode, 200)
+  assert.deepEqual(replay.json(), first.json())
+
+  const db = createContext(root).dbManager.openRawSessionDatabase('idempotent-session', { readonly: true })
+  try {
+    const row = db.prepare('SELECT COUNT(*) AS count FROM message').get() as { count: number }
+    assert.equal(row.count, 1)
+  } finally {
+    db.close()
+  }
+})
+
+test('rejects reuse of an Idempotency-Key with a different body', async (t) => {
+  const root = makeTempDir()
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }))
+  const app = Fastify()
+  registerImportRoutes(app, createContext(root))
+  await app.ready()
+  t.after(() => app.close())
+
+  const first = await app.inject({
+    method: 'POST',
+    url: '/api/v1/imports/idempotency-conflict',
+    headers: { 'content-type': 'application/json', 'idempotency-key': 'batch-1' },
+    payload: createPayload('first'),
+  })
+  const conflict = await app.inject({
+    method: 'POST',
+    url: '/api/v1/imports/idempotency-conflict',
+    headers: { 'content-type': 'application/json', 'idempotency-key': 'batch-1' },
+    payload: createPayload('second'),
+  })
+
+  assert.equal(first.statusCode, 200)
+  assert.equal(conflict.statusCode, 409)
+  assert.equal(conflict.json().error.code, 'IDEMPOTENCY_CONFLICT')
+  assert.equal(fs.existsSync(path.join(root, 'databases', 'idempotency-conflict.db')), true)
+  assert.equal(fs.existsSync(path.join(root, 'idempotency-conflict.db')), false)
+})

packages/http-routes/src/routes/imports.ts

@@ -8,11 +8,21 @@
 
 import type { FastifyInstance } from 'fastify'
 import type { HttpRouteContext } from '../context'
-import { pushImport } from '@openchatlab/node-runtime'
+import { hashImportBody, ImportIdempotencyCache, pushImport } from '@openchatlab/node-runtime'
 import type { PushImportPayload } from '@openchatlab/node-runtime'
-import { successResponse, errorResponse, invalidPayload, importInProgress, importFailed } from '../errors'
+import {
+  successResponse,
+  errorResponse,
+  invalidPayload,
+  importInProgress,
+  importFailed,
+  idempotencyConflict,
+  idempotencyPending,
+} from '../errors'
 
 export function registerImportRoutes(server: FastifyInstance, ctx: HttpRouteContext): void {
+  const idempotencyCache = new ImportIdempotencyCache<ReturnType<typeof successResponse>>()
+
   server.post<{ Params: { sessionId: string }; Body: PushImportPayload }>(
     '/api/v1/imports/:sessionId',
     async (request, reply) => {
@@ -29,22 +39,45 @@ export function registerImportRoutes(server: FastifyInstance, ctx: HttpRouteCont
         return reply.code(err.statusCode).send(errorResponse(err))
       }
 
-      const outcome = await pushImport(ctx.dbManager, sessionId, request.body ?? {})
-
-      if (!outcome.ok) {
-        if (outcome.reason === 'import_in_progress') {
-          const err = importInProgress()
+      const idempotencyKey = request.headers['idempotency-key']
+      const cacheKey = typeof idempotencyKey === 'string' && idempotencyKey ? `${idempotencyKey}:${sessionId}` : null
+      if (cacheKey) {
+        const start = idempotencyCache.start(cacheKey, hashImportBody(request.body))
+        if (start.status === 'conflict') {
+          const err = idempotencyConflict()
           return reply.code(err.statusCode).send(errorResponse(err))
         }
-        if (outcome.reason === 'invalid_payload') {
-          const err = invalidPayload(outcome.message)
+        if (start.status === 'pending') {
+          const err = idempotencyPending()
           return reply.code(err.statusCode).send(errorResponse(err))
         }
-        const err = importFailed(outcome.message)
-        return reply.code(err.statusCode).send(errorResponse(err))
+        if (start.status === 'success') return start.response
       }
 
-      return successResponse(outcome.result)
+      try {
+        const outcome = await pushImport(ctx.dbManager, sessionId, request.body ?? {})
+
+        if (!outcome.ok) {
+          if (cacheKey) idempotencyCache.fail(cacheKey)
+          if (outcome.reason === 'import_in_progress') {
+            const err = importInProgress()
+            return reply.code(err.statusCode).send(errorResponse(err))
+          }
+          if (outcome.reason === 'invalid_payload') {
+            const err = invalidPayload(outcome.message)
+            return reply.code(err.statusCode).send(errorResponse(err))
+          }
+          const err = importFailed(outcome.message)
+          return reply.code(err.statusCode).send(errorResponse(err))
+        }
+
+        const response = successResponse(outcome.result)
+        if (cacheKey) idempotencyCache.success(cacheKey, response)
+        return response
+      } catch (error) {
+        if (cacheKey) idempotencyCache.fail(cacheKey)
+        throw error
+      }
     }
   )
 }