第14章：分布式事务

package tcc

import (
	"context"
	"errors"
	"sync"
)

// TCCTransaction TCC事务
type TCCTransaction struct {
	id           string
	participants []*TCCParticipant
}

// TCCParticipant TCC参与者
type TCCParticipant struct {
	serviceName string
	tryFunc     func(ctx context.Context) error
	confirmFunc func(ctx context.Context) error
	cancelFunc  func(ctx context.Context) error
}

// TCCCoordinator TCC协调器
type TCCCoordinator struct {
	transactions sync.Map
}

func NewTCCCoordinator() *TCCCoordinator {
	return &TCCCoordinator{}
}

// Begin 开启TCC事务
func (tc *TCCCoordinator) Begin(txID string) *TCCTransaction {
	tx := &TCCTransaction{
		id:           txID,
		participants: make([]*TCCParticipant, 0),
	}
	tc.transactions.Store(txID, tx)
	return tx
}

// AddParticipant 添加参与者
func (tx *TCCTransaction) AddParticipant(p *TCCParticipant) {
	tx.participants = append(tx.participants, p)
}

// Execute 执行TCC事务
func (tc *TCCCoordinator) Execute(ctx context.Context, txID string) error {
	txInterface, ok := tc.transactions.Load(txID)
	if !ok {
		return errors.New("transaction not found")
	}

	tx := txInterface.(*TCCTransaction)

	// Phase 1: Try
	if err := tc.tryPhase(ctx, tx); err != nil {
		// Try失败，执行Cancel
		tc.cancelPhase(ctx, tx)
		return err
	}

	// Phase 2: Confirm
	if err := tc.confirmPhase(ctx, tx); err != nil {
		// Confirm失败，执行Cancel
		tc.cancelPhase(ctx, tx)
		return err
	}

	return nil
}

// tryPhase Try阶段
func (tc *TCCCoordinator) tryPhase(ctx context.Context, tx *TCCTransaction) error {
	errChan := make(chan error, len(tx.participants))
	var wg sync.WaitGroup

	for _, p := range tx.participants {
		wg.Add(1)
		go func(participant *TCCParticipant) {
			defer wg.Done()
			if err := participant.tryFunc(ctx); err != nil {
				errChan <- err
			}
		}(p)
	}

	wg.Wait()
	close(errChan)

	for err := range errChan {
		if err != nil {
			return err
		}
	}

	return nil
}

// confirmPhase Confirm阶段
func (tc *TCCCoordinator) confirmPhase(ctx context.Context, tx *TCCTransaction) error {
	errChan := make(chan error, len(tx.participants))
	var wg sync.WaitGroup

	for _, p := range tx.participants {
		wg.Add(1)
		go func(participant *TCCParticipant) {
			defer wg.Done()
			if err := participant.confirmFunc(ctx); err != nil {
				errChan <- err
			}
		}(p)
	}

	wg.Wait()
	close(errChan)

	for err := range errChan {
		if err != nil {
			return err
		}
	}

	return nil
}

// cancelPhase Cancel阶段
func (tc *TCCCoordinator) cancelPhase(ctx context.Context, tx *TCCTransaction) {
	var wg sync.WaitGroup

	for _, p := range tx.participants {
		wg.Add(1)
		go func(participant *TCCParticipant) {
			defer wg.Done()
			participant.cancelFunc(ctx)
		}(p)
	}

	wg.Wait()
}

// 使用示例：下单扣库存
type OrderTCCService struct {
	db *sql.DB
}

// Try: 创建订单（pending状态）
func (s *OrderTCCService) TryCreateOrder(ctx context.Context, order *Order) error {
	_, err := s.db.ExecContext(ctx,
		"INSERT INTO orders (order_id, user_id, amount, status) VALUES (?, ?, ?, 'try')",
		order.OrderID, order.UserID, order.Amount)
	return err
}

// Confirm: 确认订单
func (s *OrderTCCService) ConfirmOrder(ctx context.Context, orderID string) error {
	_, err := s.db.ExecContext(ctx,
		"UPDATE orders SET status = 'confirmed' WHERE order_id = ? AND status = 'try'",
		orderID)
	return err
}

