wangsli/hyb-oa
1
0
派生 0
代码 工单 合并请求 工作流 5 软件包 项目 发布 百科 活动

feat(P19ERP2-1,P19ERP2-2): 实现采购ERP凭证与摘要出站服务

浏览代码
这个提交包含在：
wangsli
2026-03-08 23:50:02 +08:00
父节点 8f4a50cdd7
当前提交 486660df06
共修改 15 个文件，包含 1912 行新增和 181 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件
server/src/config.js
+4
查看文件
@@ -109,8 +109,12 @@ function getConfig() {
erpSyncProvider: process.env.ERP_SYNC_PROVIDER ?? "mock",
erpSyncEndpoint: process.env.ERP_SYNC_ENDPOINT ?? "",
erpJournalExportEndpoint: process.env.ERP_JOURNAL_EXPORT_ENDPOINT ?? "",
erpProcurementExportEndpoint: process.env.ERP_PROCUREMENT_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,
),
erpMasterDataTimeoutMs: Number(
process.env.ERP_MASTER_DATA_TIMEOUT_MS ?? process.env.ERP_SYNC_TIMEOUT_MS ?? 5000,
),
server/src/jobs/integrationOutboxJob.js
+11
查看文件
@@ -41,6 +41,17 @@ async function runIntegrationOutboxOnce(prisma, config, logger) {
reimbursementId:
item.payload?.reimbursementId ??
(item.payload?.entityType === 'reimbursement_form' ? item.payload?.entityId ?? null : null),
purchaseRequestId:
item.payload?.purchaseRequestId ??
item.payload?.references?.purchaseRequestId ??
(item.payload?.entityType === 'purchase_request' ? item.payload?.entityId ?? null : null),
purchaseOrderArchiveId:
item.payload?.purchaseOrderArchiveId ??
item.payload?.references?.purchaseOrderArchiveId ??
(item.payload?.entityType === 'purchase_order_archive' ? item.payload?.entityId ?? null : null),
journalType: item.payload?.journalType ?? null,
poNo: item.payload?.poNo ?? item.payload?.references?.poNo ?? null,
contractNo: item.payload?.contractNo ?? item.payload?.references?.contractNo ?? null,
entityType: item.payload?.entityType ?? null,
entityId: item.payload?.entityId ?? null,
channelId: item.payload?.channelId ?? null,
server/src/routes/purchaseRequests.js
+144
查看文件
@@ -11,6 +11,13 @@ const { signPayload, verifyToken } = require("../utils/signedUrl");
const { parseWorkflowSteps, resolveActionableStep, toMoneyString } = require("../utils/workflow");
const { resolveApplicableBudgetPoolByContext } = require("../services/budgetPool");
const { issueDocumentNumber } = require("../services/documentSequence");
const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require("../services/featureFlags");
const {
PROCUREMENT_ERP_TARGETS,
enqueuePurchaseAccrualJournalExport,
enqueuePurchaseArchiveSummaryExport,
loadPurchaseRequestErpContext,
} = require("../services/integrations/erpProcurement");
const { reserveBudgetForEntity, releaseReservedBudgetIfAny } = require("../services/purchaseBudget");
const { sendNotificationEmailSafe } = require("../services/email");
const { createNotifications, resolveRecipientsForAssignee } = require("../services/notifications");
@@ -837,6 +844,116 @@ async function loadReadablePurchaseRequest(tx, purchaseRequestId, request) {
return purchaseRequest;
}
async function handlePurchaseArchiveErpSync({
tx,
config,
request,
actorId,
purchaseRequestId,
archiveId,
isFirstArchive,
loadContext = loadPurchaseRequestErpContext,
evaluateGate = evaluateFeatureFlag,
enqueueAccrual = enqueuePurchaseAccrualJournalExport,
enqueueSummary = enqueuePurchaseArchiveSummaryExport,
}) {
const context = await loadContext(tx, purchaseRequestId);
if (!context?.purchaseRequest?.order_archive) {
return { skipped: true, reason: 'purchase_order_archive_missing' };
}
const purchaseRequest = context.purchaseRequest;
const archive = purchaseRequest.order_archive;
const departmentId = purchaseRequest.department_id ?? request.employee.department_id ?? null;
const gate = await evaluateGate({
prisma: tx,
config,
flagKey: FEATURE_FLAG_KEYS.erpSync,
departmentId,
strictDepartment: true,
});
if (!gate.enabled) {
const skipTargets = isFirstArchive
? [PROCUREMENT_ERP_TARGETS.accrualJournal, PROCUREMENT_ERP_TARGETS.archiveSummary]
: [PROCUREMENT_ERP_TARGETS.archiveSummary];
for (const erpTarget of skipTargets) {
await tx.auditLog.create({
data: buildAuditLogData(request, {
actorId,
action: 'ERP_SYNC_SKIPPED_BY_FEATURE_FLAG',
targetType: 'purchase_order_archive',
targetId: archiveId,
extra: {
erpTarget,
eventType: erpTarget === PROCUREMENT_ERP_TARGETS.archiveSummary ? 'erp.procurement.export' : 'erp.journal.export',
purchaseRequestId,
purchaseOrderArchiveId: archiveId,
gateReason: gate.reason,
featureFlag: FEATURE_FLAG_KEYS.erpSync,
departmentId,
},
}),
});
}
return { skipped: true, gate, purchaseRequest, archive };
}
const enqueuedItems = [];
if (isFirstArchive) {
const accrual = await enqueueAccrual(tx, config, {
purchaseRequest,
archive,
occurredAt: archive.archived_at ?? new Date(),
actorId,
requeueOnFailure: false,
});
enqueuedItems.push(accrual);
}
const summary = await enqueueSummary(tx, config, {
purchaseRequest,
archive,
archivedAt: archive.archived_at ?? new Date(),
actorId,
requeueOnFailure: false,
});
enqueuedItems.push(summary);
for (const item of enqueuedItems) {
await tx.auditLog.create({
data: buildAuditLogData(request, {
actorId,
action: 'ERP_SYNC_ENQUEUED',
targetType: 'purchase_order_archive',
targetId: archiveId,
extra: {
erpTarget: item.erpTarget,
eventType: item.eventType,
purchaseRequestId,
purchaseOrderArchiveId: archiveId,
outboxId: item.id,
duplicated: item.duplicated,
requeued: item.requeued,
gateReason: gate.reason,
featureFlag: FEATURE_FLAG_KEYS.erpSync,
departmentId,
},
}),
});
}
return {
skipped: false,
gate,
purchaseRequest,
archive,
items: enqueuedItems,
};
}
async function purchaseRequestRoutes(fastify) {
fastify.get(
"/purchase-requests",
@@ -1527,6 +1644,11 @@ async function purchaseRequestRoutes(fastify) {
throw buildValidationError({ status: "only_approved_archivable" }, 409);
}
const existingArchive = await tx.purchaseOrderArchive.findUnique({
where: { purchase_request_id: purchaseRequestId },
select: { id: true },
});
const archive = await tx.purchaseOrderArchive.upsert({
where: { purchase_request_id: purchaseRequestId },
update: {
@@ -1575,6 +1697,23 @@ async function purchaseRequestRoutes(fastify) {
}),
});
try {
await handlePurchaseArchiveErpSync({
tx,
config: fastify.config,
request,
actorId: request.employee.id,
purchaseRequestId,
archiveId: archive.id,
isFirstArchive: !existingArchive,
});
} catch (error) {
fastify.log.error(
{ err: error, purchaseRequestId, purchaseOrderArchiveId: archive.id },
"Failed to enqueue purchase archive ERP events",
);
}
response = {
archiveOrder: mapPurchaseOrderArchive(archive),
};
@@ -1942,3 +2081,8 @@ async function purchaseRequestRoutes(fastify) {
}
module.exports = purchaseRequestRoutes;
module.exports.__testables = {
handlePurchaseArchiveErpSync,
parseArchiveBody,
};
server/src/routes/workflow.js
+106
查看文件
@@ -1,10 +1,17 @@
const { z } = require("zod");
const { AppError } = require("../utils/AppError");
const { buildAuditLogData } = require("../utils/audit");
const { sendData } = require("../utils/reply");
const { resolvePermissionsForRoles, hasPermission, ROLE_PERMISSIONS, resolveEntityPermission } = require("../constants/permissions");
const { getWorkflowStepsSnapshot, resolveActionableStep, toMoneyString } = require("../utils/workflow");
const { consumeReservedLeave, releaseReservedLeave, formatDateOnly } = require("../services/leave");
const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require("../services/featureFlags");
const { confirmBudgetForEntity, releaseReservedBudgetIfAny } = require("../services/purchaseBudget");
const {
PROCUREMENT_ERP_TARGETS,
enqueuePurchaseCommitmentJournalExport,
loadPurchaseRequestErpContext,
} = require("../services/integrations/erpProcurement");
const {
createNotifications,
resolveRecipientsForAssignee,
@@ -97,6 +104,89 @@ function parseTaskId(taskId) {
return parsed.data;
}
async function handlePurchaseApprovalErpSync({
tx,
config,
request,
actorId,
purchaseRequestId,
loadContext = loadPurchaseRequestErpContext,
evaluateGate = evaluateFeatureFlag,
enqueueCommitment = enqueuePurchaseCommitmentJournalExport,
}) {
const context = await loadContext(tx, purchaseRequestId);
if (!context?.purchaseRequest) {
return { skipped: true, reason: 'purchase_request_not_found' };
}
const purchaseRequest = context.purchaseRequest;
const departmentId = purchaseRequest.department_id ?? request.employee.department_id ?? null;
const gate = await evaluateGate({
prisma: tx,
config,
flagKey: FEATURE_FLAG_KEYS.erpSync,
departmentId,
strictDepartment: true,
});
if (!gate.enabled) {
await tx.auditLog.create({
data: buildAuditLogData(request, {
actorId,
action: 'ERP_SYNC_SKIPPED_BY_FEATURE_FLAG',
targetType: 'purchase_request',
targetId: purchaseRequest.id,
extra: {
erpTarget: PROCUREMENT_ERP_TARGETS.commitmentJournal,
eventType: 'erp.journal.export',
purchaseRequestId: purchaseRequest.id,
purchaseOrderArchiveId: null,
gateReason: gate.reason,
featureFlag: FEATURE_FLAG_KEYS.erpSync,
departmentId,
},
}),
});
return { skipped: true, gate, purchaseRequest };
}
const enqueued = await enqueueCommitment(tx, config, {
purchaseRequest,
occurredAt: context.approvalOccurredAt ?? new Date(),
actorId,
requeueOnFailure: false,
});
await tx.auditLog.create({
data: buildAuditLogData(request, {
actorId,
action: 'ERP_SYNC_ENQUEUED',
targetType: 'purchase_request',
targetId: purchaseRequest.id,
extra: {
erpTarget: PROCUREMENT_ERP_TARGETS.commitmentJournal,
eventType: enqueued.eventType,
purchaseRequestId: purchaseRequest.id,
purchaseOrderArchiveId: null,
outboxId: enqueued.id,
duplicated: enqueued.duplicated,
requeued: enqueued.requeued,
gateReason: gate.reason,
featureFlag: FEATURE_FLAG_KEYS.erpSync,
departmentId,
},
}),
});
return {
skipped: false,
gate,
purchaseRequest,
enqueued,
};
}
function parseTaskListQuery(query) {
const parsed = TASK_LIST_QUERY_SCHEMA.safeParse(query ?? {});
if (!parsed.success) {
@@ -1372,6 +1462,21 @@ async function workflowRoutes(fastify) {
actorId: request.employee.id,
note: "approved_confirm_purchase_budget",
});
try {
await handlePurchaseApprovalErpSync({
tx,
config: fastify.config,
request,
actorId: request.employee.id,
purchaseRequestId: workflowInstance.entity_id,
});
} catch (error) {
fastify.log.error(
{ err: error, purchaseRequestId: workflowInstance.entity_id },
"Failed to enqueue purchase approval ERP events",
);
}
}
await tx.workflowHistory.create({
@@ -1905,5 +2010,6 @@ async function workflowRoutes(fastify) {
module.exports = workflowRoutes;
module.exports.__testables = {
handlePurchaseApprovalErpSync,
listAllowedActionsForTask,
};
server/src/services/integrations/dispatcher.js
+146 -41
查看文件
@@ -1,4 +1,8 @@
const { pushPaidReimbursementToErp, exportJournalToErp } = require('./erp');
const {
pushPaidReimbursementToErp,
exportJournalToErp,
exportProcurementToErp,
} = require('./erp');
const { sendInstantMessageByChannel } = require('../notifications/channels');
class IntegrationDispatchError extends Error {
@@ -22,71 +26,168 @@ function toDispatchError(result) {
});
}
function extractExternalId(response) {
if (!response) {
function extractResponseValue(response, key) {
if (!response || !key) {
return null;
}
if (typeof response.externalId === 'string' && response.externalId.trim()) {
return response.externalId.trim();
const directValue = response[key];
if (typeof directValue === 'string' && directValue.trim()) {
return directValue.trim();
}
if (
typeof response.body?.externalId === 'string' &&
response.body.externalId.trim()
) {
return response.body.externalId.trim();
const bodyValue = response.body?.[key];
if (typeof bodyValue === 'string' && bodyValue.trim()) {
return bodyValue.trim();
}
return null;
}
async function upsertErpBindingForPayload(prisma, payload, response) {
const entityType = String(payload?.entityType ?? '').trim();
const entityId = String(payload?.entityId ?? '').trim();
if (!prisma || !entityType || !entityId) {
function createBindingSpec(entityType, localId, externalId, externalCode) {
const normalizedEntityType = String(entityType ?? '').trim();
const normalizedLocalId = String(localId ?? '').trim();
if (!normalizedEntityType || !normalizedLocalId) {
return null;
}
return {
entityType: normalizedEntityType,
localId: normalizedLocalId,
externalId: typeof externalId === 'string' ? externalId.trim() || null : null,
externalCode: typeof externalCode === 'string' ? externalCode.trim() || null : null,
};
}
function resolveBindingSpecs(eventType, payload, response) {
if (eventType === 'erp.reimbursement.paid') {
return [
createBindingSpec(
'reimbursement_journal',
payload?.reimbursementId ?? payload?.entityId,
extractResponseValue(response, 'externalId'),
payload?.formNumber ?? payload?.documentNo ?? null,
),
].filter(Boolean);
}
if (eventType === 'erp.journal.export') {
const journalType = String(payload?.journalType ?? '').trim();
if (journalType === 'reimbursement_paid') {
return [
createBindingSpec(
'reimbursement_journal',
payload?.entityId,
extractResponseValue(response, 'externalId'),
payload?.documentNo ?? payload?.formNumber ?? null,
),
].filter(Boolean);
}
if (journalType === 'purchase_request_commitment') {
return [
createBindingSpec(
'purchase_request_commitment_journal',
payload?.references?.purchaseRequestId ?? payload?.entityId,
extractResponseValue(response, 'externalId'),
extractResponseValue(response, 'externalCode') ?? payload?.references?.requestNo ?? payload?.documentNo ?? null,
),
].filter(Boolean);
}
if (journalType === 'purchase_order_accrual') {
return [
createBindingSpec(
'purchase_order_accrual_journal',
payload?.references?.purchaseOrderArchiveId ?? payload?.entityId,
extractResponseValue(response, 'externalId'),
extractResponseValue(response, 'externalCode') ?? payload?.references?.poNo ?? payload?.documentNo ?? null,
),
].filter(Boolean);
}
}
if (eventType === 'erp.procurement.export') {
const archiveId = payload?.entityId;
const specs = [];
if (archiveId && (payload?.poNo || extractResponseValue(response, 'externalPoId') || extractResponseValue(response, 'externalPoNo'))) {
specs.push(
createBindingSpec(
'purchase_order_po',
archiveId,
extractResponseValue(response, 'externalPoId'),
extractResponseValue(response, 'externalPoNo') ?? payload?.poNo ?? null,
),
);
}
if (archiveId && (payload?.contractNo || extractResponseValue(response, 'externalContractId') || extractResponseValue(response, 'externalContractNo'))) {
specs.push(
createBindingSpec(
'purchase_order_contract',
archiveId,
extractResponseValue(response, 'externalContractId'),
extractResponseValue(response, 'externalContractNo') ?? payload?.contractNo ?? null,
),
);
}
return specs.filter(Boolean);
}
return [];
}
async function upsertErpBindings(prisma, eventType, payload, response) {
if (!prisma) {
return;
}
if (entityType !== 'reimbursement_form') {
const specs = resolveBindingSpecs(eventType, payload, response);
if (!specs.length) {
return;
}
await prisma.integrationBinding.upsert({
where: {
provider_entity_type_local_id: {
provider: 'erp',
entity_type: 'reimbursement_journal',
local_id: entityId,
for (const spec of specs) {
await prisma.integrationBinding.upsert({
where: {
provider_entity_type_local_id: {
provider: 'erp',
entity_type: spec.entityType,
local_id: spec.localId,
},
},
},
create: {
provider: 'erp',
entity_type: 'reimbursement_journal',
local_id: entityId,
external_id: extractExternalId(response),
external_code: payload?.documentNo ?? payload?.formNumber ?? null,
last_synced_at: new Date(),
},
update: {
external_id: extractExternalId(response),
external_code: payload?.documentNo ?? payload?.formNumber ?? null,
last_synced_at: new Date(),
},
});
create: {
provider: 'erp',
entity_type: spec.entityType,
local_id: spec.localId,
external_id: spec.externalId,
external_code: spec.externalCode,
last_synced_at: new Date(),
},
update: {
external_id: spec.externalId,
external_code: spec.externalCode,
last_synced_at: new Date(),
},
});
}
}
async function dispatchErpEvent({ prisma, config, eventType, payload }) {
const result =
eventType === 'erp.journal.export'
? await exportJournalToErp(config, payload)
: await pushPaidReimbursementToErp(config, payload);
eventType === 'erp.procurement.export'
? await exportProcurementToErp(config, payload)
: eventType === 'erp.journal.export'
? await exportJournalToErp(config, payload)
: await pushPaidReimbursementToErp(config, payload);
if (!result.ok) {
throw toDispatchError(result);
}
await upsertErpBindingForPayload(prisma, payload, result.response ?? null);
await upsertErpBindings(prisma, eventType, payload, result.response ?? null);
return {
channel: 'erp',
@@ -96,7 +197,11 @@ async function dispatchErpEvent({ prisma, config, eventType, payload }) {
}
async function dispatchOutboxEvent({ prisma, config, eventType, payload }) {
if (eventType === 'erp.reimbursement.paid' || eventType === 'erp.journal.export') {
if (
eventType === 'erp.reimbursement.paid' ||
eventType === 'erp.journal.export' ||
eventType === 'erp.procurement.export'
) {
return dispatchErpEvent({ prisma, config, eventType, payload });
}
server/src/services/integrations/erp.js
+64 -1
查看文件
@@ -34,6 +34,13 @@ function resolveJournalExportEndpoint(config) {
);
}
function resolveProcurementExportEndpoint(config) {
return (
String(config.erpProcurementExportEndpoint ?? '').trim() ||
String(config.erpSyncEndpoint ?? '').trim()
);
}
async function parseResponseBody(response) {
const contentType = response.headers.get('content-type') || '';
if (contentType.includes('application/json')) {
@@ -192,12 +199,14 @@ async function exportJournalToErp(config, payload) {
const provider = normalizeProvider(config);
if (provider === 'mock') {
const identity = payload?.documentNo ?? payload?.formNumber ?? payload?.entityId ?? payload?.reimbursementId ?? 'UNKNOWN';
return {
ok: true,
code: 'mock_synced',
message: 'mock ERP synced',
response: {
externalId: `ERP-${payload?.documentNo ?? payload?.formNumber ?? payload?.entityId ?? payload?.reimbursementId ?? 'UNKNOWN'}`,
externalId: `ERP-${identity}`,
externalCode: `ERP-${identity}`,
},
};
}
@@ -227,6 +236,59 @@ async function exportJournalToErp(config, payload) {
};
}
async function exportProcurementToErp(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?.poNo ?? payload?.contractNo ?? payload?.entityId ?? 'UNKNOWN';
return {
ok: true,
code: 'mock_synced',
message: 'mock ERP procurement synced',
response: {
externalId: `ERP-PROC-${identity}`,
externalCode: `ERP-PROC-${identity}`,
externalPoId: payload?.poNo ? `ERP-PO-${payload.poNo}` : null,
externalPoNo: payload?.poNo ? `ERP-PO-${payload.poNo}` : null,
externalContractId: payload?.contractNo ? `ERP-CT-${payload.contractNo}` : null,
externalContractNo: payload?.contractNo ? `ERP-CT-${payload.contractNo}` : null,
},
};
}
if (provider === 'http') {
return postJsonToErp(config, {
endpoint: resolveProcurementExportEndpoint(config),
timeoutMs: coerceTimeout(config.erpProcurementTimeoutMs, config.erpSyncTimeoutMs),
payload,
successCode: 'http_synced',
successMessage: 'ERP procurement synced',
missingEndpointCode: 'erp_procurement_endpoint_missing',
missingEndpointMessage: 'ERP procurement endpoint missing',
http4xxCode: 'erp_procurement_http_4xx',
http5xxCode: 'erp_procurement_http_5xx',
timeoutCode: 'erp_procurement_timeout',
networkCode: 'erp_procurement_network_error',
messagePrefix: 'ERP procurement sync',
});
}
return {
ok: false,
code: 'erp_sync_provider_invalid',
message: `Unsupported ERP sync provider: ${provider}`,
retryable: false,
};
}
function buildErpPayload(input) {
return buildReimbursementJournalPayload(input);
}
@@ -238,6 +300,7 @@ async function pushPaidReimbursementToErp(config, payload) {
module.exports = {
buildErpPayload,
exportJournalToErp,
exportProcurementToErp,
normalizeProvider,
pullMasterDataFromErp,
pushPaidReimbursementToErp,
server/src/services/integrations/erpProcurement.js
+704
查看文件
@@ -0,0 +1,704 @@
const crypto = require('node:crypto');
const { enqueueIntegrationEvent } = require('./outbox');
const PROCUREMENT_TRACK_STATUSES = Object.freeze({
notTriggered: 'not_triggered',
skipped: 'skipped',
});
const PROCUREMENT_ERP_TARGETS = Object.freeze({
commitmentJournal: 'commitment_journal',
accrualJournal: 'accrual_journal',
archiveSummary: 'archive_summary',
});
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 normalizeMoney(value) {
if (value == null) {
return '0.00';
}
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed || '0.00';
}
if (typeof value?.toString === 'function') {
return value.toString();
}
return '0.00';
}
function resolveCurrencyCode(purchaseRequest) {
return normalizeString(purchaseRequest?.currency_code) ?? 'CNY';
}
function resolveLegalEntityCode(purchaseRequest) {
return normalizeString(purchaseRequest?.legal_entity?.code);
}
function resolveCostCenterCode(purchaseRequest) {
return normalizeString(purchaseRequest?.cost_center?.code);
}
function resolveSupplierCode(purchaseRequest) {
return normalizeString(purchaseRequest?.supplier?.code);
}
function resolveBudgetPoolCode(purchaseRequest) {
return normalizeString(purchaseRequest?.budget_pool?.code);
}
function resolveDocumentCategoryCode(purchaseRequest) {
return normalizeString(purchaseRequest?.document_category?.code);
}
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 hashProcurementPayload(value) {
return crypto.createHash('sha256').update(stableStringifyJson(value)).digest('hex');
}
function buildPurchaseCommitmentJournalPayload(input) {
const purchaseRequest = input?.purchaseRequest ?? {};
return {
journalType: 'purchase_request_commitment',
sourceEventName: 'purchase_request.approved',
entityType: 'purchase_request',
entityId: purchaseRequest.id ?? null,
documentNo: purchaseRequest.request_no ?? null,
currencyCode: resolveCurrencyCode(purchaseRequest),
totalAmount: normalizeMoney(purchaseRequest.total_amount),
occurredAt:
toIsoString(input?.occurredAt) ??
toIsoString(purchaseRequest.updated_at) ??
toIsoString(purchaseRequest.created_at),
dimensions: {
departmentId: purchaseRequest.department_id ?? null,
applicantEmployeeId: purchaseRequest.applicant_id ?? null,
submitterEmployeeId: purchaseRequest.submitter_id ?? null,
legalEntityCode: resolveLegalEntityCode(purchaseRequest),
costCenterCode: resolveCostCenterCode(purchaseRequest),
supplierCode: resolveSupplierCode(purchaseRequest),
budgetPoolCode: resolveBudgetPoolCode(purchaseRequest),
},
references: {
purchaseRequestId: purchaseRequest.id ?? null,
documentCategoryCode: resolveDocumentCategoryCode(purchaseRequest),
requestNo: purchaseRequest.request_no ?? null,
},
};
}
function buildPurchaseAccrualJournalPayload(input) {
const purchaseRequest = input?.purchaseRequest ?? {};
const archive = input?.archive ?? purchaseRequest.order_archive ?? {};
return {
journalType: 'purchase_order_accrual',
sourceEventName: 'purchase_order.archived',
entityType: 'purchase_order_archive',
entityId: archive.id ?? null,
documentNo: archive.po_no ?? null,
currencyCode: resolveCurrencyCode(purchaseRequest),
totalAmount: normalizeMoney(purchaseRequest.total_amount),
occurredAt:
toIsoString(input?.occurredAt) ??
toIsoString(archive.archived_at) ??
toIsoString(purchaseRequest.updated_at) ??
toIsoString(purchaseRequest.created_at),
dimensions: {
departmentId: purchaseRequest.department_id ?? null,
applicantEmployeeId: purchaseRequest.applicant_id ?? null,
submitterEmployeeId: purchaseRequest.submitter_id ?? null,
legalEntityCode: resolveLegalEntityCode(purchaseRequest),
costCenterCode: resolveCostCenterCode(purchaseRequest),
supplierCode: resolveSupplierCode(purchaseRequest),
budgetPoolCode: resolveBudgetPoolCode(purchaseRequest),
},
references: {
purchaseRequestId: purchaseRequest.id ?? null,
purchaseOrderArchiveId: archive.id ?? null,
requestNo: purchaseRequest.request_no ?? null,
poNo: archive.po_no ?? null,
contractNo: archive.contract_no ?? null,
},
};
}
function buildPurchaseArchiveSummaryPayload(input) {
const purchaseRequest = input?.purchaseRequest ?? {};
const archive = input?.archive ?? purchaseRequest.order_archive ?? {};
const archivedFiles = Array.isArray(archive.archived_files)
? archive.archived_files.map((file) => ({
storagePath: file?.storagePath ?? null,
originalName: file?.originalName ?? null,
mimeType: file?.mimeType ?? null,
fileSize: Number(file?.fileSize ?? 0),
}))
: [];
return {
exportType: 'purchase_order_archive',
sourceEventName: 'purchase_order.archived',
entityType: 'purchase_order_archive',
entityId: archive.id ?? null,
purchaseRequestId: purchaseRequest.id ?? null,
requestNo: purchaseRequest.request_no ?? null,
poNo: archive.po_no ?? null,
contractNo: archive.contract_no ?? null,
currencyCode: resolveCurrencyCode(purchaseRequest),
totalAmount: normalizeMoney(purchaseRequest.total_amount),
archivedAt:
toIsoString(input?.archivedAt) ??
toIsoString(archive.archived_at) ??
toIsoString(purchaseRequest.updated_at) ??
toIsoString(purchaseRequest.created_at),
supplier: purchaseRequest.supplier
? {
id: purchaseRequest.supplier.id,
code: purchaseRequest.supplier.code ?? null,
name: purchaseRequest.supplier.name ?? null,
}
: {
id: purchaseRequest.supplier_id ?? null,
code: null,
name: null,
},
dimensions: {
legalEntityCode: resolveLegalEntityCode(purchaseRequest),
costCenterCode: resolveCostCenterCode(purchaseRequest),
departmentId: purchaseRequest.department_id ?? null,
},
files: archivedFiles,
};
}
function buildPurchaseCommitmentJournalDedupeKey(purchaseRequestId) {
return `purchase_request:${purchaseRequestId}:commitment`;
}
function buildPurchaseAccrualJournalDedupeKey(archiveId) {
return `purchase_order_archive:${archiveId}:accrual`;
}
function buildPurchaseArchiveSummaryDedupeKey(archiveId) {
return `purchase_order_archive:${archiveId}:summary`;
}
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 enqueuePurchaseCommitmentJournalExport(prisma, config, input) {
const payload =
input?.payload ??
buildPurchaseCommitmentJournalPayload({
purchaseRequest: input?.purchaseRequest,
occurredAt: input?.occurredAt,
});
const enqueued = await enqueueIntegrationEvent(prisma, {
eventType: 'erp.journal.export',
payload,
dedupeKey: buildPurchaseCommitmentJournalDedupeKey(payload.entityId),
maxAttempts: config.integrationOutboxMaxAttempts,
actorId: input?.actorId ?? null,
requeueOnFailure: input?.requeueOnFailure === true,
});
return {
...enqueued,
eventType: 'erp.journal.export',
erpTarget: PROCUREMENT_ERP_TARGETS.commitmentJournal,
payload,
};
}
async function enqueuePurchaseAccrualJournalExport(prisma, config, input) {
const payload =
input?.payload ??
buildPurchaseAccrualJournalPayload({
purchaseRequest: input?.purchaseRequest,
archive: input?.archive,
occurredAt: input?.occurredAt,
});
const enqueued = await enqueueIntegrationEvent(prisma, {
eventType: 'erp.journal.export',
payload,
dedupeKey: buildPurchaseAccrualJournalDedupeKey(payload.entityId),
maxAttempts: config.integrationOutboxMaxAttempts,
actorId: input?.actorId ?? null,
requeueOnFailure: input?.requeueOnFailure === true,
});
return {
...enqueued,
eventType: 'erp.journal.export',
erpTarget: PROCUREMENT_ERP_TARGETS.accrualJournal,
payload,
};
}
async function enqueuePurchaseArchiveSummaryExport(prisma, config, input) {
const payload =
input?.payload ??
buildPurchaseArchiveSummaryPayload({
purchaseRequest: input?.purchaseRequest,
archive: input?.archive,
archivedAt: input?.archivedAt,
});
const enqueued = await enqueueIntegrationEvent(prisma, {
eventType: 'erp.procurement.export',
payload,
dedupeKey: buildPurchaseArchiveSummaryDedupeKey(payload.entityId),
maxAttempts: config.integrationOutboxMaxAttempts,
actorId: input?.actorId ?? null,
requeueOnFailure: input?.requeueOnFailure === true,
resetOnPayloadChange: true,
});
return {
...enqueued,
eventType: 'erp.procurement.export',
erpTarget: PROCUREMENT_ERP_TARGETS.archiveSummary,
payload,
};
}
async function loadPurchaseRequestErpContext(prisma, purchaseRequestId) {
const purchaseRequest = await prisma.purchaseRequest.findUnique({
where: { id: purchaseRequestId },
select: {
id: true,
request_no: true,
applicant_id: true,
submitter_id: true,
department_id: true,
legal_entity_id: true,
cost_center_id: true,
supplier_id: true,
document_category_id: true,
budget_pool_id: true,
status: true,
currency_code: true,
total_amount: true,
created_at: true,
updated_at: true,
applicant: {
select: {
id: true,
employee_no: true,
name: true,
},
},
submitter: {
select: {
id: true,
employee_no: true,
name: true,
},
},
legal_entity: {
select: {
id: true,
code: true,
name: true,
},
},
cost_center: {
select: {
id: true,
code: true,
name: true,
},
},
supplier: {
select: {
id: true,
code: true,
name: true,
},
},
document_category: {
select: {
id: true,
code: true,
name: true,
},
},
budget_pool: {
select: {
id: true,
code: true,
name: true,
},
},
order_archive: {
select: {
id: true,
po_no: true,
contract_no: true,
archived_files: true,
archived_at: true,
archived_by: true,
},
},
workflow_instances: {
where: { entity_type: 'purchase_request' },
orderBy: { created_at: 'desc' },
take: 1,
select: { id: true },
},
},
});
if (!purchaseRequest) {
return null;
}
const workflowInstanceId = purchaseRequest.workflow_instances?.[0]?.id ?? null;
const approvalHistory = workflowInstanceId
? await prisma.workflowHistory.findFirst({
where: {
instance_id: workflowInstanceId,
action: 'approve',
},
orderBy: { created_at: 'desc' },
select: { created_at: true },
})
: null;
return {
purchaseRequest,
approvalOccurredAt:
approvalHistory?.created_at ?? purchaseRequest.updated_at ?? purchaseRequest.created_at ?? null,
};
}
function hasAnyBinding(binding) {
if (!binding) {
return false;
}
return Boolean(binding.externalId || binding.externalCode || binding.lastSyncedAt);
}
function resolveTrackStageStatus({ eligible, outbox, binding, bindings, gateEnabled }) {
if (outbox?.status) {
return outbox.status;
}
if (binding && hasAnyBinding(binding)) {
return 'succeeded';
}
if (bindings && Object.values(bindings).some((item) => hasAnyBinding(item))) {
return 'succeeded';
}
if (!eligible) {
return PROCUREMENT_TRACK_STATUSES.notTriggered;
}
if (!gateEnabled) {
return PROCUREMENT_TRACK_STATUSES.skipped;
}
return PROCUREMENT_TRACK_STATUSES.notTriggered;
}
function buildPurchaseTrackSummary(context) {
const purchaseRequest = context?.purchaseRequest ?? {};
const archive = purchaseRequest.order_archive ?? null;
return {
id: purchaseRequest.id,
requestNo: purchaseRequest.request_no ?? null,
status: purchaseRequest.status,
totalAmount: normalizeMoney(purchaseRequest.total_amount),
supplier: purchaseRequest.supplier
? {
id: purchaseRequest.supplier.id,
code: purchaseRequest.supplier.code ?? null,
name: purchaseRequest.supplier.name ?? null,
}
: null,
archiveOrder: archive
? {
id: archive.id,
poNo: archive.po_no,
contractNo: archive.contract_no ?? null,
archivedAt: toIsoString(archive.archived_at),
}
: null,
};
}
async function buildPurchaseRequestErpTrack(prisma, config, purchaseRequestId, options = {}) {
const context = options.context ?? (await loadPurchaseRequestErpContext(prisma, purchaseRequestId));
if (!context) {
return null;
}
const purchaseRequest = context.purchaseRequest;
const archive = purchaseRequest.order_archive ?? null;
const canManage = Boolean(options.canManage);
const gateEnabled = Boolean(options.gate?.effectiveEnabled ?? options.gate?.enabled ?? true);
const [
commitmentOutboxRow,
accrualOutboxRow,
summaryOutboxRow,
commitmentBindingRow,
accrualBindingRow,
poBindingRow,
contractBindingRow,
] = await Promise.all([
findOutboxByKey(
prisma,
'erp.journal.export',
buildPurchaseCommitmentJournalDedupeKey(purchaseRequest.id),
),
archive
? findOutboxByKey(
prisma,
'erp.journal.export',
buildPurchaseAccrualJournalDedupeKey(archive.id),
)
: Promise.resolve(null),
archive
? findOutboxByKey(
prisma,
'erp.procurement.export',
buildPurchaseArchiveSummaryDedupeKey(archive.id),
)
: Promise.resolve(null),
findBindingByLocalId(prisma, 'purchase_request_commitment_journal', purchaseRequest.id),
archive
? findBindingByLocalId(prisma, 'purchase_order_accrual_journal', archive.id)
: Promise.resolve(null),
archive ? findBindingByLocalId(prisma, 'purchase_order_po', archive.id) : Promise.resolve(null),
archive
? findBindingByLocalId(prisma, 'purchase_order_contract', archive.id)
: Promise.resolve(null),
]);
const commitmentPreview = buildPurchaseCommitmentJournalPayload({
purchaseRequest,
occurredAt: context.approvalOccurredAt,
});
const accrualPreview = archive
? buildPurchaseAccrualJournalPayload({
purchaseRequest,
archive,
occurredAt: archive.archived_at,
})
: {};
const summaryPreview = archive
? buildPurchaseArchiveSummaryPayload({
purchaseRequest,
archive,
archivedAt: archive.archived_at,
})
: {};
const commitmentBinding = redactBinding(mapIntegrationBinding(commitmentBindingRow), canManage);
const accrualBinding = redactBinding(mapIntegrationBinding(accrualBindingRow), canManage);
const poBinding = redactBinding(mapIntegrationBinding(poBindingRow), canManage);
const contractBinding = redactBinding(mapIntegrationBinding(contractBindingRow), canManage);
const commitmentOutbox = mapIntegrationOutboxState(commitmentOutboxRow);
const accrualOutbox = mapIntegrationOutboxState(accrualOutboxRow);
const summaryOutbox = mapIntegrationOutboxState(summaryOutboxRow);
const commitmentEligible = purchaseRequest.status === 'approved';
const archiveEligible = Boolean(archive);
return {
purchaseRequest: buildPurchaseTrackSummary(context),
stages: {
commitmentJournal: {
eligible: commitmentEligible,
reason: commitmentEligible ? null : 'purchase_request_not_approved',
status: resolveTrackStageStatus({
eligible: commitmentEligible,
outbox: commitmentOutbox,
binding: commitmentBinding,
gateEnabled,
}),
preview: commitmentPreview,
outbox: commitmentOutbox,
binding: commitmentBinding,
},
accrualJournal: {
eligible: archiveEligible,
reason: archiveEligible ? null : 'purchase_order_archive_missing',
status: resolveTrackStageStatus({
eligible: archiveEligible,
outbox: accrualOutbox,
binding: accrualBinding,
gateEnabled,
}),
preview: accrualPreview,
outbox: accrualOutbox,
binding: accrualBinding,
},
archiveSummary: {
eligible: archiveEligible,
reason: archiveEligible ? null : 'purchase_order_archive_missing',
status: resolveTrackStageStatus({
eligible: archiveEligible,
outbox: summaryOutbox,
bindings: {
po: poBinding,
contract: contractBinding,
},
gateEnabled,
}),
preview: summaryPreview,
outbox: summaryOutbox,
bindings: {
po: poBinding,
contract: contractBinding,
},
},
},
permissions: {
canRetry: Boolean(canManage && gateEnabled),
},
};
}
module.exports = {
PROCUREMENT_ERP_TARGETS,
PROCUREMENT_TRACK_STATUSES,
buildPurchaseAccrualJournalDedupeKey,
buildPurchaseAccrualJournalPayload,
buildPurchaseArchiveSummaryDedupeKey,
buildPurchaseArchiveSummaryPayload,
buildPurchaseCommitmentJournalDedupeKey,
buildPurchaseCommitmentJournalPayload,
buildPurchaseRequestErpTrack,
enqueuePurchaseAccrualJournalExport,
enqueuePurchaseArchiveSummaryExport,
enqueuePurchaseCommitmentJournalExport,
hashProcurementPayload,
loadPurchaseRequestErpContext,
stableStringifyJson,
};
server/src/services/integrations/outbox.js
+59 -26
查看文件
@@ -35,6 +35,55 @@ function formatDispatchError(error) {
return parts.join(' | ');
}
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 hasIntegrationPayloadChanged(currentPayload, nextPayload) {
return JSON.stringify(stableNormalizeJson(currentPayload)) !== JSON.stringify(stableNormalizeJson(nextPayload));
}
async function resetExistingOutbox(prisma, existed, input) {
const reset = await prisma.integrationOutbox.update({
where: { id: existed.id },
data: {
status: OUTBOX_STATUSES.pending,
payload: input.payload,
attempts: 0,
max_attempts: input.maxAttempts,
last_error: null,
next_retry_at: null,
processed_at: null,
dead_letter_at: null,
updated_by: input.actorId ?? null,
},
select: {
id: true,
status: true,
},
});
return {
id: reset.id,
status: reset.status,
duplicated: false,
requeued: true,
};
}
async function enqueueIntegrationEvent(prisma, input) {
const dedupeKey = input.dedupeKey ?? null;
@@ -47,40 +96,23 @@ async function enqueueIntegrationEvent(prisma, input) {
dedupe_key: dedupeKey,
},
},
select: { id: true, status: true },
select: { id: true, status: true, payload: true },
})
: null;
if (existed) {
if (
input.resetOnPayloadChange &&
hasIntegrationPayloadChanged(existed.payload, input.payload)
) {
return resetExistingOutbox(prisma, existed, input);
}
if (
input.requeueOnFailure &&
[OUTBOX_STATUSES.failed, OUTBOX_STATUSES.dead_letter].includes(existed.status)
) {
const reset = await prisma.integrationOutbox.update({
where: { id: existed.id },
data: {
status: OUTBOX_STATUSES.pending,
payload: input.payload,
attempts: 0,
max_attempts: input.maxAttempts,
last_error: null,
next_retry_at: null,
processed_at: null,
dead_letter_at: null,
updated_by: input.actorId ?? null,
},
select: {
id: true,
status: true,
},
});
return {
id: reset.id,
status: reset.status,
duplicated: false,
requeued: true,
};
return resetExistingOutbox(prisma, existed, input);
}
return {
@@ -188,6 +220,7 @@ module.exports = {
OUTBOX_STATUSES,
enqueueIntegrationEvent,
fetchPendingOutboxEvents,
hasIntegrationPayloadChanged,
markOutboxProcessing,
markOutboxSucceeded,
markOutboxFailed,