Xây nền tảng media với NestJS — Part 4: Tách microservice, gRPC & outbox pattern

Xac thuc, danh sach video, phat lai
  - moi request: vai mili giay, gan nhu khong ton CPU
  - can: NHIEU ban nho, phan hoi nhanh
  - luu luong: 10.000 request/phut la binh thuong

Chuyen ma
  - moi job: vai phut, an tron mot nhan CPU
  - can: IT ban nhung MAY MANH, co the co GPU
  - luu luong: 50 job/gio da la nhieu

=> Hai loai nay khong the dung chung mot cach nhan ban.
   Nhan ban de phuc vu HTTP thi thua CPU. Nhan ban de chuyen ma thi
   tra tien cho mot dong may manh ngoi cho request nhe.

media-forge/
├── proto/                    # HOP DONG chung, khong thuoc ve rieng ai
│   ├── auth.proto
│   ├── media.proto
│   └── billing.proto
├── apps/
│   ├── gateway/              # HTTP + WebSocket, khong co nghiep vu
│   ├── auth-svc/             # users, refresh_tokens
│   ├── media-svc/            # videos, media_assets, jobs
│   └── billing-svc/          # credit_entries
└── libs/
    └── proto-types/          # kieu TypeScript sinh tu .proto

syntax = "proto3";
package billing;

service BillingService {
  rpc GetBalance (GetBalanceRequest) returns (BalanceReply);
  rpc Charge (ChargeRequest) returns (ChargeReply);
}

message GetBalanceRequest {
  string user_id = 1;
}

message BalanceReply {
  int64 balance = 1;
}

message ChargeRequest {
  string user_id = 1;
  int64 amount  = 2;
  // Khoa binh than: goi lai voi cung job_id thi khong tru them lan nua.
  string job_id = 3;
}

message ChargeReply {
  bool   charged     = 1;
  int64  new_balance = 2;
}

npm i @nestjs/microservices @grpc/grpc-js @grpc/proto-loader
npm i -D ts-proto

# ts-proto chi la plugin sinh kieu TypeScript cho protoc, khong thay the duoc
# ban than trinh bien dich protoc. npx tu tai mot binary thay the neu may
# chua co, nhung binary do la mot goi npm khong ro nguon goc — cai qua trinh
# quan ly goi he dieu hanh de kiem soat dung phien ban dang dung:
brew install protobuf                    # macOS
sudo apt install -y protobuf-compiler    # Ubuntu/Debian

