09 - 支付系统设计

package main

import (
    "database/sql"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
    
    "github.com/google/uuid"
    "github.com/gorilla/mux"
    _ "github.com/go-sql-driver/mysql"
)

// PaymentOrder 支付订单
type PaymentOrder struct {
    ID              int64     `json:"id"`
    PaymentOrderID  string    `json:"payment_order_id"`
    OrderID         string    `json:"order_id"`
    UserID          int64     `json:"user_id"`
    Amount          float64   `json:"amount"`
    Currency        string    `json:"currency"`
    PaymentMethod   string    `json:"payment_method"`
    Status          string    `json:"status"`
    Subject         string    `json:"subject"`
    Description     string    `json:"description"`
    TransactionID   string    `json:"transaction_id,omitempty"`
    NotifyURL       string    `json:"notify_url"`
    ReturnURL       string    `json:"return_url"`
    ClientIP        string    `json:"client_ip"`
    PaidAt          *time.Time `json:"paid_at,omitempty"`
    ExpiresAt       time.Time `json:"expires_at"`
    CreatedAt       time.Time `json:"created_at"`
    UpdatedAt       time.Time `json:"updated_at"`
}

// CreatePaymentRequest 创建支付请求
type CreatePaymentRequest struct {
    OrderID       string  `json:"order_id"`
    UserID        int64   `json:"user_id"`
    Amount        float64 `json:"amount"`
    Currency      string  `json:"currency"`
    PaymentMethod string  `json:"payment_method"`
    Subject       string  `json:"subject"`
    Description   string  `json:"description"`
    NotifyURL     string  `json:"notify_url"`
    ReturnURL     string  `json:"return_url"`
    Timeout       int     `json:"timeout"`
    ClientIP      string  `json:"client_ip"`
}

// PaymentService 支付服务
type PaymentService struct {
    db *sql.DB
}

// NewPaymentService 创建支付服务
func NewPaymentService(db *sql.DB) *PaymentService {
    return &PaymentService{db: db}
}

// CreatePaymentOrder 创建支付订单
func (s *PaymentService) CreatePaymentOrder(req *CreatePaymentRequest) (*PaymentOrder, error) {
    // 1. 生成支付订单号
    paymentOrderID := generatePaymentOrderID()
    
    // 2. 检查业务订单是否已支付
    var existingOrderID string
    err := s.db.QueryRow(`
        SELECT payment_order_id 
        FROM payment_orders 
        WHERE order_id = ? AND status = 'success'
        LIMIT 1
    `, req.OrderID).Scan(&existingOrderID)
    
    if err == nil {
        return nil, fmt.Errorf("订单已支付: %s", existingOrderID)
    }
    
    // 3. 创建支付订单
    expiresAt := time.Now().Add(time.Duration(req.Timeout) * time.Second)
    
    result, err := s.db.Exec(`
        INSERT INTO payment_orders (
            payment_order_id, order_id, user_id, amount, currency,
            payment_method, status, subject, description,
            notify_url, return_url, client_ip, expires_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    `, paymentOrderID, req.OrderID, req.UserID, req.Amount, req.Currency,
        req.PaymentMethod, "pending", req.Subject, req.Description,
        req.NotifyURL, req.ReturnURL, req.ClientIP, expiresAt)
    
    if err != nil {
        return nil, fmt.Errorf("创建支付订单失败: %w", err)
    }
    
    orderID, _ := result.LastInsertId()
    
    // 4. 调用第三方支付接口（这里简化处理）
    qrCode := ""
    switch req.PaymentMethod {
    case "wechat":
        qrCode = callWeChatPay(paymentOrderID, req.Amount)
    case "alipay":
        qrCode = callAlipay(paymentOrderID, req.Amount)
    case "balance":
        // 余额支付直接扣款
        err = s.deductBalance(req.UserID, req.Amount, paymentOrderID)
        if err != nil {
            return nil, err
        }
    }
    
    // 5. 返回支付订单
    order := &PaymentOrder{
        ID:             orderID,
        PaymentOrderID: paymentOrderID,
        OrderID:        req.OrderID,
        UserID:         req.UserID,
        Amount:         req.Amount,
        Currency:       req.Currency,
        PaymentMethod:  req.PaymentMethod,
        Status:         "pending",
        Subject:        req.Subject,
        Description:    req.Description,
        NotifyURL:      req.NotifyURL,
        ReturnURL:      req.ReturnURL,
        ClientIP:       req.ClientIP,
        ExpiresAt:      expiresAt,
        CreatedAt:      time.Now(),
        UpdatedAt:      time.Now(),
    }
    
    // 如果有二维码，添加到响应中
    if qrCode != "" {
        order.Description = qrCode // 简化：实际应该用单独字段
    }
    
    return order, nil
}

