10 - 电商系统设计

package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
    
    "github.com/google/uuid"
    "github.com/gorilla/mux"
)

// ==================== 数据结构 ====================

// Product 商品
type Product struct {
    SkuID         int64   `json:"sku_id"`
    Title         string  `json:"title"`
    Price         float64 `json:"price"`
    Stock         int     `json:"stock"`
    Sales         int     `json:"sales"`
    ImageURL      string  `json:"image_url"`
}

// CartItem 购物车项
type CartItem struct {
    ID       int64   `json:"id"`
    UserID   int64   `json:"user_id"`
    SkuID    int64   `json:"sku_id"`
    Quantity int     `json:"quantity"`
    Selected bool    `json:"selected"`
}

// Order 订单
type Order struct {
    OrderID        string    `json:"order_id"`
    OrderSn        string    `json:"order_sn"`
    UserID         int64     `json:"user_id"`
    Status         string    `json:"status"`
    TotalAmount    float64   `json:"total_amount"`
    DiscountAmount float64   `json:"discount_amount"`
    FinalAmount    float64   `json:"final_amount"`
    CreatedAt      time.Time `json:"created_at"`
}

// OrderItem 订单商品
type OrderItem struct {
    OrderID     string  `json:"order_id"`
    SkuID       int64   `json:"sku_id"`
    Title       string  `json:"title"`
    Price       float64 `json:"price"`
    Quantity    int     `json:"quantity"`
    TotalAmount float64 `json:"total_amount"`
}

// ==================== 商品服务 ====================

type ProductService struct {
    db *sql.DB
}

// GetProducts 获取商品列表
func (s *ProductService) GetProducts(categoryID int, page, pageSize int) ([]*Product, error) {
    offset := (page - 1) * pageSize
    
    rows, err := s.db.Query(`
        SELECT sku_id, title, price, stock, sales, image_url
        FROM products_sku
        WHERE status = 'active'
        ORDER BY sales DESC
        LIMIT ? OFFSET ?
    `, pageSize, offset)
    
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    
    products := make([]*Product, 0)
    for rows.Next() {
        var p Product
        rows.Scan(&p.SkuID, &p.Title, &p.Price, &p.Stock, &p.Sales, &p.ImageURL)
        products = append(products, &p)
    }
    
    return products, nil
}

// GetProduct 获取商品详情
func (s *ProductService) GetProduct(skuID int64) (*Product, error) {
    var p Product
    err := s.db.QueryRow(`
        SELECT sku_id, title, price, stock, sales, image_url
        FROM products_sku
        WHERE sku_id = ? AND status = 'active'
    `, skuID).Scan(&p.SkuID, &p.Title, &p.Price, &p.Stock, &p.Sales, &p.ImageURL)
    
    return &p, err
}

// ==================== 购物车服务 ====================

type CartService struct {
    db *sql.DB
}

// AddToCart 添加到购物车
func (s *CartService) AddToCart(userID, skuID int64, quantity int) error {
    // 检查商品是否存在
    var stock int
    err := s.db.QueryRow(`SELECT stock FROM products_sku WHERE sku_id = ?`, skuID).Scan(&stock)
    if err != nil {
        return fmt.Errorf("商品不存在")
    }
    
    if stock < quantity {
        return fmt.Errorf("库存不足")
    }
    
    // 插入或更新购物车
    _, err = s.db.Exec(`
        INSERT INTO shopping_cart (user_id, sku_id, quantity)
        VALUES (?, ?, ?)
        ON DUPLICATE KEY UPDATE quantity = quantity + ?
    `, userID, skuID, quantity, quantity)
    
    return err
}

// GetCartItems 获取购物车列表
func (s *CartService) GetCartItems(userID int64) ([]*CartItem, error) {
    rows, err := s.db.Query(`
        SELECT id, user_id, sku_id, quantity, selected
        FROM shopping_cart
        WHERE user_id = ?
    `, userID)
    
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    
    items := make([]*CartItem, 0)
    for rows.Next() {
        var item CartItem
        rows.Scan(&item.ID, &item.UserID, &item.SkuID, &item.Quantity, &item.Selected)
        items = append(items, &item)
    }
    
    return items, nil
}

// ==================== 订单服务 ====================

type OrderService struct {
    db             *sql.DB
    inventoryService *InventoryService
}

