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

@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';
}

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 });

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

  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 code = await new Promise<number>((resolve) => child.on('close', resolve));

  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.
}

function stop(child: ChildProcess): void {
  child.kill('SIGTERM'); // Xin dung, cho don dep

  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));
}

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

interface HashTask {
  readonly filePath: string;
}

const task = workerData as 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);
  }
}

                    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);
  }
}

// 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;

    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();

{
  "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: .
    command: npm run start:worker
    init: true # PID 1 tu te — don tien trinh ffmpeg mo coi, xem canh bao muc 2.3
    environment:
      DATABASE_URL: postgres://app:secret@postgres:5432/media
      REDIS_URL: redis://redis:6379
    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

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 {}

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 }));
  }
}

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

export async function bootstrap(): Promise<void> {
  const app = await NestFactory.create<NestExpressApplication>(AppModule);
  app.set('trust proxy', 1); // xem Part 2, muc 4.2
  await app.listen(process.env.PORT ?? 3000);
}

// 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();
}

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;
}

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);
    });
  }
}

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 },
    });

    socket.on('connect', () => {
      void api('/jobs/active')
        .then((res) => res.json() as Promise<JobState[]>)
        .then(setJobs);
    });

    socket.on('job:progress', ({ jobId, percent }: { jobId: string; percent: number }) => {
      setJobs((prev) =>
        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;
}

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>
  );
}