// deductBalance 扣除余额
func (s *PaymentService) deductBalance(userID int64, amount float64, paymentOrderID string) error {
    // 1. 开启事务
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 2. 锁定账户（使用 FOR UPDATE）
    var balance float64
    var version int
    err = tx.QueryRow(`
        SELECT balance, version 
        FROM accounts 
        WHERE user_id = ? 
        FOR UPDATE
    `, userID).Scan(&balance, &version)
    
    if err != nil {
        return fmt.Errorf("账户不存在: %w", err)
    }
    
    // 3. 检查余额是否足够
    if balance < amount {
        return fmt.Errorf("余额不足: 当前余额 %.2f, 需要 %.2f", balance, amount)
    }
    
    // 4. 扣减余额（使用乐观锁）
    result, err := tx.Exec(`
        UPDATE accounts 
        SET balance = balance - ?, 
            total_balance = total_balance - ?,
            version = version + 1,
            updated_at = NOW()
        WHERE user_id = ? AND version = ?
    `, amount, amount, userID, version)
    
    if err != nil {
        return fmt.Errorf("扣款失败: %w", err)
    }
    
    rowsAffected, _ := result.RowsAffected()
    if rowsAffected == 0 {
        return fmt.Errorf("扣款失败: 版本冲突")
    }
    
    // 5. 记录流水
    transactionID := generateTransactionID()
    _, err = tx.Exec(`
        INSERT INTO account_transactions (
            transaction_id, user_id, account_id, type, amount,
            balance_before, balance_after, business_type, business_id, remark
        ) VALUES (?, ?, (SELECT id FROM accounts WHERE user_id = ?), ?, ?, ?, ?, ?, ?, ?)
    `, transactionID, userID, userID, "debit", amount,
        balance, balance-amount, "payment", paymentOrderID, "支付扣款")
    
    if err != nil {
        return fmt.Errorf("记录流水失败: %w", err)
    }
    
    // 6. 更新支付订单状态
    _, err = tx.Exec(`
        UPDATE payment_orders 
        SET status = 'success', paid_at = NOW(), updated_at = NOW()
        WHERE payment_order_id = ?
    `, paymentOrderID)
    
    if err != nil {
        return fmt.Errorf("更新订单状态失败: %w", err)
    }
    
    // 7. 提交事务
    if err = tx.Commit(); err != nil {
        return fmt.Errorf("提交事务失败: %w", err)
    }
    
    return nil
}

// QueryPaymentOrder 查询支付订单
func (s *PaymentService) QueryPaymentOrder(paymentOrderID string) (*PaymentOrder, error) {
    var order PaymentOrder
    err := s.db.QueryRow(`
        SELECT id, payment_order_id, order_id, user_id, amount, currency,
               payment_method, status, subject, description, transaction_id,
               notify_url, return_url, client_ip, paid_at, expires_at,
               created_at, updated_at
        FROM payment_orders
        WHERE payment_order_id = ?
    `, paymentOrderID).Scan(
        &order.ID, &order.PaymentOrderID, &order.OrderID, &order.UserID,
        &order.Amount, &order.Currency, &order.PaymentMethod, &order.Status,
        &order.Subject, &order.Description, &order.TransactionID,
        &order.NotifyURL, &order.ReturnURL, &order.ClientIP,
        &order.PaidAt, &order.ExpiresAt, &order.CreatedAt, &order.UpdatedAt,
    )
    
    if err == sql.ErrNoRows {
        return nil, fmt.Errorf("订单不存在")
    }
    if err != nil {
        return nil, err
    }
    
    return &order, nil
}

// HandleCallback 处理支付回调
func (s *PaymentService) HandleCallback(paymentMethod string, data map[string]interface{}) error {
    // 1. 验证签名
    if !verifySign(paymentMethod, data) {
        return fmt.Errorf("签名验证失败")
    }
    
    // 2. 提取订单号和交易号
    paymentOrderID := data["out_trade_no"].(string)
    transactionID := data["transaction_id"].(string)
    resultCode := data["result_code"].(string)
    
    // 3. 查询订单
    order, err := s.QueryPaymentOrder(paymentOrderID)
    if err != nil {
        return err
    }
    
    // 4. 幂等性检查
    if order.Status == "success" {
        return nil // 已经处理过
    }
    
    // 5. 更新订单状态
    var newStatus string
    if resultCode == "SUCCESS" {
        newStatus = "success"
    } else {
        newStatus = "failed"
    }
    
    _, err = s.db.Exec(`
        UPDATE payment_orders 
        SET status = ?, transaction_id = ?, paid_at = NOW(), updated_at = NOW()
        WHERE payment_order_id = ? AND status = 'pending'
    `, newStatus, transactionID, paymentOrderID)
    
    if err != nil {
        return fmt.Errorf("更新订单状态失败: %w", err)
    }
    
    // 6. 异步通知业务系统
    go s.notifyBusiness(order.NotifyURL, order)
    
    return nil
}