// CreateOrder 创建订单
func (s *OrderService) CreateOrder(userID int64, cartItemIDs []int64, addressID int64) (*Order, error) {
    // 1. 开启事务
    tx, err := s.db.Begin()
    if err != nil {
        return nil, err
    }
    defer tx.Rollback()
    
    // 2. 查询购物车商品
    items, err := s.getCartItemsByIDs(tx, cartItemIDs)
    if err != nil {
        return nil, err
    }
    
    // 3. 检查并锁定库存
    for _, item := range items {
        if err := s.inventoryService.LockStock(tx, item.SkuID, item.Quantity); err != nil {
            return nil, fmt.Errorf("库存不足: %w", err)
        }
    }
    
    // 4. 计算金额
    totalAmount := 0.0
    for _, item := range items {
        var price float64
        tx.QueryRow(`SELECT price FROM products_sku WHERE sku_id = ?`, item.SkuID).Scan(&price)
        totalAmount += price * float64(item.Quantity)
    }
    
    // 5. 创建订单
    orderID := generateOrderID()
    orderSn := generateOrderSn()
    
    _, err = tx.Exec(`
        INSERT INTO orders (
            order_id, order_sn, user_id, status, 
            total_amount, final_amount, address_id
        ) VALUES (?, ?, ?, 'unpaid', ?, ?, ?)
    `, orderID, orderSn, userID, totalAmount, totalAmount, addressID)
    
    if err != nil {
        return nil, err
    }
    
    // 6. 创建订单商品
    for _, item := range items {
        var title string
        var price float64
        tx.QueryRow(`SELECT title, price FROM products_sku WHERE sku_id = ?`, item.SkuID).
            Scan(&title, &price)
        
        _, err = tx.Exec(`
            INSERT INTO order_items (
                order_id, sku_id, title, price, quantity, total_amount
            ) VALUES (?, ?, ?, ?, ?, ?)
        `, orderID, item.SkuID, title, price, item.Quantity, price*float64(item.Quantity))
        
        if err != nil {
            return nil, err
        }
    }
    
    // 7. 删除购物车
    _, err = tx.Exec(`DELETE FROM shopping_cart WHERE id IN (?)`, cartItemIDs)
    if err != nil {
        return nil, err
    }
    
    // 8. 提交事务
    if err = tx.Commit(); err != nil {
        return nil, err
    }
    
    // 9. 设置订单超时（15分钟未支付自动取消）
    go s.scheduleOrderTimeout(orderID, 15*time.Minute)
    
    order := &Order{
        OrderID:     orderID,
        OrderSn:     orderSn,
        UserID:      userID,
        Status:      "unpaid",
        TotalAmount: totalAmount,
        FinalAmount: totalAmount,
        CreatedAt:   time.Now(),
    }
    
    return order, nil
}

// PayOrder 支付订单
func (s *OrderService) PayOrder(orderID string, paymentMethod string) error {
    // 1. 开启事务
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 2. 查询订单
    var status string
    err = tx.QueryRow(`SELECT status FROM orders WHERE order_id = ? FOR UPDATE`, orderID).Scan(&status)
    if err != nil {
        return fmt.Errorf("订单不存在")
    }
    
    if status != "unpaid" {
        return fmt.Errorf("订单状态异常: %s", status)
    }
    
    // 3. 扣减库存（从预占转为实际扣减）
    items, err := s.getOrderItems(tx, orderID)
    if err != nil {
        return err
    }
    
    for _, item := range items {
        if err := s.inventoryService.DeductStock(tx, item.SkuID, item.Quantity); err != nil {
            return err
        }
    }
    
    // 4. 更新订单状态
    _, err = tx.Exec(`
        UPDATE orders 
        SET status = 'paid', paid_at = NOW()
        WHERE order_id = ?
    `, orderID)
    
    if err != nil {
        return err
    }
    
    // 5. 提交事务
    return tx.Commit()
}

