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.
This commit is contained in:
digua committed 2026-07-13 01:01:42 +08:00
1 parent ec822cb1b0
commit 381cfd05fd
3 files changed
+72 -11

No files matched your search

@@ -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,
}
)
})
@@ -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,
}
}
+6 -11
View File
@@ -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<string>()
@@ -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))
}
}