const { z } = require('zod'); const { AppError } = require('../utils/AppError'); const { sendData } = require('../utils/reply'); const { buildAuditLogData } = require('../utils/audit'); const { buildDataEnvelopeSchema, buildErrorEnvelopeSchema, } = require('../utils/openapi'); const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require('../services/featureFlags'); const { runIntegrationOutboxOnce } = require('../jobs/integrationOutboxJob'); const { ERP_MASTER_DATA_RESOURCES, listMasterDataSyncStates, normalizeRequestedResources, syncErpMasterData, } = require('../services/integrations/erpMasterData'); const { OUTBOX_STATUSES } = require('../services/integrations/outbox'); const { buildReimbursementJournalPreview, enqueueReimbursementJournalExport, } = require('../services/integrations/erpJournal'); const { PROCUREMENT_ERP_TARGETS, buildPurchaseRequestErpTrack, enqueuePurchaseAccrualJournalExport, enqueuePurchaseArchiveSummaryExport, enqueuePurchaseCommitmentJournalExport, loadPurchaseRequestErpContext, } = require('../services/integrations/erpProcurement'); const { ASSET_ERP_EVENT_TYPE, ASSET_ERP_TARGET, buildAssetErpTrack, buildAssetLedgerSummaryDedupeKey, enqueueAssetLedgerSummaryExport, loadAssetErpContext, resolveAssetLedgerSummaryDrift, } = require('../services/integrations/erpAsset'); const { TRAVEL_ERP_EVENT_TYPE, buildTravelReconciliationErpTrack, buildTravelReconciliationSummaryDedupeKey, enqueueTravelReconciliationSummaryExport, loadTravelReconciliationErpContext, } = require('../services/integrations/erpTravel'); const UUID_SCHEMA = z.string().uuid(); const MASTER_DATA_RESOURCE_ENUM = z.enum(['legal_entity', 'cost_center', 'supplier']); const MASTER_DATA_SYNC_BODY_SCHEMA = z.object({ resources: z.array(MASTER_DATA_RESOURCE_ENUM).max(3).optional(), }); const PROCUREMENT_REQUEUE_TARGET_ENUM = z.enum([ 'commitment_journal', 'accrual_journal', 'archive_summary', ]); 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', required: ['outboxId', 'status', 'duplicated'], properties: { outboxId: { type: 'string', format: 'uuid' }, status: { type: 'string' }, duplicated: { type: 'boolean' }, }, }; const ERP_JOURNAL_EXPORT_RESPONSE_SCHEMA = { type: 'object', required: ['outboxId', 'status', 'duplicated', 'requeued', 'eventType'], properties: { outboxId: { type: 'string', format: 'uuid' }, status: { type: 'string' }, duplicated: { type: 'boolean' }, requeued: { type: 'boolean' }, eventType: { type: 'string' }, }, }; const ERP_MASTER_DATA_STATUS_SCHEMA = { type: 'object', required: ['items', 'gate'], properties: { gate: { type: 'object', required: ['envEnabled', 'effectiveEnabled', 'effectiveReason', 'provider'], properties: { envEnabled: { type: 'boolean' }, effectiveEnabled: { type: 'boolean' }, effectiveReason: { type: 'string' }, provider: { type: 'string' }, masterDataEndpointConfigured: { type: 'boolean' }, journalEndpointConfigured: { type: 'boolean' }, masterDataSyncCron: { anyOf: [{ type: 'string' }, { type: 'null' }] }, }, }, items: { type: 'array', items: { type: 'object', required: ['resourceType', 'status'], properties: { resourceType: { type: 'string' }, status: { type: 'string' }, lastStartedAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] }, lastFinishedAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] }, lastSuccessAt: { anyOf: [{ type: 'string', format: 'date-time' }, { type: 'null' }] }, lastError: { anyOf: [{ type: 'string' }, { type: 'null' }] }, lastResult: { anyOf: [{ type: 'object' }, { type: 'null' }] }, }, }, }, }, }; function parseRunNow(query) { return ( query?.runNow === '1' || query?.runNow === 1 || query?.runNow === true || query?.runNow === 'true' ); } 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, config, flagKey: FEATURE_FLAG_KEYS.erpSync, departmentId: null, strictDepartment: false, }); const envEnabled = Boolean(config.erpSyncEnabled); const featureFlagEnabled = Boolean(evaluation.enabled); return { envEnabled, featureFlagEnabled, effectiveEnabled: envEnabled && featureFlagEnabled, effectiveReason: envEnabled ? evaluation.reason : 'env_disabled', provider: config.erpSyncProvider, masterDataEndpointConfigured: Boolean(String(config.erpMasterDataEndpoint ?? '').trim()), journalEndpointConfigured: Boolean( String(config.erpJournalExportEndpoint ?? config.erpSyncEndpoint ?? '').trim(), ), masterDataSyncCron: String(config.erpMasterDataSyncCron ?? '').trim() || null, featureFlag: FEATURE_FLAG_KEYS.erpSync, }; } async function requireErpManagementGate(prisma, config) { const gate = await resolveErpGate(prisma, config); if (gate.effectiveEnabled) { return gate; } throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 409, details: { integration: gate.envEnabled ? 'feature_flag_blocked' : 'erp_sync_disabled', featureFlag: gate.featureFlag, reason: gate.effectiveReason, }, }); } async function fetchReimbursementForErp(prisma, reimbursementId) { const reimbursement = await prisma.reimbursementForm.findUnique({ where: { id: reimbursementId }, select: { id: true, form_number: true, owner_id: true, submitter_id: true, department_id: true, status: true, total_amount: true, payment_voucher_no: true, payment_date: true, paid_at: true, owner: { select: { legal_entity: { select: { code: true, }, }, }, }, submitter: { select: { legal_entity: { select: { code: true, }, }, }, }, }, }); if (!reimbursement) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } return reimbursement; } function ensurePaidReimbursement(reimbursement) { if (reimbursement.status !== 'paid') { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 409, details: { reimbursement: 'erp_sync_requires_paid' }, }); } } function canReadPurchaseRequest(request, purchaseRequest) { if (!purchaseRequest) { return false; } const permissionCodes = request?.employee?.permissionCodes ?? []; if (permissionCodes.includes('*') || permissionCodes.includes('procurement.view_all') || permissionCodes.includes('procurement.manage')) { return true; } return ( purchaseRequest.applicant_id === request.employee.id || purchaseRequest.submitter_id === request.employee.id ); } function normalizeProcurementTargets(body) { const parsed = PROCUREMENT_REQUEUE_BODY_SCHEMA.safeParse(body ?? {}); if (!parsed.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: parsed.error.flatten(), }); } const orderedTargets = [ PROCUREMENT_ERP_TARGETS.commitmentJournal, PROCUREMENT_ERP_TARGETS.accrualJournal, PROCUREMENT_ERP_TARGETS.archiveSummary, ]; const requested = new Set(parsed.data.targets); return orderedTargets.filter((target) => requested.has(target)); } async function fetchReadablePurchaseRequestContext(prisma, purchaseRequestId, request) { const context = await loadPurchaseRequestErpContext(prisma, purchaseRequestId); if (!context?.purchaseRequest) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } if (!canReadPurchaseRequest(request, context.purchaseRequest)) { throw new AppError({ code: 'PERM_001', message: '无权限', statusCode: 403, }); } return context; } function canReadAsset(request, asset) { if (!asset) { return false; } const permissionCodes = request?.employee?.permissionCodes ?? []; if ( permissionCodes.includes('*') || permissionCodes.includes('asset.view_all') || permissionCodes.includes('asset.manage') || permissionCodes.includes('asset.transfer') || permissionCodes.includes('asset.inventory') ) { return true; } return ( permissionCodes.includes('asset.view_own') && asset.assignee_employee_id === request?.employee?.id ); } async function fetchReadableAssetContext(prisma, assetId, request) { const context = await loadAssetErpContext(prisma, assetId); if (!context?.asset) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } if (!canReadAsset(request, context.asset)) { throw new AppError({ code: 'PERM_001', message: '无权限', statusCode: 403, }); } return context; } function canReadTravelReconciliationBatch(request) { const permissionCodes = request?.employee?.permissionCodes ?? []; return ( permissionCodes.includes('*') || permissionCodes.includes('travel.payment.manage') || permissionCodes.includes('integration.manage') ); } async function fetchReadableTravelReconciliationContext(prisma, batchId, request) { const context = await loadTravelReconciliationErpContext(prisma, batchId); if (!context) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } if (!canReadTravelReconciliationBatch(request)) { throw new AppError({ code: 'PERM_001', message: '无权限', statusCode: 403, }); } return context; } async function findTravelReconciliationOutboxByDedupeKey(prisma, batchId) { return prisma.integrationOutbox.findUnique({ where: { event_type_dedupe_key: { event_type: TRAVEL_ERP_EVENT_TYPE, dedupe_key: buildTravelReconciliationSummaryDedupeKey(batchId), }, }, select: { id: true, status: true, payload: true, }, }); } async function enqueueTravelReconciliationExportAndMaybeRun({ prisma, config, logger, context, actorId, runNow, findExistingOutbox = findTravelReconciliationOutboxByDedupeKey, enqueueExport = enqueueTravelReconciliationSummaryExport, runOutboxOnce = runIntegrationOutboxOnce, }) { const existingOutbox = await findExistingOutbox(prisma, context.id); const enqueued = await enqueueExport(prisma, config, { batch: context, payload: existingOutbox?.payload, actorId, requeueOnFailure: true, }); if (runNow && (!enqueued.duplicated || enqueued.requeued)) { await runOutboxOnce(prisma, config, logger); } return enqueued; } async function findAssetOutboxByDedupeKey(prisma, assetId) { return prisma.integrationOutbox.findUnique({ where: { event_type_dedupe_key: { event_type: ASSET_ERP_EVENT_TYPE, dedupe_key: buildAssetLedgerSummaryDedupeKey(assetId), }, }, select: { id: true, status: true, payload: true, }, }); } async function enqueueAssetExportAndMaybeRun({ prisma, config, logger, context, actorId, runNow, findExistingOutbox = findAssetOutboxByDedupeKey, enqueueExport = enqueueAssetLedgerSummaryExport, runOutboxOnce = runIntegrationOutboxOnce, }) { const asset = context?.asset; const existingOutbox = await findExistingOutbox(prisma, asset.id); const drift = resolveAssetLedgerSummaryDrift(asset, existingOutbox?.payload ?? null); const enqueued = await enqueueExport(prisma, config, { asset, payload: existingOutbox?.payload && !drift ? existingOutbox.payload : undefined, sourceEventName: drift?.sourceEventName, changeType: drift?.changeType, changedFields: drift?.changedFields, occurredAt: new Date(), actorId, requeueOnFailure: true, }); if (runNow && (!enqueued.duplicated || enqueued.requeued)) { await runOutboxOnce(prisma, config, logger); } return enqueued; } function ensureProcurementTargetEligible(target, context) { const purchaseRequest = context?.purchaseRequest; const archive = purchaseRequest?.order_archive ?? null; if (target === PROCUREMENT_ERP_TARGETS.commitmentJournal) { if (purchaseRequest?.status !== 'approved') { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 409, details: { purchaseRequest: 'purchase_request_not_approved', target }, }); } return; } if (!archive) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 409, details: { purchaseOrderArchive: 'purchase_order_archive_missing', target }, }); } } async function enqueueProcurementTargetsAndMaybeRun({ prisma, config, logger, context, actorId, targets, runNow, }) { const purchaseRequest = context.purchaseRequest; const archive = purchaseRequest.order_archive ?? null; const items = []; for (const target of targets) { ensureProcurementTargetEligible(target, context); if (target === PROCUREMENT_ERP_TARGETS.commitmentJournal) { const enqueued = await enqueuePurchaseCommitmentJournalExport(prisma, config, { purchaseRequest, occurredAt: context.approvalOccurredAt ?? new Date(), actorId, requeueOnFailure: true, }); items.push({ target, ...enqueued }); continue; } if (target === PROCUREMENT_ERP_TARGETS.accrualJournal) { const enqueued = await enqueuePurchaseAccrualJournalExport(prisma, config, { purchaseRequest, archive, occurredAt: archive.archived_at ?? new Date(), actorId, requeueOnFailure: true, }); items.push({ target, ...enqueued }); continue; } const enqueued = await enqueuePurchaseArchiveSummaryExport(prisma, config, { purchaseRequest, archive, archivedAt: archive.archived_at ?? new Date(), actorId, requeueOnFailure: true, }); items.push({ target, ...enqueued }); } if (runNow && items.some((item) => !item.duplicated || item.requeued)) { await runIntegrationOutboxOnce(prisma, config, logger); } return items; } async function enqueueReimbursementExportAndMaybeRun({ prisma, config, logger, reimbursement, actorId, runNow, }) { const enqueued = await enqueueReimbursementJournalExport(prisma, config, { reimbursement, actorId, }); if (runNow) { await runIntegrationOutboxOnce(prisma, config, logger); } return enqueued; } function normalizeResourcesInput(value) { const parsed = MASTER_DATA_SYNC_BODY_SCHEMA.safeParse(value ?? {}); if (!parsed.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: parsed.error.flatten(), }); } return normalizeRequestedResources(parsed.data.resources ?? []); } 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', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '查询 ERP 主数据同步状态', security: [{ bearerAuth: [] }], response: { 200: buildDataEnvelopeSchema(ERP_MASTER_DATA_STATUS_SCHEMA, 'ERP 主数据同步状态'), }, }, }, async (request, reply) => { const [items, gate] = await Promise.all([ listMasterDataSyncStates(fastify.prisma), resolveErpGate(fastify.prisma, fastify.config), ]); return sendData(reply, request, { gate, items }); }, ); fastify.post( '/integrations/erp/master-data/sync', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '手动触发 ERP 主数据同步', security: [{ bearerAuth: [] }], response: { 200: buildDataEnvelopeSchema(ERP_MASTER_DATA_STATUS_SCHEMA, 'ERP 主数据同步结果'), 400: buildErrorEnvelopeSchema({ description: '参数校验失败', codeExample: 'VALID_001', messageExample: '参数校验失败', }), 409: buildErrorEnvelopeSchema({ description: 'ERP 总闸关闭或同步失败', codeExample: 'VALID_001', messageExample: '参数校验失败', }), }, }, }, async (request, reply) => { const resources = normalizeResourcesInput(request.body); const gate = await requireErpManagementGate(fastify.prisma, fastify.config); const result = await syncErpMasterData(fastify.prisma, fastify.config, { resources, actorId: request.employee.id, logger: fastify.log, }); await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, action: 'ERP_MASTER_DATA_SYNC_TRIGGERED', targetType: 'integration_master_data', extra: { resources, gateReason: gate.effectiveReason, provider: gate.provider, ok: result.ok, errorCode: result.error?.code ?? null, }, }), }); if (!result.ok) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 409, details: { integration: result.error?.code ?? 'erp_master_data_sync_failed', reason: result.error?.message ?? 'erp_master_data_sync_failed', }, }); } return sendData(reply, request, { gate, items: result.items.filter((item) => resources.includes(item.resourceType)), triggerMode: 'manual', }); }, ); fastify.get( '/integrations/erp/journals/reimbursements/:reimbursementId/preview', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '预览报销付款凭证切片', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['reimbursementId'], properties: { reimbursementId: { type: 'string', format: 'uuid' }, }, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.reimbursementId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { reimbursementId: 'invalid_uuid' }, }); } const reimbursement = await fetchReimbursementForErp(fastify.prisma, parsedId.data); ensurePaidReimbursement(reimbursement); const preview = await buildReimbursementJournalPreview(fastify.prisma, reimbursement); return sendData(reply, request, { journal: preview.journal, exportState: preview.exportState, }); }, ); fastify.post( '/integrations/erp/journals/reimbursements/:reimbursementId/export', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '导出报销付款凭证到 ERP', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['reimbursementId'], properties: { reimbursementId: { type: 'string', format: 'uuid' }, }, }, querystring: { type: 'object', properties: { runNow: { anyOf: [ { type: 'string', enum: ['0', '1', 'false', 'true'] }, { type: 'boolean' }, ], }, }, additionalProperties: true, }, response: { 200: buildDataEnvelopeSchema(ERP_JOURNAL_EXPORT_RESPONSE_SCHEMA, 'ERP 导出任务已入队'), 400: buildErrorEnvelopeSchema({ description: '参数校验失败', codeExample: 'VALID_001', messageExample: '参数校验失败', }), 404: buildErrorEnvelopeSchema({ description: '报销单不存在', codeExample: 'SYS_404', messageExample: 'Not Found', }), 409: buildErrorEnvelopeSchema({ description: '状态不满足或开关阻断', codeExample: 'VALID_001', messageExample: '参数校验失败', }), }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.reimbursementId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { reimbursementId: 'invalid_uuid' }, }); } const reimbursement = await fetchReimbursementForErp(fastify.prisma, parsedId.data); ensurePaidReimbursement(reimbursement); const gate = await requireErpManagementGate(fastify.prisma, fastify.config); const enqueued = await enqueueReimbursementExportAndMaybeRun({ prisma: fastify.prisma, config: fastify.config, logger: fastify.log, reimbursement, actorId: request.employee.id, runNow: parseRunNow(request.query), }); await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, action: 'ERP_SYNC_ENQUEUED', targetType: 'reimbursement_form', targetId: reimbursement.id, extra: { outboxId: enqueued.id, duplicated: enqueued.duplicated, requeued: enqueued.requeued, eventType: enqueued.eventType, featureFlag: FEATURE_FLAG_KEYS.erpSync, featureFlagReason: gate.effectiveReason, }, }), }); return sendData(reply, request, { outboxId: enqueued.id, status: enqueued.status, duplicated: enqueued.duplicated, requeued: enqueued.requeued, eventType: enqueued.eventType, }); }, ); fastify.get( '/integrations/erp/procurement/purchase-requests/:purchaseRequestId/track', { preHandler: fastify.authenticate, schema: { tags: ['integrations'], summary: '查询采购 ERP 同步轨道', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['purchaseRequestId'], properties: { purchaseRequestId: { type: 'string', format: 'uuid' }, }, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.purchaseRequestId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { purchaseRequestId: 'invalid_uuid' }, }); } const [context, gate] = await Promise.all([ fetchReadablePurchaseRequestContext(fastify.prisma, parsedId.data, request), resolveErpGate(fastify.prisma, fastify.config), ]); const canManage = fastify.hasPermission(request.employee, 'integration.manage'); const track = await buildPurchaseRequestErpTrack(fastify.prisma, fastify.config, parsedId.data, { context, canManage, gate, }); return sendData(reply, request, track); }, ); fastify.post( '/integrations/erp/procurement/purchase-requests/:purchaseRequestId/requeue', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '重试或补发采购 ERP 同步', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['purchaseRequestId'], properties: { purchaseRequestId: { type: 'string', format: 'uuid' }, }, }, querystring: { type: 'object', properties: { runNow: { anyOf: [ { type: 'string', enum: ['0', '1', 'false', 'true'] }, { type: 'boolean' }, ], }, }, additionalProperties: true, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.purchaseRequestId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { purchaseRequestId: 'invalid_uuid' }, }); } const targets = normalizeProcurementTargets(request.body); const gate = await requireErpManagementGate(fastify.prisma, fastify.config); const context = await loadPurchaseRequestErpContext(fastify.prisma, parsedId.data); if (!context?.purchaseRequest) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } const items = await enqueueProcurementTargetsAndMaybeRun({ prisma: fastify.prisma, config: fastify.config, logger: fastify.log, context, actorId: request.employee.id, targets, runNow: parseRunNow(request.query), }); for (const item of items) { await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, action: 'ERP_SYNC_ENQUEUED', targetType: item.target === PROCUREMENT_ERP_TARGETS.commitmentJournal ? 'purchase_request' : 'purchase_order_archive', targetId: item.target === PROCUREMENT_ERP_TARGETS.commitmentJournal ? context.purchaseRequest.id : context.purchaseRequest.order_archive?.id ?? null, extra: { erpTarget: item.target, eventType: item.eventType, purchaseRequestId: context.purchaseRequest.id, purchaseOrderArchiveId: context.purchaseRequest.order_archive?.id ?? null, outboxId: item.id, duplicated: item.duplicated, requeued: item.requeued, gateReason: gate.effectiveReason, featureFlag: FEATURE_FLAG_KEYS.erpSync, }, }), }); } return sendData(reply, request, { items: items.map((item) => ({ target: item.target, outboxId: item.id, status: item.status, duplicated: item.duplicated, requeued: item.requeued, eventType: item.eventType, })), }); }, ); fastify.get( '/integrations/erp/assets/:assetId/track', { preHandler: fastify.authenticate, schema: { tags: ['integrations'], summary: '查询资产 ERP 同步轨道', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['assetId'], properties: { assetId: { type: 'string', format: 'uuid' }, }, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.assetId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { assetId: 'invalid_uuid' }, }); } const [context, gate] = await Promise.all([ fetchReadableAssetContext(fastify.prisma, parsedId.data, request), resolveErpGate(fastify.prisma, fastify.config), ]); const canManage = fastify.hasPermission(request.employee, 'integration.manage'); const track = await buildAssetErpTrack(fastify.prisma, fastify.config, parsedId.data, { context, canManage, gate, }); return sendData(reply, request, track); }, ); fastify.post( '/integrations/erp/assets/:assetId/requeue', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '补发或重试资产 ERP 同步', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['assetId'], properties: { assetId: { type: 'string', format: 'uuid' }, }, }, querystring: { type: 'object', properties: { runNow: { anyOf: [ { type: 'string', enum: ['0', '1', 'false', 'true'] }, { type: 'boolean' }, ], }, }, additionalProperties: true, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.assetId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { assetId: 'invalid_uuid' }, }); } const gate = await requireErpManagementGate(fastify.prisma, fastify.config); const context = await loadAssetErpContext(fastify.prisma, parsedId.data); if (!context?.asset) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } const enqueued = await enqueueAssetExportAndMaybeRun({ prisma: fastify.prisma, config: fastify.config, logger: fastify.log, context, actorId: request.employee.id, runNow: parseRunNow(request.query), }); await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, action: 'ERP_SYNC_ENQUEUED', targetType: 'asset_card', targetId: context.asset.id, extra: { erpTarget: ASSET_ERP_TARGET, eventType: enqueued.eventType, assetId: context.asset.id, assetNo: context.asset.asset_no ?? null, outboxId: enqueued.id, duplicated: enqueued.duplicated, requeued: enqueued.requeued, gateReason: gate.effectiveReason, featureFlag: FEATURE_FLAG_KEYS.erpSync, }, }), }); return sendData(reply, request, { outboxId: enqueued.id, status: enqueued.status, duplicated: enqueued.duplicated, requeued: enqueued.requeued, eventType: enqueued.eventType, }); }, ); fastify.get( '/integrations/erp/travel/reconciliation-batches/:batchId/track', { preHandler: fastify.authenticate, schema: { tags: ['integrations'], summary: '查询差旅对账批次 ERP 同步轨道', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['batchId'], properties: { batchId: { type: 'string', format: 'uuid' }, }, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.batchId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { batchId: 'invalid_uuid' }, }); } const [context, gate] = await Promise.all([ fetchReadableTravelReconciliationContext(fastify.prisma, parsedId.data, request), resolveErpGate(fastify.prisma, fastify.config), ]); const canManage = fastify.hasPermission(request.employee, 'integration.manage'); const track = await buildTravelReconciliationErpTrack(fastify.prisma, fastify.config, parsedId.data, { context, canManage, gate, }); return sendData(reply, request, track); }, ); fastify.post( '/integrations/erp/travel/reconciliation-batches/:batchId/requeue', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '重试差旅对账批次 ERP 摘要桥同步', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['batchId'], properties: { batchId: { type: 'string', format: 'uuid' }, }, }, querystring: { type: 'object', properties: { runNow: { anyOf: [ { type: 'string', enum: ['0', '1', 'false', 'true'] }, { type: 'boolean' }, ], }, }, additionalProperties: true, }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.batchId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { batchId: 'invalid_uuid' }, }); } const gate = await requireErpManagementGate(fastify.prisma, fastify.config); const context = await loadTravelReconciliationErpContext(fastify.prisma, parsedId.data); if (!context) { throw new AppError({ code: 'SYS_404', message: 'Not Found', statusCode: 404, }); } if (context.status !== 'reconciled') { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 409, details: { batch: 'travel_reconciliation_batch_not_reconciled' }, }); } const enqueued = await enqueueTravelReconciliationExportAndMaybeRun({ prisma: fastify.prisma, config: fastify.config, logger: fastify.log, context, actorId: request.employee.id, runNow: parseRunNow(request.query), }); await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, action: 'ERP_SYNC_ENQUEUED', targetType: 'travel_reconciliation_batch', targetId: context.id, extra: { erpTarget: 'travel_reconciliation_summary', eventType: enqueued.eventType, batchId: context.id, batchNo: context.batch_no, outboxId: enqueued.id, duplicated: enqueued.duplicated, requeued: enqueued.requeued, gateReason: gate.effectiveReason, featureFlag: FEATURE_FLAG_KEYS.erpSync, }, }), }); return sendData(reply, request, { outboxId: enqueued.id, status: enqueued.status, duplicated: enqueued.duplicated, requeued: enqueued.requeued, eventType: enqueued.eventType, }); }, ); fastify.post( '/integrations/erp/sync/:reimbursementId', { preHandler: fastify.requirePermission('integration.manage'), schema: { tags: ['integrations'], summary: '手动触发 ERP 同步', security: [{ bearerAuth: [] }], params: { type: 'object', required: ['reimbursementId'], properties: { reimbursementId: { type: 'string', format: 'uuid' }, }, }, querystring: { type: 'object', properties: { runNow: { anyOf: [ { type: 'string', enum: ['0', '1', 'false', 'true'] }, { type: 'boolean' }, ], }, }, additionalProperties: true, }, response: { 200: buildDataEnvelopeSchema(ERP_SYNC_RESPONSE_SCHEMA, 'ERP 同步任务已入队'), 400: buildErrorEnvelopeSchema({ description: '参数校验失败', codeExample: 'VALID_001', messageExample: '参数校验失败', }), 404: buildErrorEnvelopeSchema({ description: '报销单不存在', codeExample: 'SYS_404', messageExample: 'Not Found', }), 409: buildErrorEnvelopeSchema({ description: '状态不满足或开关阻断', codeExample: 'VALID_001', messageExample: '参数校验失败', detailsExample: { integration: 'feature_flag_blocked', featureFlag: 'erp_sync', reason: 'department_not_allowed', }, }), }, }, }, async (request, reply) => { const parsedId = UUID_SCHEMA.safeParse(request.params?.reimbursementId); if (!parsedId.success) { throw new AppError({ code: 'VALID_001', message: '参数校验失败', statusCode: 400, details: { reimbursementId: 'invalid_uuid' }, }); } const reimbursement = await fetchReimbursementForErp(fastify.prisma, parsedId.data); ensurePaidReimbursement(reimbursement); const gate = await requireErpManagementGate(fastify.prisma, fastify.config); const enqueued = await enqueueReimbursementExportAndMaybeRun({ prisma: fastify.prisma, config: fastify.config, logger: fastify.log, reimbursement, actorId: request.employee.id, runNow: parseRunNow(request.query), }); await fastify.prisma.auditLog.create({ data: buildAuditLogData(request, { actorId: request.employee.id, action: 'ERP_SYNC_ENQUEUED', targetType: 'reimbursement_form', targetId: reimbursement.id, extra: { outboxId: enqueued.id, duplicated: enqueued.duplicated, requeued: enqueued.requeued, featureFlag: FEATURE_FLAG_KEYS.erpSync, featureFlagReason: gate.effectiveReason, }, }), }); return sendData(reply, request, { outboxId: enqueued.id, status: enqueued.status, duplicated: enqueued.duplicated, }); }, ); } integrationsRoutes.__testables = { canReadAsset, canReadPurchaseRequest, canReadTravelReconciliationBatch, enqueueAssetExportAndMaybeRun, enqueueTravelReconciliationExportAndMaybeRun, ensureProcurementTargetEligible, normalizeProcurementTargets, normalizeResourcesInput, parseRunNow, }; module.exports = integrationsRoutes;