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 @nestjs/event-emitter
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

import 'reflect-metadata';
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { join } from 'node:path';
import { BillingModule } from './billing.module';

async function bootstrap(): Promise<void> {
  // createMicroservice, KHONG phai create(): khong co cong HTTP, khong
  // controller REST — chi mot server gRPC nghe theo hop dong .proto.
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(BillingModule, {
    transport: Transport.GRPC,
    options: {
      package: 'billing', // PHAI khop `package billing;` trong .proto
      protoPath: join(process.cwd(), 'proto/billing.proto'),
      url: process.env['GRPC_URL'] ?? '0.0.0.0:50051',
    },
  });

  await app.listen();
}
void bootstrap();

{
  "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);

  // Worker cua media-svc goi khi chuyen ma xong — day la duong vao that su
  // cua outbox o muc 4.1. Thieu RPC nay thi completeJob() khong ai goi duoc.
  rpc CompleteJob (CompleteJobRequest) returns (CompleteJobReply);
}

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

// Part 3 — hai kenh, hai hinh dang khac nhau:
await this.redis.publish('progress', JSON.stringify({ userId, jobId, percent })); // percent toi da 99
// ... roi rieng, sau khi that su xong:
await this.redis.publish('job:done', JSON.stringify({ userId, jobId })); // khong percent, khong status

// media-svc — mot kenh, mot hinh dang, di het toi 100:
await this.redis.publish('progress', JSON.stringify({ jobId, percent: 0, status: 'processing' }));
// ... trong luc chay, moi lan ffmpeg bao tien do:
await this.redis.publish('progress', JSON.stringify({ jobId, percent, status: 'processing' }));
// ... xong that (sau khi completeJob() da ghi outbox — muc 4.1):
await this.redis.publish('progress', JSON.stringify({ jobId, percent: 100, status: 'completed' }));

import { Injectable, OnModuleInit } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import type { Redis } from 'ioredis';
import { RedisService } from '../redis/redis.service';

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

// Nguoi lam cau noi. Worker publish len Redis (khong biet gi ve EventEmitter2),
// class nay nghe roi phat lai NOI BO tien trinh — dung vai tro cua
// ProgressSubscriber o Part 3, nhung gio dat trong chinh media-svc.
@Injectable()
export class ProgressBridge implements OnModuleInit {
  private readonly subscriber: Redis;

  constructor(
    private readonly redis: RedisService,
    private readonly emitter: EventEmitter2,
  ) {
    this.subscriber = this.redis.duplicate();
  }

  async onModuleInit(): Promise<void> {
    await this.subscriber.subscribe('progress');
    this.subscriber.on('message', (_channel, raw) => {
      const event = JSON.parse(raw) as ProgressEvent;
      this.emitter.emit('progress', event);
    });
  }
}

syntax = "proto3";
package auth;

service AuthService {
  rpc Login (LoginRequest) returns (TokenPair);
  rpc Refresh (RefreshRequest) returns (TokenPair);
  rpc Logout (LogoutRequest) returns (Empty);
  rpc LogoutEverywhere (LogoutEverywhereRequest) returns (Empty);
}

message LoginRequest { string email = 1; string password = 2; }
message RefreshRequest { string refresh_token = 1; }
message LogoutRequest { string refresh_token = 1; }
message LogoutEverywhereRequest { string user_id = 1; }
message TokenPair { string access_token = 1; string refresh_token = 2; }
message Empty {}

@GrpcMethod('AuthService', 'Refresh')
async refresh(request: RefreshRequest): Promise<TokenPair> {
  const pair = await this.auth.refresh(request.refreshToken); // co the nem UnauthorizedException
  return { accessToken: pair.accessToken, refreshToken: pair.refreshToken };
}

code: 2 (UNKNOWN)
details: "Internal server error"

import { Controller, UnauthorizedException } from '@nestjs/common';
import { GrpcMethod, RpcException } from '@nestjs/microservices';
import { status as GrpcStatus } from '@grpc/grpc-js';
import type {
  LoginRequest,
  RefreshRequest,
  LogoutRequest,
  LogoutEverywhereRequest,
  TokenPair,
  Empty,
} from '@app/proto-types/auth';
import { AuthService } from './auth.service';

@Controller()
export class AuthGrpcController {
  constructor(private readonly auth: AuthService) {}

  @GrpcMethod('AuthService', 'Login')
  async login(request: LoginRequest): Promise<TokenPair> {
    const pair = await this.callAuth(() => this.auth.login(request.email, request.password));
    return { accessToken: pair.accessToken, refreshToken: pair.refreshToken };
  }