// CancelOrder 取消订单
func (s *OrderService) CancelOrder(orderID string, reason string) error {
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 1. 查询订单
    var status string
    tx.QueryRow(`SELECT status FROM orders WHERE order_id = ?`, orderID).Scan(&status)
    
    if status == "paid" || status == "shipped" {
        return fmt.Errorf("订单不能取消")
    }
    
    // 2. 释放库存
    items, _ := s.getOrderItems(tx, orderID)
    for _, item := range items {
        s.inventoryService.UnlockStock(tx, item.SkuID, item.Quantity)
    }
    
    // 3. 更新订单状态
    _, err = tx.Exec(`
        UPDATE orders 
        SET status = 'cancelled', cancelled_at = NOW(), cancel_reason = ?
        WHERE order_id = ?
    `, reason, orderID)
    
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

// scheduleOrderTimeout 订单超时处理
func (s *OrderService) scheduleOrderTimeout(orderID string, timeout time.Duration) {
    time.Sleep(timeout)
    
    // 查询订单状态
    var status string
    s.db.QueryRow(`SELECT status FROM orders WHERE order_id = ?`, orderID).Scan(&status)
    
    if status == "unpaid" {
        s.CancelOrder(orderID, "超时未支付")
    }
}

func (s *OrderService) getCartItemsByIDs(tx *sql.Tx, ids []int64) ([]*CartItem, error) {
    // 简化实现
    return nil, nil
}

func (s *OrderService) getOrderItems(tx *sql.Tx, orderID string) ([]*OrderItem, error) {
    rows, err := tx.Query(`
        SELECT order_id, sku_id, title, price, quantity, total_amount
        FROM order_items
        WHERE order_id = ?
    `, orderID)
    
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    
    items := make([]*OrderItem, 0)
    for rows.Next() {
        var item OrderItem
        rows.Scan(&item.OrderID, &item.SkuID, &item.Title, &item.Price, &item.Quantity, &item.TotalAmount)
        items = append(items, &item)
    }
    
    return items, nil
}

// ==================== 库存服务 ====================

type InventoryService struct {
    db *sql.DB
}

// LockStock 锁定库存（预占）
func (s *InventoryService) LockStock(tx *sql.Tx, skuID int64, quantity int) error {
    // 使用乐观锁
    result, err := tx.Exec(`
        UPDATE inventory 
        SET available_stock = available_stock - ?,
            locked_stock = locked_stock + ?,
            version = version + 1
        WHERE sku_id = ? 
          AND available_stock >= ?
          AND version = (SELECT version FROM (SELECT version FROM inventory WHERE sku_id = ?) AS tmp)
    `, quantity, quantity, skuID, quantity, skuID)
    
    if err != nil {
        return err
    }
    
    rowsAffected, _ := result.RowsAffected()
    if rowsAffected == 0 {
        return fmt.Errorf("库存不足或版本冲突")
    }
    
    // 记录流水
    s.logInventory(tx, skuID, "lock", quantity)
    
    return nil
}

// DeductStock 扣减库存（支付后）
func (s *InventoryService) DeductStock(tx *sql.Tx, skuID int64, quantity int) error {
    _, err := tx.Exec(`
        UPDATE inventory 
        SET locked_stock = locked_stock - ?,
            total_stock = total_stock - ?
        WHERE sku_id = ?
    `, quantity, quantity, skuID)
    
    if err != nil {
        return err
    }
    
    // 记录流水
    s.logInventory(tx, skuID, "deduct", quantity)
    
    return nil
}

// UnlockStock 解锁库存（取消订单）
func (s *InventoryService) UnlockStock(tx *sql.Tx, skuID int64, quantity int) error {
    _, err := tx.Exec(`
        UPDATE inventory 
        SET available_stock = available_stock + ?,
            locked_stock = locked_stock - ?
        WHERE sku_id = ?
    `, quantity, quantity, skuID)
    
    if err != nil {
        return err
    }
    
    // 记录流水
    s.logInventory(tx, skuID, "unlock", quantity)
    
    return nil
}

func (s *InventoryService) logInventory(tx *sql.Tx, skuID int64, opType string, quantity int) {
    logID := uuid.New().String()
    tx.Exec(`
        INSERT INTO inventory_logs (log_id, sku_id, operation_type, quantity, created_at)
        VALUES (?, ?, ?, ?, NOW())
    `, logID, skuID, opType, quantity)
}

// ==================== 辅助函数 ====================

func generateOrderID() string {
    return "ORD" + time.Now().Format("20060102") + uuid.New().String()[:8]
}

func generateOrderSn() string {
    return time.Now().Format("20060102150405") + fmt.Sprintf("%06d", time.Now().Nanosecond()%1000000)
}

// ==================== HTTP Handlers ====================

type EcommerceHandler struct {
    productService   *ProductService
    cartService      *CartService
    orderService     *OrderService
    inventoryService *InventoryService
}

func (h *EcommerceHandler) GetProductsHandler(w http.ResponseWriter, r *http.Request) {
    products, err := h.productService.GetProducts(0, 1, 20)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
        "data":    products,
    })
}

