feat(P19ERP1-1,P19ERP1-2): 新增ERP主数据同步模型与服务

这个提交包含在:
2026-03-08 16:28:56 +08:00
父节点 b45b8f2945
当前提交 a1b1d7cfc9
共修改 10 个文件,包含 1167 行新增和 94 行删除
@@ -0,0 +1,50 @@
-- CreateTable
CREATE TABLE "integration_bindings" (
"id" UUID NOT NULL,
"provider" VARCHAR(32) NOT NULL,
"entity_type" VARCHAR(32) NOT NULL,
"local_id" UUID NOT NULL,
"external_id" VARCHAR(128),
"external_code" VARCHAR(128),
"last_synced_at" TIMESTAMP(3),
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"created_by" UUID,
"updated_at" TIMESTAMP(3) NOT NULL,
"updated_by" UUID,
CONSTRAINT "integration_bindings_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "integration_sync_states" (
"id" UUID NOT NULL,
"provider" VARCHAR(32) NOT NULL,
"resource_type" VARCHAR(32) NOT NULL,
"status" VARCHAR(16) NOT NULL DEFAULT 'idle',
"last_result" JSONB,
"last_error" TEXT,
"last_started_at" TIMESTAMP(3),
"last_finished_at" TIMESTAMP(3),
"last_success_at" TIMESTAMP(3),
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"created_by" UUID,
"updated_at" TIMESTAMP(3) NOT NULL,
"updated_by" UUID,
CONSTRAINT "integration_sync_states_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE INDEX "idx_integration_bindings_provider_entity_external_code" ON "integration_bindings"("provider", "entity_type", "external_code");
-- CreateIndex
CREATE UNIQUE INDEX "integration_bindings_provider_entity_local_ux" ON "integration_bindings"("provider", "entity_type", "local_id");
-- CreateIndex
CREATE UNIQUE INDEX "integration_bindings_provider_entity_external_id_ux" ON "integration_bindings"("provider", "entity_type", "external_id");
-- CreateIndex
CREATE INDEX "idx_integration_sync_states_provider_status" ON "integration_sync_states"("provider", "status");
-- CreateIndex
CREATE UNIQUE INDEX "integration_sync_states_provider_resource_ux" ON "integration_sync_states"("provider", "resource_type");
+41
查看文件
@@ -1205,6 +1205,47 @@ model IntegrationOutbox {
@@map("integration_outbox")
}
model IntegrationBinding {
id String @id @default(uuid()) @db.Uuid
provider String @db.VarChar(32)
entity_type String @db.VarChar(32)
local_id String @db.Uuid
external_id String? @db.VarChar(128)
external_code String? @db.VarChar(128)
last_synced_at DateTime?
created_at DateTime @default(now())
created_by String? @db.Uuid
updated_at DateTime @updatedAt
updated_by String? @db.Uuid
@@unique([provider, entity_type, local_id], map: "integration_bindings_provider_entity_local_ux")
@@unique([provider, entity_type, external_id], map: "integration_bindings_provider_entity_external_id_ux")
@@index([provider, entity_type, external_code], map: "idx_integration_bindings_provider_entity_external_code")
@@map("integration_bindings")
}
model IntegrationSyncState {
id String @id @default(uuid()) @db.Uuid
provider String @db.VarChar(32)
resource_type String @db.VarChar(32)
status String @default("idle") @db.VarChar(16)
last_result Json?
last_error String?
last_started_at DateTime?
last_finished_at DateTime?
last_success_at DateTime?
created_at DateTime @default(now())
created_by String? @db.Uuid
updated_at DateTime @updatedAt
updated_by String? @db.Uuid
@@unique([provider, resource_type], map: "integration_sync_states_provider_resource_ux")
@@index([provider, status], map: "idx_integration_sync_states_provider_status")
@@map("integration_sync_states")
}
model TravelOrder {
id String @id @default(uuid()) @db.Uuid
source String @default("manual") @db.VarChar(32)
+6
查看文件
@@ -108,7 +108,13 @@ function getConfig() {
erpSyncFlagEnv: process.env.ERP_SYNC_ENABLED ?? "false",
erpSyncProvider: process.env.ERP_SYNC_PROVIDER ?? "mock",
erpSyncEndpoint: process.env.ERP_SYNC_ENDPOINT ?? "",
erpJournalExportEndpoint: process.env.ERP_JOURNAL_EXPORT_ENDPOINT ?? "",
erpMasterDataEndpoint: process.env.ERP_MASTER_DATA_ENDPOINT ?? "",
erpSyncTimeoutMs: Number(process.env.ERP_SYNC_TIMEOUT_MS ?? 5000),
erpMasterDataTimeoutMs: Number(
process.env.ERP_MASTER_DATA_TIMEOUT_MS ?? process.env.ERP_SYNC_TIMEOUT_MS ?? 5000,
),
erpMasterDataSyncCron: process.env.ERP_MASTER_DATA_SYNC_CRON ?? "",
erpSyncAuthToken: process.env.ERP_SYNC_AUTH_TOKEN ?? "",
travelSyncEnabled: process.env.TRAVEL_SYNC_ENABLED === "true",
imNotifierEnabled: process.env.IM_NOTIFIER_ENABLED === "true",
+91
查看文件
@@ -0,0 +1,91 @@
const cron = require('node-cron');
const { FEATURE_FLAG_KEYS, evaluateFeatureFlag } = require('../services/featureFlags');
const { syncErpMasterData } = require('../services/integrations/erpMasterData');
async function runErpMasterDataSyncOnce(prisma, config, logger) {
if (!config.erpMasterDataSyncCron) {
return {
skipped: true,
reason: 'cron_disabled',
items: [],
};
}
const gate = await evaluateFeatureFlag({
prisma,
config,
flagKey: FEATURE_FLAG_KEYS.erpSync,
departmentId: null,
strictDepartment: false,
});
if (!gate.enabled) {
logger.info(
{
featureFlag: FEATURE_FLAG_KEYS.erpSync,
reason: gate.reason,
},
'ERP master data sync skipped by ERP gate',
);
return {
skipped: true,
reason: gate.reason,
items: [],
};
}
return syncErpMasterData(prisma, config, { logger });
}
function startErpMasterDataSyncJob(fastify) {
if (!fastify.config.erpMasterDataSyncCron) {
fastify.log.info('ERP master data sync cron disabled');
return null;
}
let running = false;
const task = cron.schedule(fastify.config.erpMasterDataSyncCron, async () => {
if (running) {
return;
}
running = true;
try {
await runErpMasterDataSyncOnce(fastify.prisma, fastify.config, fastify.log);
fastify.metrics?.jobRunsTotal?.inc?.({
job: 'erp.master-data.sync',
driver: 'cron',
outcome: 'success',
});
} catch (error) {
fastify.log.error({ err: error }, 'ERP master data sync job failed');
fastify.metrics?.jobRunsTotal?.inc?.({
job: 'erp.master-data.sync',
driver: 'cron',
outcome: 'failure',
});
} finally {
running = false;
}
});
fastify.addHook('onClose', async () => {
task.stop();
task.destroy();
});
fastify.log.info(
{
cron: fastify.config.erpMasterDataSyncCron,
},
'ERP master data sync job scheduled',
);
return task;
}
module.exports = {
runErpMasterDataSyncOnce,
startErpMasterDataSyncJob,
};
+2
查看文件
@@ -2,6 +2,7 @@ const { startDelegationRevokeJob } = require("./delegationRevokeJob");
const { startInvoiceArchiveJob } = require("./invoiceArchiveJob");
const { startOcrJob } = require("./ocrJob");
const { startIntegrationOutboxWorker } = require("./integrationOutboxWorker");
const { startErpMasterDataSyncJob } = require("./erpMasterDataSyncJob");
const { startPgBossJobs } = require("./pgBossJobs");
async function startJobs(fastify) {
@@ -19,6 +20,7 @@ async function startJobs(fastify) {
startOcrJob(fastify);
startInvoiceArchiveJob(fastify);
startIntegrationOutboxWorker(fastify);
startErpMasterDataSyncJob(fastify);
return { driver: "cron" };
}
+48
查看文件
@@ -5,6 +5,7 @@ const { archiveUnusedInvoices } = require("./invoiceArchiveJob");
const { runOcrOnce } = require("./ocrJob");
const { runVerificationForRecord } = require("./verificationJob");
const { runIntegrationOutboxOnce } = require("./integrationOutboxJob");
const { runErpMasterDataSyncOnce } = require("./erpMasterDataSyncJob");
const {
FEATURE_FLAG_KEYS,
evaluateFeatureFlag,
@@ -15,6 +16,7 @@ const JOB_DELEGATION_REVOKE = "delegation.revoke-expired";
const JOB_INVOICE_ARCHIVE = "invoice.archive-unused";
const JOB_INVOICE_VERIFY = "invoice.verify";
const JOB_INTEGRATION_OUTBOX = "integration.outbox.dispatch";
const JOB_ERP_MASTER_DATA_SYNC = "erp.master-data.sync";
function resolveDatabaseConnectionString() {
const connectionString = process.env.DATABASE_URL;
@@ -45,6 +47,9 @@ async function startPgBossJobs(fastify) {
await boss.createQueue(JOB_INVOICE_ARCHIVE);
await boss.createQueue(JOB_INVOICE_VERIFY);
await boss.createQueue(JOB_INTEGRATION_OUTBOX);
if (fastify.config.erpMasterDataSyncCron) {
await boss.createQueue(JOB_ERP_MASTER_DATA_SYNC);
}
await boss.work(JOB_DELEGATION_REVOKE, { batchSize: 1 }, async (jobs) => {
const job = jobs?.[0];
@@ -198,6 +203,37 @@ async function startPgBossJobs(fastify) {
}
});
if (fastify.config.erpMasterDataSyncCron) {
await boss.work(JOB_ERP_MASTER_DATA_SYNC, { batchSize: 1 }, async (jobs) => {
const job = jobs?.[0];
try {
const result = await runErpMasterDataSyncOnce(
fastify.prisma,
fastify.config,
fastify.log,
);
fastify.metrics?.jobRunsTotal?.inc?.({
job: JOB_ERP_MASTER_DATA_SYNC,
driver: "pgboss",
outcome: "success",
});
if (!result?.skipped) {
fastify.log.info({ jobId: job?.id, result }, "pg-boss ERP master data sync processed");
}
} catch (error) {
fastify.metrics?.jobRunsTotal?.inc?.({
job: JOB_ERP_MASTER_DATA_SYNC,
driver: "pgboss",
outcome: "failure",
});
fastify.log.error({ err: error, jobId: job?.id }, "pg-boss ERP master data sync failed");
throw error;
}
});
}
await boss.work(JOB_INTEGRATION_OUTBOX, { batchSize: 1 }, async (jobs) => {
const job = jobs?.[0];
@@ -294,6 +330,16 @@ async function startPgBossJobs(fastify) {
singletonSeconds: 55,
});
if (fastify.config.erpMasterDataSyncCron) {
await boss.schedule(JOB_ERP_MASTER_DATA_SYNC, fastify.config.erpMasterDataSyncCron, null, {
tz: fastify.config.appTimezone,
singletonKey: JOB_ERP_MASTER_DATA_SYNC,
singletonSeconds: 55,
});
} else {
await boss.unschedule(JOB_ERP_MASTER_DATA_SYNC).catch(() => {});
}
fastify.log.info(
{
schema: fastify.config.pgBossSchema,
@@ -305,6 +351,7 @@ async function startPgBossJobs(fastify) {
delegationCron: fastify.config.delegationRevokeCron,
invoiceArchiveCron: fastify.config.invoiceArchiveCron,
integrationOutboxCron: fastify.config.integrationOutboxCron,
erpMasterDataSyncCron: fastify.config.erpMasterDataSyncCron || null,
},
"pg-boss job system started",
);
@@ -318,5 +365,6 @@ module.exports = {
JOB_INVOICE_ARCHIVE,
JOB_INVOICE_VERIFY,
JOB_INTEGRATION_OUTBOX,
JOB_ERP_MASTER_DATA_SYNC,
startPgBossJobs,
};
+149 -62
查看文件
@@ -1,24 +1,11 @@
function buildErpPayload(input) {
return {
reimbursementId: input.reimbursement.id,
formNumber: input.reimbursement.form_number,
ownerEmployeeId: input.reimbursement.owner_id,
submitterEmployeeId: input.reimbursement.submitter_id,
totalAmount: input.reimbursement.total_amount?.toString?.() ?? null,
paymentVoucherNo: input.reimbursement.payment_voucher_no ?? null,
paymentDate: input.reimbursement.payment_date
? new Date(input.reimbursement.payment_date).toISOString().slice(0, 10)
: null,
paidAt: input.reimbursement.paid_at ? new Date(input.reimbursement.paid_at).toISOString() : null,
};
}
const { buildReimbursementJournalPayload } = require('./erpJournal');
function normalizeProvider(config) {
return String(config.erpSyncProvider ?? "mock").trim().toLowerCase() || "mock";
return String(config.erpSyncProvider ?? 'mock').trim().toLowerCase() || 'mock';
}
function buildAuthHeaders(config) {
const token = String(config.erpSyncAuthToken ?? "").trim();
const token = String(config.erpSyncAuthToken ?? '').trim();
if (!token) {
return {};
}
@@ -28,67 +15,87 @@ function buildAuthHeaders(config) {
};
}
async function postToErpHttp(config, payload) {
const endpoint = String(config.erpSyncEndpoint ?? "").trim();
function coerceTimeout(value, fallback) {
const parsed = Number(value ?? fallback ?? 5000);
if (Number.isFinite(parsed) && parsed >= 100) {
return parsed;
}
return Math.max(100, Number(fallback ?? 5000) || 5000);
}
function resolveMasterDataEndpoint(config) {
return String(config.erpMasterDataEndpoint ?? '').trim();
}
function resolveJournalExportEndpoint(config) {
return (
String(config.erpJournalExportEndpoint ?? '').trim() ||
String(config.erpSyncEndpoint ?? '').trim()
);
}
async function parseResponseBody(response) {
const contentType = response.headers.get('content-type') || '';
if (contentType.includes('application/json')) {
try {
return await response.json();
} catch {
return null;
}
}
try {
return await response.text();
} catch {
return null;
}
}
async function postJsonToErp(config, options) {
const endpoint = String(options.endpoint ?? '').trim();
if (!endpoint) {
return {
ok: false,
code: "erp_sync_endpoint_missing",
message: "ERP sync endpoint missing",
code: options.missingEndpointCode,
message: options.missingEndpointMessage,
retryable: false,
};
}
const timeoutMs = Math.max(100, Number(config.erpSyncTimeoutMs ?? 5000));
const timeoutMs = coerceTimeout(options.timeoutMs, config.erpSyncTimeoutMs);
const controller = new AbortController();
const timer = setTimeout(() => {
controller.abort();
}, timeoutMs);
const timer = setTimeout(() => controller.abort(), timeoutMs);
try {
const response = await fetch(endpoint, {
method: "POST",
method: 'POST',
headers: {
"content-type": "application/json",
'content-type': 'application/json',
...buildAuthHeaders(config),
},
body: JSON.stringify(payload),
body: JSON.stringify(options.payload ?? {}),
signal: controller.signal,
});
let responseBody = null;
const contentType = response.headers.get("content-type") || "";
if (contentType.includes("application/json")) {
try {
responseBody = await response.json();
} catch {
responseBody = null;
}
} else {
try {
responseBody = await response.text();
} catch {
responseBody = null;
}
}
const responseBody = await parseResponseBody(response);
if (response.ok) {
return {
ok: true,
code: "http_synced",
message: "ERP synced",
code: options.successCode,
message: options.successMessage,
response: {
status: response.status,
body: responseBody,
},
data: responseBody,
};
}
const retryable = response.status >= 500;
return {
ok: false,
code: retryable ? "erp_sync_http_5xx" : "erp_sync_http_4xx",
message: `ERP sync HTTP ${response.status}`,
code: retryable ? options.http5xxCode : options.http4xxCode,
message: `${options.messagePrefix} HTTP ${response.status}`,
retryable,
httpStatus: response.status,
response: {
@@ -97,19 +104,19 @@ async function postToErpHttp(config, payload) {
},
};
} catch (error) {
if (error?.name === "AbortError") {
if (error?.name === 'AbortError') {
return {
ok: false,
code: "erp_sync_timeout",
message: `ERP sync timeout (${timeoutMs}ms)`,
code: options.timeoutCode,
message: `${options.messagePrefix} timeout (${timeoutMs}ms)`,
retryable: true,
};
}
return {
ok: false,
code: "erp_sync_network_error",
message: String(error?.message ?? error ?? "ERP sync network error"),
code: options.networkCode,
message: String(error?.message ?? error ?? `${options.messagePrefix} network error`),
retryable: true,
};
} finally {
@@ -117,41 +124,121 @@ async function postToErpHttp(config, payload) {
}
}
async function pushPaidReimbursementToErp(config, payload) {
async function pullMasterDataFromErp(config, options = {}) {
if (!config.erpSyncEnabled) {
return {
ok: false,
code: "erp_sync_disabled",
message: "ERP sync disabled",
code: 'erp_sync_disabled',
message: 'ERP sync disabled',
retryable: false,
};
}
const provider = normalizeProvider(config);
if (provider === "mock") {
const resources = Array.isArray(options.resources) ? options.resources : [];
if (provider === 'mock') {
return {
ok: true,
code: "mock_synced",
message: "mock ERP synced",
code: 'mock_synced',
message: 'mock ERP master data pulled',
response: {
externalId: `ERP-${payload.formNumber ?? payload.reimbursementId}`,
provider,
},
data: {
ok: true,
legalEntities: [],
costCenters: [],
suppliers: [],
requestedResources: resources,
},
};
}
if (provider === "http") {
return postToErpHttp(config, payload);
if (provider === 'http') {
return postJsonToErp(config, {
endpoint: resolveMasterDataEndpoint(config),
timeoutMs: coerceTimeout(config.erpMasterDataTimeoutMs, config.erpSyncTimeoutMs),
payload: { resources },
successCode: 'http_synced',
successMessage: 'ERP master data pulled',
missingEndpointCode: 'erp_master_data_endpoint_missing',
missingEndpointMessage: 'ERP master data endpoint missing',
http4xxCode: 'erp_master_data_http_4xx',
http5xxCode: 'erp_master_data_http_5xx',
timeoutCode: 'erp_master_data_timeout',
networkCode: 'erp_master_data_network_error',
messagePrefix: 'ERP master data pull',
});
}
return {
ok: false,
code: "erp_sync_provider_invalid",
code: 'erp_master_data_provider_invalid',
message: `Unsupported ERP master data provider: ${provider}`,
retryable: false,
};
}
async function exportJournalToErp(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') {
return {
ok: true,
code: 'mock_synced',
message: 'mock ERP synced',
response: {
externalId: `ERP-${payload?.documentNo ?? payload?.formNumber ?? payload?.entityId ?? payload?.reimbursementId ?? 'UNKNOWN'}`,
},
};
}
if (provider === 'http') {
return postJsonToErp(config, {
endpoint: resolveJournalExportEndpoint(config),
timeoutMs: coerceTimeout(config.erpSyncTimeoutMs, config.erpSyncTimeoutMs),
payload,
successCode: 'http_synced',
successMessage: 'ERP synced',
missingEndpointCode: 'erp_sync_endpoint_missing',
missingEndpointMessage: 'ERP sync endpoint missing',
http4xxCode: 'erp_sync_http_4xx',
http5xxCode: 'erp_sync_http_5xx',
timeoutCode: 'erp_sync_timeout',
networkCode: 'erp_sync_network_error',
messagePrefix: 'ERP sync',
});
}
return {
ok: false,
code: 'erp_sync_provider_invalid',
message: `Unsupported ERP sync provider: ${provider}`,
retryable: false,
};
}
function buildErpPayload(input) {
return buildReimbursementJournalPayload(input);
}
async function pushPaidReimbursementToErp(config, payload) {
return exportJournalToErp(config, payload);
}
module.exports = {
buildErpPayload,
exportJournalToErp,
normalizeProvider,
pullMasterDataFromErp,
pushPaidReimbursementToErp,
};
@@ -0,0 +1,534 @@
const { pullMasterDataFromErp } = require('./erp');
const ERP_MASTER_DATA_PROVIDER = 'erp';
const ERP_MASTER_DATA_RESOURCES = Object.freeze(['legal_entity', 'cost_center', 'supplier']);
function normalizeRequestedResources(resources) {
if (!Array.isArray(resources) || !resources.length) {
return [...ERP_MASTER_DATA_RESOURCES];
}
const seen = new Set();
const result = [];
for (const resource of ERP_MASTER_DATA_RESOURCES) {
if (resources.includes(resource) && !seen.has(resource)) {
seen.add(resource);
result.push(resource);
}
}
return result;
}
function makeSummary(requested) {
return {
requested,
created: 0,
updated: 0,
archived: 0,
skipped: 0,
sampleSkipped: [],
};
}
function addSkip(summary, key, reason) {
summary.skipped += 1;
if (summary.sampleSkipped.length < 20) {
summary.sampleSkipped.push({ key: key ?? 'unknown', reason });
}
}
function mapSyncState(record) {
return {
resourceType: record.resource_type,
status: record.status,
lastStartedAt: record.last_started_at?.toISOString?.() ?? null,
lastFinishedAt: record.last_finished_at?.toISOString?.() ?? null,
lastSuccessAt: record.last_success_at?.toISOString?.() ?? null,
lastError: record.last_error ?? null,
lastResult: record.last_result ?? null,
};
}
async function upsertSyncState(tx, resourceType, data) {
return tx.integrationSyncState.upsert({
where: {
provider_resource_type: {
provider: ERP_MASTER_DATA_PROVIDER,
resource_type: resourceType,
},
},
create: {
provider: ERP_MASTER_DATA_PROVIDER,
resource_type: resourceType,
...data,
},
update: data,
});
}
async function markSyncStateRunning(tx, resourceType, actorId) {
return upsertSyncState(tx, resourceType, {
status: 'running',
last_started_at: new Date(),
last_finished_at: null,
last_error: null,
updated_by: actorId ?? null,
created_by: actorId ?? null,
});
}
async function markSyncStateSucceeded(tx, resourceType, actorId, result) {
const now = new Date();
return upsertSyncState(tx, resourceType, {
status: 'succeeded',
last_result: result,
last_error: null,
last_finished_at: now,
last_success_at: now,
updated_by: actorId ?? null,
created_by: actorId ?? null,
});
}
async function markSyncStateFailed(tx, resourceType, actorId, error) {
return upsertSyncState(tx, resourceType, {
status: 'failed',
last_error: String(error?.message ?? error ?? 'erp_master_data_sync_failed').slice(0, 500),
last_finished_at: new Date(),
updated_by: actorId ?? null,
created_by: actorId ?? null,
});
}
async function listMasterDataSyncStates(prisma) {
const rows = await prisma.integrationSyncState.findMany({
where: {
provider: ERP_MASTER_DATA_PROVIDER,
resource_type: { in: ERP_MASTER_DATA_RESOURCES },
},
orderBy: [{ resource_type: 'asc' }],
});
const map = new Map(rows.map((row) => [row.resource_type, row]));
return ERP_MASTER_DATA_RESOURCES.map((resourceType) => {
const row = map.get(resourceType);
if (!row) {
return {
resourceType,
status: 'idle',
lastStartedAt: null,
lastFinishedAt: null,
lastSuccessAt: null,
lastError: null,
lastResult: null,
};
}
return mapSyncState(row);
});
}
async function ensureBindingNotConflicted(tx, entityType, localId, externalId) {
if (!externalId) {
return { ok: true };
}
const binding = await tx.integrationBinding.findFirst({
where: {
provider: ERP_MASTER_DATA_PROVIDER,
entity_type: entityType,
external_id: externalId,
},
select: { local_id: true },
});
if (binding && binding.local_id !== localId) {
return { ok: false, reason: 'binding_conflict' };
}
return { ok: true };
}
async function upsertBinding(tx, entityType, localId, externalId, externalCode, actorId) {
return tx.integrationBinding.upsert({
where: {
provider_entity_type_local_id: {
provider: ERP_MASTER_DATA_PROVIDER,
entity_type: entityType,
local_id: localId,
},
},
create: {
provider: ERP_MASTER_DATA_PROVIDER,
entity_type: entityType,
local_id: localId,
external_id: externalId ?? null,
external_code: externalCode ?? null,
last_synced_at: new Date(),
created_by: actorId ?? null,
updated_by: actorId ?? null,
},
update: {
external_id: externalId ?? null,
external_code: externalCode ?? null,
last_synced_at: new Date(),
updated_by: actorId ?? null,
},
});
}
function normalizePayloadRows(rows, resolveKey) {
const summary = makeSummary(Array.isArray(rows) ? rows.length : 0);
const uniqueRows = [];
const seen = new Set();
for (const row of Array.isArray(rows) ? rows : []) {
const key = resolveKey(row);
if (!key) {
addSkip(summary, null, 'external_code_missing');
continue;
}
if (seen.has(key)) {
addSkip(summary, key, 'duplicate_payload_key');
continue;
}
seen.add(key);
uniqueRows.push(row);
}
return { summary, rows: uniqueRows };
}
function resolveArchivedState(enabled, archivedAt) {
if (enabled === false) {
return archivedAt ?? new Date();
}
return null;
}
async function syncLegalEntities(tx, rows, actorId) {
const prepared = normalizePayloadRows(rows, (row) => String(row?.code ?? '').trim() || null);
const summary = prepared.summary;
const codes = prepared.rows.map((row) => String(row.code).trim());
const existingRows = await tx.legalEntity.findMany({
where: { code: { in: codes } },
select: { id: true, code: true, enabled: true, archived_at: true },
});
const existingByCode = new Map(existingRows.map((row) => [row.code, row]));
for (const row of prepared.rows) {
const code = String(row.code).trim();
const existing = existingByCode.get(code);
const enabled = row.enabled !== false;
const archivedAt = resolveArchivedState(enabled, existing?.archived_at ?? null);
if (existing) {
const bindingCheck = await ensureBindingNotConflicted(tx, 'legal_entity', existing.id, row.externalId ?? null);
if (!bindingCheck.ok) {
addSkip(summary, code, bindingCheck.reason);
continue;
}
await tx.legalEntity.update({
where: { id: existing.id },
data: {
name: row.name,
short_name: row.shortName ?? null,
currency_code: row.currencyCode ?? 'CNY',
tax_no: row.taxNo ?? null,
enabled,
archived_at: archivedAt,
updated_by: actorId ?? null,
},
});
summary.updated += 1;
if (!enabled && (existing.enabled || !existing.archived_at)) {
summary.archived += 1;
}
await upsertBinding(tx, 'legal_entity', existing.id, row.externalId ?? null, code, actorId);
continue;
}
const created = await tx.legalEntity.create({
data: {
code,
name: row.name,
short_name: row.shortName ?? null,
currency_code: row.currencyCode ?? 'CNY',
tax_no: row.taxNo ?? null,
enabled,
archived_at: enabled ? null : new Date(),
created_by: actorId ?? null,
updated_by: actorId ?? null,
},
select: { id: true },
});
summary.created += 1;
await upsertBinding(tx, 'legal_entity', created.id, row.externalId ?? null, code, actorId);
}
return summary;
}
async function syncCostCenters(tx, rows, actorId) {
const prepared = normalizePayloadRows(
rows,
(row) => {
const legalEntityCode = String(row?.legalEntityCode ?? '').trim();
const code = String(row?.code ?? '').trim();
return legalEntityCode && code ? `${legalEntityCode}::${code}` : null;
},
);
const summary = prepared.summary;
const legalEntityCodes = Array.from(new Set(prepared.rows.map((row) => String(row.legalEntityCode).trim())));
const legalEntities = await tx.legalEntity.findMany({
where: { code: { in: legalEntityCodes } },
select: { id: true, code: true },
});
const legalEntityByCode = new Map(legalEntities.map((row) => [row.code, row]));
const existingRows = await tx.costCenter.findMany({
where: {
OR: prepared.rows.map((row) => ({
code: String(row.code).trim(),
legal_entity: { code: String(row.legalEntityCode).trim() },
})),
},
select: { id: true, code: true, legal_entity_id: true, enabled: true, archived_at: true, legal_entity: { select: { code: true } } },
}).catch(() => []);
const existingByKey = new Map(existingRows.map((row) => [`${row.legal_entity?.code ?? ''}::${row.code}`, row]));
for (const row of prepared.rows) {
const legalEntityCode = String(row.legalEntityCode).trim();
const code = String(row.code).trim();
const key = `${legalEntityCode}::${code}`;
const legalEntity = legalEntityByCode.get(legalEntityCode);
if (!legalEntity) {
addSkip(summary, key, 'legal_entity_missing');
continue;
}
const existing = existingByKey.get(key);
const enabled = row.enabled !== false;
const archivedAt = resolveArchivedState(enabled, existing?.archived_at ?? null);
if (existing) {
const bindingCheck = await ensureBindingNotConflicted(tx, 'cost_center', existing.id, row.externalId ?? null);
if (!bindingCheck.ok) {
addSkip(summary, key, bindingCheck.reason);
continue;
}
await tx.costCenter.update({
where: { id: existing.id },
data: {
name: row.name,
enabled,
archived_at: archivedAt,
updated_by: actorId ?? null,
},
});
summary.updated += 1;
if (!enabled && (existing.enabled || !existing.archived_at)) {
summary.archived += 1;
}
await upsertBinding(tx, 'cost_center', existing.id, row.externalId ?? null, code, actorId);
continue;
}
const created = await tx.costCenter.create({
data: {
legal_entity_id: legalEntity.id,
code,
name: row.name,
enabled,
archived_at: enabled ? null : new Date(),
created_by: actorId ?? null,
updated_by: actorId ?? null,
},
select: { id: true },
});
summary.created += 1;
await upsertBinding(tx, 'cost_center', created.id, row.externalId ?? null, code, actorId);
}
return summary;
}
async function syncSuppliers(tx, rows, actorId) {
const prepared = normalizePayloadRows(rows, (row) => String(row?.code ?? '').trim() || null);
const summary = prepared.summary;
const legalEntityCodes = Array.from(
new Set(
prepared.rows
.map((row) => String(row.legalEntityCode ?? '').trim())
.filter(Boolean),
),
);
const legalEntities = await tx.legalEntity.findMany({
where: { code: { in: legalEntityCodes } },
select: { id: true, code: true },
});
const legalEntityByCode = new Map(legalEntities.map((row) => [row.code, row]));
const codes = prepared.rows.map((row) => String(row.code).trim());
const existingRows = await tx.supplier.findMany({
where: { code: { in: codes } },
select: { id: true, code: true, enabled: true, archived_at: true },
});
const existingByCode = new Map(existingRows.map((row) => [row.code, row]));
for (const row of prepared.rows) {
const code = String(row.code).trim();
const legalEntityCode = String(row.legalEntityCode ?? '').trim() || null;
const legalEntity = legalEntityCode ? legalEntityByCode.get(legalEntityCode) : null;
if (legalEntityCode && !legalEntity) {
addSkip(summary, code, 'legal_entity_missing');
continue;
}
const existing = existingByCode.get(code);
const enabled = row.enabled !== false;
const archivedAt = resolveArchivedState(enabled, existing?.archived_at ?? null);
if (existing) {
const bindingCheck = await ensureBindingNotConflicted(tx, 'supplier', existing.id, row.externalId ?? null);
if (!bindingCheck.ok) {
addSkip(summary, code, bindingCheck.reason);
continue;
}
await tx.supplier.update({
where: { id: existing.id },
data: {
name: row.name,
short_name: row.shortName ?? null,
tax_no: row.taxNo ?? null,
contact_name: row.contactName ?? null,
contact_phone: row.contactPhone ?? null,
payment_terms: row.paymentTerms ?? null,
legal_entity_id: legalEntity?.id ?? null,
enabled,
archived_at: archivedAt,
updated_by: actorId ?? null,
},
});
summary.updated += 1;
if (!enabled && (existing.enabled || !existing.archived_at)) {
summary.archived += 1;
}
await upsertBinding(tx, 'supplier', existing.id, row.externalId ?? null, code, actorId);
continue;
}
const created = await tx.supplier.create({
data: {
code,
name: row.name,
short_name: row.shortName ?? null,
tax_no: row.taxNo ?? null,
contact_name: row.contactName ?? null,
contact_phone: row.contactPhone ?? null,
payment_terms: row.paymentTerms ?? null,
legal_entity_id: legalEntity?.id ?? null,
enabled,
archived_at: enabled ? null : new Date(),
created_by: actorId ?? null,
updated_by: actorId ?? null,
},
select: { id: true },
});
summary.created += 1;
await upsertBinding(tx, 'supplier', created.id, row.externalId ?? null, code, actorId);
}
return summary;
}
function resolveResourceRows(data, resourceType) {
if (resourceType === 'legal_entity') {
return data?.legalEntities ?? [];
}
if (resourceType === 'cost_center') {
return data?.costCenters ?? [];
}
return data?.suppliers ?? [];
}
async function runResourceSync(prisma, resourceType, rows, actorId) {
return prisma.$transaction(async (tx) => {
if (resourceType === 'legal_entity') {
return syncLegalEntities(tx, rows, actorId);
}
if (resourceType === 'cost_center') {
return syncCostCenters(tx, rows, actorId);
}
return syncSuppliers(tx, rows, actorId);
});
}
async function syncErpMasterData(prisma, config, options = {}) {
const logger = options.logger ?? console;
const actorId = options.actorId ?? null;
const resources = normalizeRequestedResources(options.resources);
for (const resourceType of resources) {
await prisma.$transaction((tx) => markSyncStateRunning(tx, resourceType, actorId));
}
const pulled = await pullMasterDataFromErp(config, { resources });
if (!pulled.ok) {
for (const resourceType of resources) {
await prisma.$transaction((tx) => markSyncStateFailed(tx, resourceType, actorId, pulled.message));
}
logger.error(
{
provider: 'erp',
resourceType: resources.join(','),
endpoint: config.erpMasterDataEndpoint ?? null,
reason: pulled.code,
},
'ERP master data pull failed',
);
return {
ok: false,
error: pulled,
items: await listMasterDataSyncStates(prisma),
};
}
for (const resourceType of resources) {
try {
const summary = await runResourceSync(
prisma,
resourceType,
resolveResourceRows(pulled.data, resourceType),
actorId,
);
await prisma.$transaction((tx) => markSyncStateSucceeded(tx, resourceType, actorId, summary));
} catch (error) {
logger.error(
{
err: error,
provider: 'erp',
resourceType,
endpoint: config.erpMasterDataEndpoint ?? null,
reason: error?.message ?? 'sync_failed',
},
'ERP master data resource sync failed',
);
await prisma.$transaction((tx) => markSyncStateFailed(tx, resourceType, actorId, error));
}
}
return {
ok: true,
error: null,
items: await listMasterDataSyncStates(prisma),
};
}
module.exports = {
ERP_MASTER_DATA_PROVIDER,
ERP_MASTER_DATA_RESOURCES,
listMasterDataSyncStates,
mapSyncState,
normalizeRequestedResources,
syncErpMasterData,
};
@@ -0,0 +1,145 @@
const {
normalizeRequestedResources,
syncErpMasterData,
} = require('../../../src/services/integrations/erpMasterData');
function createPrismaMock() {
const syncStates = new Map();
const prisma = {
$transaction: async (fn) => fn(prisma),
integrationSyncState: {
upsert: vi.fn().mockImplementation(async ({ where, create, update }) => {
const key = `${where.provider_resource_type.provider}:${where.provider_resource_type.resource_type}`;
const current = syncStates.get(key) ?? null;
const next = current ? { ...current, ...update } : { ...create };
syncStates.set(key, next);
return next;
}),
findMany: vi.fn().mockImplementation(async () => Array.from(syncStates.values())),
},
integrationBinding: {
findFirst: vi.fn().mockResolvedValue(null),
upsert: vi.fn().mockResolvedValue({ id: 'binding-1' }),
},
legalEntity: {
findMany: vi.fn().mockResolvedValue([]),
create: vi.fn().mockResolvedValue({ id: 'le-1' }),
update: vi.fn().mockResolvedValue({ id: 'le-1' }),
},
costCenter: {
findMany: vi.fn().mockResolvedValue([]),
create: vi.fn().mockResolvedValue({ id: 'cc-1' }),
update: vi.fn().mockResolvedValue({ id: 'cc-1' }),
},
supplier: {
findMany: vi.fn().mockResolvedValue([]),
create: vi.fn().mockResolvedValue({ id: 'sup-1' }),
update: vi.fn().mockResolvedValue({ id: 'sup-1' }),
},
};
return prisma;
}
describe('ERP Master Data Service', () => {
const originalFetch = global.fetch;
afterEach(() => {
global.fetch = originalFetch;
vi.restoreAllMocks();
});
it('normalizes resource order and dedupes input', () => {
expect(normalizeRequestedResources(['supplier', 'legal_entity', 'supplier'])).toEqual([
'legal_entity',
'supplier',
]);
expect(normalizeRequestedResources([])).toEqual(['legal_entity', 'cost_center', 'supplier']);
});
it('skips supplier rows without stable code', async () => {
const prisma = createPrismaMock();
global.fetch = vi.fn().mockResolvedValue({
ok: true,
status: 200,
headers: { get: () => 'application/json' },
json: vi.fn().mockResolvedValue({
suppliers: [{ code: '', name: 'bad supplier' }],
}),
});
const result = await syncErpMasterData(prisma, {
erpSyncEnabled: true,
erpSyncProvider: 'http',
erpMasterDataEndpoint: 'https://example.com/master-data',
erpMasterDataTimeoutMs: 500,
}, {
resources: ['supplier'],
actorId: 'actor-1',
logger: { error: vi.fn() },
});
const supplierState = result.items.find((item) => item.resourceType === 'supplier');
expect(result.ok).toBe(true);
expect(supplierState.lastResult.skipped).toBe(1);
expect(prisma.supplier.create).not.toHaveBeenCalled();
});
it('skips cost centers when legal entity is missing', async () => {
const prisma = createPrismaMock();
global.fetch = vi.fn().mockResolvedValue({
ok: true,
status: 200,
headers: { get: () => 'application/json' },
json: vi.fn().mockResolvedValue({
costCenters: [{ code: 'IT001', legalEntityCode: 'MISSING', name: 'IT中心' }],
}),
});
const result = await syncErpMasterData(prisma, {
erpSyncEnabled: true,
erpSyncProvider: 'http',
erpMasterDataEndpoint: 'https://example.com/master-data',
erpMasterDataTimeoutMs: 500,
}, {
resources: ['cost_center'],
actorId: 'actor-1',
logger: { error: vi.fn() },
});
const costCenterState = result.items.find((item) => item.resourceType === 'cost_center');
expect(costCenterState.lastResult.skipped).toBe(1);
expect(costCenterState.lastResult.sampleSkipped[0].reason).toBe('legal_entity_missing');
});
it('skips rows when integration binding conflicts', async () => {
const prisma = createPrismaMock();
prisma.supplier.findMany.mockResolvedValue([{ id: 'sup-local', code: 'SUP001', enabled: true, archived_at: null }]);
prisma.integrationBinding.findFirst.mockResolvedValue({ local_id: 'another-local' });
global.fetch = vi.fn().mockResolvedValue({
ok: true,
status: 200,
headers: { get: () => 'application/json' },
json: vi.fn().mockResolvedValue({
suppliers: [{ code: 'SUP001', externalId: 'ERP-SUP-001', name: '供应商' }],
}),
});
const result = await syncErpMasterData(prisma, {
erpSyncEnabled: true,
erpSyncProvider: 'http',
erpMasterDataEndpoint: 'https://example.com/master-data',
erpMasterDataTimeoutMs: 500,
}, {
resources: ['supplier'],
actorId: 'actor-1',
logger: { error: vi.fn() },
});
const supplierState = result.items.find((item) => item.resourceType === 'supplier');
expect(supplierState.lastResult.skipped).toBe(1);
expect(supplierState.lastResult.sampleSkipped[0].reason).toBe('binding_conflict');
expect(prisma.supplier.update).not.toHaveBeenCalled();
});
});
@@ -1,40 +1,109 @@
const erp = require('../../../../src/services/integrations/erp');
const erp = require("../../../../src/services/integrations/erp");
describe('ERP Sync Services', () => {
const originalFetch = global.fetch;
describe("ERP Sync Services", () => {
it("builds correct payload map", () => {
const payload = erp.buildErpPayload({
reimbursement: {
id: "r1",
form_number: "F001",
payment_date: "2026-03-01T12:00:00.000Z",
},
});
expect(payload.formNumber).toBe("F001");
expect(payload.paymentDate).toBe("2026-03-01");
afterEach(() => {
global.fetch = originalFetch;
vi.restoreAllMocks();
});
it('builds reimbursement journal payload via compat helper', () => {
const payload = erp.buildErpPayload({
reimbursement: {
id: 'r1',
form_number: 'F001',
owner_id: 'owner-1',
submitter_id: 'submitter-1',
department_id: 'dept-1',
total_amount: { toString: () => '100.00' },
payment_voucher_no: 'PV001',
payment_date: '2026-03-01T12:00:00.000Z',
paid_at: '2026-03-02T08:00:00.000Z',
owner: { legal_entity: { code: 'HY' } },
},
});
it("returns mock synced when provider is mock", async () => {
const result = await erp.pushPaidReimbursementToErp(
{ erpSyncEnabled: true, erpSyncProvider: "mock" },
{ formNumber: "F001" }
);
expect(result.ok).toBe(true);
expect(result.code).toBe("mock_synced");
expect(payload.journalType).toBe('reimbursement_paid');
expect(payload.documentNo).toBe('F001');
expect(payload.paymentDate).toBe('2026-03-01');
expect(payload.dimensions.legalEntityCode).toBe('HY');
});
it('returns mock synced when journal export provider is mock', async () => {
const result = await erp.exportJournalToErp(
{ erpSyncEnabled: true, erpSyncProvider: 'mock' },
{ documentNo: 'BX-001' },
);
expect(result.ok).toBe(true);
expect(result.code).toBe('mock_synced');
expect(result.response.externalId).toContain('BX-001');
});
it('returns mock master data result when provider is mock', async () => {
const result = await erp.pullMasterDataFromErp(
{ erpSyncEnabled: true, erpSyncProvider: 'mock' },
{ resources: ['legal_entity', 'supplier'] },
);
expect(result.ok).toBe(true);
expect(result.code).toBe('mock_synced');
expect(result.data.legalEntities).toEqual([]);
expect(result.data.suppliers).toEqual([]);
});
it('handles http master data 4xx errors', async () => {
global.fetch = vi.fn().mockResolvedValue({
ok: false,
status: 422,
headers: { get: () => 'application/json' },
json: vi.fn().mockResolvedValue({ code: 'invalid_payload' }),
});
it("returns invalid provider if unsupported", async () => {
const result = await erp.pushPaidReimbursementToErp(
{ erpSyncEnabled: true, erpSyncProvider: "invalid_xyz" },
{}
);
expect(result.ok).toBe(false);
expect(result.code).toBe("erp_sync_provider_invalid");
});
const result = await erp.pullMasterDataFromErp(
{
erpSyncEnabled: true,
erpSyncProvider: 'http',
erpMasterDataEndpoint: 'https://example.com/master-data',
erpMasterDataTimeoutMs: 500,
},
{ resources: ['legal_entity'] },
);
it("returns disabled if disabled in config", async () => {
const result = await erp.pushPaidReimbursementToErp({ erpSyncEnabled: false }, {});
expect(result.ok).toBe(false);
expect(result.code).toBe("erp_sync_disabled");
});
expect(result.ok).toBe(false);
expect(result.code).toBe('erp_master_data_http_4xx');
expect(result.retryable).toBe(false);
});
it('handles http journal export timeout', async () => {
global.fetch = vi.fn().mockRejectedValue({ name: 'AbortError' });
const result = await erp.exportJournalToErp(
{
erpSyncEnabled: true,
erpSyncProvider: 'http',
erpJournalExportEndpoint: 'https://example.com/journals',
erpSyncTimeoutMs: 500,
},
{ documentNo: 'BX-001' },
);
expect(result.ok).toBe(false);
expect(result.code).toBe('erp_sync_timeout');
expect(result.retryable).toBe(true);
});
it('returns invalid provider if unsupported', async () => {
const result = await erp.exportJournalToErp(
{ erpSyncEnabled: true, erpSyncProvider: 'invalid_xyz' },
{},
);
expect(result.ok).toBe(false);
expect(result.code).toBe('erp_sync_provider_invalid');
});
it('returns disabled if ERP sync is disabled', async () => {
const result = await erp.exportJournalToErp({ erpSyncEnabled: false }, {});
expect(result.ok).toBe(false);
expect(result.code).toBe('erp_sync_disabled');
});
});