Xây nền tảng media với NestJS — Part 3: child_process, worker_threads, cluster & realtime

import { Controller, Get } from '@nestjs/common';

@Controller()
export class BlockController {
  @Get('block')
  block(): string {
    const end = Date.now() + 5000;
    while (Date.now() < end) {
      // Nam giay tinh toan thuan tuy — dai dien cho bat ky viec nang CPU nao.
    }
    return 'xong';
  }

  @Get('ping')
  ping(): string {
    return 'pong';
  }
}

# Tab 1 — chay nen roi doi mot chut cho no kip khoi dong
( curl -s -o /dev/null -w 'block: %{time_total}s\n' http://localhost:3000/block & ); sleep 0.4

# Tab 2 — goi /ping trong luc /block dang chay
curl -s -o /dev/null -w 'ping:  %{time_total}s\n' http://localhost:3000/ping
# -> block: 5.005s
# -> ping:  4.600s      <-- mot ham `return 'pong'` ma mat 4,6 giay

# Doi chung: goi lai /ping luc event loop ranh
curl -s -o /dev/null -w 'ping:  %{time_total}s\n' http://localhost:3000/ping
# -> ping:  0.001s

brew install ffmpeg              # macOS
sudo apt install -y ffmpeg       # Ubuntu/Debian

ffmpeg -version                  # phai in ra so phien ban, khong phai "command not found"

exec    chay qua shell, gom TOAN BO output vao bo nho roi moi tra ve
        -> ffmpeg in hang nghin dong tien do => phinh bo nho
        -> chuoi lenh di qua shell => lo hong chen lenh

execFile khong qua shell (tot hon), nhung van gom het output

fork    tao mot tien trinh NODE moi, co san kenh IPC
        -> dung cho worker Node, khong dung cho ffmpeg

spawn   tra ve STREAM stdout/stderr, khong gom vao bo nho
        -> dung cho tien trinh chay lau va noi nhieu. Chinh la ffmpeg.

import { spawn } from 'node:child_process';
import { createInterface } from 'node:readline';

export interface TranscodeOptions {
  readonly input: string;
  readonly output: string;
  readonly height: 480 | 720 | 1080;
  readonly durationSeconds: number;
  readonly onProgress: (percent: number) => void;
  readonly signal: AbortSignal;
}

export async function transcode(options: TranscodeOptions): Promise<void> {
  const args = [
    '-hide_banner',
    '-nostdin', // KHONG doi input tu ban phim. Thieu co nay ffmpeg co the treo im lang.
    '-i',
    options.input,
    '-vf',
    `scale=-2:${options.height}`, // -2 = tu tinh chieu rong sao cho chia het cho 2
    '-c:v',
    'libx264',
    '-preset',
    'veryfast',
    '-crf',
    '23',
    '-c:a',
    'aac',
    '-movflags',
    '+faststart', // day metadata len dau file => trinh duyet phat duoc ngay
    '-progress',
    'pipe:2',
    '-y',
    options.output,
  ];

  const child = spawn('ffmpeg', args, { signal: options.signal });

  // spawn bao loi KHOI DONG qua event 'error', KHONG qua ma thoat. Khong bat o
  // day thi thieu ffmpeg nem ra mot loi khong ai bat va giet ca worker, thay vi
  // hong mot job roi duoc thu lai nhu muc 4.1.
  const failed = new Promise<never>((_, reject) => {
    child.on('error', reject);
  });

  // Doc theo DONG. Doc theo chunk thi mot dong co the bi cat lam doi.
  const lines = createInterface({ input: child.stderr });

  const readProgress = (async () => {
    for await (const line of lines) {
      // ffmpeg in "out_time_ms=12345678" — don vi thuc te la MICRO giay,
      // bat ke ten bien noi gi.
      const match = /^out_time_ms=(\d+)$/.exec(line);
      if (match?.[1] !== undefined) {
        const seconds = Number(match[1]) / 1_000_000;
        const percent = Math.min(99, Math.round((seconds / options.durationSeconds) * 100));
        options.onProgress(percent);
      }
    }
  })();

  const closed = new Promise<number>((resolve) => child.on('close', resolve));

  // Chay dua voi `failed`: hong ngay luc khoi dong thi bat duoc lien, khong treo.
  const code = await Promise.race([
    (async () => {
      await readProgress;
      return closed;
    })(),
    failed,
  ]);

  if (code !== 0) {
    throw new Error(`ffmpeg thoat voi ma ${code}`);
  }
}

const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), 30 * 60 * 1000);

try {
  await transcode({ ...options, signal: controller.signal });
} finally {
  clearTimeout(timer); // Quen dong nay thi tien trinh song them 30 phut vo ich.
}

// Leo thang hai buoc. `signal` chi gui SIGTERM — xin dung, cho don dep.
options.signal.addEventListener('abort', () => {
  const force = setTimeout(() => {
    // Van con song sau 10 giay -> SIGKILL. Khong the bat, khong the bo qua.
    if (child.exitCode === null) child.kill('SIGKILL');
  }, 10_000);

  child.on('close', () => clearTimeout(force));
});

# Trong khi mot job dang chuyen ma, dem so tien trinh ffmpeg:
ps -ax -o command | grep -c '[f]fmpeg -hide_banner'
# -> 1        (dang chay)

