From 381cfd05fd8d60f49a31e77a06f0d07923ff6cf6 Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 13 Jul 2026 00:01:06 +0800 Subject: [PATCH] fix(api): preserve import result contracts Map data-directory import lock conflicts to HTTP 409 across REST import paths and report stream duplicate counts from the dedicated diagnostics field. --- .../main/api/routes/import-helpers.test.ts | 44 +++++++++++++++++++ .../desktop/main/api/routes/import-helpers.ts | 22 ++++++++++ apps/desktop/main/api/routes/import.ts | 17 +++---- 3 files changed, 72 insertions(+), 11 deletions(-) create mode 100644 apps/desktop/main/api/routes/import-helpers.test.ts create mode 100644 apps/desktop/main/api/routes/import-helpers.ts diff --git a/apps/desktop/main/api/routes/import-helpers.test.ts b/apps/desktop/main/api/routes/import-helpers.test.ts new file mode 100644 index 00000000..961d4444 --- /dev/null +++ b/apps/desktop/main/api/routes/import-helpers.test.ts @@ -0,0 +1,44 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock' +import { ApiErrorCode } from '../errors' +import { apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers' + +test('maps an import lock result to the documented 409 API error', () => { + const error = apiErrorFromImportResult(IMPORT_IN_PROGRESS_ERROR_KEY, 'Import failed') + + assert.equal(error.code, ApiErrorCode.IMPORT_IN_PROGRESS) + assert.equal(error.statusCode, 409) +}) + +test('keeps ordinary import failures as IMPORT_FAILED', () => { + const error = apiErrorFromImportResult('database write failed', 'Import failed') + + assert.equal(error.code, ApiErrorCode.IMPORT_FAILED) + assert.equal(error.statusCode, 500) + assert.equal(error.message, 'database write failed') +}) + +test('maps stream duplicate and invalid-row counts to their distinct API fields', () => { + assert.deepEqual( + batchFromStreamDiagnostics({ + logFile: null, + detectedFormat: 'fixture', + messagesReceived: 12, + messagesWritten: 7, + duplicateCount: 3, + messagesSkipped: 2, + skipReasons: { + noSenderId: 1, + noAccountName: 0, + invalidTimestamp: 1, + noType: 0, + }, + }), + { + receivedCount: 12, + writtenCount: 7, + duplicateCount: 3, + } + ) +}) diff --git a/apps/desktop/main/api/routes/import-helpers.ts b/apps/desktop/main/api/routes/import-helpers.ts new file mode 100644 index 00000000..096225f3 --- /dev/null +++ b/apps/desktop/main/api/routes/import-helpers.ts @@ -0,0 +1,22 @@ +import { IMPORT_IN_PROGRESS_ERROR_KEY } from '@openchatlab/node-runtime/src/import/import-lock' +import type { ImportDiagnostics } from '@openchatlab/node-runtime/src/import/streaming-importer' +import { importFailed, importInProgress, type ApiError } from '../errors' + +export function apiErrorFromImportResult(error: string | undefined, fallbackMessage: string): ApiError { + return error === IMPORT_IN_PROGRESS_ERROR_KEY ? importInProgress() : importFailed(error || fallbackMessage) +} + +export function batchFromStreamDiagnostics(diagnostics: ImportDiagnostics | undefined): + | { + receivedCount: number + writtenCount: number + duplicateCount: number + } + | undefined { + if (!diagnostics) return undefined + return { + receivedCount: diagnostics.messagesReceived, + writtenCount: diagnostics.messagesWritten, + duplicateCount: diagnostics.duplicateCount, + } +} diff --git a/apps/desktop/main/api/routes/import.ts b/apps/desktop/main/api/routes/import.ts index 7cfe55a4..4cc749b3 100644 --- a/apps/desktop/main/api/routes/import.ts +++ b/apps/desktop/main/api/routes/import.ts @@ -33,6 +33,7 @@ import { errorResponse, } from '../errors' import { apiLogger } from '../logger' +import { apiErrorFromImportResult, batchFromStreamDiagnostics } from './import-helpers' // Per-session lock: different sessionIds can import in parallel. const isImporting = new Set() @@ -280,7 +281,7 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply, reply.send(responsePayload) } else { idempotencyFail(cacheKey) - const err = importFailed(result.error || 'Incremental import failed') + const err = apiErrorFromImportResult(result.error, 'Incremental import failed') reply.code(err.statusCode).send(errorResponse(err)) } } else { @@ -308,20 +309,14 @@ async function handleUnifiedImport(request: FastifyRequest, reply: FastifyReply, const responsePayload = successResponse({ sessionId: result.sessionId, created: true, - batch: diag - ? { - receivedCount: diag.messagesReceived, - writtenCount: diag.messagesWritten, - duplicateCount: diag.messagesSkipped, - } - : undefined, + batch: batchFromStreamDiagnostics(diag), session: sessionInfo, }) idempotencySuccess(cacheKey, responsePayload) reply.send(responsePayload) } else { idempotencyFail(cacheKey) - const err = importFailed(result.error || 'Import failed') + const err = apiErrorFromImportResult(result.error, 'Import failed') reply.code(err.statusCode).send(errorResponse(err)) } } @@ -401,7 +396,7 @@ async function handleLegacyImport(request: FastifyRequest, reply: FastifyReply, }) ) } else { - const err = importFailed(result.error || 'Incremental import failed') + const err = apiErrorFromImportResult(result.error, 'Incremental import failed') reply.code(err.statusCode).send(errorResponse(err)) } } else { @@ -416,7 +411,7 @@ async function handleLegacyImport(request: FastifyRequest, reply: FastifyReply, }) ) } else { - const err = importFailed(result.error || 'Import failed') + const err = apiErrorFromImportResult(result.error, 'Import failed') reply.code(err.statusCode).send(errorResponse(err)) } }