func (h *EcommerceHandler) AddToCartHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        SkuID    int64 `json:"sku_id"`
        Quantity int   `json:"quantity"`
    }
    
    json.NewDecoder(r.Body).Decode(&req)
    
    userID := int64(12345) // 从 token 获取
    
    err := h.cartService.AddToCart(userID, req.SkuID, req.Quantity)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
    })
}

func (h *EcommerceHandler) CreateOrderHandler(w http.ResponseWriter, r *http.Request) {
    var req struct {
        CartItemIDs []int64 `json:"cart_item_ids"`
        AddressID   int64   `json:"address_id"`
    }
    
    json.NewDecoder(r.Body).Decode(&req)
    
    userID := int64(12345) // 从 token 获取
    
    order, err := h.orderService.CreateOrder(userID, req.CartItemIDs, req.AddressID)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
        "data":    order,
    })
}

func main() {
    db, _ := sql.Open("mysql", "root:password@tcp(localhost:3306)/ecommerce")
    defer db.Close()
    
    productService := &ProductService{db: db}
    cartService := &CartService{db: db}
    inventoryService := &InventoryService{db: db}
    orderService := &OrderService{
        db:               db,
        inventoryService: inventoryService,
    }
    
    handler := &EcommerceHandler{
        productService:   productService,
        cartService:      cartService,
        orderService:     orderService,
        inventoryService: inventoryService,
    }
    
    r := mux.NewRouter()
    r.HandleFunc("/api/v1/products", handler.GetProductsHandler).Methods("GET")
    r.HandleFunc("/api/v1/cart/items", handler.AddToCartHandler).Methods("POST")
    r.HandleFunc("/api/v1/orders", handler.CreateOrderHandler).Methods("POST")
    
    fmt.Println("电商系统启动在 :8080")
    http.ListenAndServe(":8080", r)
}

// ==================== Redis 库存缓存 ====================

package cache

import (
    "context"
    "fmt"
    "strconv"
    
    "github.com/go-redis/redis/v8"
)

// InventoryCacheService 库存缓存服务
type InventoryCacheService struct {
    redis *redis.Client
}

// DecrStock 扣减库存（原子操作）
func (s *InventoryCacheService) DecrStock(skuID int64, quantity int) error {
    ctx := context.Background()
    key := fmt.Sprintf("inventory:sku:%d:stock", skuID)
    
    // Lua 脚本保证原子性
    script := redis.NewScript(`
        local stock = redis.call('GET', KEYS[1])
        if not stock then
            return -1
        end
        
        stock = tonumber(stock)
        local quantity = tonumber(ARGV[1])
        
        if stock < quantity then
            return -2
        end
        
        redis.call('DECRBY', KEYS[1], quantity)
        return stock - quantity
    `)
    
    result, err := script.Run(ctx, s.redis, []string{key}, quantity).Result()
    if err != nil {
        return err
    }
    
    remaining := result.(int64)
    if remaining == -1 {
        return fmt.Errorf("库存不存在")
    } else if remaining == -2 {
        return fmt.Errorf("库存不足")
    }
    
    return nil
}

// SyncToRedis 同步库存到 Redis
func (s *InventoryCacheService) SyncToRedis(skuID int64, stock int) error {
    ctx := context.Background()
    key := fmt.Sprintf("inventory:sku:%d:stock", skuID)
    
    return s.redis.Set(ctx, key, stock, 0).Err()
}

// ==================== 分布式锁 ====================

type DistributedLock struct {
    redis *redis.Client
}

// Lock 获取锁
func (l *DistributedLock) Lock(key string, expiration time.Duration) bool {
    ctx := context.Background()
    result, err := l.redis.SetNX(ctx, key, "1", expiration).Result()
    return err == nil && result
}

// Unlock 释放锁
func (l *DistributedLock) Unlock(key string) {
    ctx := context.Background()
    l.redis.Del(ctx, key)
}

// ==================== 订单创建流程（TCC）====================

type OrderTCCService struct {
    orderDB          *sql.DB
    inventoryService *InventoryCacheService
    paymentService   *PaymentService
}