# ...goi controller.abort(), roi doi vai giay va dem lai:
ps -ax -o command | grep -c '[f]fmpeg -hide_banner'
# -> 0        (khong con tien trinh mo coi nao)

import { parentPort } from 'node:worker_threads';
import { createHash } from 'node:crypto';
import { createReadStream } from 'node:fs';

interface HashTask {
  readonly filePath: string;
}

// NGHE tung viec mot, thay vi doc workerData. Ho worker o duoi tai su dung
// mot worker cho nhieu viec, ma workerData thi chi dat duoc DUNG MOT LAN
// luc tao Worker — dung no thi worker chi lam duoc dung mot viec roi thanh
// vo dung.
parentPort?.on('message', (task: HashTask) => {
  const hash = createHash('sha256');

  createReadStream(task.filePath)
    .on('data', (chunk) => hash.update(chunk))
    .on('end', () => parentPort?.postMessage({ digest: hash.digest('hex') }))
    .on('error', (error: Error) => parentPort?.postMessage({ error: error.message }));
});

import { Worker } from 'node:worker_threads';
import { availableParallelism } from 'node:os';

interface Pending<T> {
  readonly data: unknown;
  readonly resolve: (value: T) => void;
  readonly reject: (error: Error) => void;
}

export class WorkerPool<T> {
  private readonly idle: Worker[] = [];
  private readonly queue: Pending<T>[] = [];
  private spawned = 0;

  // availableParallelism ton trong gioi han CPU cua container.
  // os.cpus().length dem nhan cua CA MAY CHU — trong Docker la con so sai.
  constructor(
    private readonly script: string,
    private readonly size = availableParallelism(),
  ) {}

  private create(): Worker {
    this.spawned += 1;
    return new Worker(this.script);
  }

  run(data: unknown): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      this.queue.push({ data, resolve, reject });
      this.pump();
    });
  }

  private pump(): void {
    const job = this.queue.shift();
    if (job === undefined) return;

    const worker = this.idle.pop() ?? (this.spawned < this.size ? this.create() : undefined);
    if (worker === undefined) {
      this.queue.unshift(job); // Het worker ranh — xep lai hang, cho luot sau.
      return;
    }

    worker.once('message', (result: { digest?: string; error?: string }) => {
      if (result.error !== undefined) job.reject(new Error(result.error));
      else job.resolve(result as T);

      this.idle.push(worker);
      this.pump(); // Tra worker ve ho roi keo viec tiep theo.
    });

    worker.postMessage(job.data);
  }

  // Moi Worker la mot handle giu event loop song. Khong co ham nay thi tien
  // trinh KHONG BAO GIO tu thoat sau khi lam xong viec — xem muc 4.4, noi
  // worker phai dung tu te trong gioi han thoi gian cua Docker.
  async destroy(): Promise<void> {
    const all = [...this.idle];
    this.idle.length = 0;
    await Promise.all(all.map((worker) => worker.terminate()));
    this.spawned -= all.length;
  }
}

                    child_process           worker_threads
Bo nho              rieng hoan toan         chung, chia se duoc buffer
Khoi dong           30-50ms                 3-5ms
Truyen du lieu      serialize qua IPC       SharedArrayBuffer: 0 chi phi
                                            postMessage: van phai copy
Su co               mot tien trinh chet     mot luong chet co the keo
                    khong anh huong ai      ca tien trinh theo
Chay duoc gi        BAT KY chuong trinh nao chi JavaScript
Dung khi            goi ffmpeg, imagemagick hash, nen anh, tinh toan
                    bat cu binary nao       thuan JS trong Node

import { Controller, Get, HttpCode, HttpStatus, Param, Post, UseGuards } from '@nestjs/common';
import { JwtAuthGuard } from '../auth/jwt-auth.guard';
import { RateLimitGuard } from '../common/rate-limit/rate-limit.guard';
import { CurrentUser } from '../auth/current-user.decorator';
import type { JwtPayload } from '../auth/jwt-payload';
import { JobService, type JobState } from './job.service';

@Controller()
@UseGuards(JwtAuthGuard, RateLimitGuard)
export class JobController {
  constructor(private readonly jobs: JobService) {}

  @Post('media/:videoId/transcode')
  @HttpCode(HttpStatus.ACCEPTED) // 202: da nhan, chua lam xong
  async transcode(
    @Param('videoId') videoId: string,
    @CurrentUser() user: JwtPayload,
  ): Promise<{ jobId: string }> {
    const job = await this.jobs.enqueue(videoId, user.sub);
    return { jobId: job.id };
  }

  // Nguon su that goi lai luc connect/reconnect — xem muc 6-7.
  @Get('jobs/active')
  async active(@CurrentUser() user: JwtPayload): Promise<JobState[]> {
    return this.jobs.findActiveForUser(user.sub);
  }
}

import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { Job } from './job.entity';
import { JobQueue } from './job.queue';

export interface JobState {
  readonly id: string;
  readonly videoTitle: string;
  readonly status: 'queued' | 'processing' | 'completed' | 'failed';
  readonly percent: number;
}

