package main

import (
"context"
"fmt"
"log"
"github.com/segmentio/kafka-go"
)

func produceEvent(topic, message string) error {
 writer := kafka.NewWriter(kafka.WriterConfig{
  Brokers: []string{"localhost:9092"},
  Topic:   topic,
 })
defer writer.Close()

 err := writer.WriteMessages(context.Background(),
  kafka.Message{Value: []byte(message)},
 )
if err != nil {
return fmt.Errorf("failed to write message: %w", err)
 }
 log.Printf("Event produced: %s", message)
returnnil
}

func main() {
 err := produceEvent("user-events", `{"userID": "123", "action": "login"}`)
if err != nil {
  log.Fatalf("Error producing event: %v", err)
 }
}

// 命令處理器（寫(xiě)操作）
func handleCommand(command string) error {
 err := produceEvent("command-topic", command)
if err != nil {
return fmt.Errorf("command處理失敗: %v", err)
 }
returnnil
}

// 查詢(xún)處理器（讀操作）
func handleQuery(query string) string {
// 模擬從物化視圖查詢(xún)數(shù)據(jù)
return`{"userID": "123", "status": "active"}`
}

func main() {
// 并發(fā)處理命令與查詢(xún)
gofunc() {
  err := handleCommand(`{"action": "createUser", "userID": "123"}`)
if err != nil {
   log.Fatal(err)
  }
 }()

 result := handleQuery("GET_USER 123")
 fmt.Println("查詢(xún)結(jié)果:", result)
}

// Saga協(xié)調(diào)器監(jiān)聽(tīng)事件并觸發(fā)后續(xù)操作
func sagaOrchestrator(event string) {
switch event {
case"orderCreated":
  produceEvent("inventory-topic", `{"orderID": "123", "action": "reserve"}`)
case"inventoryReserved":
  produceEvent("payment-topic", `{"orderID": "123", "amount": 100}`)
case"paymentCompleted":
  log.Println("訂單處理完成")
 }
}

// 庫(kù)存服務(wù)消費(fèi)者
func consumeInventoryEvents() {
 reader := kafka.NewReader(kafka.ReaderConfig{
  Brokers: []string{"localhost:9092"},
  Topic:   "inventory-topic",
 })
defer reader.Close()

for {
  msg, _ := reader.ReadMessage(context.Background())
  sagaOrchestrator(string(msg.Value))
 }
}

func TestConsumerContract(t *testing.T) {
// 模擬生產(chǎn)者發(fā)送消息
 message := `{"userID": "123", "action": "login"}`
if !isValidContract(message) {
  t.Fatal("消息不符合契約")
 }
}

func isValidContract(message string) bool {
// 驗(yàn)證必需字段是否存在
 requiredFields := []string{"userID", "action"}
for _, field := range requiredFields {
if !strings.Contains(message, field) {
   returnfalse
  }
 }
returntrue
}

func processMessageWithRetry(message string, maxRetries int) error {
for i := 0; i < maxRetries; i++ {
  err := processMessage(message)
if err == nil {
   returnnil
  }
  log.Printf("第%d次重試失敗: %v", i+1, err)
  time.Sleep(2 * time.Second) // 指數(shù)退避可優(yōu)化此處
 }
return sendToDLQ(message)
}

func sendToDLQ(message string) error {
return produceEvent("dlq-topic", message)
}

func processMessage(message string) error {
// 模擬處理邏輯（如解析JSON并更新數(shù)據(jù)庫(kù)）
return fmt.Errorf("臨時(shí)錯(cuò)誤")
}