hope-events/backend/src/jobs/index.js

/**
 * Background jobs — one registry, run by pg-boss (a job queue stored in Postgres).
 *
 * Why a queue instead of setInterval:
 * - Schedules live in the database, so a job runs once per schedule even when more than
 *   one backend process is running (PM2 cluster, two servers on one database).
 * - Every run is recorded (state, start/end, result or error), shown on the admin
 *   Background jobs page, which can also run a job straight away.
 * - A run that fails is retried, and a slow run never overlaps the next one.
 *
 * Each job can be switched off with its env flag (e.g. WAITLIST_JOB_ENABLED=false), and
 * its schedule changed with JOB_SCHEDULE_<NAME> (cron, e.g. JOB_SCHEDULE_WAITLIST="*\/5 * * * *").
 * JOBS_ENABLED=false turns all of them off (used by the end-to-end tests).
 * If the queue can't start (e.g. the database user can't create the pgboss schema), the
 * same jobs run on in-process timers instead, as before.
 */

const cronParser = require('cron-parser');

const TIMEZONE = process.env.APP_TIMEZONE || 'Africa/Johannesburg';
const SCHEMA = 'pgboss';

const JOBS = [
  {
    name: 'daily-summary', label: 'Daily event summary email', cron: '0 7 * * *', env: 'DAILY_SUMMARY_ENABLED',
    run: async () => { await require('../utils/notifications').sendDailyEventSummaries(new Date()); return { sent: true }; },
  },
  {
    name: 'waitlist', label: 'Waitlist: expire offers, offer freed places', cron: '*/10 * * * *', env: 'WAITLIST_JOB_ENABLED',
    run: () => require('../utils/waitlist').runWaitlistJob(),
  },
  {
    name: 'feedback', label: 'Post-event feedback forms and reminders', cron: '*/15 * * * *', env: 'FEEDBACK_JOB_ENABLED',
    run: () => require('../utils/feedback').runFeedbackJob(),
  },
  {
    name: 'sessions', label: 'Session reminders and certificates', cron: '*/5 * * * *', env: 'SESSION_JOBS_ENABLED',
    run: async () => {
      const reminders = await require('../utils/sessionReminders').runSessionReminderJob();
      const certificates = await require('../utils/certificates').runCertificateJob();
      return { reminders, certificates };
    },
  },
  {
    name: 'consent', label: 'Parental consent reminders and medical-detail deletion', cron: '0 9 * * *', env: 'CONSENT_JOBS_ENABLED',
    run: async () => {
      const { sendConsentReminders, purgeConsentMedicalData } = require('../utils/consent');
      return { reminded: await sendConsentReminders(new Date()), purged: await purgeConsentMedicalData(new Date()) };
    },
  },
  {
    name: 'analytics-rollup', label: 'Funnel analytics: roll up visits older than 90 days', cron: '15 4 * * *', env: 'ANALYTICS_ROLLUP_ENABLED',
    run: async () => ({ rolledUp: await require('../utils/analytics').rollupOldEvents() }),
  },
  {
    name: 'backup', label: 'Database backup', cron: '0 2 * * *', env: 'BACKUP_ENABLED',
    run: async () => ({ file: (await require('../utils/backupUtils').runBackup()).filename }),
  },
  {
    name: 'temp-cleanup', label: 'Delete temp files older than 7 days', cron: '0 3 * * *', env: null,
    run: async () => require('../utils/cleanupTemp').cleanupTempFiles(),
  },
];

const flag = (name, def = true) => {
  const v = process.env[name];
  return v === undefined || v === '' ? def : String(v).toLowerCase() !== 'false';
};

/** Whether a job is switched on (JOBS_ENABLED and its own flag). */
function isEnabled(job) {
  return flag('JOBS_ENABLED') && (!job.env || flag(job.env));
}

/** A job's cron schedule, with any JOB_SCHEDULE_<NAME> override (checked). */
function cronFor(job) {
  const override = process.env[`JOB_SCHEDULE_${job.name.toUpperCase().replace(/-/g, '_')}`];
  if (override) {
    try {
      cronParser.parseExpression(override, { tz: TIMEZONE });
      return override;
    } catch {
      console.warn(`[jobs] Ignoring invalid schedule for ${job.name}: "${override}"`);
    }
  }
  return job.cron;
}

/** When a cron schedule next fires after `from`. */
function nextRun(cron, from = new Date()) {
  return cronParser.parseExpression(cron, { tz: TIMEZONE, currentDate: from }).next().toDate();
}

function summarize(result) {
  try {
    const s = JSON.stringify(result ?? null);
    return s.length > 300 ? `${s.slice(0, 300)}…` : s;
  } catch {
    return '';
  }
}

/** True when a job's result says nothing happened (all zeros/false/empty). */
function isNoop(value) {
  if (value === null || value === undefined || value === 0 || value === false || value === '') return true;
  if (Array.isArray(value)) return value.length === 0;
  if (typeof value === 'object') return Object.values(value).every(isNoop);
  return false;
}

async function runJob(job) {
  const started = Date.now();
  const result = await job.run();
  // Quiet when nothing happened, like the old timers.
  if (!isNoop(result)) console.log(`[jobs] ${job.name} done in ${Date.now() - started} ms: ${summarize(result)}`);
  return result ?? null;
}

// ─── Runner ───────────────────────────────────────────────────────────────────

let boss = null;
let mode = 'off'; // 'queue' | 'timers' | 'off'
const timers = new Map();
// Timer mode keeps its own short history so the admin page still has something to show.
const timerHistory = new Map();