@Injectable()
export class JobService {
  constructor(
    @InjectRepository(Job) private readonly jobs: Repository<Job>,
    private readonly queue: JobQueue,
  ) {}

  async enqueue(videoId: string, userId: string, height: 480 | 720 | 1080 = 720): Promise<Job> {
    const job = await this.jobs.save(this.jobs.create({ videoId }));
    await this.queue.push({ id: job.id, videoId, userId, height, attempt: 0 });
    return job;
  }

  async markFailed(jobId: string, error: string): Promise<void> {
    await this.jobs.update(jobId, { status: 'failed', error, finishedAt: new Date() });
  }

  // Nguon su that cho FE khi connect/reconnect (muc 6) — WebSocket chi tang
  // toc, bang jobs moi la thu luon dung. "Active" nghia la job chua xong
  // (queued/processing) hoac vua xong/hong trong 24 gio gan nhat, de UI con
  // thay duoc buoc chuyen trang thai cuoi cung sau khi reconnect.
  async findActiveForUser(userId: string): Promise<JobState[]> {
    const cutoff = new Date(Date.now() - 24 * 60 * 60 * 1000);

    const jobs = await this.jobs
      .createQueryBuilder('job')
      .innerJoinAndSelect('job.video', 'video')
      .where('video.user_id = :userId', { userId })
      .andWhere("(job.status IN ('queued', 'processing') OR job.updated_at >= :cutoff)", { cutoff })
      .orderBy('job.created_at', 'DESC')
      .getMany();

    return jobs.map((job) => ({
      id: job.id,
      videoTitle: job.video.originalName ?? job.videoId,
      status: job.status,
      // Phan tram song chi ton tai qua Redis pub/sub trong luc dang chay —
      // khong luu vao Postgres. WebSocket se day gia tri that ngay sau khi
      // connect (xem useJobProgress.ts o muc 7).
      percent: job.status === 'completed' ? 100 : 0,
    }));
  }
}

import { Injectable, OnModuleInit } from '@nestjs/common';
import { RedisService } from '../redis/redis.service';

export interface TranscodeJob {
  readonly id: string; // trung voi Job.id trong Postgres
  readonly videoId: string;
  readonly userId: string;
  readonly height: 480 | 720 | 1080;
  readonly attempt: number;
  // Gan boi pull() sau khi doc tu Redis Streams — push() khong co gia tri nay.
  readonly messageId?: string;
  // Chi co khi day sang hang doi chet (muc 4.1).
  readonly lastError?: string;
}

const STREAM = 'jobs:transcode';
const GROUP = 'workers';

function parseMessage(fields: string[]): TranscodeJob {
  // xadd luu 'data' <json> — fields la mang phang [key, value, key, value, ...].
  const raw = fields[fields.indexOf('data') + 1] as string;
  return JSON.parse(raw) as TranscodeJob;
}

@Injectable()
export class JobQueue implements OnModuleInit {
  constructor(private readonly redis: RedisService) {}

  async onModuleInit(): Promise<void> {
    // MKSTREAM: tao stream neu chua co. '$' = bo qua nhung message co tu
    // truoc, chi doc nhung gi den sau khi group nay duoc tao.
    try {
      await this.redis.xgroup('CREATE', STREAM, GROUP, '$', 'MKSTREAM');
    } catch (err) {
      // BUSYGROUP: group da ton tai tu lan khoi dong truoc — khong phai loi that.
      if (!String(err).includes('BUSYGROUP')) throw err;
    }
  }

  async push(payload: TranscodeJob): Promise<string> {
    const id = await this.redis.xadd(STREAM, '*', 'data', JSON.stringify(payload));
    if (id === null) throw new Error('XADD khong tra ve id.');
    return id;
  }

  async pull(consumerName: string): Promise<TranscodeJob | null> {
    // BLOCK 5000: cho toi 5 giay neu hang doi rong, khong quay vong hoi lien tuc.
    // Chu thich kieu tra ve bang tay: chu ky bien the nay cua xreadgroup khong
    // duoc ioredis suy ra dung, TypeScript thay "unknown" neu khong ep kieu.
    const result = (await this.redis.xreadgroup(
      'GROUP', GROUP, consumerName,
      'COUNT', 1,
      'BLOCK', 5000,
      'STREAMS', STREAM, '>',
    )) as [string, [string, string[]][]][] | null;

    if (result === null || result.length === 0) return null;

    const [messageId, fields] = result[0]![1][0]!;
    return { ...parseMessage(fields), messageId };
  }

  async ack(messageId: string): Promise<void> {
    await this.redis.xack(STREAM, GROUP, messageId);
  }
}

import { Module } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { Job } from './job.entity';
import { JobQueue } from './job.queue';
import { JobService } from './job.service';
import { JobController } from './job.controller';
import { AuthModule } from '../auth/auth.module';

@Module({
  // AuthModule vi JobController dung JwtAuthGuard — cung ly do nhu MediaModule
  // o Part 2 muc 6.
  imports: [TypeOrmModule.forFeature([Job]), AuthModule],
  controllers: [JobController],
  providers: [JobQueue, JobService],
  // WorkerModule o muc 4.3 tiem lai ca hai.
  exports: [JobQueue, JobService],
})
export class JobModule {}

