535 行
16 KiB
JavaScript
535 行
16 KiB
JavaScript
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,
|
|
};
|