// notifyBusiness 通知业务系统
func (s *PaymentService) notifyBusiness(notifyURL string, order *PaymentOrder) {
    // 实现重试逻辑
    maxRetries := 5
    for i := 0; i < maxRetries; i++ {
        resp, err := http.Post(notifyURL, "application/json", nil)
        if err == nil && resp.StatusCode == 200 {
            return
        }
        
        // 指数退避
        time.Sleep(time.Duration(1<<uint(i)) * time.Second)
    }
}

// 辅助函数
func generatePaymentOrderID() string {
    return "PAY" + time.Now().Format("20060102") + uuid.New().String()[:8]
}

func generateTransactionID() string {
    return "TXN" + time.Now().Format("20060102150405") + uuid.New().String()[:6]
}

func callWeChatPay(orderID string, amount float64) string {
    // 调用微信支付统一下单接口
    // 返回二维码链接
    return "weixin://wxpay/bizpayurl?pr=abc123"
}

func callAlipay(orderID string, amount float64) string {
    // 调用支付宝支付接口
    return "https://qr.alipay.com/abc123"
}

func verifySign(paymentMethod string, data map[string]interface{}) bool {
    // 验证第三方签名
    return true
}

// HTTP Handlers
func (s *PaymentService) CreatePaymentHandler(w http.ResponseWriter, r *http.Request) {
    var req CreatePaymentRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    order, err := s.CreatePaymentOrder(&req)
    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 (s *PaymentService) QueryPaymentHandler(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    paymentOrderID := vars["payment_order_id"]
    
    order, err := s.QueryPaymentOrder(paymentOrderID)
    if err != nil {
        http.Error(w, err.Error(), http.StatusNotFound)
        return
    }
    
    json.NewEncoder(w).Encode(map[string]interface{}{
        "code":    0,
        "message": "success",
        "data":    order,
    })
}

func main() {
    // 连接数据库
    db, err := sql.Open("mysql", "user:password@tcp(localhost:3306)/payment?parseTime=true")
    if err != nil {
        panic(err)
    }
    defer db.Close()
    
    // 创建服务
    service := NewPaymentService(db)
    
    // 设置路由
    r := mux.NewRouter()
    r.HandleFunc("/api/v1/payment/orders", service.CreatePaymentHandler).Methods("POST")
    r.HandleFunc("/api/v1/payment/orders/{payment_order_id}", service.QueryPaymentHandler).Methods("GET")
    
    // 启动服务
    fmt.Println("支付服务启动在 :8080")
    http.ListenAndServe(":8080", r)
}

// ==================== TCC 分布式事务实现 ====================

package tcc

import (
    "context"
    "fmt"
    "time"
)

// TCCTransaction TCC事务
type TCCTransaction struct {
    TransactionID string
    Participants  []*TCCParticipant
    Status        string // "trying" | "confirmed" | "cancelled"
    CreatedAt     time.Time
    UpdatedAt     time.Time
}

// TCCParticipant TCC参与者
type TCCParticipant struct {
    ServiceName string
    TryFunc     func(ctx context.Context, txID string) error
    ConfirmFunc func(ctx context.Context, txID string) error
    CancelFunc  func(ctx context.Context, txID string) error
}

// TCCCoordinator TCC协调器
type TCCCoordinator struct {
    transactions map[string]*TCCTransaction
}

// NewTCCCoordinator 创建TCC协调器
func NewTCCCoordinator() *TCCCoordinator {
    return &TCCCoordinator{
        transactions: make(map[string]*TCCTransaction),
    }
}

// ExecuteTCC 执行TCC事务
func (c *TCCCoordinator) ExecuteTCC(ctx context.Context, txID string, participants []*TCCParticipant) error {
    // 1. 创建事务记录
    tx := &TCCTransaction{
        TransactionID: txID,
        Participants:  participants,
        Status:        "trying",
        CreatedAt:     time.Now(),
        UpdatedAt:     time.Now(),
    }
    c.transactions[txID] = tx
    
    // 2. Try 阶段：预留资源
    for _, participant := range participants {
        if err := participant.TryFunc(ctx, txID); err != nil {
            // Try失败，执行Cancel
            c.cancelAll(ctx, txID, participants)
            return fmt.Errorf("try失败 [%s]: %w", participant.ServiceName, err)
        }
    }
    
    // 3. Confirm 阶段：确认提交
    tx.Status = "confirming"
    for _, participant := range participants {
        if err := participant.ConfirmFunc(ctx, txID); err != nil {
            // Confirm失败，需要人工介入或重试
            return fmt.Errorf("confirm失败 [%s]: %w", participant.ServiceName, err)
        }
    }
    
    tx.Status = "confirmed"
    tx.UpdatedAt = time.Now()
    
    return nil
}

// cancelAll 取消所有参与者
func (c *TCCCoordinator) cancelAll(ctx context.Context, txID string, participants []*TCCParticipant) {
    tx := c.transactions[txID]
    tx.Status = "cancelling"
    
    for _, participant := range participants {
        // Cancel 必须成功，失败需要重试
        for i := 0; i < 3; i++ {
            if err := participant.CancelFunc(ctx, txID); err == nil {
                break
            }
            time.Sleep(time.Second * time.Duration(i+1))
        }
    }
    
    tx.Status = "cancelled"
    tx.UpdatedAt = time.Now()
}

// ==================== 支付服务 TCC 实现 ====================

// PaymentTCCService 支付TCC服务
type PaymentTCCService struct {
    db    *sql.DB
    redis *redis.Client
}

// TryPayment Try阶段：冻结余额
func (s *PaymentTCCService) TryPayment(ctx context.Context, txID string, userID int64, amount float64) error {
    // 1. 获取分布式锁
    lockKey := fmt.Sprintf("payment:lock:user:%d", userID)
    lock := s.acquireLock(lockKey, 10*time.Second)
    if !lock {
        return fmt.Errorf("获取锁失败")
    }
    defer s.releaseLock(lockKey)
    
    // 2. 开启事务
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 3. 检查余额并冻结
    var balance float64
    err = tx.QueryRow(`
        SELECT balance FROM accounts WHERE user_id = ? FOR UPDATE
    `, userID).Scan(&balance)
    
    if err != nil {
        return err
    }
    
    if balance < amount {
        return fmt.Errorf("余额不足")
    }
    
    // 4. 冻结金额
    _, err = tx.Exec(`
        UPDATE accounts 
        SET balance = balance - ?,
            frozen_balance = frozen_balance + ?,
            version = version + 1
        WHERE user_id = ?
    `, amount, amount, userID)
    
    if err != nil {
        return err
    }
    
    // 5. 记录TCC事务
    _, err = tx.Exec(`
        INSERT INTO tcc_transactions (tx_id, user_id, amount, status, phase)
        VALUES (?, ?, ?, 'trying', 'try')
    `, txID, userID, amount)
    
    if err != nil {
        return err
    }
    
    return tx.Commit()
}

// ConfirmPayment Confirm阶段：确认扣款
func (s *PaymentTCCService) ConfirmPayment(ctx context.Context, txID string) error {
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 1. 查询TCC事务
    var userID int64
    var amount float64
    err = tx.QueryRow(`
        SELECT user_id, amount FROM tcc_transactions 
        WHERE tx_id = ? AND status = 'trying'
    `, txID).Scan(&userID, &amount)
    
    if err != nil {
        return err
    }
    
    // 2. 扣减冻结金额
    _, err = tx.Exec(`
        UPDATE accounts 
        SET frozen_balance = frozen_balance - ?,
            total_balance = total_balance - ?
        WHERE user_id = ?
    `, amount, amount, userID)
    
    if err != nil {
        return err
    }
    
    // 3. 更新TCC事务状态
    _, err = tx.Exec(`
        UPDATE tcc_transactions 
        SET status = 'confirmed', phase = 'confirm', updated_at = NOW()
        WHERE tx_id = ?
    `, txID)
    
    if err != nil {
        return err
    }
    
    // 4. 记录流水
    _, err = tx.Exec(`
        INSERT INTO account_transactions (
            transaction_id, user_id, account_id, type, amount,
            balance_before, balance_after, business_type, business_id
        ) SELECT ?, user_id, id, 'debit', ?,
                 balance + ?, balance, 'payment', ?
          FROM accounts WHERE user_id = ?
    `, generateTransactionID(), amount, amount, txID, userID)
    
    return tx.Commit()
}

// CancelPayment Cancel阶段：解冻余额
func (s *PaymentTCCService) CancelPayment(ctx context.Context, txID string) error {
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 1. 查询TCC事务
    var userID int64
    var amount float64
    err = tx.QueryRow(`
        SELECT user_id, amount FROM tcc_transactions 
        WHERE tx_id = ? AND status = 'trying'
    `, txID).Scan(&userID, &amount)
    
    if err != nil {
        return err
    }
    
    // 2. 解冻金额
    _, err = tx.Exec(`
        UPDATE accounts 
        SET balance = balance + ?,
            frozen_balance = frozen_balance - ?
        WHERE user_id = ?
    `, amount, amount, userID)
    
    if err != nil {
        return err
    }
    
    // 3. 更新TCC事务状态
    _, err = tx.Exec(`
        UPDATE tcc_transactions 
        SET status = 'cancelled', phase = 'cancel', updated_at = NOW()
        WHERE tx_id = ?
    `, txID)
    
    return tx.Commit()
}

// ==================== 分布式锁实现 ====================

func (s *PaymentTCCService) acquireLock(key string, expiration time.Duration) bool {
    // 使用 Redis SET NX EX 实现
    result, err := s.redis.SetNX(context.Background(), key, "1", expiration).Result()
    return err == nil && result
}

func (s *PaymentTCCService) releaseLock(key string) {
    s.redis.Del(context.Background(), key)
}

// ==================== 完整支付流程 ====================

func (s *PaymentTCCService) ProcessPayment(ctx context.Context, orderID string, userID int64, amount float64) error {
    txID := generateTxID()
    
    // 创建TCC协调器
    coordinator := NewTCCCoordinator()
    
    // 定义参与者
    participants := []*TCCParticipant{
        // 账户服务
        {
            ServiceName: "AccountService",
            TryFunc: func(ctx context.Context, txID string) error {
                return s.TryPayment(ctx, txID, userID, amount)
            },
            ConfirmFunc: func(ctx context.Context, txID string) error {
                return s.ConfirmPayment(ctx, txID)
            },
            CancelFunc: func(ctx context.Context, txID string) error {
                return s.CancelPayment(ctx, txID)
            },
        },
        // 可以添加更多参与者（如订单服务、库存服务）
    }
    
    // 执行TCC事务
    return coordinator.ExecuteTCC(ctx, txID, participants)
}

func generateTxID() string {
    return "TCC" + time.Now().Format("20060102150405") + uuid.New().String()[:6]
}

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

package eventual

import (
    "context"
    "database/sql"
    "encoding/json"
    "fmt"
    "time"
)

// LocalMessage 本地消息
type LocalMessage struct {
    ID            int64
    MessageID     string
    BusinessType  string
    BusinessID    string
    MessageBody   string
    Status        string // "pending" | "sent" | "consumed"
    RetryCount    int
    MaxRetry      int
    NextRetryTime time.Time
    CreatedAt     time.Time
    UpdatedAt     time.Time
}

// EventualConsistencyPayment 最终一致性支付
type EventualConsistencyPayment struct {
    db  *sql.DB
    mq  MessageQueue
}

// CreatePaymentWithMessage 创建支付订单并发送消息
func (p *EventualConsistencyPayment) CreatePaymentWithMessage(
    ctx context.Context,
    orderID string,
    userID int64,
    amount float64,
) error {
    // 1. 开启数据库事务
    tx, err := p.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()
    
    // 2. 创建支付订单
    paymentOrderID := generatePaymentOrderID()
    _, err = tx.Exec(`
        INSERT INTO payment_orders (
            payment_order_id, order_id, user_id, amount, status
        ) VALUES (?, ?, ?, ?, 'pending')
    `, paymentOrderID, orderID, userID, amount)
    
    if err != nil {
        return fmt.Errorf("创建订单失败: %w", err)
    }
    
    // 3. 插入本地消息表
    messageID := generateMessageID()
    messageBody, _ := json.Marshal(map[string]interface{}{
        "payment_order_id": paymentOrderID,
        "order_id":         orderID,
        "user_id":          userID,
        "amount":           amount,
        "event_type":       "payment_created",
    })
    
    _, err = tx.Exec(`
        INSERT INTO local_messages (
            message_id, business_type, business_id, message_body,
            status, retry_count, max_retry, next_retry_time
        ) VALUES (?, 'payment', ?, ?, 'pending', 0, 5, NOW())
    `, messageID, paymentOrderID, string(messageBody))
    
    if err != nil {
        return fmt.Errorf("插入消息失败: %w", err)
    }
    
    // 4. 提交事务（原子性保证）
    if err = tx.Commit(); err != nil {
        return fmt.Errorf("提交事务失败: %w", err)
    }
    
    // 5. 异步发送消息（定时任务处理）
    // 这里只是记录到本地消息表，由定时任务负责发送
    
    return nil
}

// MessageSendingJob 消息发送定时任务
func (p *EventualConsistencyPayment) MessageSendingJob() {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for range ticker.C {
        p.sendPendingMessages()
    }
}

// sendPendingMessages 发送待发送消息
func (p *EventualConsistencyPayment) sendPendingMessages() {
    // 1. 查询待发送消息
    rows, err := p.db.Query(`
        SELECT id, message_id, business_type, business_id, message_body, retry_count
        FROM local_messages
        WHERE status = 'pending' 
          AND next_retry_time <= NOW()
          AND retry_count < max_retry
        LIMIT 100
        FOR UPDATE SKIP LOCKED
    `)
    if err != nil {
        return
    }
    defer rows.Close()
    
    // 2. 遍历消息
    for rows.Next() {
        var msg LocalMessage
        err := rows.Scan(&msg.ID, &msg.MessageID, &msg.BusinessType,
            &msg.BusinessID, &msg.MessageBody, &msg.RetryCount)
        if err != nil {
            continue
        }
        
        // 3. 发送消息到MQ
        err = p.mq.SendMessage(msg.BusinessType, msg.MessageBody)
        if err != nil {
            // 发送失败，更新重试次数
            p.updateMessageRetry(&msg)
            continue
        }
        
        // 4. 更新消息状态
        _, err = p.db.Exec(`
            UPDATE local_messages 
            SET status = 'sent', updated_at = NOW()
            WHERE id = ?
        `, msg.ID)
    }
}

// updateMessageRetry 更新消息重试信息
func (p *EventualConsistencyPayment) updateMessageRetry(msg *LocalMessage) {
    // 指数退避：1s, 2s, 4s, 8s, 16s
    nextRetryTime := time.Now().Add(time.Duration(1<<uint(msg.RetryCount)) * time.Second)
    
    _, err := p.db.Exec(`
        UPDATE local_messages 
        SET retry_count = retry_count + 1,
            next_retry_time = ?,
            updated_at = NOW()
        WHERE id = ?
    `, nextRetryTime, msg.ID)
    
    if err != nil {
        fmt.Printf("更新消息重试失败: %v\n", err)
    }
}

// ==================== RocketMQ 事务消息 ====================

type RocketMQTransactionProducer struct {
    producer *rocketmq.TransactionProducer
}

// SendTransactionMessage 发送事务消息
func (p *RocketMQTransactionProducer) SendTransactionMessage(
    topic string,
    body []byte,
    localTx func() error,
) error {
    // 1. 发送半消息（Half Message）
    msg := &rocketmq.Message{
        Topic: topic,
        Body:  body,
    }
    
    result, err := p.producer.SendMessageInTransaction(msg)
    if err != nil {
        return err
    }
    
    // 2. 执行本地事务
    if err := localTx(); err != nil {
        // 本地事务失败，回滚消息
        p.producer.Rollback(result.TransactionId)
        return err
    }
    
    // 3. 提交消息
    p.producer.Commit(result.TransactionId)
    
    return nil
}

// CheckLocalTransaction 回查本地事务
func (p *RocketMQTransactionProducer) CheckLocalTransaction(msg *rocketmq.MessageExt) rocketmq.LocalTransactionState {
    // 从消息中提取订单ID
    var data map[string]interface{}
    json.Unmarshal(msg.Body, &data)
    paymentOrderID := data["payment_order_id"].(string)
    
    // 查询支付订单状态
    var status string
    err := p.db.QueryRow(`
        SELECT status FROM payment_orders WHERE payment_order_id = ?
    `, paymentOrderID).Scan(&status)
    
    if err != nil {
        return rocketmq.UnknownTransaction
    }
    
    if status == "success" {
        return rocketmq.CommitTransaction
    } else if status == "failed" {
        return rocketmq.RollbackTransaction
    }
    
    return rocketmq.UnknownTransaction
}

// ==================== 最大努力通知 ====================

type MaxEffortNotifier struct {
    db *sql.DB
}

// NotifyWithRetry 带重试的通知
func (n *MaxEffortNotifier) NotifyWithRetry(notifyURL string, data interface{}) {
    // 重试策略：1m, 5m, 10m, 30m, 1h, 2h, 6h, 24h
    retryIntervals := []time.Duration{
        1 * time.Minute,
        5 * time.Minute,
        10 * time.Minute,
        30 * time.Minute,
        1 * time.Hour,
        2 * time.Hour,
        6 * time.Hour,
        24 * time.Hour,
    }
    
    for i := 0; i < len(retryIntervals); i++ {
        // 发送通知
        if n.sendNotification(notifyURL, data) {
            return // 成功
        }
        
        // 失败，等待后重试
        time.Sleep(retryIntervals[i])
    }
    
    // 所有重试都失败，记录到失败表，需要人工处理
    n.recordFailedNotification(notifyURL, data)
}

func (n *MaxEffortNotifier) sendNotification(url string, data interface{}) bool {
    body, _ := json.Marshal(data)
    resp, err := http.Post(url, "application/json", bytes.NewBuffer(body))
    
    if err != nil {
        return false
    }
    defer resp.Body.Close()
    
    return resp.StatusCode == 200
}

func (n *MaxEffortNotifier) recordFailedNotification(url string, data interface{}) {
    body, _ := json.Marshal(data)
    n.db.Exec(`
        INSERT INTO failed_notifications (url, data, created_at)
        VALUES (?, ?, NOW())
    `, url, string(body))
}

// ==================== 对账系统 ====================

type ReconciliationService struct {
    db             *sql.DB
    paymentGateway PaymentGateway
}

// DailyReconciliation 每日对账
func (s *ReconciliationService) DailyReconciliation(date time.Time) error {
    // 1. 获取内部订单列表
    internalOrders, err := s.getInternalOrders(date)
    if err != nil {
        return err
    }
    
    // 2. 获取第三方订单列表
    externalOrders, err := s.paymentGateway.GetOrders(date)
    if err != nil {
        return err
    }
    
    // 3. 对账
    internalMap := make(map[string]float64)
    for _, order := range internalOrders {
        internalMap[order.PaymentOrderID] = order.Amount
    }
    
    externalMap := make(map[string]float64)
    for _, order := range externalOrders {
        externalMap[order.OutTradeNo] = order.Amount
    }
    
    // 4. 找出差异
    for orderID, internalAmount := range internalMap {
        externalAmount, exists := externalMap[orderID]
        
        if !exists {
            // 内部有，第三方没有（可能是退款）
            s.recordDifference(orderID, internalAmount, 0, "missing", date)
        } else if internalAmount != externalAmount {
            // 金额不一致
            s.recordDifference(orderID, internalAmount, externalAmount, "unmatched", date)
        } else {
            // 一致
            s.recordDifference(orderID, internalAmount, externalAmount, "matched", date)
        }
    }
    
    // 5. 找出多余订单（第三方有，内部没有）
    for orderID, externalAmount := range externalMap {
        if _, exists := internalMap[orderID]; !exists {
            s.recordDifference(orderID, 0, externalAmount, "excess", date)
        }
    }
    
    return nil
}

func (s *ReconciliationService) getInternalOrders(date time.Time) ([]*PaymentOrder, error) {
    rows, err := s.db.Query(`
        SELECT payment_order_id, amount
        FROM payment_orders
        WHERE DATE(paid_at) = ? AND status = 'success'
    `, date.Format("2006-01-02"))
    
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    
    orders := make([]*PaymentOrder, 0)
    for rows.Next() {
        var order PaymentOrder
        rows.Scan(&order.PaymentOrderID, &order.Amount)
        orders = append(orders, &order)
    }
    
    return orders, nil
}

func (s *ReconciliationService) recordDifference(
    orderID string,
    internalAmount, externalAmount float64,
    status string,
    date time.Time,
) {
    differenceAmount := internalAmount - externalAmount
    
    _, err := s.db.Exec(`
        INSERT INTO reconciliation_records (
            payment_order_id, internal_amount, external_amount,
            status, reconciliation_date, difference_amount, handled
        ) VALUES (?, ?, ?, ?, ?, ?, FALSE)
    `, orderID, internalAmount, externalAmount, status, date, differenceAmount)
    
    if err != nil {
        fmt.Printf("记录对账差异失败: %v\n", err)
    }
}

// PaymentGateway 支付网关接口
type PaymentGateway interface {
    GetOrders(date time.Time) ([]*ExternalOrder, error)
}

type ExternalOrder struct {
    OutTradeNo string
    Amount     float64
}

type MessageQueue interface {
    SendMessage(topic string, body string) error
}

func generateMessageID() string {
    return "MSG" + time.Now().Format("20060102150405") + uuid.New().String()[:6]
}

func (s *PaymentService) CreatePaymentOrderIdempotent(req *CreatePaymentRequest) (*PaymentOrder, error) {
    lockKey := fmt.Sprintf("payment:create:order:%s", req.OrderID)
    
    // 获取分布式锁
    lock := s.redis.SetNX(context.Background(), lockKey, "1", 10*time.Second)
    if !lock.Val() {
        return nil, fmt.Errorf("订单正在处理中")
    }
    defer s.redis.Del(context.Background(), lockKey)
    
    // 检查订单是否已存在
    existing, err := s.QueryPaymentOrderByOrderID(req.OrderID)
    if err == nil {
        return existing, nil // 幂等返回
    }
    
    // 创建订单
    return s.CreatePaymentOrder(req)
}

func (s *PaymentService) ProcessWithIdempotent(idempotentKey string, fn func() (interface{}, error)) (interface{}, error) {
    // 1. 查询幂等记录
    var record IdempotentRecord
    err := s.db.QueryRow(`
        SELECT status, response_data FROM idempotent_records WHERE idempotent_key = ?
    `, idempotentKey).Scan(&record.Status, &record.ResponseData)
    
    if err == nil {
        // 已处理过
        if record.Status == "success" {
            return record.ResponseData, nil
        } else if record.Status == "processing" {
            return nil, fmt.Errorf("请求正在处理中")
        }
    }
    
    // 2. 插入处理中记录
    _, err = s.db.Exec(`
        INSERT INTO idempotent_records (idempotent_key, request_id, status, expires_at)
        VALUES (?, ?, 'processing', DATE_ADD(NOW(), INTERVAL 1 HOUR))
        ON DUPLICATE KEY UPDATE status = 'processing'
    `, idempotentKey, generateRequestID())
    
    if err != nil {
        return nil, err
    }
    
    // 3. 执行业务逻辑
    result, err := fn()
    
    // 4. 更新结果
    if err != nil {
        s.db.Exec(`UPDATE idempotent_records SET status = 'failed' WHERE idempotent_key = ?`, idempotentKey)
        return nil, err
    }
    
    responseData, _ := json.Marshal(result)
    s.db.Exec(`
        UPDATE idempotent_records 
        SET status = 'success', response_data = ? 
        WHERE idempotent_key = ?
    `, string(responseData), idempotentKey)
    
    return result, nil
}

// 使用
func (s *PaymentService) CreatePayment(req *CreatePaymentRequest) (*PaymentOrder, error) {
    idempotentKey := fmt.Sprintf("payment:create:%s", req.OrderID)
    
    result, err := s.ProcessWithIdempotent(idempotentKey, func() (interface{}, error) {
        return s.CreatePaymentOrder(req)
    })
    
    if err != nil {
        return nil, err
    }
    
    return result.(*PaymentOrder), nil
}

type PaymentWithTimeout struct {
    db             *sql.DB
    paymentGateway PaymentGateway
}

// ProcessPaymentWithRetry 带重试的支付处理
func (p *PaymentWithTimeout) ProcessPaymentWithRetry(
    ctx context.Context,
    orderID string,
    amount float64,
) error {
    // 1. 创建支付订单
    paymentOrderID := generatePaymentOrderID()
    
    // 2. 调用第三方支付（带超时）
    ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()
    
    // 重试策略：最多3次，指数退避
    var lastErr error
    for i := 0; i < 3; i++ {
        err := p.callPaymentGateway(ctx, paymentOrderID, amount)
        if err == nil {
            return nil // 成功
        }
        
        lastErr = err
        
        // 判断是否可重试
        if !isRetryableError(err) {
            return err // 不可重试的错误，直接返回
        }
        
        // 查询订单状态（避免重复扣款）
        status, err := p.queryPaymentStatus(ctx, paymentOrderID)
        if err == nil && status == "success" {
            return nil // 已支付成功
        }
        
        // 指数退避
        time.Sleep(time.Duration(1<<uint(i)) * time.Second)
    }
    
    // 3. 所有重试都失败，标记订单为待处理
    p.markOrderAsPending(paymentOrderID)
    
    return fmt.Errorf("支付失败: %w", lastErr)
}

func (p *PaymentWithTimeout) callPaymentGateway(ctx context.Context, orderID string, amount float64) error {
    // 模拟调用第三方
    select {
    case <-ctx.Done():
        return fmt.Errorf("支付超时")
    case <-time.After(2 * time.Second):
        return nil
    }
}

func (p *PaymentWithTimeout) queryPaymentStatus(ctx context.Context, orderID string) (string, error) {
    // 查询第三方订单状态
    return "unknown", nil
}

func (p *PaymentWithTimeout) markOrderAsPending(orderID string) {
    p.db.Exec(`
        UPDATE payment_orders 
        SET status = 'pending_confirm'
        WHERE payment_order_id = ?
    `, orderID)
}

func isRetryableError(err error) bool {
    // 超时、网络错误可重试
    // 余额不足、订单不存在等不可重试
    return err.Error() == "支付超时" || err.Error() == "网络错误"
}

func (s *PaymentService) DeductBalanceSafely(userID int64, amount float64) error {
    tx, _ := s.db.Begin()
    defer tx.Rollback()
    
    // 行锁
    var balance float64
    tx.QueryRow(`SELECT balance FROM accounts WHERE user_id = ? FOR UPDATE`, userID).Scan(&balance)
    
    if balance < amount {
        return fmt.Errorf("余额不足")
    }
    
    // 扣款
    tx.Exec(`UPDATE accounts SET balance = balance - ? WHERE user_id = ?`, amount, userID)
    
    return tx.Commit()
}

func (s *PaymentService) DeductBalanceOptimistic(userID int64, amount float64) error {
    for i := 0; i < 3; i++ {
        // 读取余额和版本号
        var balance float64
        var version int
        s.db.QueryRow(`SELECT balance, version FROM accounts WHERE user_id = ?`, userID).
            Scan(&balance, &version)
        
        if balance < amount {
            return fmt.Errorf("余额不足")
        }
        
        // 更新（带版本号）
        result, _ := s.db.Exec(`
            UPDATE accounts 
            SET balance = balance - ?, version = version + 1
            WHERE user_id = ? AND version = ?
        `, amount, userID, version)
        
        rowsAffected, _ := result.RowsAffected()
        if rowsAffected > 0 {
            return nil // 成功
        }
        
        // 版本冲突，重试
        time.Sleep(time.Millisecond * 10)
    }
    
    return fmt.Errorf("扣款失败: 并发冲突")
}

func (s *PaymentService) DailyBalanceCheck() {
    // 每日对账：余额 = 期初余额 + 收入 - 支出
    rows, _ := s.db.Query(`
        SELECT user_id, balance FROM accounts
    `)
    
    for rows.Next() {
        var userID int64
        var currentBalance float64
        rows.Scan(&userID, &currentBalance)
        
        // 计算理论余额
        var totalCredit, totalDebit float64
        s.db.QueryRow(`
            SELECT 
                COALESCE(SUM(CASE WHEN type = 'credit' THEN amount ELSE 0 END), 0),
                COALESCE(SUM(CASE WHEN type = 'debit' THEN amount ELSE 0 END), 0)
            FROM account_transactions
            WHERE user_id = ?
        `, userID).Scan(&totalCredit, &totalDebit)
        
        expectedBalance := totalCredit - totalDebit
        
        // 检查差异
        if math.Abs(currentBalance-expectedBalance) > 0.01 {
            fmt.Printf("账户 %d 余额异常: 当前=%.2f, 期望=%.2f\n",
                userID, currentBalance, expectedBalance)
        }
    }
}