npx protoc \
  --proto_path=./proto \
  --plugin=./node_modules/.bin/protoc-gen-ts_proto \
  --ts_proto_out=./libs/proto-types \
  --ts_proto_opt=nestJs=true,useOptionals=messages \
  ./proto/*.proto

{
  "compilerOptions": {
    "baseUrl": ".",
    "paths": {
      "@app/proto-types/*": ["libs/proto-types/*"]
    }
  }
}

import { Controller } from '@nestjs/common';
import { GrpcMethod } from '@nestjs/microservices';
import type {
  ChargeRequest,
  ChargeReply,
  GetBalanceRequest,
  BalanceReply,
} from '@app/proto-types/billing';

@Controller()
export class BillingGrpcController {
  constructor(private readonly billing: BillingService) {}

  // Ten service va ten method phai khop CHINH XAC voi file .proto.
  @GrpcMethod('BillingService', 'Charge')
  async charge(request: ChargeRequest): Promise<ChargeReply> {
    const result = await this.billing.chargeForJob(
      request.jobId,
      request.userId,
      request.amount,
    );
    return { charged: result.charged, newBalance: result.balance };
  }

  @GrpcMethod('BillingService', 'GetBalance')
  async getBalance(request: GetBalanceRequest): Promise<BalanceReply> {
    const balance = await this.billing.getBalance(request.userId);
    return { balance };
  }
}

import { Injectable, OnModuleInit } from '@nestjs/common';
import { Client, ClientGrpc, Transport } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';
import type { BillingServiceClient } from '@app/proto-types/billing';

@Injectable()
export class BillingClient implements OnModuleInit {
  @Client({
    transport: Transport.GRPC,
    options: { package: 'billing', protoPath: 'proto/billing.proto', url: 'billing-svc:50051' },
  })
  private readonly client!: ClientGrpc;

  private service!: BillingServiceClient;

  onModuleInit(): void {
    this.service = this.client.getService<BillingServiceClient>('BillingService');
  }

  async charge(userId: string, amount: number, jobId: string): Promise<boolean> {
    // gRPC tra ve Observable. KHONG bao gio goi ma khong dat han thoi gian —
    // thieu no thi mot dich vu treo se keo theo ca chuoi goi phia tren.
    const reply = await firstValueFrom(
      this.service.charge({ userId, amount, jobId }).pipe(timeout(3000)),
    );
    return reply.charged;
  }

  async getBalance(userId: string): Promise<number> {
    const reply = await firstValueFrom(
      this.service.getBalance({ userId }).pipe(timeout(3000)),
    );
    return reply.balance;
  }
}

service MediaService {
  // Mot request, NHIEU reply chay ve dan.
  rpc WatchJob (WatchJobRequest) returns (stream JobProgress);
}

message WatchJobRequest {
  string job_id = 1;
}

message JobProgress {
  string job_id  = 1;
  int32  percent = 2;
  string status  = 3;
}

import { Controller } from '@nestjs/common';
import { GrpcMethod } from '@nestjs/microservices';
import { EventEmitter2 } from '@nestjs/event-emitter';
import { Observable } from 'rxjs';
import type { WatchJobRequest, JobProgress } from '@app/proto-types/media';

// Ten rieng, KHONG trung voi ProgressEvent co san cua DOM.
interface JobProgressEvent {
  jobId: string;
  percent: number;
  status: string;
}

@Controller()
export class MediaGrpcController {
  constructor(private readonly progress: EventEmitter2) {}

  @GrpcMethod('MediaService', 'WatchJob')
  watchJob(request: WatchJobRequest): Observable<JobProgress> {
    return new Observable<JobProgress>((subscriber) => {
      const handler = (event: JobProgressEvent): void => {
        if (event.jobId !== request.jobId) return;
        subscriber.next({ jobId: event.jobId, percent: event.percent, status: event.status });
        if (event.percent === 100) subscriber.complete();
      };

      this.progress.on('progress', handler);

      // Ham don dep — chay khi ben goi ngat luong hoac het han.
      // Thieu no thi moi ket noi ro ra mot listener.
      return () => this.progress.off('progress', handler);
    });
  }
}

await this.dataSource.transaction(async (manager) => {
  await manager.update(Job, jobId, { status: 'completed' });
  await manager.insert(CreditEntry, { userId, jobId, delta: -amount });
});
// Chay het hoac khong chay gi. Database bao dam.

await this.jobs.update(jobId, { status: 'completed' }); // DB cua media-svc
await this.billingClient.charge(userId, amount, jobId);  // qua mang, DB khac

// Hong 1: charge nem loi -> job "completed" nhung khong ai bi tru tien.
// Hong 2: charge thanh cong, phan hoi mat tren duong ve -> ta tuong that bai,
//         thu lai. Khoa binh than cuu duoc lan nay.
// Hong 3: tien trinh chet giua hai dong -> khong con ai biet buoc 2 con no.

@Entity('outbox')
export class OutboxMessage {
  @PrimaryGeneratedColumn('uuid')
  id!: string;

  @Column()
  topic!: string; // 'billing.charge'

  @Column({ type: 'jsonb' })
  payload!: Record<string, unknown>;

  // null = chua gui. Co gia tri = da gui xong.
  @Index()
  @Column({ name: 'sent_at', type: 'timestamptz', nullable: true })
  sentAt!: Date | null;

  @Column({ default: 0 })
  attempts!: number;

  @CreateDateColumn({ name: 'created_at', type: 'timestamptz' })
  createdAt!: Date;
}

async completeJob(jobId: string, userId: string, amount: number): Promise<void> {
  await this.dataSource.transaction(async (manager) => {
    await manager.update(Job, jobId, { status: 'completed' });

    // KHONG goi mang o day. Chi ghi y dinh xuong cung mot database.
    await manager.insert(OutboxMessage, {
      topic: 'billing.charge',
      payload: { userId, amount, jobId },
      sentAt: null,
    });
  });
  // Ra khoi transaction: hoac ca hai dong cung ton tai, hoac khong dong nao.
}

import { Injectable } from '@nestjs/common';
import { Interval } from '@nestjs/schedule';
import { DataSource } from 'typeorm';
import { OutboxMessage } from './outbox.entity';
import { BillingClient } from '../billing.client';

@Injectable()
export class OutboxRelay {
  constructor(
    private readonly dataSource: DataSource,
    private readonly billingClient: BillingClient,
  ) {}

  @Interval(1000)
  async flush(): Promise<void> {
    await this.dataSource.transaction(async (manager) => {
      const pending = await manager
        .createQueryBuilder(OutboxMessage, 'o')
        .setLock('pessimistic_write')
        .setOnLocked('skip_locked') // relay khac dang giu dong nao thi bo qua dong do
        .where('o.sent_at IS NULL')
        .orderBy('o.created_at', 'ASC')
        .limit(50)
        .getMany();

      for (const message of pending) {
        try {
          await this.dispatch(message);
          await manager.update(OutboxMessage, message.id, { sentAt: new Date() });
        } catch {
          // Khong danh dau da gui. Vong sau lay lai dung ban ghi nay.
          await manager.increment(OutboxMessage, { id: message.id }, 'attempts', 1);
        }
      }
    });
  }

  private async dispatch(message: OutboxMessage): Promise<void> {
    if (message.topic === 'billing.charge') {
      const { userId, amount, jobId } = message.payload as {
        userId: string;
        amount: number;
        jobId: string;
      };
      await this.billingClient.charge(userId, amount, jobId);
      return;
    }
    // Topic la chuoi tu do, khong phai enum — mot ban ghi voi topic sai
    // (do bug hoac do doi migration) khong duoc am tham bo qua.
    throw new Error(`Unknown outbox topic: ${message.topic}`);
  }
}

async getBalance(userId: string): Promise<number> {
  const key = `balance:${userId}`;
  const cached = await this.redis.get(key);
  if (cached !== null) return Number(cached);

  const balance = await this.billingClient.getBalance(userId);
  await this.redis.set(key, String(balance), 'EX', 60);
  return balance;
}

import { Injectable } from '@nestjs/common';
import { RedisService } from '../../redis/redis.service';
import { BillingClient } from '../billing.client';

@Injectable()
export class BalanceCache {
  constructor(
    private readonly redis: RedisService,
    private readonly billingClient: BillingClient,
  ) {}

  private readonly inFlight = new Map<string, Promise<number>>();

  async getBalance(userId: string): Promise<number> {
    const key = `balance:${userId}`;

    const cached = await this.redis.get(key);
    if (cached !== null) return Number(cached);

    // Trong MOT tien trinh: moi nguoi cho chung mot lan goi.
    // Cung y tuong voi bien `refreshing` cua Part 2.
    const existing = this.inFlight.get(key);
    if (existing !== undefined) return existing;

    const promise = this.loadAndStore(key, userId).finally(() => {
      this.inFlight.delete(key);
    });
    this.inFlight.set(key, promise);
    return promise;
  }

  private async loadAndStore(key: string, userId: string): Promise<number> {
    const balance = await this.billingClient.getBalance(userId);

    // Han dung ngau nhien 60-75s. Neu tat ca cung dat 60s thi cac khoa
    // tao ra trong cung mot dot se cung het han trong cung mot dot.
    const ttl = 60 + Math.floor(Math.random() * 15);
    await this.redis.set(key, String(balance), 'EX', ttl);
    return balance;
  }
}

import { AsyncLocalStorage } from 'node:async_hooks';

export const requestContext = new AsyncLocalStorage<{ correlationId: string }>();

@Injectable()
export class CorrelationMiddleware implements NestMiddleware {
  use(request: Request, response: Response, next: NextFunction): void {
    const correlationId = String(request.headers['x-correlation-id'] ?? randomUUID());
    response.setHeader('X-Correlation-Id', correlationId);

    // AsyncLocalStorage giu gia tri nay xuyen suot moi await phia sau,
    // ma khong phai truyen tay qua tung ham.
    requestContext.run({ correlationId }, () => next());
  }
}

import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
import { Metadata } from '@grpc/grpc-js';
import { requestContext } from './correlation.middleware';

@Injectable()
export class CorrelationInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler) {
    // Dinh correlation id vao metadata cua MOI loi goi gRPC di ra.
    const metadata = new Metadata();
    metadata.set('x-correlation-id', requestContext.getStore()?.correlationId ?? 'unknown');
    context.switchToRpc().getContext().add(metadata);

    return next.handle();
  }
}

gateway     [8f3a...] POST /media/abc/transcode -> 202 (12ms)
media-svc   [8f3a...] job d41 queued
media-svc   [8f3a...] job d41 processing 0%
media-svc   [8f3a...] job d41 completed (194s)
media-svc   [8f3a...] outbox billing.charge queued
billing-svc [8f3a...] charge user=42 amount=10 job=d41 -> ok

Service discovery
  Ba dich vu voi ten co dinh trong docker-compose la du.
  Can khi so ban thay doi lien tuc va dia chi khong doan truoc duoc.

Circuit breaker
  Deadline + thu lai da xu ly phan lon truong hop. Ngat mach chi dang
  gia khi mot dich vu hong LAU va viec cu goi vao no lam hong ca ben goi.

Service mesh
  Them mot lop ha tang nua de van hanh. Voi ba dich vu, chi phi hoc va
  van hanh lon hon loi ich.

Event sourcing
  So cai credit da la append-only — do la phan duy nhat can lich su day du.
  Ap cho toan he thong la doi mot van de de gap lay mot van de kho hon.

Part 1  Hai request cung tru tien mot tai khoan
        -> SELECT FOR UPDATE

Part 2  Hai tab cung xoay mot refresh token
        -> khoa dong + khoang an han
        Nhieu request cung nhan 401 va cung doi refresh
        -> gop vao MOT promise

Part 3  Mot thong diep bi xu ly hai lan sau khi worker chet
        -> unique index + ON CONFLICT DO NOTHING

Part 4  Mot lenh tru tien duoc gui hai lan tu outbox
        -> van la unique index do
        500 request cung nap lai mot khoa cache vua het han
        -> gop vao MOT promise + han dung ngau nhien

Ba loai loi giai, dung cho ca nam truong hop:
  1. Khoa lai        — cho xep hang
  2. Lam nguyen tu   — de khong co khe ho o giua
  3. Gop thanh mot   — de chi co mot ban chay that