function startTimers(jobs) {
  mode = 'timers';
  for (const job of jobs) {
    const cron = cronFor(job);
    const scheduleNext = () => {
      const delay = Math.max(1000, nextRun(cron).getTime() - Date.now());
      timers.set(job.name, setTimeout(async () => {
        const entry = { state: 'active', startedOn: new Date(), completedOn: null, output: null };
        try {
          entry.output = await runJob(job);
          entry.state = 'completed';
        } catch (e) {
          entry.state = 'failed';
          entry.output = { message: e?.message || String(e) };
          console.warn(`[jobs] ${job.name} failed:`, e?.message || e);
        } finally {
          entry.completedOn = new Date();
          timerHistory.set(job.name, [entry, ...(timerHistory.get(job.name) || [])].slice(0, 10));
          scheduleNext();
        }
      }, Math.min(delay, 2 ** 31 - 1)));
    };
    scheduleNext();
  }
  console.log(`[jobs] Running ${jobs.length} job(s) on in-process timers.`);
}

/**
 * Starts the background jobs. Call once after the server is listening.
 * @returns {Promise<'queue'|'timers'|'off'>}
 */
async function startJobs() {
  const enabled = JOBS.filter(isEnabled);
  if (!flag('JOBS_ENABLED')) {
    console.log('[jobs] All background jobs disabled by env JOBS_ENABLED=false');
    mode = 'off';
    return mode;
  }
  try {
    const PgBoss = require('pg-boss');
    boss = new PgBoss({ connectionString: process.env.DATABASE_URL, schema: SCHEMA, application_name: 'hope-events-jobs' });
    boss.on('error', (e) => console.warn('[jobs] Queue error:', e?.message || e));
    await boss.start();
    const existing = new Set((await boss.getQueues()).map(q => q.name));
    for (const job of JOBS) {
      if (!existing.has(job.name)) {
        // stately: at most one waiting and one running, so a slow run never piles up.
        await boss.createQueue(job.name, { policy: 'stately', retryLimit: 2, retryDelay: 60, expireInSeconds: 60 * 60 });
      }
      if (isEnabled(job)) {
        await boss.schedule(job.name, cronFor(job), null, { tz: TIMEZONE });
        await boss.work(job.name, async () => {
          try {
            return await runJob(job);
          } catch (e) {
            console.warn(`[jobs] ${job.name} failed:`, e?.message || e);
            throw e; // recorded as failed, and retried
          }
        });
      } else {
        await boss.unschedule(job.name).catch(() => {});
      }
    }
    mode = 'queue';
    const off = JOBS.filter(j => !isEnabled(j)).map(j => j.name);
    console.log(`[jobs] Job queue started: ${enabled.map(j => j.name).join(', ')}${off.length ? ` (off: ${off.join(', ')})` : ''}.`);
  } catch (e) {
    console.warn('[jobs] Job queue unavailable, falling back to timers:', e?.message || e);
    if (boss) { try { await boss.stop({ graceful: false }); } catch { /* ignore */ } }
    boss = null;
    startTimers(enabled);
  }
  return mode;
}

async function stopJobs() {
  for (const t of timers.values()) clearTimeout(t);
  timers.clear();
  if (boss) {
    await boss.stop({ graceful: true, timeout: 10000 }).catch(() => {});
    boss = null;
  }
  mode = 'off';
}

/** Runs a job now (admin "Run now"). Queue mode: queued for a worker; timer mode: inline. */
async function runNow(name) {
  const job = JOBS.find(j => j.name === name);
  if (!job) {
    const err = new Error('Unknown job');
    err.statusCode = 404;
    throw err;
  }
  if (mode === 'queue' && isEnabled(job)) {
    const id = await boss.send(job.name, null, { singletonKey: `${job.name}-manual` });
    return { queued: true, id };
  }
  const output = await runJob(job);
  const entry = { state: 'completed', startedOn: new Date(), completedOn: new Date(), output };
  timerHistory.set(job.name, [entry, ...(timerHistory.get(job.name) || [])].slice(0, 10));
  return { queued: false, output };
}

/** Each job with its schedule, next run and recent runs, for the admin page. */
async function jobStatus(prisma) {
  const recent = new Map();
  if (mode === 'queue') {
    const rows = await prisma.$queryRawUnsafe(`
      SELECT name, state, created_on AS "createdOn", started_on AS "startedOn", completed_on AS "completedOn", output
      FROM (
        SELECT name, state, created_on, started_on, completed_on, output,
               row_number() OVER (PARTITION BY name ORDER BY created_on DESC) AS rn
        FROM (
          SELECT name, state::text, created_on, started_on, completed_on, output FROM ${SCHEMA}.job
          UNION ALL
          SELECT name, state::text, created_on, started_on, completed_on, output FROM ${SCHEMA}.archive
        ) all_jobs
        WHERE name = ANY($1::text[])
      ) ranked
      WHERE rn <= 10
      ORDER BY "createdOn" DESC`, JOBS.map(j => j.name));
    for (const r of rows) recent.set(r.name, [...(recent.get(r.name) || []), r]);
  } else {
    for (const [name, list] of timerHistory) recent.set(name, list);
  }
  return {
    mode,
    timezone: TIMEZONE,
    jobs: JOBS.map(job => {
      const cron = cronFor(job);
      const enabled = isEnabled(job);
      return {
        name: job.name, label: job.label, cron, enabled, envFlag: job.env,
        nextRun: enabled && mode !== 'off' ? nextRun(cron) : null,
        recent: recent.get(job.name) || [],
      };
    }),
  };
}

module.exports = { JOBS, TIMEZONE, isEnabled, cronFor, nextRun, isNoop, startJobs, stopJobs, runNow, jobStatus };