feat(P19ERP1-3,P19ERP1-4): 实现ERP凭证导出与管理入口

这个提交包含在:
2026-03-08 16:29:05 +08:00
父节点 a1b1d7cfc9
当前提交 855bcdffeb
共修改 14 个文件,包含 1361 行新增和 145 行删除
+5
查看文件
@@ -38,6 +38,11 @@ async function runIntegrationOutboxOnce(prisma, config, logger) {
retryable,
outboxId: item.id,
eventType: item.event_type,
reimbursementId:
item.payload?.reimbursementId ??
(item.payload?.entityType === 'reimbursement_form' ? item.payload?.entityId ?? null : null),
entityType: item.payload?.entityType ?? null,
entityId: item.payload?.entityId ?? null,
channelId: item.payload?.channelId ?? null,
channelType: item.payload?.channelType ?? null,
notificationType: item.payload?.notificationType ?? null,
+492 -115
查看文件
@@ -1,178 +1,550 @@
const { z } = require("zod");
const { z } = require('zod');
const { AppError } = require("../utils/AppError");
const { sendData } = require("../utils/reply");
const { buildAuditLogData } = require("../utils/audit");
const { AppError } = require('../utils/AppError');
const { sendData } = require('../utils/reply');
const { buildAuditLogData } = require('../utils/audit');
const {
buildDataEnvelopeSchema,
buildErrorEnvelopeSchema,
} = require("../utils/openapi");
const { enqueueIntegrationEvent } = require("../services/integrations/outbox");
const { buildErpPayload } = require("../services/integrations/erp");
const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require("../services/featureFlags");
const { runIntegrationOutboxOnce } = require("../jobs/integrationOutboxJob");
} = 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 {
buildReimbursementJournalPreview,
enqueueReimbursementJournalExport,
} = require('../services/integrations/erpJournal');
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 ERP_SYNC_RESPONSE_SCHEMA = {
type: "object",
required: ["outboxId", "status", "duplicated"],
type: 'object',
required: ['outboxId', 'status', 'duplicated'],
properties: {
outboxId: { type: "string", format: "uuid" },
status: { type: "string" },
duplicated: { type: "boolean" },
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'
);
}
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' },
});
}
}
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.post(
"/integrations/erp/sync/:reimbursementId",
fastify.get(
'/integrations/erp/master-data/status',
{
preHandler: fastify.requirePermission("integration.manage"),
preHandler: fastify.requirePermission('integration.manage'),
schema: {
tags: ["integrations"],
summary: "手动触发 ERP 同步",
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"],
type: 'object',
required: ['reimbursementId'],
properties: {
reimbursementId: { type: "string", format: "uuid" },
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",
type: 'object',
properties: {
runNow: {
anyOf: [{ type: "string", enum: ["1", "true"] }, { type: "boolean" }],
anyOf: [
{ type: 'string', enum: ['0', '1', 'false', 'true'] },
{ type: 'boolean' },
],
},
},
additionalProperties: true,
},
response: {
200: buildDataEnvelopeSchema(ERP_SYNC_RESPONSE_SCHEMA, "ERP 同步任务已入队"),
200: buildDataEnvelopeSchema(ERP_JOURNAL_EXPORT_RESPONSE_SCHEMA, 'ERP 导出任务已入队'),
400: buildErrorEnvelopeSchema({
description: "参数校验失败",
codeExample: "VALID_001",
messageExample: "参数校验失败",
description: '参数校验失败',
codeExample: 'VALID_001',
messageExample: '参数校验失败',
}),
404: buildErrorEnvelopeSchema({
description: "报销单不存在",
codeExample: "SYS_404",
messageExample: "Not Found",
description: '报销单不存在',
codeExample: 'SYS_404',
messageExample: 'Not Found',
}),
409: buildErrorEnvelopeSchema({
description: "状态不满足或开关阻断",
codeExample: "VALID_001",
messageExample: "参数校验失败",
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.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",
integration: 'feature_flag_blocked',
featureFlag: 'erp_sync',
reason: 'department_not_allowed',
},
}),
},
},
},
async (request, reply) => {
const parsedId = z.string().uuid().safeParse(request.params?.reimbursementId);
const parsedId = UUID_SCHEMA.safeParse(request.params?.reimbursementId);
if (!parsedId.success) {
throw new AppError({
code: "VALID_001",
message: "参数校验失败",
code: 'VALID_001',
message: '参数校验失败',
statusCode: 400,
details: { reimbursementId: "invalid_uuid" },
details: { reimbursementId: 'invalid_uuid' },
});
}
const reimbursementId = parsedId.data;
const reimbursement = await fastify.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,
},
});
if (!reimbursement) {
throw new AppError({
code: "SYS_404",
message: "Not Found",
statusCode: 404,
});
}
if (reimbursement.status !== "paid") {
throw new AppError({
code: "VALID_001",
message: "参数校验失败",
statusCode: 409,
details: { reimbursement: "erp_sync_requires_paid" },
});
}
const erpGate = await evaluateFeatureFlag({
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,
flagKey: FEATURE_FLAG_KEYS.erpSync,
departmentId: reimbursement.department_id ?? request.employee.department_id ?? null,
strictDepartment: true,
});
if (!erpGate.enabled) {
throw new AppError({
code: "VALID_001",
message: "参数校验失败",
statusCode: 409,
details: {
integration: "feature_flag_blocked",
featureFlag: FEATURE_FLAG_KEYS.erpSync,
reason: erpGate.reason,
},
});
}
const payload = buildErpPayload({ reimbursement });
const enqueued = await enqueueIntegrationEvent(fastify.prisma, {
eventType: "erp.reimbursement.paid",
payload,
dedupeKey: reimbursement.id,
maxAttempts: fastify.config.integrationOutboxMaxAttempts,
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: reimbursementId,
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,
departmentId: reimbursement.department_id ?? null,
featureFlagReason: erpGate.reason,
featureFlagReason: gate.effectiveReason,
},
}),
});
const runNow =
request.query?.runNow === "1" ||
request.query?.runNow === 1 ||
request.query?.runNow === true ||
request.query?.runNow === "true";
if (runNow) {
await runIntegrationOutboxOnce(fastify.prisma, fastify.config, fastify.log);
}
return sendData(reply, request, {
outboxId: enqueued.id,
status: enqueued.status,
@@ -182,4 +554,9 @@ async function integrationsRoutes(fastify) {
);
}
integrationsRoutes.__testables = {
normalizeResourcesInput,
parseRunNow,
};
module.exports = integrationsRoutes;
+21 -7
查看文件
@@ -11,8 +11,7 @@ const {
const { enqueueImNotifications } = require("../services/notifications/imEnqueue");
const { NOTIFICATION_TYPES } = require("../constants/notificationTypes");
const { sendNotificationEmailSafe } = require("../services/email");
const { enqueueIntegrationEvent } = require("../services/integrations/outbox");
const { buildErpPayload } = require("../services/integrations/erp");
const { enqueueReimbursementJournalExport } = require("../services/integrations/erpJournal");
const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require("../services/featureFlags");
const { resolveApplicableBudgetPoolByContext } = require("../services/budgetPool");
const { issueDocumentNumber } = require("../services/documentSequence");
@@ -1771,6 +1770,24 @@ async function reimbursementRoutes(fastify) {
payment_voucher_no: true,
payment_date: true,
paid_at: true,
owner: {
select: {
legal_entity: {
select: {
code: true,
},
},
},
},
submitter: {
select: {
legal_entity: {
select: {
code: true,
},
},
},
},
},
});
@@ -1786,11 +1803,8 @@ async function reimbursementRoutes(fastify) {
});
if (erpGate.enabled) {
await enqueueIntegrationEvent(tx, {
eventType: "erp.reimbursement.paid",
dedupeKey: refreshed.id,
payload: buildErpPayload({ reimbursement: refreshed }),
maxAttempts: fastify.config.integrationOutboxMaxAttempts,
await enqueueReimbursementJournalExport(tx, fastify.config, {
reimbursement: refreshed,
actorId: request.employee.id,
});
} else {
+76 -12
查看文件
@@ -1,4 +1,4 @@
const { pushPaidReimbursementToErp } = require('./erp');
const { pushPaidReimbursementToErp, exportJournalToErp } = require('./erp');
const { sendInstantMessageByChannel } = require('../notifications/channels');
class IntegrationDispatchError extends Error {
@@ -22,18 +22,82 @@ function toDispatchError(result) {
});
}
async function dispatchOutboxEvent({ prisma, config, eventType, payload }) {
if (eventType === 'erp.reimbursement.paid') {
const result = await pushPaidReimbursementToErp(config, payload);
if (!result.ok) {
throw toDispatchError(result);
}
function extractExternalId(response) {
if (!response) {
return null;
}
return {
channel: 'erp',
code: result.code,
response: result.response ?? null,
};
if (typeof response.externalId === 'string' && response.externalId.trim()) {
return response.externalId.trim();
}
if (
typeof response.body?.externalId === 'string' &&
response.body.externalId.trim()
) {
return response.body.externalId.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) {
return;
}
if (entityType !== 'reimbursement_form') {
return;
}
await prisma.integrationBinding.upsert({
where: {
provider_entity_type_local_id: {
provider: 'erp',
entity_type: 'reimbursement_journal',
local_id: entityId,
},
},
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(),
},
});
}
async function dispatchErpEvent({ prisma, config, eventType, payload }) {
const result =
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);
return {
channel: 'erp',
code: result.code,
response: result.response ?? null,
};
}
async function dispatchOutboxEvent({ prisma, config, eventType, payload }) {
if (eventType === 'erp.reimbursement.paid' || eventType === 'erp.journal.export') {
return dispatchErpEvent({ prisma, config, eventType, payload });
}
if (eventType === 'notification.im') {
@@ -0,0 +1,182 @@
const { enqueueIntegrationEvent } = require('./outbox');
const ERP_JOURNAL_EVENT_TYPES = Object.freeze(['erp.journal.export', 'erp.reimbursement.paid']);
const REIMBURSEMENT_JOURNAL_DEDUPE_PREFIX = 'reimbursement';
function toDateOnly(value) {
if (!value) {
return null;
}
try {
return new Date(value).toISOString().slice(0, 10);
} catch {
return null;
}
}
function toIsoString(value) {
if (!value) {
return null;
}
try {
return new Date(value).toISOString();
} catch {
return null;
}
}
function resolveLegalEntityCode(reimbursement) {
return (
reimbursement?.owner?.legal_entity?.code ??
reimbursement?.submitter?.legal_entity?.code ??
null
);
}
function buildReimbursementJournalPayload(input) {
const reimbursement = input?.reimbursement ?? {};
const occurredAt =
toIsoString(reimbursement.paid_at) ??
(reimbursement.payment_date ? `${toDateOnly(reimbursement.payment_date)}T00:00:00.000Z` : null);
const payload = {
journalType: 'reimbursement_paid',
sourceEventName: 'reimbursement.paid',
entityType: 'reimbursement_form',
entityId: reimbursement.id,
documentNo: reimbursement.form_number ?? null,
currencyCode: 'CNY',
totalAmount: reimbursement.total_amount?.toString?.() ?? null,
occurredAt,
paymentVoucherNo: reimbursement.payment_voucher_no ?? null,
paymentDate: toDateOnly(reimbursement.payment_date),
dimensions: {
departmentId: reimbursement.department_id ?? null,
ownerEmployeeId: reimbursement.owner_id ?? null,
submitterEmployeeId: reimbursement.submitter_id ?? null,
legalEntityCode: resolveLegalEntityCode(reimbursement),
costCenterCode: null,
},
reimbursementId: reimbursement.id,
formNumber: reimbursement.form_number ?? null,
};
return payload;
}
function buildReimbursementJournalDedupeKey(reimbursementId) {
return `${REIMBURSEMENT_JOURNAL_DEDUPE_PREFIX}:${reimbursementId}:paid`;
}
async function enqueueReimbursementJournalExport(prisma, config, input) {
const payload =
input?.payload ?? buildReimbursementJournalPayload({ reimbursement: input?.reimbursement });
const enqueued = await enqueueIntegrationEvent(prisma, {
eventType: 'erp.journal.export',
payload,
dedupeKey: buildReimbursementJournalDedupeKey(payload.entityId),
maxAttempts: config.integrationOutboxMaxAttempts,
actorId: input?.actorId ?? null,
requeueOnFailure: true,
});
return {
...enqueued,
eventType: 'erp.journal.export',
payload,
};
}
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 mapIntegrationOutboxState(record) {
if (!record) {
return null;
}
return {
id: record.id,
eventType: record.event_type,
status: record.status,
lastError: record.last_error ?? null,
processedAt: record.processed_at?.toISOString?.() ?? null,
updatedAt: record.updated_at?.toISOString?.() ?? null,
};
}
async function findLatestReimbursementJournalOutbox(prisma, reimbursementId) {
const current = await prisma.integrationOutbox.findFirst({
where: {
event_type: 'erp.journal.export',
dedupe_key: buildReimbursementJournalDedupeKey(reimbursementId),
},
orderBy: { updated_at: 'desc' },
});
if (current) {
return current;
}
return prisma.integrationOutbox.findFirst({
where: {
event_type: 'erp.reimbursement.paid',
dedupe_key: reimbursementId,
},
orderBy: { updated_at: 'desc' },
});
}
async function buildReimbursementJournalPreview(prisma, reimbursement) {
const [binding, outbox] = await Promise.all([
prisma.integrationBinding.findUnique({
where: {
provider_entity_type_local_id: {
provider: 'erp',
entity_type: 'reimbursement_journal',
local_id: reimbursement.id,
},
},
}),
findLatestReimbursementJournalOutbox(prisma, reimbursement.id),
]);
const journal = buildReimbursementJournalPayload({ reimbursement });
const exportState = {
externalId: binding?.external_id ?? null,
lastOutboxId: outbox?.id ?? null,
lastOutboxStatus: outbox?.status ?? null,
lastError: outbox?.last_error ?? null,
exportedAt: binding?.last_synced_at?.toISOString?.() ?? outbox?.processed_at?.toISOString?.() ?? null,
};
return {
journal,
exportState,
binding: mapIntegrationBinding(binding),
outbox: mapIntegrationOutboxState(outbox),
};
}
module.exports = {
ERP_JOURNAL_EVENT_TYPES,
buildReimbursementJournalDedupeKey,
buildReimbursementJournalPayload,
buildReimbursementJournalPreview,
enqueueReimbursementJournalExport,
findLatestReimbursementJournalOutbox,
};
+44 -11
查看文件
@@ -1,9 +1,9 @@
const OUTBOX_STATUSES = {
pending: "pending",
processing: "processing",
succeeded: "succeeded",
failed: "failed",
dead_letter: "dead_letter",
pending: 'pending',
processing: 'processing',
succeeded: 'succeeded',
failed: 'failed',
dead_letter: 'dead_letter',
};
function nextRetryAt(config, attempts) {
@@ -15,7 +15,7 @@ function nextRetryAt(config, attempts) {
function formatDispatchError(error) {
const code = error?.code ? String(error.code) : null;
const statusCode = Number.isFinite(error?.statusCode) ? Number(error.statusCode) : null;
const message = String(error?.message ?? error ?? "integration_outbox_error").trim();
const message = String(error?.message ?? error ?? 'integration_outbox_error').trim();
const parts = [];
if (code) {
@@ -29,10 +29,10 @@ function formatDispatchError(error) {
}
if (!parts.length) {
return "integration_outbox_error";
return 'integration_outbox_error';
}
return parts.join(" | ");
return parts.join(' | ');
}
async function enqueueIntegrationEvent(prisma, input) {
@@ -52,10 +52,42 @@ async function enqueueIntegrationEvent(prisma, input) {
: null;
if (existed) {
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 {
id: existed.id,
status: existed.status,
duplicated: true,
requeued: false,
};
}
@@ -79,6 +111,7 @@ async function enqueueIntegrationEvent(prisma, input) {
id: created.id,
status: created.status,
duplicated: false,
requeued: false,
};
}
@@ -93,7 +126,7 @@ async function fetchPendingOutboxEvents(prisma, options) {
},
OR: [{ next_retry_at: null }, { next_retry_at: { lte: now } }],
},
orderBy: [{ created_at: "asc" }],
orderBy: [{ created_at: 'asc' }],
take: batchSize,
});
}
@@ -128,9 +161,9 @@ async function markOutboxFailed(prisma, config, item, error, actorId, options =
);
const retryable =
typeof options.retryable === "boolean"
typeof options.retryable === 'boolean'
? options.retryable
: typeof error?.retryable === "boolean"
: typeof error?.retryable === 'boolean'
? error.retryable
: true;