// Cancel: 取消订单
func (s *OrderTCCService) CancelOrder(ctx context.Context, orderID string) error {
	_, err := s.db.ExecContext(ctx,
		"DELETE FROM orders WHERE order_id = ? AND status = 'try'",
		orderID)
	return err
}

// 库存TCC服务
type InventoryTCCService struct {
	db *sql.DB
}

// Try: 冻结库存
func (s *InventoryTCCService) TryFreezeInventory(ctx context.Context, productID string, quantity int) error {
	// 检查库存
	var available int
	err := s.db.QueryRowContext(ctx,
		"SELECT available FROM inventory WHERE product_id = ?", productID).Scan(&available)
	if err != nil {
		return err
	}

	if available < quantity {
		return errors.New("insufficient inventory")
	}

	// 冻结库存
	_, err = s.db.ExecContext(ctx,
		"UPDATE inventory SET available = available - ?, frozen = frozen + ? WHERE product_id = ?",
		quantity, quantity, productID)
	return err
}

// Confirm: 扣减库存
func (s *InventoryTCCService) ConfirmInventory(ctx context.Context, productID string, quantity int) error {
	_, err := s.db.ExecContext(ctx,
		"UPDATE inventory SET frozen = frozen - ? WHERE product_id = ?",
		quantity, productID)
	return err
}

// Cancel: 解冻库存
func (s *InventoryTCCService) CancelInventory(ctx context.Context, productID string, quantity int) error {
	_, err := s.db.ExecContext(ctx,
		"UPDATE inventory SET available = available + ?, frozen = frozen - ? WHERE product_id = ?",
		quantity, quantity, productID)
	return err
}

// 完整流程
func PlaceOrderWithTCC(order *Order) error {
	ctx := context.Background()
	txID := generateTxID()

	coordinator := NewTCCCoordinator()
	tx := coordinator.Begin(txID)

	orderService := &OrderTCCService{db: db}
	inventoryService := &InventoryTCCService{db: db}

	// 添加订单参与者
	tx.AddParticipant(&TCCParticipant{
		serviceName: "order",
		tryFunc: func(ctx context.Context) error {
			return orderService.TryCreateOrder(ctx, order)
		},
		confirmFunc: func(ctx context.Context) error {
			return orderService.ConfirmOrder(ctx, order.OrderID)
		},
		cancelFunc: func(ctx context.Context) error {
			return orderService.CancelOrder(ctx, order.OrderID)
		},
	})

	// 添加库存参与者
	tx.AddParticipant(&TCCParticipant{
		serviceName: "inventory",
		tryFunc: func(ctx context.Context) error {
			return inventoryService.TryFreezeInventory(ctx, order.ProductID, order.Quantity)
		},
		confirmFunc: func(ctx context.Context) error {
			return inventoryService.ConfirmInventory(ctx, order.ProductID, order.Quantity)
		},
		cancelFunc: func(ctx context.Context) error {
			return inventoryService.CancelInventory(ctx, order.ProductID, order.Quantity)
		},
	})

	// 执行TCC事务
	return coordinator.Execute(ctx, txID)
}

package saga

import (
	"context"
	"errors"
)

// SagaStep Saga步骤
type SagaStep struct {
	name         string
	action       func(ctx context.Context) error  // 正向操作
	compensation func(ctx context.Context) error  // 补偿操作
}

// Saga事务
type Saga struct {
	steps          []*SagaStep
	executedSteps  []*SagaStep
}

func NewSaga() *Saga {
	return &Saga{
		steps:         make([]*SagaStep, 0),
		executedSteps: make([]*SagaStep, 0),
	}
}

// AddStep 添加步骤
func (s *Saga) AddStep(name string, action, compensation func(ctx context.Context) error) {
	s.steps = append(s.steps, &SagaStep{
		name:         name,
		action:       action,
		compensation: compensation,
	})
}

// Execute 执行Saga
func (s *Saga) Execute(ctx context.Context) error {
	// 正向执行所有步骤
	for _, step := range s.steps {
		if err := step.action(ctx); err != nil {
			// 失败，执行补偿
			s.compensate(ctx)
			return err
		}

		// 记录已执行步骤
		s.executedSteps = append(s.executedSteps, step)
	}

	return nil
}