curl -sk --resolve media-forge.local:8443:127.0.0.1 \
  -X POST "https://media-forge.local:8443/media/$VID/transcode" \
  -H "Authorization: Bearer $TOK" -w '\nhttp=%{http_code}\n'
# -> {"jobId":"06f95a5e-..."}
# -> http=202

# Thong diep da nam trong Redis Streams chua?
docker exec forge-redis redis-cli XLEN jobs:transcode
# -> 1

curl -sk --resolve media-forge.local:8443:127.0.0.1 \
  "https://media-forge.local:8443/jobs/active" -H "Authorization: Bearer $TOK"
# -> [{"id":"06f95a5e-...","videoTitle":"...","status":"queued","percent":0}]

// them vao class JobQueue o muc 4
private readonly DELAYED_KEY = 'jobs:delayed'; // sorted set: score = thoi diem (ms) duoc phep chay
private readonly DEAD_LETTER_STREAM = 'jobs:dead';

// Khong day thang vao stream chinh — xep vao sorted-set voi diem la thoi
// diem duoc phep chay. promoteDelayed() (goi dinh ky) day nhung job da toi
// han sang stream chinh.
async pushDelayed(payload: TranscodeJob, delayMs: number): Promise<void> {
  await this.redis.zadd(this.DELAYED_KEY, Date.now() + delayMs, JSON.stringify(payload));
}

async promoteDelayed(): Promise<number> {
  const due = await this.redis.zrangebyscore(this.DELAYED_KEY, 0, Date.now());
  for (const raw of due) {
    // ZREM truoc: hai worker cung quet dung luc thi chi mot ben con xoa
    // duoc, ben kia nhan 0 — khong day trung.
    const removed = await this.redis.zrem(this.DELAYED_KEY, raw);
    if (removed > 0) await this.push(JSON.parse(raw) as TranscodeJob);
  }
  return due.length;
}

// Khong dung consumer group — day la noi con nguoi doc lai bang XRANGE,
// khong ai XACK ca.
async pushDeadLetter(payload: TranscodeJob): Promise<string> {
  const id = await this.redis.xadd(this.DEAD_LETTER_STREAM, '*', 'data', JSON.stringify(payload));
  if (id === null) throw new Error('XADD khong tra ve id.');
  return id;
}

import { Injectable } from '@nestjs/common';
import { Interval } from '@nestjs/schedule';
import { JobQueue } from './job.queue';

@Injectable()
export class DelayedJobPromoter {
  constructor(private readonly queue: JobQueue) {}

  @Interval(1000)
  async flush(): Promise<void> {
    await this.queue.promoteDelayed();
  }
}

const MAX_ATTEMPTS = 3;

// Method cua JobRunner, KHONG PHAI mot ham doc lap — mot function() thuong
// khong co `this` gan san, trong khi than ham ben duoi dung this.queue,
// this.jobs. Xem class JobRunner day du o muc 4.4.
private async processOne(job: TranscodeJob): Promise<void> {
  try {
    await this.runTranscode(job);
    await this.queue.ack(job.messageId!);
  } catch (error) {
    const attempt = job.attempt + 1;

    // Ghi so lan da thu vao CSDL. Redis giu con so nay trong thong diep, nhung
    // nguoi di dieu tra mot job hong thi nhin vao bang `jobs` truoc tien —
    // de cot `attempts` dung im o 0 la de no noi doi.
    await this.jobRepo.update(job.id, { attempts: attempt });

    if (attempt >= MAX_ATTEMPTS) {
      // Het luot: day sang hang doi chet. KHONG xoa — con nguoi can xem lai.
      await this.queue.pushDeadLetter({ ...job, lastError: String(error) });
      await this.queue.ack(job.messageId!);
      await this.jobs.markFailed(job.id, String(error));
      return;
    }

    // Cho lau dan: 2s, 4s, 8s. Thu lai ngay lap tuc chi lam nang them
    // dung cai he thong dang gap su co.
    const delayMs = 2 ** attempt * 1000;
    await this.queue.pushDelayed({ ...job, attempt }, delayMs);
    await this.queue.ack(job.messageId!);
  }
}

export interface ChargeForJobResult {
  readonly charged: boolean;
  readonly balance: number;
}

async chargeForJob(jobId: string, userId: string, amount: number): Promise<ChargeForJobResult> {
  return this.dataSource.transaction(async (manager) => {
    // So cai append-only tu Part 1, gio them mot rang buoc duy nhat tren job_id.
    // Lan chay thu hai va vao rang buoc nay va bi tu choi.
    const result = await manager
      .createQueryBuilder()
      .insert()
      .into(CreditEntry)
      .values({ userId, jobId, delta: -amount, reason: 'transcode' })
      .orIgnore() // ON CONFLICT DO NOTHING
      .execute();

    // Bi bo qua vi trung job_id thi RETURNING khong tra dong nao — result.raw rong.
    const charged = result.raw.length > 0;

    const row = await manager
      .createQueryBuilder(CreditEntry, 'e')
      .select('COALESCE(SUM(e.delta), 0)', 'balance')
      .where('e.user_id = :userId', { userId })
      .getRawOne<{ balance: string }>();

    return { charged, balance: Number(row?.balance ?? 0) };
  });
}

