예약 작업과 큐 운영

import { IsDateString, IsInt, IsString, Max, Min } from 'class-validator';

export class ScheduleReportDto {
  @IsString()
  reportId: string;

  @IsString()
  requestedBy: string;

  @IsInt()
  @Min(1)
  @Max(100_000)
  rows: number;

  @IsDateString()
  runAt: string;
}

import { ScheduleReportDto } from './dto/schedule-report.dto';

async enqueueAt(dto: ScheduleReportDto) {
  const runAt = new Date(dto.runAt).getTime();
  const delay = runAt - Date.now();

  if (delay <= 0) {
    throw new BadRequestException('runAt must be in the future');
  }

  const job = await this.reportsQueue.add(
    GENERATE_REPORT_JOB,
    {
      reportId: dto.reportId,
      requestedBy: dto.requestedBy,
      rows: dto.rows,
    },
    {
      delay,
      attempts: 4,
      backoff: { type: 'exponential', delay: 1_000, jitter: 0.5 },
      deduplication: { id: dto.reportId },
      removeOnComplete: 100,
      removeOnFail: 100,
    },
  );

  return {
    jobId: job.id,
    state: await job.getState(),
    runAt: new Date(runAt).toISOString(),
  };
}

import { ScheduleReportDto } from './dto/schedule-report.dto';

@Post('scheduled')
@HttpCode(HttpStatus.ACCEPTED)
schedule(@Body() dto: ScheduleReportDto) {
  return this.reportsService.enqueueAt(dto);
}

node -e "const runAt=new Date(Date.now()+60000).toISOString(); fetch('http://localhost:3000/reports/scheduled',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({reportId:'scheduled-demo',requestedBy:'user-42',rows:3000,runAt})}).then(async r=>console.log(r.status,await r.text()))"

export const REPORT_QUEUE = 'reports';
export const GENERATE_REPORT_JOB = 'report.generate';
export const CLEANUP_REPORTS_JOB = 'reports.cleanup';

export interface CleanupReportsData {
  keepDays: number;
}

export interface CleanupReportsResult {
  removed: number;
}

import { InjectQueue } from '@nestjs/bullmq';
import { Injectable, OnApplicationBootstrap } from '@nestjs/common';
import { Queue } from 'bullmq';
import {
  CLEANUP_REPORTS_JOB,
  REPORT_QUEUE,
} from './reports.constants';

@Injectable()
export class ReportSchedulerService implements OnApplicationBootstrap {
  constructor(
    @InjectQueue(REPORT_QUEUE)
    private readonly reportsQueue: Queue,
  ) {}

  async onApplicationBootstrap() {
    await this.reportsQueue.upsertJobScheduler(
      'daily-report-cleanup',
      {
        pattern: '0 0 3 * * *',
        tz: 'Asia/Seoul',
      },
      {
        name: CLEANUP_REPORTS_JOB,
        data: {
          keepDays: 30,
        },
        opts: {
          attempts: 3,
          backoff: { type: 'exponential', delay: 5_000 },
          removeOnComplete: 30,
          removeOnFail: 30,
        },
      },
    );
  }
}

getSchedulers() {
  return this.reportsQueue.getJobSchedulers(0, 20, true);
}

removeDailyCleanup() {
  return this.reportsQueue.removeJobScheduler('daily-report-cleanup');
}

import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job, UnrecoverableError } from 'bullmq';
import { ReportArtifactsService } from './report-artifacts.service';
import {
  CleanupReportsData,
  CleanupReportsResult,
  GenerateReportData,
  GenerateReportResult,
} from './report-job.types';
import {
  CLEANUP_REPORTS_JOB,
  GENERATE_REPORT_JOB,
  REPORT_QUEUE,
} from './reports.constants';

const wait = (milliseconds: number) =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

@Processor(REPORT_QUEUE, { concurrency: 3 })
export class ReportsProcessor extends WorkerHost {
  private readonly logger = new Logger(ReportsProcessor.name);

