R53 A1-A6:ERP 集成运维可视化与业务同步状态接入

- ErpIntegrationPage Tab 化:主数据同步 / Outbox 队列管理 / ID 映射查看器
- Outbox 分页查询、详情、重试与统计端点
- IntegrationBinding 分页查询、routePath 推导与采购/资产深链
- 报销/采购/资产详情区接入 ERP 同步状态卡片
- 单测覆盖 Outbox 与 Binding 端点
这个提交包含在:
2026-04-07 10:33:58 +08:00
父节点 64d9ded024
当前提交 7cc6dc3397
共修改 9 个文件,包含 1989 行新增和 146 行删除
+586
查看文件
@@ -15,6 +15,7 @@ const {
normalizeRequestedResources,
syncErpMasterData,
} = require('../services/integrations/erpMasterData');
const { OUTBOX_STATUSES } = require('../services/integrations/outbox');
const {
buildReimbursementJournalPreview,
enqueueReimbursementJournalExport,
@@ -57,6 +58,155 @@ const PROCUREMENT_REQUEUE_TARGET_ENUM = z.enum([
const PROCUREMENT_REQUEUE_BODY_SCHEMA = z.object({
targets: z.array(PROCUREMENT_REQUEUE_TARGET_ENUM).min(1).max(3),
});
const OUTBOX_STATUS_ENUM = z.enum([
OUTBOX_STATUSES.pending,
OUTBOX_STATUSES.processing,
OUTBOX_STATUSES.succeeded,
OUTBOX_STATUSES.failed,
OUTBOX_STATUSES.dead_letter,
]);
const OUTBOX_LIST_QUERY_SCHEMA = z.object({
page: z.coerce.number().int().min(1).optional(),
pageSize: z.coerce.number().int().min(1).max(100).optional(),
status: OUTBOX_STATUS_ENUM.optional(),
event_type: z.string().trim().max(64).optional(),
});
const BINDINGS_LIST_QUERY_SCHEMA = z.object({
page: z.coerce.number().int().min(1).optional(),
pageSize: z.coerce.number().int().min(1).max(100).optional(),
provider: z.string().trim().max(32).optional(),
entity_type: z.string().trim().max(64).optional(),
});
const OUTBOX_SUMMARY_SCHEMA = {
type: 'object',
required: ['entityType', 'entityId', 'sourceEventName', 'journalType', 'exportType', 'identityCode'],
properties: {
entityType: { anyOf: [{ type: 'string' }, { type: 'null' }] },
entityId: { anyOf: [{ type: 'string' }, { type: 'null' }] },
sourceEventName: { anyOf: [{ type: 'string' }, { type: 'null' }] },
journalType: { anyOf: [{ type: 'string' }, { type: 'null' }] },
exportType: { anyOf: [{ type: 'string' }, { type: 'null' }] },
identityCode: { anyOf: [{ type: 'string' }, { type: 'null' }] },
},
};
const OUTBOX_ITEM_SCHEMA = {
type: 'object',
required: [
'id',
'eventType',
'status',
'payloadSummary',
'attempts',
'maxAttempts',
'nextRetryAt',
'createdAt',
'processedAt',
'deadLetterAt',
],
properties: {
id: { type: 'string', format: 'uuid' },
eventType: { type: 'string' },
status: { type: 'string' },
payloadSummary: OUTBOX_SUMMARY_SCHEMA,
attempts: { type: 'integer' },
maxAttempts: { type: 'integer' },
nextRetryAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] },
createdAt: { type: 'string', format: 'date-time' },
processedAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] },
deadLetterAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] },
},
};
const OUTBOX_DETAIL_SCHEMA = {
type: 'object',
required: ['item'],
properties: {
item: {
allOf: [
OUTBOX_ITEM_SCHEMA,
{
type: 'object',
required: ['payload', 'lastError'],
properties: {
payload: {
anyOf: [
{ type: 'object', additionalProperties: true },
{ type: 'null' },
],
},
lastError: { anyOf: [{ type: 'string' }, { type: 'null' }] },
},
},
],
},
},
};
const OUTBOX_LIST_RESPONSE_SCHEMA = {
type: 'object',
required: ['page', 'pageSize', 'total', 'items'],
properties: {
page: { type: 'integer' },
pageSize: { type: 'integer' },
total: { type: 'integer' },
items: {
type: 'array',
items: OUTBOX_ITEM_SCHEMA,
},
},
};
const OUTBOX_STATS_RESPONSE_SCHEMA = {
type: 'object',
required: ['pending', 'processing', 'succeeded', 'failed', 'deadLetter'],
properties: {
pending: { type: 'integer' },
processing: { type: 'integer' },
succeeded: { type: 'integer' },
failed: { type: 'integer' },
deadLetter: { type: 'integer' },
},
};
const BINDING_ITEM_SCHEMA = {
type: 'object',
required: [
'id',
'provider',
'entityType',
'localId',
'externalId',
'externalCode',
'lastSyncedAt',
'routePath',
],
properties: {
id: { type: 'string', format: 'uuid' },
provider: { type: 'string' },
entityType: { type: 'string' },
localId: { type: 'string', format: 'uuid' },
externalId: { anyOf: [{ type: 'string' }, { type: 'null' }] },
externalCode: { anyOf: [{ type: 'string' }, { type: 'null' }] },
lastSyncedAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] },
routePath: { anyOf: [{ type: 'string' }, { type: 'null' }] },
},
};
const BINDINGS_LIST_RESPONSE_SCHEMA = {
type: 'object',
required: ['page', 'pageSize', 'total', 'items'],
properties: {
page: { type: 'integer' },
pageSize: { type: 'integer' },
total: { type: 'integer' },
items: {
type: 'array',
items: BINDING_ITEM_SCHEMA,
},
},
};
const ERP_SYNC_RESPONSE_SCHEMA = {
type: 'object',
@@ -125,6 +275,195 @@ function parseRunNow(query) {
);
}
function buildValidationError(details, statusCode = 400) {
return new AppError({
code: 'VALID_001',
message: '参数校验失败',
statusCode,
details,
});
}
function parseOutboxListQuery(query) {
const parsed = OUTBOX_LIST_QUERY_SCHEMA.safeParse(query ?? {});
if (!parsed.success) {
throw buildValidationError(parsed.error.flatten());
}
return {
page: parsed.data.page ?? 1,
pageSize: parsed.data.pageSize ?? 20,
status: parsed.data.status ?? null,
eventType: String(parsed.data.event_type ?? '').trim() || null,
};
}
function parseBindingsListQuery(query) {
const parsed = BINDINGS_LIST_QUERY_SCHEMA.safeParse(query ?? {});
if (!parsed.success) {
throw buildValidationError(parsed.error.flatten());
}
return {
page: parsed.data.page ?? 1,
pageSize: parsed.data.pageSize ?? 20,
provider: String(parsed.data.provider ?? '').trim() || null,
entityType: String(parsed.data.entity_type ?? '').trim() || null,
};
}
function parseUuidParam(value, fieldName) {
const parsed = UUID_SCHEMA.safeParse(value);
if (!parsed.success) {
throw buildValidationError({ [fieldName]: 'invalid_uuid' });
}
return parsed.data;
}
function toIsoString(value) {
return value?.toISOString?.() ?? null;
}
function resolveOutboxIdentityCode(payload) {
return (
payload?.documentNo ??
payload?.formNumber ??
payload?.assetNo ??
payload?.batchNo ??
payload?.poNo ??
payload?.contractNo ??
payload?.requestNo ??
payload?.externalOrderNo ??
payload?.references?.poNo ??
payload?.references?.contractNo ??
payload?.references?.requestNo ??
payload?.references?.externalOrderNo ??
null
);
}
function buildOutboxPayloadSummary(payload) {
return {
entityType: typeof payload?.entityType === 'string' ? payload.entityType : null,
entityId: typeof payload?.entityId === 'string' ? payload.entityId : null,
sourceEventName: typeof payload?.sourceEventName === 'string' ? payload.sourceEventName : null,
journalType: typeof payload?.journalType === 'string' ? payload.journalType : null,
exportType: typeof payload?.exportType === 'string' ? payload.exportType : null,
identityCode: resolveOutboxIdentityCode(payload),
};
}
function mapOutboxItem(record) {
return {
id: record.id,
eventType: record.event_type,
status: record.status,
payloadSummary: buildOutboxPayloadSummary(record.payload ?? null),
attempts: Number(record.attempts ?? 0),
maxAttempts: Number(record.max_attempts ?? 0),
nextRetryAt: toIsoString(record.next_retry_at),
createdAt: toIsoString(record.created_at),
processedAt: toIsoString(record.processed_at),
deadLetterAt: toIsoString(record.dead_letter_at),
};
}
function mapOutboxDetail(record) {
return {
...mapOutboxItem(record),
payload: record.payload ?? null,
lastError: record.last_error ?? null,
};
}
function mapIntegrationBindingListItem(record, routePath) {
return {
id: record.id,
provider: record.provider,
entityType: record.entity_type,
localId: record.local_id,
externalId: record.external_id ?? null,
externalCode: record.external_code ?? null,
lastSyncedAt: toIsoString(record.last_synced_at),
routePath: routePath ?? null,
};
}
async function resolveBindingRoutePaths(prisma, rows) {
const archiveEntityTypes = new Set([
'purchase_order_accrual_journal',
'purchase_order_po',
'purchase_order_contract',
]);
const archiveIds = Array.from(
new Set(
(rows ?? [])
.filter((row) => archiveEntityTypes.has(row.entity_type))
.map((row) => row.local_id)
.filter(Boolean),
),
);
const purchaseRequestByArchiveId = new Map();
if (archiveIds.length && prisma?.purchaseOrderArchive?.findMany) {
const archiveRows = await prisma.purchaseOrderArchive.findMany({
where: {
id: {
in: archiveIds,
},
},
select: {
id: true,
purchase_request_id: true,
},
});
for (const item of archiveRows) {
purchaseRequestByArchiveId.set(item.id, item.purchase_request_id);
}
}
const routePathByBindingId = new Map();
for (const row of rows ?? []) {
let routePath = null;
if (row.entity_type === 'reimbursement_journal') {
routePath = `/reimbursements/${row.local_id}`;
} else if (row.entity_type === 'purchase_request_commitment_journal') {
routePath = `/purchase-requests?id=${row.local_id}`;
} else if (archiveEntityTypes.has(row.entity_type)) {
const purchaseRequestId = purchaseRequestByArchiveId.get(row.local_id) ?? null;
routePath = purchaseRequestId ? `/purchase-requests?id=${purchaseRequestId}` : null;
} else if (row.entity_type === 'asset_erp_card') {
routePath = `/assets?id=${row.local_id}`;
}
routePathByBindingId.set(row.id, routePath);
}
return routePathByBindingId;
}
async function buildOutboxStats(prisma) {
const [pending, processing, succeeded, failed, deadLetter] = await Promise.all([
prisma.integrationOutbox.count({ where: { status: OUTBOX_STATUSES.pending } }),
prisma.integrationOutbox.count({ where: { status: OUTBOX_STATUSES.processing } }),
prisma.integrationOutbox.count({ where: { status: OUTBOX_STATUSES.succeeded } }),
prisma.integrationOutbox.count({ where: { status: OUTBOX_STATUSES.failed } }),
prisma.integrationOutbox.count({ where: { status: OUTBOX_STATUSES.dead_letter } }),
]);
return {
pending,
processing,
succeeded,
failed,
deadLetter,
};
}
async function resolveErpGate(prisma, config) {
const evaluation = await evaluateFeatureFlag({
prisma,
@@ -565,6 +904,253 @@ function normalizeResourcesInput(value) {
}
async function integrationsRoutes(fastify) {
fastify.get(
'/integrations/outbox/stats',
{
preHandler: fastify.requirePermission('integration.manage'),
schema: {
tags: ['integrations'],
summary: '查询 ERP Outbox 队列状态统计',
security: [{ bearerAuth: [] }],
response: {
200: buildDataEnvelopeSchema(OUTBOX_STATS_RESPONSE_SCHEMA, 'ERP Outbox 状态统计'),
},
},
},
async (request, reply) => {
const stats = await buildOutboxStats(fastify.prisma);
return sendData(reply, request, stats);
},
);
fastify.get(
'/integrations/outbox',
{
preHandler: fastify.requirePermission('integration.manage'),
schema: {
tags: ['integrations'],
summary: '分页查询 ERP Outbox 事件',
security: [{ bearerAuth: [] }],
response: {
200: buildDataEnvelopeSchema(OUTBOX_LIST_RESPONSE_SCHEMA, 'ERP Outbox 列表'),
400: buildErrorEnvelopeSchema({
description: '参数校验失败',
codeExample: 'VALID_001',
messageExample: '参数校验失败',
}),
},
},
},
async (request, reply) => {
const query = parseOutboxListQuery(request.query);
const skip = (query.page - 1) * query.pageSize;
const where = {};
if (query.status) {
where.status = query.status;
}
if (query.eventType) {
where.event_type = query.eventType;
}
const [items, total] = await Promise.all([
fastify.prisma.integrationOutbox.findMany({
where,
orderBy: [{ created_at: 'desc' }],
skip,
take: query.pageSize,
}),
fastify.prisma.integrationOutbox.count({ where }),
]);
return sendData(reply, request, {
page: query.page,
pageSize: query.pageSize,
total,
items: items.map(mapOutboxItem),
});
},
);
fastify.get(
'/integrations/outbox/:id',
{
preHandler: fastify.requirePermission('integration.manage'),
schema: {
tags: ['integrations'],
summary: '查询单条 ERP Outbox 事件详情',
security: [{ bearerAuth: [] }],
response: {
200: buildDataEnvelopeSchema(OUTBOX_DETAIL_SCHEMA, 'ERP Outbox 详情'),
400: buildErrorEnvelopeSchema({
description: '参数校验失败',
codeExample: 'VALID_001',
messageExample: '参数校验失败',
}),
404: buildErrorEnvelopeSchema({
description: 'Outbox 事件不存在',
codeExample: 'SYS_404',
messageExample: 'Not Found',
}),
},
},
},
async (request, reply) => {
const outboxId = parseUuidParam(request.params?.id, 'id');
const item = await fastify.prisma.integrationOutbox.findUnique({
where: { id: outboxId },
});
if (!item) {
throw new AppError({
code: 'SYS_404',
message: 'Not Found',
statusCode: 404,
});
}
return sendData(reply, request, {
item: mapOutboxDetail(item),
});
},
);
fastify.post(
'/integrations/outbox/:id/retry',
{
preHandler: fastify.requirePermission('integration.manage'),
schema: {
tags: ['integrations'],
summary: '重试 failed/dead_letter 的 ERP Outbox 事件',
security: [{ bearerAuth: [] }],
response: {
200: buildDataEnvelopeSchema(OUTBOX_DETAIL_SCHEMA, 'ERP Outbox 已重置为 pending'),
400: buildErrorEnvelopeSchema({
description: '参数校验失败',
codeExample: 'VALID_001',
messageExample: '参数校验失败',
}),
404: buildErrorEnvelopeSchema({
description: 'Outbox 事件不存在',
codeExample: 'SYS_404',
messageExample: 'Not Found',
}),
409: buildErrorEnvelopeSchema({
description: '当前状态不允许重试',
codeExample: 'VALID_001',
messageExample: '参数校验失败',
}),
},
},
},
async (request, reply) => {
const outboxId = parseUuidParam(request.params?.id, 'id');
const existed = await fastify.prisma.integrationOutbox.findUnique({
where: { id: outboxId },
});
if (!existed) {
throw new AppError({
code: 'SYS_404',
message: 'Not Found',
statusCode: 404,
});
}
if (![OUTBOX_STATUSES.failed, OUTBOX_STATUSES.dead_letter].includes(existed.status)) {
throw buildValidationError(
{
outbox: 'integration_outbox_retry_status_invalid',
status: existed.status,
},
409,
);
}
const retried = await fastify.prisma.integrationOutbox.update({
where: { id: outboxId },
data: {
status: OUTBOX_STATUSES.pending,
attempts: 0,
last_error: null,
next_retry_at: new Date(),
processed_at: null,
dead_letter_at: null,
updated_by: request.employee.id,
},
});
await fastify.prisma.auditLog.create({
data: buildAuditLogData(request, {
actorId: request.employee.id,
action: 'ERP_OUTBOX_RETRIED',
targetType: 'integration_outbox',
targetId: retried.id,
extra: {
previousStatus: existed.status,
nextStatus: retried.status,
eventType: retried.event_type,
outboxId: retried.id,
},
}),
});
return sendData(reply, request, {
item: mapOutboxDetail(retried),
});
},
);
fastify.get(
'/integrations/bindings',
{
preHandler: fastify.requirePermission('integration.manage'),
schema: {
tags: ['integrations'],
summary: '分页查询 ERP IntegrationBinding',
security: [{ bearerAuth: [] }],
response: {
200: buildDataEnvelopeSchema(BINDINGS_LIST_RESPONSE_SCHEMA, 'ERP Binding 列表'),
400: buildErrorEnvelopeSchema({
description: '参数校验失败',
codeExample: 'VALID_001',
messageExample: '参数校验失败',
}),
},
},
},
async (request, reply) => {
const query = parseBindingsListQuery(request.query);
const skip = (query.page - 1) * query.pageSize;
const where = {};
if (query.provider) {
where.provider = query.provider;
}
if (query.entityType) {
where.entity_type = query.entityType;
}
const [items, total] = await Promise.all([
fastify.prisma.integrationBinding.findMany({
where,
orderBy: [{ last_synced_at: { sort: 'desc', nulls: 'last' } }, { created_at: 'desc' }],
skip,
take: query.pageSize,
}),
fastify.prisma.integrationBinding.count({ where }),
]);
const routePathByBindingId = await resolveBindingRoutePaths(fastify.prisma, items);
return sendData(reply, request, {
page: query.page,
pageSize: query.pageSize,
total,
items: items.map((item) => mapIntegrationBindingListItem(item, routePathByBindingId.get(item.id) ?? null)),
});
},
);
fastify.get(
'/integrations/erp/master-data/status',
{