Producer, Worker와 작업 상태

export interface GenerateReportData {
  reportId: string;
  tenantId: string;
  requestedBy: string;
  rows: number;
}

export interface GenerateReportResult {
  reportId: string;
  objectKey: string;
  generatedRows: number;
}

export interface ReportProgress {
  step: 'load' | 'render' | 'upload';
  percent: number;
}

export interface ReportPrincipal {
  tenantId: string;
  userId: string;
}

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

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

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

import { ValidationPipe } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';

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

void bootstrap();

import { InjectQueue } from '@nestjs/bullmq';
import { Injectable, NotFoundException } from '@nestjs/common';
import { Queue } from 'bullmq';
import { CreateReportDto } from './dto/create-report.dto';
import {
  GenerateReportData,
  GenerateReportResult,
  ReportPrincipal,
} from './report-job.types';
import { GENERATE_REPORT_JOB, REPORT_QUEUE } from './reports.constants';

@Injectable()
export class ReportsService {
  constructor(
    @InjectQueue(REPORT_QUEUE)
    private readonly reportsQueue: Queue<
      GenerateReportData,
      GenerateReportResult,
      typeof GENERATE_REPORT_JOB
    >,
  ) {}

  async enqueue(dto: CreateReportDto, principal: ReportPrincipal) {
    const job = await this.reportsQueue.add(
      GENERATE_REPORT_JOB,
      {
        reportId: dto.reportId,
        tenantId: principal.tenantId,
        requestedBy: principal.userId,
        rows: dto.rows,
      },
      {
        attempts: 3,
        backoff: { type: 'exponential', delay: 1_000 },
        removeOnComplete: 100,
        removeOnFail: 100,
      },
    );

    return {
      jobId: job.id,
      accepted: true,
      statusUrl: `/reports/jobs/${job.id}`,
    };
  }

  async getStatus(jobId: string, principal: ReportPrincipal) {
    const job = await this.reportsQueue.getJob(jobId);

    if (
      !job ||
      job.data.tenantId !== principal.tenantId ||
      job.data.requestedBy !== principal.userId
    ) {
      throw new NotFoundException('report job not found');
    }

    const state = await job.getState();
    if (state === 'unknown') {
      throw new NotFoundException('report job not found');
    }

    return {
      jobId: job.id,
      name: job.name,
      state,
      progress: job.progress,
      attemptsMade: job.attemptsMade,
      result: job.returnvalue ?? null,
      failedReason: job.failedReason || null,
    };
  }
}

import {
  Body,
  Controller,
  Get,
  HttpCode,
  HttpStatus,
  Param,
  Post,
  Req,
} from '@nestjs/common';
import { CreateReportDto } from './dto/create-report.dto';
import { ReportPrincipal } from './report-job.types';
import { ReportsService } from './reports.service';

interface AuthenticatedRequest {
  user: ReportPrincipal;
}

@Controller('reports')
export class ReportsController {
  constructor(private readonly reportsService: ReportsService) {}

  @Post()
  @HttpCode(HttpStatus.ACCEPTED)
  create(@Body() dto: CreateReportDto, @Req() request: AuthenticatedRequest) {
    return this.reportsService.enqueue(dto, request.user);
  }

  @Get('jobs/:jobId')
  getStatus(
    @Param('jobId') jobId: string,
    @Req() request: AuthenticatedRequest,
  ) {
    return this.reportsService.getStatus(jobId, request.user);
  }
}

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import {
  GenerateReportData,
  GenerateReportResult,
} from './report-job.types';
import { GENERATE_REPORT_JOB, REPORT_QUEUE } from './reports.constants';

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

@Processor(REPORT_QUEUE)
export class ReportsProcessor extends WorkerHost {
  async process(
    job: Job<
      GenerateReportData,
      GenerateReportResult,
      typeof GENERATE_REPORT_JOB
    >,
  ): Promise<GenerateReportResult> {
    if (job.name !== GENERATE_REPORT_JOB) {
      throw new Error(`unsupported report job: ${job.name}`);
    }

    await job.updateProgress({ step: 'load', percent: 20 });
    await wait(700);

    await job.updateProgress({ step: 'render', percent: 70 });
    await wait(700);

    await job.updateProgress({ step: 'upload', percent: 95 });
    await wait(700);
    await job.updateProgress({ step: 'upload', percent: 100 });

    return {
      reportId: job.data.reportId,
      objectKey: `reports/${job.data.reportId}.csv`,
      generatedRows: job.data.rows,
    };
  }
}

import { BullModule } from '@nestjs/bullmq';
import { Module } from '@nestjs/common';
import { ReportsController } from './reports.controller';
import { ReportsProcessor } from './reports.processor';
import { ReportsService } from './reports.service';
import { REPORT_QUEUE } from './reports.constants';

@Module({
  imports: [BullModule.registerQueue({ name: REPORT_QUEUE })],
  controllers: [ReportsController],
  providers: [ReportsService, ReportsProcessor],
})
export class ReportsModule {}

curl -i -X POST http://localhost:3000/reports \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <access-token>" \
  -d '{"reportId":"sales-2026-07","rows":5000}'

{
  "jobId": "1",
  "accepted": true,
  "statusUrl": "/reports/jobs/1"
}

curl -H "Authorization: Bearer <access-token>" \
  http://localhost:3000/reports/jobs/1

{
  "jobId": "1",
  "name": "report.generate",
  "state": "completed",
  "progress": { "step": "upload", "percent": 100 },
  "attemptsMade": 1,
  "result": {
    "reportId": "sales-2026-07",
    "objectKey": "reports/sales-2026-07.csv",
    "generatedRows": 5000
  },
  "failedReason": null
}