feat(P19ERP3-1,P19ERP3-2): 实现资产ERP摘要服务与建卡/流转出站联动

这个提交包含在:
2026-03-09 11:15:21 +08:00
父节点 7613d7d912
当前提交 0168ebfe2e
共修改 10 个文件,包含 1128 行新增和 4 行删除
+4
查看文件
@@ -110,11 +110,15 @@ function getConfig() {
erpSyncEndpoint: process.env.ERP_SYNC_ENDPOINT ?? "",
erpJournalExportEndpoint: process.env.ERP_JOURNAL_EXPORT_ENDPOINT ?? "",
erpProcurementExportEndpoint: process.env.ERP_PROCUREMENT_EXPORT_ENDPOINT ?? "",
erpAssetExportEndpoint: process.env.ERP_ASSET_EXPORT_ENDPOINT ?? "",
erpMasterDataEndpoint: process.env.ERP_MASTER_DATA_ENDPOINT ?? "",
erpSyncTimeoutMs: Number(process.env.ERP_SYNC_TIMEOUT_MS ?? 5000),
erpProcurementTimeoutMs: Number(
process.env.ERP_PROCUREMENT_TIMEOUT_MS ?? process.env.ERP_SYNC_TIMEOUT_MS ?? 5000,
),
erpAssetTimeoutMs: Number(
process.env.ERP_ASSET_TIMEOUT_MS ?? process.env.ERP_SYNC_TIMEOUT_MS ?? 5000,
),
erpMasterDataTimeoutMs: Number(
process.env.ERP_MASTER_DATA_TIMEOUT_MS ?? process.env.ERP_SYNC_TIMEOUT_MS ?? 5000,
),
+5
查看文件
@@ -52,6 +52,11 @@ async function runIntegrationOutboxOnce(prisma, config, logger) {
journalType: item.payload?.journalType ?? null,
poNo: item.payload?.poNo ?? item.payload?.references?.poNo ?? null,
contractNo: item.payload?.contractNo ?? item.payload?.references?.contractNo ?? null,
assetId:
item.payload?.assetId ??
(item.payload?.entityType === 'asset_card' ? item.payload?.entityId ?? null : null),
assetNo: item.payload?.assetNo ?? null,
changeType: item.payload?.changeType ?? null,
entityType: item.payload?.entityType ?? null,
entityId: item.payload?.entityId ?? null,
channelId: item.payload?.channelId ?? null,
+120
查看文件
@@ -4,6 +4,12 @@ const { AppError } = require("../utils/AppError");
const { buildAuditLogData } = require("../utils/audit");
const { parseDateOnlyString } = require("../utils/date");
const { issueDocumentNumber } = require("../services/documentSequence");
const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require('../services/featureFlags');
const {
ASSET_ERP_TARGET,
enqueueAssetLedgerSummaryExport,
resolveAssetLedgerSummaryChange,
} = require('../services/integrations/erpAsset');
const { sendData } = require("../utils/reply");
const UUID_SCHEMA = z.string().uuid();
@@ -557,6 +563,92 @@ async function ensureApprovedPurchaseRequest(tx, purchaseRequestId) {
return record;
}
async function handleAssetLedgerSummaryErpSync({
tx,
config,
request,
actorId,
asset,
sourceEventName,
changeType,
changedFields,
evaluateGate = evaluateFeatureFlag,
enqueueExport = enqueueAssetLedgerSummaryExport,
}) {
const gate = await evaluateGate({
prisma: tx,
config,
flagKey: FEATURE_FLAG_KEYS.erpSync,
departmentId: null,
strictDepartment: false,
});
if (!gate.enabled) {
await tx.auditLog.create({
data: buildAuditLogData(request, {
actorId,
action: 'ERP_SYNC_SKIPPED_BY_FEATURE_FLAG',
targetType: 'asset_card',
targetId: asset.id,
extra: {
erpTarget: ASSET_ERP_TARGET,
eventType: 'erp.asset.export',
assetId: asset.id,
assetNo: asset.asset_no,
changeType,
gateReason: gate.reason,
featureFlag: FEATURE_FLAG_KEYS.erpSync,
},
}),
});
return {
skipped: true,
gate,
asset,
};
}
const enqueued = await enqueueExport(tx, config, {
asset,
sourceEventName,
changeType,
changedFields,
occurredAt: new Date(),
actorId,
requeueOnFailure: false,
});
await tx.auditLog.create({
data: buildAuditLogData(request, {
actorId,
action: 'ERP_SYNC_ENQUEUED',
targetType: 'asset_card',
targetId: asset.id,
extra: {
erpTarget: ASSET_ERP_TARGET,
eventType: enqueued.eventType,
assetId: asset.id,
assetNo: asset.asset_no,
outboxId: enqueued.id,
duplicated: enqueued.duplicated,
requeued: enqueued.requeued,
changeType,
changedFields: Array.isArray(changedFields) ? changedFields : [],
gateReason: gate.reason,
featureFlag: FEATURE_FLAG_KEYS.erpSync,
},
}),
});
return {
skipped: false,
gate,
asset,
enqueued,
};
}
async function assetsRoutes(fastify) {
fastify.get(
"/asset-categories",
@@ -843,6 +935,17 @@ async function assetsRoutes(fastify) {
}),
});
await handleAssetLedgerSummaryErpSync({
tx,
config: fastify.config,
request,
actorId: request.employee.id,
asset: created,
sourceEventName: 'asset.card.created',
changeType: 'created',
changedFields: [],
});
return created;
});
@@ -953,6 +1056,20 @@ async function assetsRoutes(fastify) {
}),
});
const assetLedgerSummaryChange = resolveAssetLedgerSummaryChange(current, updated);
if (assetLedgerSummaryChange) {
await handleAssetLedgerSummaryErpSync({
tx,
config: fastify.config,
request,
actorId: request.employee.id,
asset: updated,
sourceEventName: assetLedgerSummaryChange.sourceEventName,
changeType: assetLedgerSummaryChange.changeType,
changedFields: assetLedgerSummaryChange.changedFields,
});
}
return updated;
});
@@ -962,3 +1079,6 @@ async function assetsRoutes(fastify) {
}
module.exports = assetsRoutes;
module.exports.__testables = {
handleAssetLedgerSummaryErpSync,
};
+19 -4
查看文件
@@ -1,5 +1,6 @@
const {
pushPaidReimbursementToErp,
exportAssetToErp,
exportJournalToErp,
exportProcurementToErp,
} = require('./erp');
@@ -136,6 +137,17 @@ function resolveBindingSpecs(eventType, payload, response) {
return specs.filter(Boolean);
}
if (eventType === 'erp.asset.export') {
return [
createBindingSpec(
'asset_erp_card',
payload?.entityId,
extractResponseValue(response, 'externalAssetId') ?? extractResponseValue(response, 'externalId'),
extractResponseValue(response, 'externalAssetNo') ?? extractResponseValue(response, 'externalCode') ?? payload?.assetNo ?? null,
),
].filter(Boolean);
}
return [];
}
@@ -179,9 +191,11 @@ async function dispatchErpEvent({ prisma, config, eventType, payload }) {
const result =
eventType === 'erp.procurement.export'
? await exportProcurementToErp(config, payload)
: eventType === 'erp.journal.export'
? await exportJournalToErp(config, payload)
: await pushPaidReimbursementToErp(config, payload);
: eventType === 'erp.asset.export'
? await exportAssetToErp(config, payload)
: eventType === 'erp.journal.export'
? await exportJournalToErp(config, payload)
: await pushPaidReimbursementToErp(config, payload);
if (!result.ok) {
throw toDispatchError(result);
@@ -200,7 +214,8 @@ async function dispatchOutboxEvent({ prisma, config, eventType, payload }) {
if (
eventType === 'erp.reimbursement.paid' ||
eventType === 'erp.journal.export' ||
eventType === 'erp.procurement.export'
eventType === 'erp.procurement.export' ||
eventType === 'erp.asset.export'
) {
return dispatchErpEvent({ prisma, config, eventType, payload });
}
+57
查看文件
@@ -41,6 +41,13 @@ function resolveProcurementExportEndpoint(config) {
);
}
function resolveAssetExportEndpoint(config) {
return (
String(config.erpAssetExportEndpoint ?? '').trim() ||
String(config.erpSyncEndpoint ?? '').trim()
);
}
async function parseResponseBody(response) {
const contentType = response.headers.get('content-type') || '';
if (contentType.includes('application/json')) {
@@ -236,6 +243,55 @@ async function exportJournalToErp(config, payload) {
};
}
async function exportAssetToErp(config, payload) {
if (!config.erpSyncEnabled) {
return {
ok: false,
code: 'erp_sync_disabled',
message: 'ERP sync disabled',
retryable: false,
};
}
const provider = normalizeProvider(config);
if (provider === 'mock') {
const identity = payload?.assetNo ?? payload?.entityId ?? 'UNKNOWN';
return {
ok: true,
code: 'mock_synced',
message: 'mock ERP asset synced',
response: {
externalAssetId: `ERP-ASSET-${identity}`,
externalAssetNo: `ERP-ASSET-${identity}`,
},
};
}
if (provider === 'http') {
return postJsonToErp(config, {
endpoint: resolveAssetExportEndpoint(config),
timeoutMs: coerceTimeout(config.erpAssetTimeoutMs, config.erpSyncTimeoutMs),
payload,
successCode: 'http_synced',
successMessage: 'ERP asset synced',
missingEndpointCode: 'erp_asset_endpoint_missing',
missingEndpointMessage: 'ERP asset endpoint missing',
http4xxCode: 'erp_asset_http_4xx',
http5xxCode: 'erp_asset_http_5xx',
timeoutCode: 'erp_asset_timeout',
networkCode: 'erp_asset_network_error',
messagePrefix: 'ERP asset sync',
});
}
return {
ok: false,
code: 'erp_sync_provider_invalid',
message: `Unsupported ERP sync provider: ${provider}`,
retryable: false,
};
}
async function exportProcurementToErp(config, payload) {
if (!config.erpSyncEnabled) {
return {
@@ -299,6 +355,7 @@ async function pushPaidReimbursementToErp(config, payload) {
module.exports = {
buildErpPayload,
exportAssetToErp,
exportJournalToErp,
exportProcurementToErp,
normalizeProvider,
@@ -0,0 +1,552 @@
const crypto = require('node:crypto');
const { enqueueIntegrationEvent } = require('./outbox');
const ASSET_ERP_EVENT_TYPE = 'erp.asset.export';
const ASSET_ERP_BINDING_ENTITY_TYPE = 'asset_erp_card';
const ASSET_ERP_TARGET = 'asset_ledger_summary';
const ASSET_TRACK_STATUSES = Object.freeze({
notTriggered: 'not_triggered',
skipped: 'skipped',
});
const ASSET_TRACK_ACTIONS = Object.freeze({
none: 'none',
backfill: 'backfill',
retry: 'retry',
resync: 'resync',
});
const ASSET_EXPORTABLE_STATUS_CHANGES = new Set(['retired', 'archived']);
const ASSET_ERP_CONTEXT_SELECT = {
id: true,
asset_no: true,
name: true,
category_id: true,
legal_entity_id: true,
cost_center_id: true,
purchase_request_id: true,
serial_no: true,
location: true,
status: true,
assignee_employee_id: true,
activated_at: true,
retired_at: true,
created_at: true,
updated_at: true,
category: {
select: {
id: true,
code: true,
name: true,
document_category: {
select: {
code: true,
},
},
},
},
legal_entity: {
select: {
id: true,
code: true,
name: true,
},
},
cost_center: {
select: {
id: true,
code: true,
name: true,
},
},
purchase_request: {
select: {
id: true,
request_no: true,
title: true,
status: true,
},
},
assignee: {
select: {
id: true,
employee_no: true,
name: true,
},
},
};
function normalizeString(value) {
const text = String(value ?? '').trim();
return text || null;
}
function toIsoString(value) {
if (!value) {
return null;
}
try {
return new Date(value).toISOString();
} catch {
return null;
}
}
function toDateOnly(value) {
const iso = toIsoString(value);
return iso ? iso.slice(0, 10) : null;
}
function stableNormalizeJson(value) {
if (Array.isArray(value)) {
return value.map((item) => stableNormalizeJson(item));
}
if (value && typeof value === 'object' && value.constructor === Object) {
return Object.keys(value)
.sort()
.reduce((result, key) => {
result[key] = stableNormalizeJson(value[key]);
return result;
}, {});
}
return value;
}
function stableStringifyJson(value) {
return JSON.stringify(stableNormalizeJson(value));
}
function hashAssetPayload(payload) {
return crypto.createHash('sha256').update(stableStringifyJson(payload)).digest('hex');
}
function buildAssetLedgerSummaryDedupeKey(assetId) {
return `asset_card:${assetId}:ledger_summary`;
}
function resolveAssetChangeType(status, changedFields) {
const normalizedStatus = normalizeString(status);
if (normalizedStatus === 'retired') {
return 'retired';
}
if (normalizedStatus === 'archived') {
return 'archived';
}
return changedFields?.length ? 'financial_dimensions_changed' : 'created';
}
function buildAssetLedgerSummaryPayload(input) {
const asset = input?.asset ?? {};
const changedFields = Array.isArray(input?.changedFields)
? Array.from(new Set(input.changedFields.filter(Boolean)))
: [];
const sourceEventName = normalizeString(input?.sourceEventName) ?? 'asset.card.created';
const changeType = normalizeString(input?.changeType) ?? resolveAssetChangeType(asset.status, changedFields);
return {
exportType: 'asset_ledger_summary',
sourceEventName,
changeType,
entityType: 'asset_card',
entityId: asset.id ?? null,
assetNo: asset.asset_no ?? null,
name: asset.name ?? null,
status: asset.status ?? null,
location: normalizeString(asset.location),
activatedAt: toDateOnly(asset.activated_at),
retiredAt: toDateOnly(asset.retired_at),
occurredAt:
toIsoString(input?.occurredAt) ??
toIsoString(asset.updated_at) ??
toIsoString(asset.created_at),
changedFields,
category: asset.category
? {
id: asset.category.id,
code: asset.category.code ?? null,
name: asset.category.name ?? null,
documentCategoryCode: asset.category.document_category?.code ?? null,
}
: {
id: asset.category_id ?? null,
code: null,
name: null,
documentCategoryCode: null,
},
legalEntity: asset.legal_entity
? {
id: asset.legal_entity.id,
code: asset.legal_entity.code ?? null,
name: asset.legal_entity.name ?? null,
}
: {
id: asset.legal_entity_id ?? null,
code: null,
name: null,
},
costCenter: asset.cost_center
? {
id: asset.cost_center.id,
code: asset.cost_center.code ?? null,
name: asset.cost_center.name ?? null,
}
: {
id: asset.cost_center_id ?? null,
code: null,
name: null,
},
purchaseRequest: asset.purchase_request
? {
id: asset.purchase_request.id,
requestNo: asset.purchase_request.request_no ?? null,
title: asset.purchase_request.title ?? null,
}
: {
id: asset.purchase_request_id ?? null,
requestNo: null,
title: null,
},
assignee: asset.assignee
? {
id: asset.assignee.id,
employeeNo: asset.assignee.employee_no ?? null,
name: asset.assignee.name ?? null,
}
: {
id: asset.assignee_employee_id ?? null,
employeeNo: null,
name: null,
},
};
}
function resolveAssetLedgerSummaryChange(previousAsset, nextAsset) {
if (!previousAsset || !nextAsset) {
return null;
}
const changedFields = [];
if (previousAsset.legal_entity_id !== nextAsset.legal_entity_id) {
changedFields.push('legalEntityId');
}
if (previousAsset.cost_center_id !== nextAsset.cost_center_id) {
changedFields.push('costCenterId');
}
const nextStatus = normalizeString(nextAsset.status);
if (previousAsset.status !== nextAsset.status && ASSET_EXPORTABLE_STATUS_CHANGES.has(nextStatus)) {
changedFields.push('status');
}
if (!changedFields.length) {
return null;
}
return {
sourceEventName: 'asset.ledger.summary.updated',
changeType: resolveAssetChangeType(nextAsset.status, changedFields),
changedFields,
};
}
function resolveAssetLedgerSummaryDrift(asset, previousPayload) {
if (!asset || !previousPayload) {
return null;
}
const changedFields = [];
const previousLegalEntityId = previousPayload.legalEntity?.id ?? null;
const previousCostCenterId = previousPayload.costCenter?.id ?? null;
const previousStatus = previousPayload.status ?? null;
if ((asset.legal_entity_id ?? null) !== previousLegalEntityId) {
changedFields.push('legalEntityId');
}
if ((asset.cost_center_id ?? null) !== previousCostCenterId) {
changedFields.push('costCenterId');
}
const assetStatus = normalizeString(asset.status);
if (ASSET_EXPORTABLE_STATUS_CHANGES.has(assetStatus) && assetStatus !== previousStatus) {
changedFields.push('status');
}
if (!changedFields.length) {
return null;
}
return {
sourceEventName: 'asset.ledger.summary.updated',
changeType: resolveAssetChangeType(asset.status, changedFields),
changedFields,
};
}
function mapIntegrationBinding(record) {
if (!record) {
return null;
}
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: record.last_synced_at?.toISOString?.() ?? null,
};
}
function redactBinding(binding, canManage) {
if (!binding) {
return null;
}
if (canManage) {
return binding;
}
return {
...binding,
externalId: null,
externalCode: null,
};
}
function mapIntegrationOutboxState(record) {
if (!record) {
return null;
}
return {
id: record.id,
eventType: record.event_type,
status: record.status,
attempts: Number(record.attempts ?? 0),
lastError: record.last_error ?? null,
processedAt: record.processed_at?.toISOString?.() ?? null,
updatedAt: record.updated_at?.toISOString?.() ?? null,
};
}
async function findOutboxByKey(prisma, eventType, dedupeKey) {
if (!prisma || !eventType || !dedupeKey) {
return null;
}
return prisma.integrationOutbox.findUnique({
where: {
event_type_dedupe_key: {
event_type: eventType,
dedupe_key: dedupeKey,
},
},
});
}
async function findBindingByLocalId(prisma, entityType, localId) {
if (!prisma || !entityType || !localId) {
return null;
}
return prisma.integrationBinding.findUnique({
where: {
provider_entity_type_local_id: {
provider: 'erp',
entity_type: entityType,
local_id: localId,
},
},
});
}
async function enqueueAssetLedgerSummaryExport(prisma, config, input) {
const payload =
input?.payload ??
buildAssetLedgerSummaryPayload({
asset: input?.asset,
sourceEventName: input?.sourceEventName,
changeType: input?.changeType,
occurredAt: input?.occurredAt,
changedFields: input?.changedFields,
});
const enqueued = await enqueueIntegrationEvent(prisma, {
eventType: ASSET_ERP_EVENT_TYPE,
payload,
dedupeKey: buildAssetLedgerSummaryDedupeKey(payload.entityId),
maxAttempts: config.integrationOutboxMaxAttempts,
actorId: input?.actorId ?? null,
requeueOnFailure: input?.requeueOnFailure === true,
resetOnPayloadChange: true,
});
return {
...enqueued,
eventType: ASSET_ERP_EVENT_TYPE,
erpTarget: ASSET_ERP_TARGET,
payload,
};
}
async function loadAssetErpContext(prisma, assetId) {
const asset = await prisma.assetCard.findUnique({
where: { id: assetId },
select: ASSET_ERP_CONTEXT_SELECT,
});
if (!asset) {
return null;
}
return { asset };
}
function buildAssetTrackSummary(context) {
const asset = context?.asset ?? {};
return {
id: asset.id,
assetNo: asset.asset_no ?? null,
name: asset.name ?? null,
status: asset.status ?? null,
legalEntity: asset.legal_entity
? {
id: asset.legal_entity.id,
code: asset.legal_entity.code ?? null,
name: asset.legal_entity.name ?? null,
}
: null,
costCenter: asset.cost_center
? {
id: asset.cost_center.id,
code: asset.cost_center.code ?? null,
name: asset.cost_center.name ?? null,
}
: null,
purchaseRequest: asset.purchase_request
? {
id: asset.purchase_request.id,
requestNo: asset.purchase_request.request_no ?? null,
title: asset.purchase_request.title ?? null,
}
: null,
};
}
function buildAssetTrackPreview(asset, options = {}) {
return {
changeType: options.changeType ?? 'created',
sourceEventName: options.sourceEventName ?? 'asset.card.created',
categoryCode: asset?.category?.code ?? null,
legalEntityCode: asset?.legal_entity?.code ?? null,
costCenterCode: asset?.cost_center?.code ?? null,
status: asset?.status ?? null,
};
}
function resolveTrackStatus({ outbox, binding, gateEnabled }) {
if (outbox?.status) {
return outbox.status;
}
if (binding) {
return 'succeeded';
}
if (!gateEnabled) {
return ASSET_TRACK_STATUSES.skipped;
}
return ASSET_TRACK_STATUSES.notTriggered;
}
function resolveTrackAction({ canManage, gateEnabled, outbox, binding, drift }) {
if (!canManage || !gateEnabled) {
return ASSET_TRACK_ACTIONS.none;
}
if (outbox && ['failed', 'dead_letter'].includes(outbox.status)) {
return ASSET_TRACK_ACTIONS.retry;
}
if (!outbox && !binding) {
return ASSET_TRACK_ACTIONS.backfill;
}
if (drift) {
return ASSET_TRACK_ACTIONS.resync;
}
return ASSET_TRACK_ACTIONS.none;
}
async function buildAssetErpTrack(prisma, config, assetId, options = {}) {
const context = options.context ?? (await loadAssetErpContext(prisma, assetId));
if (!context?.asset) {
return null;
}
const asset = context.asset;
const canManage = Boolean(options.canManage);
const gateEnabled = Boolean(options.gate?.effectiveEnabled ?? options.gate?.enabled ?? true);
const dedupeKey = buildAssetLedgerSummaryDedupeKey(asset.id);
const [outboxRow, bindingRow] = await Promise.all([
findOutboxByKey(prisma, ASSET_ERP_EVENT_TYPE, dedupeKey),
findBindingByLocalId(prisma, ASSET_ERP_BINDING_ENTITY_TYPE, asset.id),
]);
const outbox = mapIntegrationOutboxState(outboxRow);
const binding = redactBinding(mapIntegrationBinding(bindingRow), canManage);
const drift = resolveAssetLedgerSummaryDrift(asset, outboxRow?.payload ?? null);
const preview = buildAssetTrackPreview(asset, {
changeType: drift?.changeType ?? outboxRow?.payload?.changeType ?? 'created',
sourceEventName:
drift?.sourceEventName ??
outboxRow?.payload?.sourceEventName ??
'asset.card.created',
});
return {
asset: buildAssetTrackSummary(context),
stage: {
eligible: true,
reason: !gateEnabled && !outboxRow && !bindingRow ? (options.gate?.effectiveReason ?? options.gate?.reason ?? 'erp_sync_disabled') : null,
deliveryStatus: resolveTrackStatus({ outbox, binding: bindingRow, gateEnabled }),
actionHint: resolveTrackAction({
canManage,
gateEnabled,
outbox,
binding: bindingRow,
drift,
}),
preview,
outbox,
binding,
},
permissions: {
canRetry: Boolean(canManage && gateEnabled),
canSeeExternalRefs: canManage,
},
};
}
module.exports = {
ASSET_ERP_BINDING_ENTITY_TYPE,
ASSET_ERP_CONTEXT_SELECT,
ASSET_ERP_EVENT_TYPE,
ASSET_ERP_TARGET,
ASSET_TRACK_ACTIONS,
ASSET_TRACK_STATUSES,
buildAssetErpTrack,
buildAssetLedgerSummaryDedupeKey,
buildAssetLedgerSummaryPayload,
buildAssetTrackPreview,
enqueueAssetLedgerSummaryExport,
hashAssetPayload,
loadAssetErpContext,
resolveAssetLedgerSummaryChange,
resolveAssetLedgerSummaryDrift,
stableStringifyJson,
};