import { NestFactory } from '@nestjs/core';
import { WorkerModule } from './worker.module';
import { JobRunner } from './job.runner';

async function bootstrap(): Promise<void> {
  // createApplicationContext, KHONG phai create(). Co DI, co ConfigService,
  // co TypeORM — nhung khong mo cong nao, khong co controller nao.
  const app = await NestFactory.createApplicationContext(WorkerModule);
  app.enableShutdownHooks();

  const runner = app.get(JobRunner);
  await runner.loop();
}

void bootstrap();

import { Module } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { TypeOrmModule } from '@nestjs/typeorm';
import { ScheduleModule } from '@nestjs/schedule';
import { AppConfigModule } from '../config/config.module';
import { sharedOptions } from '../database/typeorm.options';
import type { AppConfig } from '../config/configuration';
import { RedisModule } from '../redis/redis.module';
import { BillingModule } from '../billing/billing.module';
import { JobModule } from '../job/job.module';
import { Video } from '../media/video.entity';
import { Job } from '../job/job.entity';
import { DelayedJobPromoter } from '../job/delayed-job-promoter';
import { JobRunner } from './job.runner';

// Dung lai toan bo module nghiep vu — chia thu muc theo mien nghiep vu
// tra co tuc dung o day (xem Part 1, muc 1).
@Module({
  imports: [
    AppConfigModule,
    TypeOrmModule.forRootAsync({
      inject: [ConfigService],
      useFactory: (config: ConfigService<AppConfig, true>) => ({
        ...sharedOptions,
        url: config.get('DATABASE_URL', { infer: true }),
      }),
    }),
    // Video CUNG Job — JobRunner tiem thang Repository<Job>, va JobModule
    // khong export forFeature cua rieng no nen phai khai lai o day.
    TypeOrmModule.forFeature([Video, Job]),
    ScheduleModule.forRoot(), // bat buoc de @Interval o DelayedJobPromoter chay that
    RedisModule,
    BillingModule,
    JobModule,
  ],
  providers: [JobRunner, DelayedJobPromoter],
})
export class WorkerModule {}

{
  "scripts": {
    "start:dev": "nest start --watch",
    "start:worker": "node dist/worker/main.js",
    "start:worker:dev": "nest start --watch --entryFile worker/main"
  }
}

services:
  worker:
    build: . # can mot Dockerfile — xem callout ngay duoi
    command: npm run start:worker
    init: true # PID 1 tu te — don tien trinh ffmpeg mo coi, xem canh bao muc 2.3
    environment:
      # PHAI khop voi dich vu postgres da dung o Part 1 muc 4.5:
      # user forge, mat khau forge, database media_forge.
      DATABASE_URL: postgres://forge:forge@postgres:5432/media_forge
      REDIS_URL: redis://redis:6379
      UPLOAD_DIR: /var/media # trong container, khac ./uploads o may minh
    volumes:
      - media:/var/media # dung chung thu muc voi API
    depends_on:
      postgres: { condition: service_healthy }
      redis: { condition: service_started }
    deploy:
      replicas: 2 # hai worker chay song song, Redis Streams tu chia viec

# Khai bao o cap cao nhat, ngang hang voi `services:` — thieu khoi nay thi
# Compose bao volume `media` khong ton tai.
volumes:
  forge-pgdata: # da co tu Part 1
  media: # moi, dung chung giua API va worker

import { Injectable, OnApplicationShutdown } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { hostname } from 'node:os';
import { dirname, join } from 'node:path';
import { Job } from '../job/job.entity';
import { JobQueue } from '../job/job.queue';
import { JobService } from '../job/job.service';
import type { TranscodeJob } from '../job/job.queue';
import { Video } from '../media/video.entity';
import { BillingService } from '../billing/billing.service';
import { RedisService } from '../redis/redis.service';
import { transcode } from './transcoder';

// Gia tren moi do phan giai — dung cho demo, thuc te se doc tu bang gia.
const PRICE_BY_HEIGHT: Record<TranscodeJob['height'], number> = { 480: 5, 720: 10, 1080: 20 };

// So lan thu toi da truoc khi day sang hang doi chet (muc 4.1).
const MAX_ATTEMPTS = 3;

@Injectable()
export class JobRunner implements OnApplicationShutdown {
  private stopping = false;
  private current: Promise<void> | null = null;
  private readonly consumerName = `${hostname()}-${process.pid}`;

  constructor(
    private readonly queue: JobQueue,
    private readonly jobs: JobService,
    @InjectRepository(Job) private readonly jobRepo: Repository<Job>,
    @InjectRepository(Video) private readonly videos: Repository<Video>,
    private readonly billing: BillingService,
    private readonly redis: RedisService,
  ) {}

  async loop(): Promise<void> {
    while (!this.stopping) {
      const job = await this.queue.pull(this.consumerName);
      if (job === null) continue; // het 5 giay cho ma khong co viec

      this.current = this.processOne(job);
      await this.current;
      this.current = null;
    }
  }

  // Nest goi ham nay khi nhan SIGTERM (nho enableShutdownHooks).
  async onApplicationShutdown(): Promise<void> {
    this.stopping = true; // khong nhan viec moi nua
    await this.current; // nhung lam not viec dang do
  }

