cangku/backend/src/jobs/jxStockSelectionSyncJob.js

110 lines
3.6 KiB
JavaScript

'use strict';
const cron = require('node-cron');
const database = require('../config/database');
const {
syncJxStockSelectionTables,
rebuildJxStockSelectionBase,
rebuildJxStockSelectionResult,
DEFAULT_REQUEST_TIMEOUT_MS,
} = require('../services/jxStockSelectionSync');
let baseRunning = false;
let resultRunning = false;
const JOB_LOCK_WAIT_SECONDS = Math.max(parseInt(process.env.JX_STOCK_SELECTION_JOB_LOCK_WAIT_SECONDS || '1', 10), 0);
async function withGlobalJobLock(lockKey, trigger, runner) {
const connection = await database.local.pool.getConnection();
let lockAcquired = false;
try {
const [rows] = await connection.query('SELECT GET_LOCK(?, ?) AS lockStatus', [ lockKey, JOB_LOCK_WAIT_SECONDS ]);
const lockStatus = rows && rows[0] ? Number(rows[0].lockStatus) : 0;
lockAcquired = lockStatus === 1;
if (!lockAcquired) {
console.log(`[JxStockSelectionSync] global lock not acquired, skip trigger=${trigger}, lockKey=${lockKey}`);
return { success: false, skipped: true, reason: 'global_lock_not_acquired', trigger, lockKey };
}
return await runner();
} finally {
try {
if (lockAcquired) {
await connection.query('DO RELEASE_LOCK(?)', [ lockKey ]);
}
} finally {
connection.release();
}
}
}
async function runJxStockSelectionBaseJob(trigger = 'manual_base') {
if (baseRunning) {
console.log(`[JxStockSelectionSync] base job is already running, skip trigger=${trigger}`);
return { success: false, skipped: true, reason: 'running' };
}
baseRunning = true;
try {
return await withGlobalJobLock('jx_stock_selection_base_daily', trigger, async () => {
return rebuildJxStockSelectionBase({
syncType: trigger,
requestTimeout: DEFAULT_REQUEST_TIMEOUT_MS,
});
});
} finally {
baseRunning = false;
}
}
async function runJxStockSelectionResultJob(trigger = 'manual_result') {
if (resultRunning) {
console.log(`[JxStockSelectionSync] result job is already running, skip trigger=${trigger}`);
return { success: false, skipped: true, reason: 'running' };
}
resultRunning = true;
try {
return await withGlobalJobLock('jx_stock_selection_result_daily', trigger, async () => {
return rebuildJxStockSelectionResult({
syncType: trigger,
requestTimeout: DEFAULT_REQUEST_TIMEOUT_MS,
});
});
} finally {
resultRunning = false;
}
}
function startJxStockSelectionSchedulers() {
cron.schedule('0 5 * * *', () => {
runJxStockSelectionBaseJob('daily_schedule_base').catch((error) => {
console.error('[JxStockSelectionSync] base scheduled job failed', error);
});
}, { timezone: 'Asia/Shanghai' });
cron.schedule('0 7 * * *', () => {
runJxStockSelectionResultJob('daily_schedule_result').catch((error) => {
console.error('[JxStockSelectionSync] result scheduled job failed', error);
});
}, { timezone: 'Asia/Shanghai' });
console.log('[Scheduler] 京喜条码打标底表任务已注册: 每天05:00执行');
console.log('[Scheduler] 京喜条码打标结果表任务已注册: 每天07:00执行');
}
async function manualJxStockSelectionSync() {
const baseResult = await runJxStockSelectionBaseJob('manual_base');
const resultResult = await runJxStockSelectionResultJob('manual_result');
return { success: Boolean(baseResult?.success && resultResult?.success), baseResult, resultResult };
}
module.exports = {
startJxStockSelectionSchedulers,
manualJxStockSelectionSync,
runJxStockSelectionBaseJob,
runJxStockSelectionResultJob,
runJxStockSelectionSyncJob: syncJxStockSelectionTables,
};