  constructor(private readonly artifacts: ReportArtifactsService) {
    super();
  }

  async process(job: Job): Promise<unknown> {
    switch (job.name) {
      case GENERATE_REPORT_JOB:
        return this.generateReport(
          job as Job<GenerateReportData, GenerateReportResult>,
        );
      case CLEANUP_REPORTS_JOB:
        return this.cleanupReports(
          job as Job<CleanupReportsData, CleanupReportsResult>,
        );
      default:
        throw new UnrecoverableError(`unsupported report job: ${job.name}`);
    }
  }

  @OnWorkerEvent('active')
  onActive(job: Job) {
    this.logger.log(`report job active: ${job.id}`);
  }

  @OnWorkerEvent('failed')
  onFailed(job: Job | undefined, error: Error) {
    this.logger.error(`report job failed: ${job?.id ?? 'unknown'} ${error.message}`);
  }

  private async generateReport(
    job: Job<GenerateReportData, GenerateReportResult>,
  ): Promise<GenerateReportResult> {
    if (job.data.rows < 1) {
      throw new UnrecoverableError('rows must be greater than zero');
    }

    const existing = this.artifacts.find(job.data.reportId);
    if (existing) return { ...existing, reused: true };

    if (job.attemptsMade < (job.data.failUntilAttempt ?? 0)) {
      throw new Error(`temporary storage failure on attempt ${job.attemptsMade + 1}`);
    }

    await job.updateProgress({ step: 'load', percent: 20 });
    await wait(400);
    await job.updateProgress({ step: 'render', percent: 70 });
    await wait(400);
    await job.updateProgress({ step: 'upload', percent: 95 });
    await wait(400);
    await job.updateProgress({ step: 'upload', percent: 100 });

    return this.artifacts.saveOnce({
      reportId: job.data.reportId,
      objectKey: `reports/${job.data.reportId}.csv`,
      generatedRows: job.data.rows,
      reused: false,
    });
  }

  private async cleanupReports(
    job: Job<CleanupReportsData, CleanupReportsResult>,
  ): Promise<CleanupReportsResult> {
    const removed = await this.artifacts.removeOlderThan(job.data.keepDays);
    return { removed };
  }
}

async removeOlderThan(_keepDays: number) {
  return 0;
}

import { ReportArtifactsService } from './report-artifacts.service';
import { ReportSchedulerService } from './report-scheduler.service';

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
  ReportSchedulerService,
],

import {
  OnQueueEvent,
  QueueEventsHost,
  QueueEventsListener,
} from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { REPORT_QUEUE } from './reports.constants';

@QueueEventsListener(REPORT_QUEUE)
export class ReportQueueEventsListener extends QueueEventsHost {
  private readonly logger = new Logger(ReportQueueEventsListener.name);

  @OnQueueEvent('completed')
  onCompleted({ jobId }: { jobId: string }) {
    this.logger.log(`report job completed: ${jobId}`);
  }

  @OnQueueEvent('failed')
  onFailed({ jobId, failedReason }: { jobId: string; failedReason: string }) {
    this.logger.error(`report job failed: ${jobId} ${failedReason}`);
  }

  @OnQueueEvent('deduplicated')
  onDeduplicated({
    jobId,
    deduplicatedJobId,
  }: {
    jobId: string;
    deduplicatedJobId: string;
  }) {
    this.logger.warn(`duplicate ${deduplicatedJobId} kept ${jobId}`);
  }
}

import { ReportQueueEventsListener } from './report-queue-events.listener';

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
  ReportSchedulerService,
  ReportQueueEventsListener,
],

async getQueueHealth() {
  return this.reportsQueue.getJobCounts(
    'waiting',
    'active',
    'delayed',
    'completed',
    'failed',
  );
}

@Get('queue/health')
getQueueHealth() {
  return this.reportsService.getQueueHealth();
}

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.enableShutdownHooks();
  app.useGlobalPipes(new ValidationPipe({ whitelist: true, transform: true }));
  await app.listen(process.env.PORT ?? 3000);
}