cangku/backend/src/jobs/planningDingWorkflowSyncJob.js

'use strict';

const cron = require('node-cron');

const planningDatabase = require('../services/planningDatabase');
const {
  syncPlanningDingWorkflowTables,
} = require('../services/planningDingWorkflowSyncService');

let running = false;

const JOB_ENABLED = String(process.env.DING_WORKFLOW_SYNC_ENABLED || 'true').toLowerCase() !== 'false';
const HOURLY_CRON = process.env.DING_WORKFLOW_SYNC_CRON || '5 * * * *';
const FULL_CRON = process.env.DING_WORKFLOW_SYNC_FULL_CRON || '35 3 * * *';
const TIMEZONE = process.env.DING_WORKFLOW_SYNC_TIMEZONE || 'Asia/Shanghai';
const STARTUP_SYNC_ENABLED = String(process.env.DING_WORKFLOW_SYNC_STARTUP_ENABLED || 'true').toLowerCase() !== 'false';
const STARTUP_FULL_SYNC = String(process.env.DING_WORKFLOW_SYNC_STARTUP_FULL || 'false').toLowerCase() === 'true';
const STARTUP_DELAY_MS = Math.max(
  Number.parseInt(process.env.DING_WORKFLOW_SYNC_STARTUP_DELAY_MS || '30000', 10) || 30000,
  1000
);
const JOB_LOCK_WAIT_SECONDS = Math.max(
  Number.parseInt(process.env.DING_WORKFLOW_SYNC_LOCK_WAIT_SECONDS || '1', 10) || 1,
  0
);
const CHUNK_SIZE = Math.max(Number.parseInt(process.env.DING_WORKFLOW_SYNC_CHUNK_SIZE || '2000', 10) || 2000, 100);
const LOOKBACK_MINUTES = Math.max(
  Number.parseInt(process.env.DING_WORKFLOW_SYNC_LOOKBACK_MINUTES || '180', 10) || 180,
  0
);
const LOOKBACK_ROWS = Math.max(
  Number.parseInt(process.env.DING_WORKFLOW_SYNC_LOOKBACK_ROWS || '10000', 10) || 10000,
  0
);

async function withGlobalJobLock(lockKey, trigger, runner) {
  const pool = await planningDatabase.getPool();
  const connection = await 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(`[DingWorkflowSync] global lock not acquired, skip trigger=${trigger}`);
      return { success: false, skipped: true, reason: 'global_lock_not_acquired', trigger };
    }

    return runner();
  } finally {
    try {
      if (lockAcquired) {
        await connection.query('DO RELEASE_LOCK(?)', [lockKey]);
      }
    } finally {
      connection.release();
    }
  }
}

async function runPlanningDingWorkflowSyncJob(trigger = 'manual', options = {}) {
  if (running) {
    console.log(`[DingWorkflowSync] job is already running, skip trigger=${trigger}`);
    return { success: false, skipped: true, reason: 'running', trigger };
  }

  running = true;
  const startedAt = Date.now();

  try {
    return await withGlobalJobLock('planning_ding_workflow_sync', trigger, async () => {
      const full = Boolean(options.full);
      const table = options.table || 'all';
      const result = await syncPlanningDingWorkflowTables({
        table,
        full,
        chunkSize: options.chunkSize || CHUNK_SIZE,
        lookbackMinutes: options.lookbackMinutes || LOOKBACK_MINUTES,
        lookbackRows: options.lookbackRows || LOOKBACK_ROWS,
        onProgress: (payload) => {
          console.log(
            `[DingWorkflowSync] ${trigger} ${payload.table} ${payload.fetchedRows}/${payload.totalRows}`
          );
        },
      });

      return {
        ...result,
        trigger,
        elapsedMs: Date.now() - startedAt,
      };
    });
  } finally {
    running = false;
  }
}

function startPlanningDingWorkflowSyncSchedulers() {
  if (!JOB_ENABLED) {
    console.log('[Scheduler] 钉钉流程表同步任务未启用');
    return;
  }

  cron.schedule(HOURLY_CRON, () => {
    runPlanningDingWorkflowSyncJob('hourly_incremental', { full: false }).catch((error) => {
      console.error('[DingWorkflowSync] hourly scheduled job failed', error);
    });
  }, { timezone: TIMEZONE });

  console.log(`[Scheduler] 钉钉流程表同步任务已注册: ${HOURLY_CRON} (${TIMEZONE})`);

  if (FULL_CRON && String(FULL_CRON).trim()) {
    cron.schedule(FULL_CRON, () => {
      runPlanningDingWorkflowSyncJob('daily_full_reconcile', { full: true }).catch((error) => {
        console.error('[DingWorkflowSync] daily full reconcile failed', error);
      });
    }, { timezone: TIMEZONE });

    console.log(`[Scheduler] 钉钉流程表全量校准任务已注册: ${FULL_CRON} (${TIMEZONE})`);
  }

  if (STARTUP_SYNC_ENABLED) {
    const timer = setTimeout(() => {
      runPlanningDingWorkflowSyncJob(
        STARTUP_FULL_SYNC ? 'startup_full_sync' : 'startup_incremental_sync',
        { full: STARTUP_FULL_SYNC }
      ).catch((error) => {
        console.error('[DingWorkflowSync] startup sync failed', error);
      });
    }, STARTUP_DELAY_MS);

    if (typeof timer.unref === 'function') {
      timer.unref();
    }
  }
}

module.exports = {
  startPlanningDingWorkflowSyncSchedulers,
  runPlanningDingWorkflowSyncJob,
};