재시도, 중복 방지와 실패 복구

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

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

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

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

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

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

  @IsOptional()
  @IsInt()
  @Min(0)
  @Max(5)
  failUntilAttempt?: number;
}

import { createHash } from 'node:crypto';
import type { GenerateReportData } from './report-job.types';

export function reportOperationKey(
  data: Pick<
    GenerateReportData,
    'tenantId' | 'requestedBy' | 'reportId' | 'rows'
  >,
) {
  return createHash('sha256')
    .update(
      JSON.stringify([
        'report.generate.v1',
        data.tenantId,
        data.requestedBy,
        data.reportId,
        data.rows,
      ]),
    )
    .digest('hex');
}

const data: GenerateReportData = {
  reportId: dto.reportId,
  tenantId: principal.tenantId,
  requestedBy: principal.userId,
  rows: dto.rows,
  failUntilAttempt: dto.failUntilAttempt,
};

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

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

import { Injectable } from '@nestjs/common';
import { GenerateReportResult } from './report-job.types';

@Injectable()
export class ReportArtifactsService {
  private readonly artifacts = new Map<string, GenerateReportResult>();

  find(operationKey: string) {
    return this.artifacts.get(operationKey);
  }

  saveOnce(operationKey: string, result: GenerateReportResult) {
    const existing = this.artifacts.get(operationKey);
    if (existing) return existing;

    this.artifacts.set(operationKey, result);
    return result;
  }
}

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job, UnrecoverableError } from 'bullmq';
import { ReportArtifactsService } from './report-artifacts.service';
import {
  GenerateReportData,
  GenerateReportResult,
} from './report-job.types';
import { reportOperationKey } from './report-operation-key';
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 {
  constructor(private readonly artifacts: ReportArtifactsService) {
    super();
  }

  async process(
    job: Job<
      GenerateReportData,
      GenerateReportResult,
      typeof GENERATE_REPORT_JOB
    >,
  ): Promise<GenerateReportResult> {
    if (job.name !== GENERATE_REPORT_JOB) {
      throw new UnrecoverableError(
        'unsupported report job: ' + job.name,
      );
    }

    if (job.data.rows < 1) {
      throw new UnrecoverableError('rows must be greater than zero');
    }

    const operationKey = reportOperationKey(job.data);
    const existing = this.artifacts.find(operationKey);
    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(operationKey, {
      operationKey,
      reportId: job.data.reportId,
      objectKey:
        'reports/' + job.data.tenantId + '/' + operationKey + '.csv',
      generatedRows: job.data.rows,
      reused: false,
    });
  }
}

providers: [
  ReportsService,
  ReportsProcessor,
  ReportArtifactsService,
],

curl -i -X POST http://localhost:3000/reports \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <access-token>" \
  -d '{"reportId":"retry-demo","rows":1000,"failUntilAttempt":2}'

JOB_ID='POST 응답의 jobId'
curl -H "Authorization: Bearer <access-token>" \
  "http://localhost:3000/reports/jobs/$JOB_ID"

import { IsString, MinLength } from 'class-validator';

export class RetryReportJobDto {
  @IsString()
  @MinLength(10)
  reason: string;
}

async retry(
  jobId: string,
  reason: 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 !== 'failed') {
    throw new BadRequestException(
      'only failed jobs can be retried: ' + state,
    );
  }

  const operationKey = reportOperationKey(job.data);
  const liveJobId =
    await this.reportsQueue.getDeduplicationJobId(operationKey);

  if (liveJobId && liveJobId !== job.id) {
    throw new ConflictException(
      'another report job is already in progress',
    );
  }

  if (this.artifacts.find(operationKey)) {
    throw new ConflictException('report result already exists');
  }

  await this.recoveryAudit.append({
    operationKey,
    jobId: job.id,
    tenantId: job.data.tenantId,
    requestedBy: job.data.requestedBy,
    reportId: job.data.reportId,
    rows: job.data.rows,
    failedReason: job.failedReason,
    stacktrace: job.stacktrace ?? [],
    attemptsMade: job.attemptsMade,
    attemptsStarted: job.attemptsStarted,
    retriedBy: principal.userId,
    reason,
    recordedAt: new Date().toISOString(),
  });

  await job.retry('failed', {
    resetAttemptsMade: true,
    resetAttemptsStarted: true,
  });

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

@UseGuards(ReportRetryGuard)
@Post('jobs/:jobId/retry')
@HttpCode(HttpStatus.ACCEPTED)
retry(
  @Param('jobId') jobId: string,
  @Body() dto: RetryReportJobDto,
  @Req() request: AuthenticatedRequest,
) {
  return this.reportsService.retry(
    jobId,
    dto.reason,
    request.user,
  );
}