  // Method cua JobRunner, KHONG PHAI mot ham doc lap — mot function() thuong
  // khong co `this` gan san, ma than ham duoi day dung this.queue, this.jobs.
  private async processOne(job: TranscodeJob): Promise<void> {
    try {
      await this.runTranscode(job);
      await this.queue.ack(job.messageId!);
    } catch (error) {
      const attempt = job.attempt + 1;

      if (attempt >= MAX_ATTEMPTS) {
        // Het luot: day sang hang doi chet. KHONG xoa — con nguoi can xem lai.
        await this.queue.pushDeadLetter({ ...job, lastError: String(error) });
        await this.queue.ack(job.messageId!);
        await this.jobs.markFailed(job.id, String(error));
        return;
      }

      // Cho lau dan: 2s, 4s, 8s. Thu lai ngay lap tuc chi lam nang them
      // dung cai he thong dang gap su co.
      const delayMs = 2 ** attempt * 1000;
      await this.queue.pushDelayed({ ...job, attempt }, delayMs);
      await this.queue.ack(job.messageId!);
    }
  }

  private async runTranscode(job: TranscodeJob): Promise<void> {
    const video = await this.videos.findOneOrFail({ where: { id: job.videoId } });
    await this.jobRepo.update(job.id, { status: 'processing', startedAt: new Date() });

    const controller = new AbortController();
    const output = join(dirname(video.originalKey), `${video.id}-${job.height}p.mp4`);

    await transcode({
      input: video.originalKey,
      output,
      height: job.height,
      durationSeconds: video.duration ?? 60,
      signal: controller.signal,
      onProgress: (percent) => {
        // Worker khong biet gi ve WebSocket. No chi phat ra Redis (muc 6).
        void this.redis.publish('progress', JSON.stringify({ userId: job.userId, jobId: job.id, percent }));
      },
    });

    // Tru credit BINH THAN — an toan neu job nay bi xu ly lai (muc 4.2).
    await this.billing.chargeForJob(job.id, job.userId, PRICE_BY_HEIGHT[job.height]);
    await this.jobRepo.update(job.id, { status: 'completed', finishedAt: new Date() });

    // Kenh RIENG voi 'progress' — client can biet job da xong HOAN TOAN,
    // khong chi la mot moc phan tram nua trong hang dai cac moc trung gian.
    void this.redis.publish('job:done', JSON.stringify({ userId: job.userId, jobId: job.id }));
  }
}

# Cua so 1: API (nhu tu truoc toi gio)
npm run start:dev

# Cua so 2: worker
npm run start:worker:dev

# Cua so 3: day mot job cho video da upload o Part 2
curl -sk --resolve media-forge.local:8443:127.0.0.1 \
  -X POST "https://media-forge.local:8443/media/$VID/transcode" \
  -H "Authorization: Bearer $TOK"
# -> {"jobId":"1639da5a-..."}  http=202

# Nghe tien do worker phat ra, trong luc no chay
docker exec forge-redis redis-cli SUBSCRIBE progress
# -> {"userId":"...","jobId":"1639da5a-...","percent":22}
# -> {"userId":"...","jobId":"1639da5a-...","percent":47}
# -> {"userId":"...","jobId":"1639da5a-...","percent":70}

services:
  worker:
    # ... build, command, init, environment, volumes, depends_on da chot o muc 4.3 ...
    stop_grace_period: 10m # doi toi 10 phut truoc khi Docker gui SIGKILL
    deploy:
      replicas: 2

import { NestFactory } from '@nestjs/core';
import { ConfigService } from '@nestjs/config';
import { ValidationPipe } from '@nestjs/common';
import { AppModule } from './app.module';
import { TimeoutInterceptor } from './common/timeout/timeout.interceptor';
import type { NestExpressApplication } from '@nestjs/platform-express';
import type { AppConfig } from './config/configuration';

export async function bootstrap(): Promise<void> {
  const app = await NestFactory.create<NestExpressApplication>(AppModule);

  app.set('trust proxy', 1); // Part 2, muc 4.2
  app.useGlobalPipes(
    new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }),
  ); // Part 2, muc 1
  app.useGlobalInterceptors(new TimeoutInterceptor()); // Part 2, muc 5.2

  const config = app.get(ConfigService<AppConfig, true>);
  await app.listen(config.get('PORT', { infer: true }));
}

// Chi tu chay khi file nay la entry point that su (npm run start:dev goi
// thang main.ts) — tranh chay bootstrap() hai lan khi cluster.ts import ham nay.
if (require.main === module) {
  void bootstrap();
}

import { bootstrap } from './main';
import cluster from 'node:cluster';
import { availableParallelism } from 'node:os';

if (cluster.isPrimary) {
  const count = availableParallelism();

  for (let i = 0; i < count; i += 1) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    // Chet vi loi thi dung ngay mot cai moi. Con neu la dung co y
    // (SIGTERM khi trien khai ban moi) thi de yen cho no thoat.
    if (signal !== 'SIGTERM') {
      console.error(`worker ${worker.process.pid} chet (${code}), dung lai`);
      cluster.fork();
    }
  });
} else {
  // bootstrap() CUA src/main.ts (tien trinh API, co app.listen()) — KHONG
  // PHAI bootstrap() cua src/worker/main.ts (tien trinh worker, khong HTTP,
  // xem muc 4.3). Hai file dinh nghia hai ham trung ten nhung khac nhau.
  void bootstrap();
}

