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, assetId: item.payload?.assetId ?? (item.payload?.entityType === 'asset_card' ? item.payload?.entityId ?? null : null), assetNo: item.payload?.assetNo ?? null, changeType: item.payload?.changeType ?? 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, };