// compensate 补偿
func (s *Saga) compensate(ctx context.Context) {
	// 逆序执行补偿
	for i := len(s.executedSteps) - 1; i >= 0; i-- {
		step := s.executedSteps[i]
		if err := step.compensation(ctx); err != nil {
			// 补偿失败，记录日志，人工介入
			log.Printf("Compensation failed for step %s: %v", step.name, err)
		}
	}
}

// 使用示例
func PlaceOrderWithSaga(order *Order) error {
	ctx := context.Background()
	saga := NewSaga()

	// Step 1: 创建订单
	saga.AddStep("create_order",
		func(ctx context.Context) error {
			return createOrder(ctx, order)
		},
		func(ctx context.Context) error {
			return cancelOrder(ctx, order.OrderID)
		})

	// Step 2: 扣减库存
	saga.AddStep("deduct_inventory",
		func(ctx context.Context) error {
			return deductInventory(ctx, order.ProductID, order.Quantity)
		},
		func(ctx context.Context) error {
			return restoreInventory(ctx, order.ProductID, order.Quantity)
		})

	// Step 3: 扣款
	saga.AddStep("deduct_balance",
		func(ctx context.Context) error {
			return deductBalance(ctx, order.UserID, order.Amount)
		},
		func(ctx context.Context) error {
			return refund(ctx, order.UserID, order.Amount)
		})

	// Step 4: 通知发货
	saga.AddStep("notify_shipping",
		func(ctx context.Context) error {
			return notifyShipping(ctx, order.OrderID)
		},
		func(ctx context.Context) error {
			return cancelShipping(ctx, order.OrderID)
		})

	return saga.Execute(ctx)
}

// 消息表
type LocalMessage struct {
	ID          int64
	MessageID   string
	Topic       string
	Payload     string
	Status      string  // pending/sent/failed
	RetryCount  int
	NextRetry   time.Time
	CreatedAt   time.Time
	UpdatedAt   time.Time
}

// 创建订单 + 插入消息表（本地事务）
func CreateOrderWithMessage(order *Order) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	// 1. 创建订单
	_, err = tx.Exec(
		"INSERT INTO orders (order_id, user_id, amount, status) VALUES (?, ?, ?, 'created')",
		order.OrderID, order.UserID, order.Amount)
	if err != nil {
		return err
	}

	// 2. 插入消息表
	message := &LocalMessage{
		MessageID: generateMessageID(),
		Topic:     "order-created",
		Payload:   marshalJSON(order),
		Status:    "pending",
		CreatedAt: time.Now(),
	}

	_, err = tx.Exec(
		"INSERT INTO local_messages (message_id, topic, payload, status, created_at) VALUES (?, ?, ?, ?, ?)",
		message.MessageID, message.Topic, message.Payload, message.Status, message.CreatedAt)
	if err != nil {
		return err
	}

	// 3. 提交事务
	return tx.Commit()
}

// 定时任务：扫描并发送消息
func MessageSender() {
	ticker := time.NewTicker(1 * time.Second)

	for range ticker.C {
		// 查询待发送消息
		rows, _ := db.Query(
			"SELECT id, message_id, topic, payload FROM local_messages WHERE status = 'pending' AND next_retry < NOW() LIMIT 100")

		for rows.Next() {
			var msg LocalMessage
			rows.Scan(&msg.ID, &msg.MessageID, &msg.Topic, &msg.Payload)

			// 发送到Kafka
			err := sendToKafka(msg.Topic, msg.MessageID, []byte(msg.Payload))
			if err != nil {
				// 发送失败，更新重试时间
				updateMessageStatus(msg.ID, "pending", msg.RetryCount+1)
				continue
			}

			// 发送成功，标记为已发送
			updateMessageStatus(msg.ID, "sent", msg.RetryCount)
		}

		rows.Close()
	}
}

func sendToKafka(topic, key string, value []byte) error {
	return producer.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{
			Topic:     &topic,
			Partition: kafka.PartitionAny,
		},
		Key:   []byte(key),
		Value: value,
	}, nil)
}

