145 行
4.1 KiB
JavaScript
145 行
4.1 KiB
JavaScript
const cron = require('node-cron');
|
|
|
|
const {
|
|
fetchPendingOutboxEvents,
|
|
markOutboxFailed,
|
|
markOutboxProcessing,
|
|
markOutboxSucceeded,
|
|
} = require('../services/integrations/outbox');
|
|
const { dispatchOutboxEvent } = require('../services/integrations/dispatcher');
|
|
|
|
async function runIntegrationOutboxOnce(prisma, config, logger) {
|
|
const items = await fetchPendingOutboxEvents(prisma, {
|
|
batchSize: config.integrationOutboxBatchSize,
|
|
});
|
|
|
|
let succeeded = 0;
|
|
let failed = 0;
|
|
|
|
for (const item of items) {
|
|
await markOutboxProcessing(prisma, item.id, null);
|
|
|
|
try {
|
|
await dispatchOutboxEvent({
|
|
prisma,
|
|
config,
|
|
eventType: item.event_type,
|
|
payload: item.payload,
|
|
});
|
|
await markOutboxSucceeded(prisma, item.id, null);
|
|
succeeded += 1;
|
|
} catch (error) {
|
|
const retryable = typeof error?.retryable === 'boolean' ? error.retryable : true;
|
|
await markOutboxFailed(prisma, config, item, error, null, { retryable });
|
|
failed += 1;
|
|
logger.error(
|
|
{
|
|
err: error,
|
|
retryable,
|
|
outboxId: item.id,
|
|
eventType: item.event_type,
|
|
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,
|
|
channelType: item.payload?.channelType ?? null,
|
|
notificationType: item.payload?.notificationType ?? null,
|
|
},
|
|
'Integration outbox dispatch failed',
|
|
);
|
|
}
|
|
}
|
|
|
|
return {
|
|
total: items.length,
|
|
succeeded,
|
|
failed,
|
|
};
|
|
}
|
|
|
|
function startIntegrationOutboxJob(fastify) {
|
|
let running = false;
|
|
|
|
const task = cron.schedule(fastify.config.integrationOutboxCron, async () => {
|
|
if (running) {
|
|
return;
|
|
}
|
|
|
|
running = true;
|
|
try {
|
|
const result = await runIntegrationOutboxOnce(
|
|
fastify.prisma,
|
|
fastify.config,
|
|
fastify.log,
|
|
);
|
|
fastify.metrics?.jobRunsTotal?.inc?.({
|
|
job: 'integration.outbox.dispatch',
|
|
driver: 'cron',
|
|
outcome: 'success',
|
|
});
|
|
if (result.succeeded > 0) {
|
|
fastify.metrics?.jobProcessedTotal?.inc?.(
|
|
{
|
|
job: 'integration.outbox.dispatch',
|
|
driver: 'cron',
|
|
kind: 'succeeded',
|
|
},
|
|
result.succeeded,
|
|
);
|
|
}
|
|
if (result.failed > 0) {
|
|
fastify.metrics?.jobProcessedTotal?.inc?.(
|
|
{
|
|
job: 'integration.outbox.dispatch',
|
|
driver: 'cron',
|
|
kind: 'failed',
|
|
},
|
|
result.failed,
|
|
);
|
|
}
|
|
} catch (error) {
|
|
fastify.log.error({ err: error }, 'Integration outbox job failed');
|
|
fastify.metrics?.jobRunsTotal?.inc?.({
|
|
job: 'integration.outbox.dispatch',
|
|
driver: 'cron',
|
|
outcome: 'failure',
|
|
});
|
|
} finally {
|
|
running = false;
|
|
}
|
|
});
|
|
|
|
fastify.addHook('onClose', async () => {
|
|
task.stop();
|
|
task.destroy();
|
|
});
|
|
|
|
fastify.log.info(
|
|
{
|
|
cron: fastify.config.integrationOutboxCron,
|
|
batchSize: fastify.config.integrationOutboxBatchSize,
|
|
},
|
|
'Integration outbox job scheduled',
|
|
);
|
|
|
|
return task;
|
|
}
|
|
|
|
module.exports = {
|
|
runIntegrationOutboxOnce,
|
|
startIntegrationOutboxJob,
|
|
};
|