diff --git a/docs/cn/standard/chatlab-import.md b/docs/cn/standard/chatlab-import.md index e06f8260..8912fa78 100644 --- a/docs/cn/standard/chatlab-import.md +++ b/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 | 服务内部错误 | 是 | diff --git a/docs/en/standard/chatlab-import.md b/docs/en/standard/chatlab-import.md index 8eac4729..befb07e0 100644 --- a/docs/en/standard/chatlab-import.md +++ b/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 | diff --git a/packages/http-routes/src/errors.ts b/packages/http-routes/src/errors.ts index 2b292325..81dda167 100644 --- a/packages/http-routes/src/errors.ts +++ b/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.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) } diff --git a/packages/http-routes/src/routes/imports.test.ts b/packages/http-routes/src/routes/imports.test.ts new file mode 100644 index 00000000..7d85b7f9 --- /dev/null +++ b/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) +}) diff --git a/packages/http-routes/src/routes/imports.ts b/packages/http-routes/src/routes/imports.ts index 601cc8e3..a812947a 100644 --- a/packages/http-routes/src/routes/imports.ts +++ b/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>() + 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 + } } ) }