// CreateOrderWithTCC 使用 TCC 创建订单
func (s *OrderTCCService) CreateOrderWithTCC(
    userID int64,
    items []*OrderItem,
    addressID int64,
) (*Order, error) {
    txID := generateTxID()
    
    // Try 阶段
    if err := s.tryCreateOrder(txID, userID, items, addressID); err != nil {
        s.cancelCreateOrder(txID)
        return nil, err
    }
    
    // Confirm 阶段
    if err := s.confirmCreateOrder(txID); err != nil {
        s.cancelCreateOrder(txID)
        return nil, err
    }
    
    return nil, nil
}

// tryCreateOrder Try 阶段：预占资源
func (s *OrderTCCService) tryCreateOrder(
    txID string,
    userID int64,
    items []*OrderItem,
    addressID int64,
) error {
    // 1. 检查库存并预占
    for _, item := range items {
        if err := s.inventoryService.DecrStock(item.SkuID, item.Quantity); err != nil {
            return err
        }
    }
    
    // 2. 创建预订单
    orderID := generateOrderID()
    _, err := s.orderDB.Exec(`
        INSERT INTO orders_temp (order_id, tx_id, user_id, status)
        VALUES (?, ?, ?, 'trying')
    `, orderID, txID, userID)
    
    return err
}

// confirmCreateOrder Confirm 阶段：确认提交
func (s *OrderTCCService) confirmCreateOrder(txID string) error {
    // 1. 将临时订单转为正式订单
    _, err := s.orderDB.Exec(`
        INSERT INTO orders (order_id, user_id, status)
        SELECT order_id, user_id, 'unpaid'
        FROM orders_temp
        WHERE tx_id = ? AND status = 'trying'
    `, txID)
    
    if err != nil {
        return err
    }
    
    // 2. 删除临时订单
    s.orderDB.Exec(`DELETE FROM orders_temp WHERE tx_id = ?`, txID)
    
    return nil
}

// cancelCreateOrder Cancel 阶段：回滚
func (s *OrderTCCService) cancelCreateOrder(txID string) error {
    // 1. 查询临时订单
    rows, err := s.orderDB.Query(`
        SELECT sku_id, quantity FROM order_items_temp WHERE tx_id = ?
    `, txID)
    
    if err != nil {
        return err
    }
    defer rows.Close()
    
    // 2. 释放库存
    for rows.Next() {
        var skuID int64
        var quantity int
        rows.Scan(&skuID, &quantity)
        
        // 库存回滚
        s.inventoryService.redis.IncrBy(
            context.Background(),
            fmt.Sprintf("inventory:sku:%d:stock", skuID),
            int64(quantity),
        )
    }
    
    // 3. 删除临时订单
    s.orderDB.Exec(`DELETE FROM orders_temp WHERE tx_id = ?`, txID)
    s.orderDB.Exec(`DELETE FROM order_items_temp WHERE tx_id = ?`, txID)
    
    return nil
}

func generateTxID() string {
    return "TCC_" + uuid.New().String()
}

type PaymentService struct{}

// ==================== 本地消息表方案 ====================

package eventual

import (
    "encoding/json"
)

// OrderEventPublisher 订单事件发布器
type OrderEventPublisher struct {
    db    *sql.DB
    kafka *KafkaProducer
}

// PublishOrderCreatedEvent 发布订单创建事件
func (p *OrderEventPublisher) PublishOrderCreatedEvent(order *Order) error {
    // 1. 在事务中插入订单和本地消息
    tx, err := p.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 插入订单
    _, err = tx.Exec(`INSERT INTO orders ...`, order)
    if err != nil {
        return err
    }
    
    // 插入本地消息
    messageID := uuid.New().String()
    messageBody, _ := json.Marshal(map[string]interface{}{
        "event_type": "order_created",
        "order_id":   order.OrderID,
        "user_id":    order.UserID,
        "amount":     order.FinalAmount,
    })
    
    _, err = tx.Exec(`
        INSERT INTO local_messages (message_id, topic, body, status)
        VALUES (?, 'order-events', ?, 'pending')
    `, messageID, string(messageBody))
    
    if err != nil {
        return err
    }
    
    // 2. 提交事务
    if err = tx.Commit(); err != nil {
        return err
    }
    
    // 3. 异步发送消息到 Kafka（定时任务处理）
    
    return nil
}

// MessageRelayJob 消息中继任务
func (p *OrderEventPublisher) MessageRelayJob() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for range ticker.C {
        p.sendPendingMessages()
    }
}

