Scheduled Tasks

// src/health/queue-depth.monitor.ts
import { Injectable, Logger } from '@nestjs/common';
import { Cron, CronExpression, Interval, Timeout } from '@nestjs/schedule';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class QueueDepthMonitor {
  private readonly logger = new Logger(QueueDepthMonitor.name);

  constructor(@InjectQueue('reports') private readonly reports: Queue) {}

  // Once, 10 seconds after bootstrap. Not at a wall-clock time.
  @Timeout('warm-metrics', 10_000)
  async warmMetrics(): Promise<void> {
    await this.reportDepth();
  }

  // Every 30 seconds, counted from bootstrap. setInterval under the hood,
  // so it drifts relative to the clock; @Cron does not.
  @Interval('queue-depth', 30_000)
  async reportDepth(): Promise<void> {
    const counts = await this.reports.getJobCounts('waiting', 'active', 'failed');
    this.logger.log(`reports queue: ${JSON.stringify(counts)}`);
  }

  // CronExpression.EVERY_5_MINUTES is shorthand for the string '0 */5 * * * *'.
  // Both are the same argument; the enum just avoids a typo.
  @Cron(CronExpression.EVERY_5_MINUTES, {
    name: 'stale-sweep',
    waitForCompletion: true, // a tick during a running sweep is skipped, not queued
    disabled: process.env.NODE_ENV === 'test', // registered, never fired
  })
  async sweepStale(): Promise<void> {
    // grace is in ms: remove failed jobs older than 30 days, 1000 at a time.
    await this.reports.clean(30 * 24 * 60 * 60 * 1000, 1000, 'failed');
  }
}

// src/reports/report-schedule.service.ts
import { Injectable, Logger, NotFoundException } from '@nestjs/common';
import { SchedulerRegistry } from '@nestjs/schedule';
import { CronJob } from 'cron'; // the package @nestjs/schedule is built on

@Injectable()
export class ReportScheduleService {
  private readonly logger = new Logger(ReportScheduleService.name);

  constructor(private readonly schedulerRegistry: SchedulerRegistry) {}

  // The decorators only handle schedules known at compile time. Customer-chosen
  // ones have to be built by hand and registered under a name.
  addDailyExport(customerId: string, hour: number): void {
    const name = `export-${customerId}`;

    if (this.schedulerRegistry.doesExist('cron', name)) {
      this.schedulerRegistry.deleteCronJob(name); // removes it from the registry
    }

    const job = new CronJob(`0 0 ${hour} * * *`, () => {
      this.logger.log(`exporting for ${customerId}`);
    });

    this.schedulerRegistry.addCronJob(name, job);
    job.start(); // addCronJob registers it; it does not start it for you
  }

  pause(name: string): void {
    // Throws if nothing is registered under that name.
    this.schedulerRegistry.getCronJob(name).stop();
  }

  listNextRuns(): { name: string; next: string }[] {
    // getCronJobs returns a Map keyed by name; intervals and timeouts are
    // listed as plain string[] by getIntervals() and getTimeouts().
    return [...this.schedulerRegistry.getCronJobs().entries()].map(([name, job]) => ({
      name,
      next: job.nextDate().toISO() ?? 'unknown', // a Luxon DateTime
    }));
  }
}

// src/app.module.ts
import { Module, ModuleMetadata } from '@nestjs/common';
import { ScheduleModule } from '@nestjs/schedule';
import { BullModule } from '@nestjs/bullmq';
import { ReportsModule } from './reports/reports.module';
import { SchedulerModule } from './scheduler/scheduler.module';

// Deploy the same image twice: APP_ROLE=api with several replicas, and
// APP_ROLE=scheduler with exactly one. Only the second registers timers, so
// each cron pattern fires once no matter how far the API scales.
const schedulerImports: NonNullable<ModuleMetadata['imports']> =
  process.env.APP_ROLE === 'scheduler'
    ? [ScheduleModule.forRoot(), SchedulerModule]
    : [];

@Module({
  imports: [
    BullModule.forRoot({
      connection: { host: process.env.REDIS_HOST ?? '127.0.0.1', port: 6379 },
    }),
    ...schedulerImports,
    ReportsModule, // queue producers and processors run in every role
  ],
})
export class AppModule {}