npx nest build
node dist/cluster.js

# Bao nhieu ban NestJS da khoi dong?
# -> mot dong "Nest application successfully started" cho MOI nhan CPU

# Va bay gio lam lai dung phep thu o muc 1:
( curl -s -o /dev/null -w 'block: %{time_total}s\n' http://localhost:3000/block & ); sleep 0.5
curl -s -o /dev/null -w 'ping:  %{time_total}s\n' http://localhost:3000/ping
# -> ping:  0.004s      <-- muc 1 la 4,6 giay
# -> block: 5.004s

1. Trang thai trong bo nho
   Cache bang Map trong tien trinh -> 8 ban khac nhau, khong dong bo.
   => Redis.

2. Ket noi WebSocket
   Nguoi dung noi vao tien trinh 3. Tien do phat tu worker khong den duoc.
   => Redis pub/sub (muc 6).

3. Cong viec dinh ky
   @Cron chay o CA 8 tien trinh -> gui email 8 lan.
   => Khoa trong Redis, hoac tach han ra mot tien trinh scheduler rieng.

upstream app {
    least_conn;              # gui toi instance dang it ket noi nhat
    server app1:3000 max_fails=3 fail_timeout=30s;
    server app2:3000 max_fails=3 fail_timeout=30s;
    server app3:3000 max_fails=3 fail_timeout=30s;
    keepalive 32;
}

# Cho @Interval o DelayedJobPromoter (muc 4.1)
npm i @nestjs/schedule

# Cho WebSocket o muc nay
npm i @nestjs/websockets @nestjs/platform-socket.io socket.io

import { OnGatewayConnection, WebSocketGateway, WebSocketServer } from '@nestjs/websockets';
import type { Server, Socket } from 'socket.io';
import { JwtService } from '@nestjs/jwt';
import type { JwtPayload } from '../auth/jwt-payload';

@WebSocketGateway({ path: '/ws', cors: false })
export class ProgressGateway implements OnGatewayConnection {
  @WebSocketServer() private readonly server!: Server;

  constructor(private readonly jwt: JwtService) {}

  async handleConnection(client: Socket): Promise<void> {
    try {
      // Xac thuc NGAY luc bat tay. Ket noi chua xac thuc thi dong luon —
      // de no mo la de nguoi la giu tai nguyen cua ban.
      const token = String(client.handshake.auth['token'] ?? '');
      const payload = await this.jwt.verifyAsync<JwtPayload>(token);

      // Moi nguoi dung mot phong rieng. Khong ai nghe duoc tien do cua ai.
      await client.join(`user:${payload.sub}`);
    } catch {
      client.disconnect(true);
    }
  }

  // Duoc goi khi co thong diep tu Redis pub/sub.
  emitProgress(userId: string, jobId: string, percent: number): void {
    this.server.to(`user:${userId}`).emit('job:progress', { jobId, percent });
  }

  // Su kien RIENG, khac 'job:progress' — bao cho client biet job da xong
  // hoan toan (kem tru credit), khong chi la mot moc phan tram nua.
  emitDone(userId: string, jobId: string): void {
    this.server.to(`user:${userId}`).emit('job:done', { jobId });
  }
}

// Worker khong biet gi ve WebSocket. No chi phat ra Redis.
await this.redis.publish(
  'progress',
  JSON.stringify({ userId: job.userId, jobId: job.id, percent }),
);

import { Injectable, OnModuleInit } from '@nestjs/common';
import type { Redis } from 'ioredis';
import { RedisService } from '../redis/redis.service';
import { ProgressGateway } from './progress.gateway';

interface ProgressEvent {
  readonly userId: string;
  readonly jobId: string;
  readonly percent: number;
}

interface DoneEvent {
  readonly userId: string;
  readonly jobId: string;
}

@Injectable()
export class ProgressSubscriber implements OnModuleInit {
  private readonly subscriber: Redis;

  constructor(
    private readonly redis: RedisService,
    private readonly gateway: ProgressGateway,
  ) {
    // MOT ket noi Redis RIENG chi de dang ky nghe. Ket noi dang o che do
    // subscribe khong chay duoc lenh thuong nao khac.
    this.subscriber = this.redis.duplicate();
  }

  onModuleInit(): void {
    // Dang ky ca hai kenh — tham so channel cua callback tu phan biet duoc
    // thong diep nao thuoc kenh nao, khong can hai handler rieng.
    this.subscriber.subscribe('progress', 'job:done');

    this.subscriber.on('message', (channel: string, raw: string) => {
      if (channel === 'job:done') {
        const event = JSON.parse(raw) as DoneEvent;
        this.gateway.emitDone(event.userId, event.jobId);
        return;
      }

      const event = JSON.parse(raw) as ProgressEvent;
      // Tien trinh nao dang giu ket noi cua nguoi do thi gui duoc.
      // Cac tien trinh con lai goi to() vao mot phong rong — khong sao ca.
      this.gateway.emitProgress(event.userId, event.jobId, event.percent);
    });
  }
}

import { Module } from '@nestjs/common';
import { AuthModule } from '../auth/auth.module';
import { ProgressGateway } from './progress.gateway';
import { ProgressSubscriber } from './progress.subscriber';

@Module({
  // AuthModule vi ProgressGateway tiem JwtService de xac thuc bat tay —
  // cung ly do nhu MediaModule o Part 2 muc 6.
  imports: [AuthModule],
  providers: [ProgressGateway, ProgressSubscriber],
})
export class GatewayModule {}

>>> WS da ket noi
>>> transcode 202 {"jobId":"f919b4e9-a6d5-4473-a048-d95623e8bb06"}
>>> job:progress 22
>>> job:progress 47
>>> job:progress 70
>>> job:progress 94
>>> job:progress 99
>>> job:done f919b4e9-a6d5-4473-a048-d95623e8bb06

location /ws {
    proxy_pass http://app;

    proxy_http_version 1.1;              # bat buoc: 1.0 khong nang cap duoc
    proxy_set_header Upgrade    $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host       $host;

    # Mac dinh 60s. Ket noi im lang qua 60s se bi cat — ma WebSocket thi
    # im lang la binh thuong.
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
}

import { api } from './api';

socket.on('connect', async () => {
  // Moi lan noi lai — ke ca lan dau — hoi mot phat trang thai hien tai.
  // Sau do WebSocket lo phan cap nhat.
  const res = await api('/jobs/active');
  const jobs = (await res.json()) as JobState[];
  setJobs(jobs);
});

socket.on('job:progress', (event: { jobId: string; percent: number }) => {
  setJobs((prev) =>
    prev.map((job) => (job.id === event.jobId ? { ...job, percent: event.percent } : job)),
  );
});

import { useEffect, useState } from 'react';
import { io, type Socket } from 'socket.io-client';
import { api } from '../lib/api';

export interface JobState {
  readonly id: string;
  readonly videoTitle: string;
  readonly status: 'queued' | 'processing' | 'completed' | 'failed';
  readonly percent: number;
}

export function useJobProgress(accessToken: string): readonly JobState[] {
  const [jobs, setJobs] = useState<readonly JobState[]>([]);

  useEffect(() => {
    const socket: Socket = io('/', {
      path: '/ws',
      transports: ['websocket'], // bo qua long-polling, xem canh bao o muc 6.1
      auth: { token: accessToken },
    });

    const sync = (): void => {
      void api('/jobs/active')
        .then((res) => res.json() as Promise<JobState[]>)
        .then(setJobs);
    };

    socket.on('connect', sync);

    socket.on('job:progress', ({ jobId, percent }: { jobId: string; percent: number }) => {
      setJobs((prev) => {
        // Job MOI, tao sau luc connect: chua co trong danh sach nen map() se
        // khong khop gi ca. Phai hoi lai /jobs/active de lay ten video, neu
        // khong no VO HINH cho toi khi tai lai trang.
        if (!prev.some((job) => job.id === jobId)) {
          sync();
          return prev;
        }
        return prev.map((job) => (job.id === jobId ? { ...job, percent, status: 'processing' } : job));
      });
    });

    socket.on('job:done', ({ jobId }: { jobId: string }) => {
      setJobs((prev) =>
        prev.map((job) => (job.id === jobId ? { ...job, percent: 100, status: 'completed' } : job)),
      );
    });

    // Dong ket noi khi component bi go bo. Quen dong nay thi moi lan
    // dieu huong lai mo them mot socket nua.
    return () => {
      socket.disconnect();
    };
  }, [accessToken]);

  return jobs;
}

import type { JSX } from 'react';
import { useJobProgress } from '../hooks/useJobProgress';

export function JobList({ token }: { token: string }): JSX.Element {
  const jobs = useJobProgress(token);

  return (
    <ul className="space-y-3">
      {jobs.map((job) => (
        <li key={job.id} className="rounded-lg border border-slate-200 p-4">
          <div className="flex justify-between text-sm">
            <span className="font-medium">{job.videoTitle}</span>
            <span className="text-slate-500">{job.percent}%</span>
          </div>
          <div className="mt-2 h-2 overflow-hidden rounded-full bg-slate-200">
            <div
              className="h-full bg-emerald-500 transition-[width] duration-300"
              style={{ width: `${job.percent}%` }}
            />
          </div>
        </li>
      ))}
    </ul>
  );
}

import { api, getAccessToken } from '../lib/api';
import { JobList } from './JobList';

// ...trong component, sau khi upload xong:
async function startTranscode(id: string): Promise<void> {
  await api(`/media/${id}/transcode`, { method: 'POST' });
}

const token = getAccessToken();

// ...trong phan JSX:
{videoId !== null && (
  <button type="button" onClick={() => void startTranscode(videoId)}
          className="rounded bg-emerald-600 px-3 py-2 text-sm text-white">
    Chuyển mã 720p
  </button>
)}

{token !== null && (
  <div className="space-y-2">
    <h2 className="text-sm font-semibold text-slate-700">Tiến độ chuyển mã</h2>
    <JobList token={token} />
  </div>
)}