func (p *OrderEventPublisher) sendPendingMessages() {
    rows, _ := p.db.Query(`
        SELECT message_id, topic, body
        FROM local_messages
        WHERE status = 'pending'
        LIMIT 100
        FOR UPDATE SKIP LOCKED
    `)
    defer rows.Close()
    
    for rows.Next() {
        var messageID, topic, body string
        rows.Scan(&messageID, &topic, &body)
        
        // 发送到 Kafka
        if err := p.kafka.Send(topic, body); err != nil {
            // 发送失败，更新重试次数
            continue
        }
        
        // 标记为已发送
        p.db.Exec(`
            UPDATE local_messages SET status = 'sent' WHERE message_id = ?
        `, messageID)
    }
}

type KafkaProducer struct{}

func (k *KafkaProducer) Send(topic, body string) error {
    return nil
}

// ==================== 商品搜索（Elasticsearch）====================

package search

import (
    "context"
    "encoding/json"
    
    "github.com/elastic/go-elasticsearch/v8"
)

type ProductSearchService struct {
    es *elasticsearch.Client
}

// SearchProducts 搜索商品
func (s *ProductSearchService) SearchProducts(keyword string, page, pageSize int) ([]*Product, error) {
    ctx := context.Background()
    
    // 构建搜索查询
    query := map[string]interface{}{
        "query": map[string]interface{}{
            "multi_match": map[string]interface{}{
                "query":  keyword,
                "fields": []string{"title^3", "description"},
            },
        },
        "from": (page - 1) * pageSize,
        "size": pageSize,
        "sort": []map[string]interface{}{
            {"sales": "desc"},
            {"_score": "desc"},
        },
    }
    
    queryJSON, _ := json.Marshal(query)
    
    // 执行搜索
    res, err := s.es.Search(
        s.es.Search.WithContext(ctx),
        s.es.Search.WithIndex("products"),
        s.es.Search.WithBody(strings.NewReader(string(queryJSON))),
    )
    
    if err != nil {
        return nil, err
    }
    defer res.Body.Close()
    
    // 解析结果
    var result map[string]interface{}
    json.NewDecoder(res.Body).Decode(&result)
    
    products := make([]*Product, 0)
    // 解析 hits ...
    
    return products, nil
}

// SyncProductToES 同步商品到 ES
func (s *ProductSearchService) SyncProductToES(product *Product) error {
    ctx := context.Background()
    
    body, _ := json.Marshal(product)
    
    _, err := s.es.Index(
        "products",
        strings.NewReader(string(body)),
        s.es.Index.WithContext(ctx),
        s.es.Index.WithDocumentID(fmt.Sprintf("%d", product.SkuID)),
    )
    
    return err
}

func DecrStockWithLock(db *sql.DB, skuID int64, quantity int) error {
    tx, _ := db.Begin()
    defer tx.Rollback()
    
    // 行锁
    var stock int
    tx.QueryRow(`SELECT stock FROM inventory WHERE sku_id = ? FOR UPDATE`, skuID).Scan(&stock)
    
    if stock < quantity {
        return fmt.Errorf("库存不足")
    }
    
    tx.Exec(`UPDATE inventory SET stock = stock - ? WHERE sku_id = ?`, quantity, skuID)
    
    return tx.Commit()
}

func DecrStockOptimistic(db *sql.DB, skuID int64, quantity int) error {
    for i := 0; i < 3; i++ {
        var stock, version int
        db.QueryRow(`SELECT stock, version FROM inventory WHERE sku_id = ?`, skuID).
            Scan(&stock, &version)
        
        if stock < quantity {
            return fmt.Errorf("库存不足")
        }
        
        result, _ := db.Exec(`
            UPDATE inventory 
            SET stock = stock - ?, version = version + 1
            WHERE sku_id = ? AND version = ?
        `, quantity, skuID, version)
        
        rowsAffected, _ := result.RowsAffected()
        if rowsAffected > 0 {
            return nil
        }
        
        // 版本冲突，重试
        time.Sleep(time.Millisecond * 10)
    }
    
    return fmt.Errorf("扣减库存失败")
}