  @GrpcMethod('AuthService', 'Refresh')
  async refresh(request: RefreshRequest): Promise<TokenPair> {
    const pair = await this.callAuth(() => this.auth.refresh(request.refreshToken));
    return { accessToken: pair.accessToken, refreshToken: pair.refreshToken };
  }

  @GrpcMethod('AuthService', 'Logout')
  async logout(request: LogoutRequest): Promise<Empty> {
    await this.callAuth(() => this.auth.logout(request.refreshToken));
    return {};
  }

  @GrpcMethod('AuthService', 'LogoutEverywhere')
  async logoutEverywhere(request: LogoutEverywhereRequest): Promise<Empty> {
    await this.callAuth(() => this.auth.logoutEverywhere(request.userId));
    return {};
  }

  // NestJS khong tu dich UnauthorizedException sang ma trang thai gRPC —
  // boc lai thanh RpcException de AuthClient (gateway) doc dung ma + thong diep.
  private async callAuth<T>(fn: () => Promise<T>): Promise<T> {
    try {
      return await fn();
    } catch (err) {
      if (err instanceof UnauthorizedException) {
        throw new RpcException({ code: GrpcStatus.UNAUTHENTICATED, message: err.message });
      }
      throw err;
    }
  }
}

import { Injectable, OnModuleInit, UnauthorizedException } from '@nestjs/common';
import { Client, ClientGrpc, Transport } from '@nestjs/microservices';
import { status as GrpcStatus } from '@grpc/grpc-js';
import { catchError, firstValueFrom, throwError, timeout, type Observable } from 'rxjs';
import type { AuthServiceClient, TokenPair } from '@app/proto-types/auth';

function mapAuthError(err: { code?: number; details?: string; message?: string }): Observable<never> {
  if (err?.code === GrpcStatus.UNAUTHENTICATED) {
    return throwError(() => new UnauthorizedException(err.details ?? err.message));
  }
  return throwError(() => err);
}

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

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

  async login(email: string, password: string): Promise<TokenPair> {
    return firstValueFrom(this.service.login({ email, password }).pipe(timeout(3000), catchError(mapAuthError)));
  }

  async refresh(refreshToken: string): Promise<TokenPair> {
    return firstValueFrom(this.service.refresh({ refreshToken }).pipe(timeout(3000), catchError(mapAuthError)));
  }

  async logout(refreshToken: string): Promise<void> {
    await firstValueFrom(this.service.logout({ refreshToken }).pipe(timeout(3000), catchError(mapAuthError)));
  }

  async logoutEverywhere(userId: string): Promise<void> {
    await firstValueFrom(this.service.logoutEverywhere({ userId }).pipe(timeout(3000), catchError(mapAuthError)));
  }
}

const COOKIE_NAME = 'refresh_token';

@Controller('auth')
export class AuthController {
  constructor(
    private readonly authClient: AuthClient,
    private readonly config: ConfigService<AppConfig, true>,
  ) {}

  @Post('login')
  async login(@Body() dto: LoginDto, @Res({ passthrough: true }) response: Response): Promise<{ accessToken: string }> {
    const pair = await this.authClient.login(dto.email, dto.password);
    this.setRefreshCookie(response, pair.refreshToken);
    return { accessToken: pair.accessToken };
  }

  @Post('refresh')
  async refresh(
    @Req() request: Request,
    @Body() dto: RefreshDto,
    @Res({ passthrough: true }) response: Response,
  ): Promise<{ accessToken: string }> {
    const presented = this.readPresentedToken(request, dto);
    if (presented === undefined) throw new UnauthorizedException('Thieu refresh token');

    const pair = await this.authClient.refresh(presented);
    this.setRefreshCookie(response, pair.refreshToken);
    return { accessToken: pair.accessToken };
  }

  // logout, logout-everywhere, setRefreshCookie, readPresentedToken: xem Part 2 muc 3.5 —
  // tung dong giu nguyen, chi ten thanh vien constructor doi tu "auth" sang "authClient".
}

import { Injectable, OnModuleInit } from '@nestjs/common';
import { Client, ClientGrpc, Transport } from '@nestjs/microservices';
import type { Observable } from 'rxjs';
import type { JobProgress, MediaServiceClient } from '@app/proto-types/media';

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

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

  // Tra ve Observable, khong await mot lan — day la stream, khong phai mot cau tra loi.
  watchJob(jobId: string): Observable<JobProgress> {
    return this.service.watchJob({ jobId });
  }
}

