Domain: E-commerce & Marketplace

SPU (Standard Product Unit) = Sản phẩm chung
  Ví dụ: iPhone 15 Pro

SKU (Stock Keeping Unit) = Biến thể cụ thể
  Ví dụ: iPhone 15 Pro - 256GB - Titan Tự Nhiên
         iPhone 15 Pro - 512GB - Titan Xanh
  → Mỗi SKU có price, stock, barcode riêng

-- SPU — thông tin chung của sản phẩm
CREATE TABLE products (
    id              BIGSERIAL PRIMARY KEY,
    product_id      VARCHAR(50) UNIQUE NOT NULL,   -- SPU code
    name            TEXT NOT NULL,
    description     TEXT,
    brand_id        BIGINT REFERENCES brands(id),
    category_id     BIGINT REFERENCES categories(id),
    status          VARCHAR(20) NOT NULL,          -- ACTIVE, DELISTED, DRAFT
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- SKU — biến thể cụ thể với attributes
CREATE TABLE product_variants (
    id              BIGSERIAL PRIMARY KEY,
    sku             VARCHAR(50) UNIQUE NOT NULL,
    product_id      BIGINT REFERENCES products(id),
    attributes      JSONB NOT NULL,                -- {color: "blue", size: "256GB"}
    price           BIGINT NOT NULL,               -- đơn vị: đồng
    stock           INT NOT NULL DEFAULT 0,
    weight_grams    INT,                           -- shipping calculation
    is_active       BOOLEAN NOT NULL DEFAULT true,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT chk_price CHECK (price >= 0),
    CONSTRAINT chk_stock CHECK (stock >= 0)
);

-- Index quan trọng để tránh overselling (race condition)
CREATE INDEX idx_sku_stock ON product_variants(sku) WHERE stock > 0;
CREATE INDEX idx_product_category ON products(category_id) WHERE status = 'ACTIVE';

// Elasticsearch document schema
{
  "sku": "IPH15P-256-TBL",
  "product_id": "IPH15P",
  "name": "iPhone 15 Pro Max 256GB Titan Blue",
  "price": 29990000,
  "stock": 150,
  "category": ["electronics", "phones", "iphone"],
  "brand": "Apple",
  "attributes": {
    "color": "Titan Blue",
    "storage": "256GB",
    "screen_size": "6.7 inch"
  },
  "rating_avg": 4.8,
  "sold_count": 12450,
  "boost_score": 1.5,  // manual ranking boost
  "created_at": "2024-09-22T10:00:00Z"
}

# Python với elasticsearch-py
search_query = {
    "query": {
        "function_score": {
            "query": {
                "bool": {
                    "must": [
                        {"multi_match": {
                            "query": "iphone 15 pro",
                            "fields": ["name^3", "brand^2", "category"],
                            "type": "best_fields",
                            "fuzziness": "AUTO"
                        }}
                    ],
                    "filter": [
                        {"term": {"is_active": True}},
                        {"range": {"stock": {"gt": 0}}},
                        {"range": {"price": {"gte": 10000000, "lte": 50000000}}}
                    ]
                }
            },
            "functions": [
                {"field_value_factor": {
                    "field": "sold_count",
                    "modifier": "log1p",
                    "factor": 0.1
                }},
                {"field_value_factor": {
                    "field": "rating_avg",
                    "modifier": "sqrt",
                    "factor": 0.5
                }},
                {"gauss": {  # decay score theo thời gian (ưu tiên mới)
                    "created_at": {
                        "origin": "now",
                        "scale": "30d",
                        "decay": 0.5
                    }
                }}
            ],
            "score_mode": "sum",
            "boost_mode": "multiply"
        }
    },
    "size": 20,
    "from": 0
}

Flash Sale: Còn 1 sản phẩm cuối cùng, 2 user click cùng lúc

User A                     Database                     User B
  │                            │                           │
  │─ SELECT stock─────────────►│                           │
  │◄─ stock=1 ─────────────────│                           │
  │                            │◄───────── SELECT stock ───│
  │                            │─ stock=1 ─────────────────►│
  │─ UPDATE stock=0 ───────────►│                           │
  │                            │◄───────── UPDATE stock=0 ─│
  └─ Success                   │                           └─ Success
                  
  Kết quả: Bán 2 sản phẩm nhưng chỉ có 1 → OVERSELLING!

BEGIN;

-- Lock row trong khi transaction chưa commit
SELECT stock
FROM product_variants
WHERE sku = 'IPH15P-256-TBL'
FOR UPDATE;  -- ← Pessimistic lock: User B phải đợi User A commit

-- Check stock trước khi trừ
IF stock < requested_quantity THEN
    ROLLBACK;
    RAISE EXCEPTION 'insufficient_stock';
END IF;

-- Update stock
UPDATE product_variants
SET stock = stock - requested_quantity,
    updated_at = now()
WHERE sku = 'IPH15P-256-TBL';

-- Tạo order
INSERT INTO orders (user_id, sku, quantity, price) 
VALUES (...);

COMMIT;

CREATE TABLE product_variants (
    ...
    stock       INT NOT NULL,
    version     BIGINT NOT NULL DEFAULT 0,  -- ← Version field
    ...
);

-- Go code
func DeductStock(db *sql.DB, sku string, qty int) error {
    tx, _ := db.Begin()
    defer tx.Rollback()

    var currentStock, version int64
    err := tx.QueryRow(`
        SELECT stock, version
        FROM product_variants
        WHERE sku = $1
    `, sku).Scan(&currentStock, &version)
    if err != nil {
        return err
    }

    if currentStock < int64(qty) {
        return ErrInsufficientStock
    }

    // Update chỉ thành công nếu version không thay đổi
    result, err := tx.Exec(`
        UPDATE product_variants
        SET stock = stock - $1,
            version = version + 1,
            updated_at = now()
        WHERE sku = $2
          AND version = $3  -- ← Optimistic lock check
    `, qty, sku, version)
    if err != nil {
        return err
    }

    rowsAffected, _ := result.RowsAffected()
    if rowsAffected == 0 {
        return ErrVersionMismatch // retry by client
    }

    return tx.Commit()
}

Thay vì trừ stock trực tiếp DB mỗi lần click:

1. Pre-load hot SKUs vào Redis khi flash sale bắt đầu
   
   HSET flash:IPH15P-256  stock 1000
   HSET flash:IPH15P-256  price 25990000

2. User click "Mua" → Trừ stock từ Redis (atomic DECR)
   
   stock_left = DECR flash:IPH15P-256:stock
   IF stock_left < 0:
       INCR flash:IPH15P-256:stock  -- rollback
       RETURN "sold_out"

3. Background job sync Redis → PostgreSQL mỗi 10s
   (eventual consistency acceptable trong flash sale)

4. Sau flash sale, reconcile Redis vs DB

func ReserveFlashSaleStock(ctx context.Context, rdb *redis.Client, sku string, qty int) error {
    script := `
        local stock = redis.call('HGET', KEYS[1], 'stock')
        if stock == false then
            return -1  -- key not found
        end
        
        stock = tonumber(stock)
        if stock < tonumber(ARGV[1]) then
            return 0  -- insufficient
        end
        
        redis.call('HINCRBY', KEYS[1], 'stock', -tonumber(ARGV[1]))
        return 1  -- success
    `
    
    result, err := rdb.Eval(ctx, script, []string{"flash:" + sku}, qty).Int()
    if err != nil {
        return err
    }
    
    switch result {
    case -1:
        return ErrFlashSaleNotFound
    case 0:
        return ErrSoldOut
    default:
        return nil
    }
}

Order Lifecycle:

 PENDING ────────► CONFIRMED ──────► PROCESSING ──────► SHIPPED ──────► DELIVERED
    │                   │                                                    │
    │                   │                                                    ▼
    │                   │                                               COMPLETED
    │                   │
    │                   ▼ (payment timeout 15 mins)
    └──────────────► CANCELLED
                        ▲
                        │
                        │ (user cancel before ship)
                        │
            ┌───────────┴────────────────┐
      CONFIRMED              PROCESSING

State transitions:
  PENDING → CONFIRMED:    payment success
  PENDING → CANCELLED:    payment timeout (TTL 15 mins)
  CONFIRMED → PROCESSING: merchant pack hàng
  CONFIRMED → CANCELLED:  user hoặc merchant cancel
  PROCESSING → SHIPPED:   handover to shipper
  SHIPPED → DELIVERED:    GPS confirm delivered
  DELIVERED → COMPLETED:  không return trong 7 ngày

CREATE TABLE orders (
    id              BIGSERIAL PRIMARY KEY,
    order_no        VARCHAR(20) UNIQUE NOT NULL,  -- ORD20240412-XXXXXX
    user_id         BIGINT NOT NULL,
    merchant_id     BIGINT NOT NULL,
    status          VARCHAR(20) NOT NULL,
    subtotal        BIGINT NOT NULL,
    shipping_fee    BIGINT NOT NULL DEFAULT 0,
    discount        BIGINT NOT NULL DEFAULT 0,
    total           BIGINT NOT NULL,
    paid_at         TIMESTAMPTZ,
    expired_at      TIMESTAMPTZ,                 -- payment deadline
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT chk_total CHECK (total = subtotal + shipping_fee - discount),
    CONSTRAINT chk_subtotal CHECK (subtotal >= 0)
);

CREATE TABLE order_items (
    id              BIGSERIAL PRIMARY KEY,
    order_id        BIGINT REFERENCES orders(id),
    sku             VARCHAR(50) NOT NULL,
    product_name    TEXT NOT NULL,              -- snapshot để tránh product đổi tên sau
    quantity        INT NOT NULL,
    unit_price      BIGINT NOT NULL,            -- snapshot giá tại thời điểm mua
    subtotal        BIGINT NOT NULL,
    CONSTRAINT chk_qty CHECK (quantity > 0),
    CONSTRAINT chk_subtotal CHECK (subtotal = quantity * unit_price)
);

CREATE TABLE order_status_history (
    id              BIGSERIAL PRIMARY KEY,
    order_id        BIGINT REFERENCES orders(id),
    from_status     VARCHAR(20),
    to_status       VARCHAR(20) NOT NULL,
    reason          TEXT,
    actor_id        BIGINT,                     -- user/merchant/system
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

Khi user click "Đặt hàng":

1. Validate cart (giá, stock availability)
2. Reserve inventory (trừ stock tạm thời)
3. Calculate shipping fee (gọi 3PL API)
4. Apply discount/voucher (check validity, usage limit)
5. Create order record
6. Call payment gateway (MoMo, ZaloPay, VNPay)
7. Wait payment callback
8. Confirm order & finalize inventory deduction

Nếu step nào fail → compensate các steps trước đó (Saga pattern):

type OrderSaga struct {
    db          *sql.DB
    inventory   InventoryService
    shipping    ShippingService
    payment     PaymentService
    voucher     VoucherService
}

func (s *OrderSaga) CreateOrder(ctx context.Context, req CreateOrderRequest) (*Order, error) {
    sagaID := uuid.New()
    
    // Step 1: Reserve inventory
    reservationID, err := s.inventory.Reserve(ctx, req.Items)
    if err != nil {
        return nil, err
    }
    defer func() {
        if err != nil {
            s.inventory.ReleaseReservation(ctx, reservationID) // compensate
        }
    }()
    
    // Step 2: Calculate shipping
    shippingFee, err := s.shipping.Calculate(ctx, req.Address, req.Items)
    if err != nil {
        return nil, err
    }
    
    // Step 3: Apply voucher
    var discount int64
    if req.VoucherCode != "" {
        discount, err = s.voucher.Apply(ctx, req.UserID, req.VoucherCode, req.Subtotal)
        if err != nil {
            return nil, err
        }
        defer func() {
            if err != nil {
                s.voucher.Release(ctx, req.UserID, req.VoucherCode) // compensate
            }
        }()
    }
    
    // Step 4: Create pending order
    order := &Order{
        OrderNo:     generateOrderNo(),
        UserID:      req.UserID,
        Status:      "PENDING",
        Subtotal:    req.Subtotal,
        ShippingFee: shippingFee,
        Discount:    discount,
        Total:       req.Subtotal + shippingFee - discount,
        ExpiredAt:   time.Now().Add(15 * time.Minute),
    }
    
    if err := s.db.Insert(ctx, order); err != nil {
        return nil, err
    }
    
    // Step 5: Initiate payment
    paymentURL, err := s.payment.CreatePaymentLink(ctx, PaymentRequest{
        OrderNo:     order.OrderNo,
        Amount:      order.Total,
        CallbackURL: fmt.Sprintf("https://api.shop.com/payment/callback/%s", order.OrderNo),
    })
    if err != nil {
        return nil, err
    }
    
    order.PaymentURL = paymentURL
    return order, nil
}

// Payment callback handler (webhook từ MoMo/ZaloPay)
func (s *OrderSaga) HandlePaymentCallback(ctx context.Context, callback PaymentCallback) error {
    order, err := s.db.GetOrderByNo(ctx, callback.OrderNo)
    if err != nil {
        return err
    }
    
    if callback.Status == "SUCCESS" {
        // Confirm order & finalize inventory
        if err := s.inventory.ConfirmReservation(ctx, order.ReservationID); err != nil {
            // Critical error: payment succeeded but inventory confirm failed
            // → Manual intervention hoặc auto-refund
            s.alertOncall(ctx, "payment_inventory_mismatch", order.OrderNo)
            return err
        }
        
        return s.db.UpdateOrderStatus(ctx, order.ID, "CONFIRMED")
    } else {
        // Payment failed/cancelled → release reservation
        s.inventory.ReleaseReservation(ctx, order.ReservationID)
        return s.db.UpdateOrderStatus(ctx, order.ID, "CANCELLED")
    }
}

Base Price:        Giá gốc của merchant set
List Price:        Giá hiển thị (có thể = base hoặc có markup)
Sale Price:        Giá khuyến mại (flash sale, voucher)
Final Price:       Giá user thực trả = Sale Price - Voucher + Shipping

def calculate_dynamic_price(sku: str, base_price: int) -> int:
    """
    Tăng giá khi:
    - Traffic cao (nhiều người đang xem sản phẩm này)
    - Stock thấp (khan hiếm)
    - Competitor bán hết (monopoly tạm thời)
    """
    # Fetch real-time metrics
    concurrent_viewers = redis.get(f"viewers:{sku}") or 0
    stock_left = redis.hget(f"inventory:{sku}", "stock") or 0
    competitor_stock = fetch_competitor_stock(sku)  # crawl data
    
    surge_multiplier = 1.0
    
    # Traffic surge
    if concurrent_viewers > 1000:
        surge_multiplier *= 1.1
    if concurrent_viewers > 5000:
        surge_multiplier *= 1.2
    
    # Low stock
    if stock_left < 10:
        surge_multiplier *= 1.15
    
    # Competitor out of stock
    if competitor_stock == 0:
        surge_multiplier *= 1.25
    
    # Cap at 150% base price (tránh PR disaster)
    surge_multiplier = min(surge_multiplier, 1.5)
    
    return int(base_price * surge_multiplier)

CREATE TABLE vouchers (
    id                  BIGSERIAL PRIMARY KEY,
    code                VARCHAR(20) UNIQUE NOT NULL,
    type                VARCHAR(20) NOT NULL,  -- PERCENTAGE, FIXED_AMOUNT, FREE_SHIP
    value               BIGINT NOT NULL,       -- 10 (10%) hoặc 50000 (50k VND)
    min_order_value     BIGINT NOT NULL DEFAULT 0,
    max_discount        BIGINT,                -- cap discount (10% max 100k)
    max_usage_per_user  INT DEFAULT 1,
    total_usage_limit   INT,                   -- tổng số lượt dùng toàn platform
    valid_from          TIMESTAMPTZ NOT NULL,
    valid_until         TIMESTAMPTZ NOT NULL,
    is_active           BOOLEAN NOT NULL DEFAULT true,
    created_at          TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE voucher_usages (
    id          BIGSERIAL PRIMARY KEY,
    voucher_id  BIGINT REFERENCES vouchers(id),
    user_id     BIGINT NOT NULL,
    order_id    BIGINT REFERENCES orders(id),
    used_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE(voucher_id, order_id)  -- prevent double-use
);

func ApplyVoucher(ctx context.Context, db *sql.DB, userID int64, code string, orderValue int64) (int64, error) {
    tx, _ := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
    defer tx.Rollback()
    
    // Lock voucher row
    var v Voucher
    err := tx.QueryRowContext(ctx, `
        SELECT id, type, value, min_order_value, max_discount, 
               max_usage_per_user, total_usage_limit
        FROM vouchers
        WHERE code = $1
          AND is_active = true
          AND now() BETWEEN valid_from AND valid_until
        FOR UPDATE
    `, code).Scan(&v.ID, &v.Type, &v.Value, &v.MinOrderValue, 
                   &v.MaxDiscount, &v.MaxUsagePerUser, &v.TotalUsageLimit)
    if err != nil {
        return 0, ErrVoucherNotFound
    }
    
    // Check min order value
    if orderValue < v.MinOrderValue {
        return 0, ErrMinOrderValueNotMet
    }
    
    // Check per-user usage
    var userUsageCount int
    tx.QueryRowContext(ctx, `
        SELECT COUNT(*) FROM voucher_usages
        WHERE voucher_id = $1 AND user_id = $2
    `, v.ID, userID).Scan(&userUsageCount)
    
    if userUsageCount >= v.MaxUsagePerUser {
        return 0, ErrVoucherLimitReached
    }
    
    // Check total usage
    if v.TotalUsageLimit > 0 {
        var totalUsageCount int
        tx.QueryRowContext(ctx, `
            SELECT COUNT(*) FROM voucher_usages WHERE voucher_id = $1
        `, v.ID).Scan(&totalUsageCount)
        
        if totalUsageCount >= v.TotalUsageLimit {
            return 0, ErrVoucherSoldOut
        }
    }
    
    // Calculate discount
    var discount int64
    switch v.Type {
    case "PERCENTAGE":
        discount = orderValue * v.Value / 100
        if v.MaxDiscount > 0 && discount > v.MaxDiscount {
            discount = v.MaxDiscount
        }
    case "FIXED_AMOUNT":
        discount = v.Value
    }
    
    // Reserve voucher (insert pending usage)
    _, err = tx.ExecContext(ctx, `
        INSERT INTO voucher_usages (voucher_id, user_id, order_id)
        VALUES ($1, $2, NULL)  -- order_id update sau khi order confirm
    `, v.ID, userID)
    if err != nil {
        return 0, err
    }
    
    tx.Commit()
    return discount, nil
}

User-based CF:
  "Người dùng giống bạn (similar purchase history) đã mua gì?"

Item-based CF:
  "Người mua sản phẩm này cũng mua gì?"  ← Phổ biến hơn, stable hơn

from pyspark.ml.recommendation import ALS
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ItemRecommendation").getOrCreate()

# Load order history
orders = spark.read.parquet("s3://data/orders_history")

# Prepare training data: (user_id, sku, implicit_rating)
interactions = orders.groupBy("user_id", "sku").agg(
    count("*").alias("purchase_count")
).withColumn(
    "rating", col("purchase_count").cast("float")  # implicit feedback
)

# ALS model
als = ALS(
    maxIter=10,
    regParam=0.1,
    userCol="user_id",
    itemCol="sku",
    ratingCol="rating",
    coldStartStrategy="drop",
    implicitPrefs=True  # implicit feedback (clicks/purchases)
)

model = als.fit(interactions)

# Generate item-to-item similarity
item_factors = model.itemFactors  # latent factors cho mỗi SKU
# Tính cosine similarity giữa các SKUs
# Lưu vào Redis: ZADD reco:IPH15P-256  0.95 SAMSG-S24  0.89 XIAOMI14

# Serving: Khi user xem iPhone 15 Pro → query Redis để lấy top similar items

Feature Engineering:
  User features:
    - Demographics: age, gender, location
    - Behavioral: avg_order_value, purchase_frequency, favorite_categories
    - Session: current_cart_items, browsing_history_last_30mins
  
  Item features:
    - Category, brand, price_range
    - Popularity: views_7d, orders_7d, rating_avg
    - Merchant: merchant_rating, fulfillment_speed

Model: LightGBM Ranker
  Input:  user features + item features + context (time, device, weather?!)
  Output: CTR prediction score
  
  Train trên historical click/purchase data
  Deploy qua model serving (TensorFlow Serving / Triton)

import redis
import requests

def get_personalized_recommendations(user_id: int, context: dict) -> list:
    # Bước 1: Candidate Generation (fast retrieval)
    # Lấy 200 candidates từ multiple sources:
    candidates = []
    
    # Source 1: User's favorite categories
    fav_categories = redis.smembers(f"user:{user_id}:fav_categories")
    for cat in fav_categories[:3]:
        candidates.extend(redis.zrevrange(f"popular:{cat}", 0, 30))
    
    # Source 2: Collaborative filtering
    candidates.extend(redis.zrevrange(f"cf:{user_id}", 0, 50))
    
    # Source 3: Trending items
    candidates.extend(redis.zrevrange("trending:today", 0, 50))
    
    # Deduplicate
    candidates = list(set(candidates))[:200]
    
    # Bước 2: Ranking (ML model scoring)
    user_features = get_user_features(user_id)
    item_features_batch = get_item_features(candidates)
    
    scores = []
    for sku, item_feat in zip(candidates, item_features_batch):
        features = {**user_features, **item_feat, **context}
        score = model_server.predict(features)  # gRPC call to Triton
        scores.append((sku, score))
    
    # Sort by score descending
    scores.sort(key=lambda x: x[1], reverse=True)
    
    # Bước 3: Business rules & diversity
    # - Filter out-of-stock items
    # - Đảm bảo diverse categories (không recommend toàn iPhone)
    # - Boost margin (nếu có 2 items cùng score, chọn item có margin cao hơn)
    
    final_recommendations = apply_business_rules(scores[:50])
    return final_recommendations[:20]

┌──────────────┐         ┌───────────────────┐         ┌──────────────┐
│              │         │  CDN (Cloudflare) │         │              │
│  User (app)  ├────────►│  - Static assets  ├────────►│  API Gateway │
│              │         │  - Rate limiting  │         │  (Kong/Envoy)│
└──────────────┘         └───────────────────┘         └───────┬──────┘
                                                               │
          ┌────────────────────────────────────────────────────┼────────┐
          │                                                    │        │
          ▼                                                    ▼        ▼
┌──────────────────┐                             ┌──────────────────────────┐
│ Flash Sale Svc   │                             │ Order Service            │
│ (Pre-reserve)    │                             │ (Async processing)       │
│                  │                             │                          │
│ Redis Cluster ───┼────sync every 10s──────────►│ PostgreSQL               │
│ (hot inventory)  │                             │ (source of truth)        │
└──────────────────┘                             └──────────────────────────┘
          │
          └───────► Message Queue (Kafka) ─────────► Workers (create order)

-- Redis Lua script: Token bucket rate limiter
-- KEYS[1] = "rate_limit:user:{user_id}"
-- ARGV[1] = max_tokens (e.g., 10)
-- ARGV[2] = refill_rate (e.g., 1 token per second)
-- ARGV[3] = current_timestamp

local key = KEYS[1]
local max_tokens = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or max_tokens
local last_refill = tonumber(bucket[2]) or now

-- Refill tokens dựa trên thời gian trôi qua
local elapsed = now - last_refill
local refill = math.floor(elapsed * refill_rate)
tokens = math.min(max_tokens, tokens + refill)

if tokens >= 1 then
    tokens = tokens - 1
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', key, 60)  -- TTL 60s
    return 1  -- allowed
else
    return 0  -- rate limited
end

func CheckRateLimit(ctx context.Context, rdb *redis.Client, userID int64) (bool, error) {
    script := `...` // Lua script ở trên
    
    result, err := rdb.Eval(ctx, script,
        []string{fmt.Sprintf("rate_limit:user:%d", userID)},
        10,                        // max 10 requests
        0.5,                       // refill 1 token per 2 seconds
        time.Now().Unix(),
    ).Int()
    
    return result == 1, err
}

Thay vì xử lý order synchronously trong request:

User click "Mua"
  ↓
API: Reserve stock (Redis) + Publish Kafka event
  ↓ (return immediately với order_id pending)
User nhận response: "Đơn hàng đang xử lý, vui lòng chờ..."
  ↓
Background worker consume Kafka:
  - Validate payment
  - Sync stock to DB
  - Create order record
  - Send notification
  ↓
WebSocket push update: "Đơn hàng thành công!"

// Producer: API handler
func HandleFlashSalePurchase(w http.ResponseWriter, r *http.Request) {
    var req PurchaseRequest
    json.NewDecoder(r.Body).Decode(&req)
    
    // Rate limit
    allowed, _ := CheckRateLimit(ctx, redisClient, req.UserID)
    if !allowed {
        http.Error(w, "Too many requests", http.StatusTooManyRequests)
        return
    }
    
    // Reserve stock in Redis
    err := ReserveFlashSaleStock(ctx, redisClient, req.SKU, req.Quantity)
    if err != nil {
        http.Error(w, "Sold out", http.StatusConflict)
        return
    }
    
    // Publish to Kafka (async processing)
    orderID := uuid.New().String()
    event := OrderCreatedEvent{
        OrderID:  orderID,
        UserID:   req.UserID,
        SKU:      req.SKU,
        Quantity: req.Quantity,
    }
    kafkaProducer.Publish("order-created", orderID, event)
    
    // Return immediately
    json.NewEncoder(w).Encode(map[string]string{
        "order_id": orderID,
        "status":   "PENDING",
    })
}

// Consumer: Background worker
func ProcessOrderCreated(msg kafka.Message) error {
    var event OrderCreatedEvent
    json.Unmarshal(msg.Value, &event)
    
    // Actual DB transaction
    tx, _ := db.Begin()
    defer tx.Rollback()
    
    // Deduct stock from PostgreSQL
    _, err := tx.Exec(`
        UPDATE product_variants
        SET stock = stock - $1
        WHERE sku = $2 AND stock >= $1
    `, event.Quantity, event.SKU)
    if err != nil {
        // Rollback Redis reservation
        redisClient.HIncrBy(ctx, "flash:"+event.SKU, "stock", int64(event.Quantity))
        return err
    }
    
    // Create order
    tx.Exec(`INSERT INTO orders (...) VALUES (...)`)
    tx.Commit()
    
    // Push notification via WebSocket
    websocketHub.SendToUser(event.UserID, "order_confirmed", event.OrderID)
    return nil
}