func ScanTimeoutOrders() {
    ticker := time.NewTicker(1 * time.Minute)
    defer ticker.Stop()
    
    for range ticker.C {
        // 查询超时未支付订单
        rows, _ := db.Query(`
            SELECT order_id 
            FROM orders 
            WHERE status = 'unpaid' 
              AND created_at < DATE_SUB(NOW(), INTERVAL 15 MINUTE)
            LIMIT 1000
        `)
        
        for rows.Next() {
            var orderID string
            rows.Scan(&orderID)
            
            // 取消订单
            CancelOrder(orderID, "超时未支付")
        }
    }
}

// RabbitMQ 延迟队列
func PublishOrderTimeout(orderID string, delay time.Duration) {
    message := map[string]interface{}{
        "order_id": orderID,
        "action":   "check_timeout",
    }
    
    body, _ := json.Marshal(message)
    
    // 发送到延迟队列
    channel.Publish(
        "",
        "order.timeout.queue",
        false,
        false,
        amqp.Publishing{
            ContentType: "application/json",
            Body:        body,
            Expiration:  fmt.Sprintf("%d", delay.Milliseconds()),
        },
    )
}

// 订单分片规则：按 user_id 取模
func GetOrderTableName(userID int64) string {
    tableIndex := userID % 256
    return fmt.Sprintf("orders_%03d", tableIndex)
}

func GetOrderDBIndex(userID int64) int {
    return int(userID / 256 % 32)
}

// 路由查询
func QueryOrder(userID int64, orderID string) (*Order, error) {
    dbIndex := GetOrderDBIndex(userID)
    tableName := GetOrderTableName(userID)
    
    db := dbCluster[dbIndex]
    
    var order Order
    db.QueryRow(fmt.Sprintf(`
        SELECT * FROM %s WHERE order_id = ?
    `, tableName), orderID).Scan(&order)
    
    return &order, nil
}

var (
    orderTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "order_total",
            Help: "Total orders",
        },
        []string{"status"},
    )
    
    orderLatency = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "order_latency_seconds",
            Help:    "Order creation latency",
            Buckets: []float64{0.1, 0.5, 1, 2, 5},
        },
        []string{"step"},
    )
    
    inventoryStock = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "inventory_stock",
            Help: "Current inventory stock",
        },
        []string{"sku_id"},
    )
    
    oversellCount = prometheus.NewCounter(
        prometheus.CounterOpts{
            Name: "inventory_oversell_total",
            Help: "Total oversell count",
        },
    )
)

func (o *Order) CanTransitionTo(newStatus string) bool {
    transitions := map[string][]string{
        "unpaid":    {"paid", "cancelled"},
        "paid":      {"shipped", "refunding"},
        "shipped":   {"received"},
        "received":  {"completed", "refunding"},
    }
    
    allowed := transitions[o.Status]
    for _, s := range allowed {
        if s == newStatus {
            return true
        }
    }
    return false
}

// 每分钟扫描一次
SELECT order_id FROM orders 
WHERE status = 'unpaid' 
  AND created_at < DATE_SUB(NOW(), INTERVAL 15 MINUTE)

time.AfterFunc(15*time.Minute, func() {
    CancelOrder(orderID)
})

lockKey := fmt.Sprintf("order:create:user:%d", userID)
if !redis.SetNX(lockKey, "1", 10*time.Second) {
    return errors.New("请勿重复提交")
}
defer redis.Del(lockKey)

// 基于请求ID去重
requestID := r.Header.Get("X-Request-ID")
if cache.Exists(requestID) {
    return cache.Get(requestID) // 返回已有结果
}

// Redis 原子扣减
remaining := redis.DecrBy("stock:sku:100101", quantity)
if remaining < 0 {
    redis.IncrBy("stock:sku:100101", quantity) // 回滚
    return errors.New("库存不足")
}

// 按 user_id 分片
dbIndex = user_id / 256 % 32      // 32个数据库
tableIndex = user_id % 256        // 256个表

db := dbCluster[GetDBIndex(userID)]
table := GetTableName(userID)
db.Query(fmt.Sprintf("SELECT * FROM %s WHERE user_id = ?", table), userID)

func CalculatePrice(skuID int64, quantity int, userID int64) float64 {
    // 1. 获取商品原价
    originalPrice := GetPrice(skuID)
    
    // 2. 查询适用的促销活动（按优先级）
    promotions := GetActivePromotions(skuID, userID)
    
    // 3. 计算最优价格
    bestPrice := originalPrice
    for _, promo := range promotions {
        price := promo.Calculate(originalPrice, quantity)
        if price < bestPrice {
            bestPrice = price
        }
    }
    
    return bestPrice
}