import {
  ConnectedSocket,
  MessageBody,
  OnGatewayConnection,
  OnGatewayDisconnect,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from '@nestjs/websockets';
import type { Server, Socket } from 'socket.io';
import { JwtService } from '@nestjs/jwt';
import type { Subscription } from 'rxjs';
import type { JwtPayload } from '../auth/jwt-payload';
import { MediaClient } from './media.client';

@WebSocketGateway({ path: '/ws', cors: false })
export class ProgressGateway implements OnGatewayConnection, OnGatewayDisconnect {
  @WebSocketServer() private readonly server!: Server;
  private readonly subscriptions = new Map<string, Set<Subscription>>();

  constructor(
    private readonly jwt: JwtService,
    private readonly media: MediaClient,
  ) {}

  async handleConnection(client: Socket): Promise<void> {
    try {
      const token = String(client.handshake.auth['token'] ?? '');
      const payload = await this.jwt.verifyAsync<JwtPayload>(token);
      await client.join(`user:${payload.sub}`);
      client.data.userId = payload.sub;
    } catch {
      client.disconnect(true);
    }
  }

  // Ngat ket noi: huy moi stream gRPC dang mo cho client nay. Thieu buoc nay
  // thi moi lan reconnect lai them mot stream ro ri, khong bao gio dong.
  handleDisconnect(client: Socket): void {
    const subs = this.subscriptions.get(client.id);
    if (subs === undefined) return;
    for (const sub of subs) sub.unsubscribe();
    this.subscriptions.delete(client.id);
  }

  @SubscribeMessage('job:subscribe')
  handleJobSubscribe(@ConnectedSocket() client: Socket, @MessageBody() body: { jobId: string }): void {
    const room = `user:${client.data.userId as string}`;

    const subscription = this.media.watchJob(body.jobId).subscribe({
      next: (progress) => {
        this.server.to(room).emit('job:progress', { jobId: progress.jobId, percent: progress.percent });
        if (progress.status === 'completed') {
          this.server.to(room).emit('job:done', { jobId: progress.jobId });
        }
      },
      error: () => this.forgetSubscription(client.id, subscription),
      complete: () => this.forgetSubscription(client.id, subscription),
    });

    const bucket = this.subscriptions.get(client.id) ?? new Set<Subscription>();
    bucket.add(subscription);
    this.subscriptions.set(client.id, bucket);
  }

  private forgetSubscription(clientId: string, subscription: Subscription): void {
    subscription.unsubscribe();
    this.subscriptions.get(clientId)?.delete(subscription);
  }
}

socket.on('connect', () => {
  void api('/jobs/active')
    .then((res) => res.json() as Promise<JobState[]>)
    .then((active) => {
      setJobs(active);
      // Gateway chi mo watchJob() cho job nao client CHU DONG hoi —
      // thieu dong nay thi 'job:progress'/'job:done' khong bao gio toi.
      for (const job of active) {
        if (job.status === 'queued' || job.status === 'processing') {
          socket.emit('job:subscribe', { jobId: job.id });
        }
      }
    });
});

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

@GrpcMethod('MediaService', 'CompleteJob')
async completeJob(request: CompleteJobRequest): Promise<CompleteJobReply> {
  await this.jobs.completeJob(request.jobId, request.userId, Number(request.amount));
  return { ok: true };
}

docker compose stop billing-svc      # dich vu dich CHET han

# Goi CompleteJob qua gRPC (worker se lam dung viec nay)
# -> OK {"ok":true}      <-- van thanh cong, du khong ai tru tien duoc

docker exec ...media-db-1 psql -U media -d media \
  -c "SELECT sent_at IS NULL AS chua_gui, attempts FROM outbox ORDER BY created_at DESC LIMIT 1;"
# -> chua_gui = t, attempts = 7      (relay dang thu lai moi giay)
# -> vai giay sau: attempts = 13, 30, 95 ... van chua gui duoc

docker compose start billing-svc     # dich vu song lai

docker exec ...media-db-1 psql -U media -d media \
  -c "SELECT sent_at IS NOT NULL AS da_gui FROM outbox ORDER BY created_at DESC LIMIT 1;"
# -> da_gui = t

docker exec ...billing-db-1 psql -U billing -d billing \
  -c "SELECT delta, reason, job_id FROM credit_entries ORDER BY created_at DESC LIMIT 1;"
# -> -10 | transcode | 16a4bf70-...

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

services:
  billing-db: { image: postgres:17-alpine, environment: { POSTGRES_USER: billing, POSTGRES_PASSWORD: billing, POSTGRES_DB: billing } }
  media-db:   { image: postgres:17-alpine, environment: { POSTGRES_USER: media,   POSTGRES_PASSWORD: media,   POSTGRES_DB: media } }
  auth-db:    { image: postgres:17-alpine, environment: { POSTGRES_USER: auth,    POSTGRES_PASSWORD: auth,    POSTGRES_DB: auth } }
  redis:      { image: redis:7-alpine }

  billing-svc:
    build: .
    command: ['node', '-r', 'ts-node/register', '-r', 'tsconfig-paths/register', 'apps/billing-svc/src/main.ts']
    environment:
      DATABASE_URL: postgres://billing:billing@billing-db:5432/billing
      GRPC_URL: '0.0.0.0:50051'
    depends_on: { billing-db: { condition: service_healthy } }

  media-svc:
    build: .
    command: ['node', '-r', 'ts-node/register', '-r', 'tsconfig-paths/register', 'apps/media-svc/src/main.ts']
    environment:
      DATABASE_URL: postgres://media:media@media-db:5432/media
      BILLING_GRPC_URL: 'billing-svc:50051'   # ten dich vu = ten host
      GRPC_URL: '0.0.0.0:50052'
    depends_on: { media-db: { condition: service_healthy } }

  auth-svc:
    build: .
    command: ['node', '-r', 'ts-node/register', '-r', 'tsconfig-paths/register', 'apps/auth-svc/src/main.ts']
    environment:
      DATABASE_URL: postgres://auth:auth@auth-db:5432/auth
      JWT_SECRET: ${JWT_SECRET}    # PHAI giong het gateway
      GRPC_URL: '0.0.0.0:50053'
    depends_on: { auth-db: { condition: service_healthy } }

  gateway:
    build: .
    command: ['node', '-r', 'ts-node/register', '-r', 'tsconfig-paths/register', 'apps/gateway/src/main.ts']
    environment:
      JWT_SECRET: ${JWT_SECRET}    # PHAI giong het auth-svc
      PORT: '3000'
    ports:
      - '3000:3000'                # dich vu DUY NHAT mo cong ra ngoai
    depends_on: [auth-svc, media-svc]

export JWT_SECRET=$(openssl rand -hex 32)
docker compose up -d --build

curl -s -X POST http://localhost:3000/auth/login \
  -H 'Content-Type: application/json' \
  -d '{"email":"demo@test.local","password":"matkhau123"}'
# -> {"accessToken":"eyJhbGciOi..."}  + Set-Cookie: refresh_token=...

curl -s -o /dev/null -w '%{http_code}\n' -X POST http://localhost:3000/auth/login \
  -H 'Content-Type: application/json' \
  -d '{"email":"demo@test.local","password":"saibetbet"}'
# -> 401   (GrpcStatus.UNAUTHENTICATED -> UnauthorizedException, muc 3.2)

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

// proto/media.proto
rpc GetBalance (GetBalanceRequest) returns (BalanceReply);

// apps/media-svc/src/media.controller.ts
@GrpcMethod('MediaService', 'GetBalance')
async getBalance(request: GetBalanceRequest): Promise<BalanceReply> {
  return { balance: await this.balances.getBalance(request.userId) };
}

# Xoa khoa de cache LANH, roi ban 50 loi goi GetBalance cung luc,
# vua chay vua theo doi Redis:
docker exec ...redis-1 redis-cli DEL "balance:$USER_ID"
docker exec ...redis-1 redis-cli MONITOR &

# ...ban 50 request...

# Dem lenh cham vao khoa do:
# GET : 50      <-- ca 50 deu kiem cache
# SET : 1       <-- nhung CHI MOT lan thuc su goi sang billing-svc
#
# va dong SET do:
# "set" "balance:961EC050-..." "-10" "EX" "63"
#                                          ^^ khong phai 60

import { Injectable, NestMiddleware } from '@nestjs/common';
import { AsyncLocalStorage } from 'node:async_hooks';
import { randomUUID } from 'node:crypto';
// PHAI lay tu 'express'. Thieu dong nay thi TypeScript im lang dung
// Request/Response cua DOM, va bao loi o cho khac han: "Headers has no
// index signature" va "Property 'setHeader' does not exist".
import type { Request, Response, NextFunction } from 'express';

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 { Metadata } from '@grpc/grpc-js';
import { requestContext } from './correlation.middleware';

// Dinh correlation id vao metadata cua moi loi goi gRPC di ra. Moi method
// sinh tu ts-proto deu nhan Metadata lam THAM SO THU HAI:
//   this.service.charge({ userId, amount, jobId }, correlationMetadata())
export function correlationMetadata(): Metadata {
  const metadata = new Metadata();
  metadata.set('x-correlation-id', requestContext.getStore()?.correlationId ?? 'unknown');
  return metadata;
}

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