func updateMessageStatus(id int64, status string, retryCount int) {
	nextRetry := time.Now().Add(time.Duration(math.Pow(2, float64(retryCount))) * time.Second)

	db.Exec(
		"UPDATE local_messages SET status = ?, retry_count = ?, next_retry = ?, updated_at = NOW() WHERE id = ?",
		status, retryCount, nextRetry, id)
}

// RocketMQ事务消息Producer
type TransactionProducer struct {
	producer *rocketmq.TransactionProducer
}

// 事务监听器
type OrderTransactionListener struct{}

// ExecuteLocalTransaction 执行本地事务
func (l *OrderTransactionListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	// 解析消息
	var order Order
	json.Unmarshal(msg.Body, &order)

	// 执行本地事务
	err := createOrder(&order)
	if err != nil {
		return primitive.RollbackMessageState
	}

	return primitive.CommitMessageState
}

// CheckLocalTransaction 回查本地事务
func (l *OrderTransactionListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	// 解析消息
	var order Order
	json.Unmarshal(msg.Body, &order)

	// 检查订单是否存在
	exists, _ := checkOrderExists(order.OrderID)
	if exists {
		return primitive.CommitMessageState
	}

	return primitive.RollbackMessageState
}

// 发送事务消息
func SendTransactionMessage(order *Order) error {
	// 创建事务Producer
	p, err := rocketmq.NewTransactionProducer(
		&OrderTransactionListener{},
		producer.WithNameServer([]string{"localhost:9876"}),
	)
	if err != nil {
		return err
	}

	p.Start()
	defer p.Shutdown()

	// 构造消息
	msg := primitive.NewMessage("order-created", marshalJSON(order))

	// 发送事务消息
	res, err := p.SendMessageInTransaction(context.Background(), msg)
	if err != nil {
		return err
	}

	log.Printf("Send transaction message: %s", res.MsgID)
	return nil
}

type NotificationService struct {
	db       *sql.DB
	maxRetry int
}

// 发送通知（最大努力）
func (ns *NotificationService) SendNotification(event *Event) error {
	// 1. 保存通知记录
	notification := &Notification{
		ID:         generateID(),
		EventType:  event.Type,
		Payload:    marshalJSON(event),
		Status:     "pending",
		RetryCount: 0,
		CreatedAt:  time.Now(),
	}

	ns.db.Exec(
		"INSERT INTO notifications (id, event_type, payload, status, retry_count, created_at) VALUES (?, ?, ?, ?, ?, ?)",
		notification.ID, notification.EventType, notification.Payload, notification.Status, notification.RetryCount, notification.CreatedAt)

	// 2. 尝试发送
	return ns.trySend(notification)
}

func (ns *NotificationService) trySend(notification *Notification) error {
	// 发送HTTP请求
	err := sendHTTPNotification(notification.Payload)
	if err == nil {
		// 成功
		ns.db.Exec("UPDATE notifications SET status = 'success' WHERE id = ?", notification.ID)
		return nil
	}

	// 失败，异步重试
	go ns.scheduleRetry(notification)

	return err
}

func (ns *NotificationService) scheduleRetry(notification *Notification) {
	if notification.RetryCount >= ns.maxRetry {
		// 超过最大重试次数，放弃
		ns.db.Exec("UPDATE notifications SET status = 'failed' WHERE id = ?", notification.ID)
		return
	}

	// 指数退避
	delay := time.Duration(math.Pow(2, float64(notification.RetryCount))) * time.Minute
	time.Sleep(delay)

	// 更新重试次数
	notification.RetryCount++
	ns.db.Exec("UPDATE notifications SET retry_count = ? WHERE id = ?", notification.RetryCount, notification.ID)

	// 重试
	ns.trySend(notification)
}

lock := redis.SetNX("order:"+orderID, 1, 10*time.Second)
if !lock {
  return errors.New("duplicate request")
}
defer redis.Del("order:"+orderID)

// Saga补偿
saga := NewSaga()
saga.AddStep("create_order",
  action: createOrder,
  compensation: cancelOrder)  // 补偿

saga.AddStep("deduct_inventory",
  action: deductInventory,
  compensation: restoreInventory)  // 补偿

if err := saga.Execute(); err != nil {
  // 自动执行补偿
}