main.go

package main

import (
	"context"
	"fmt"
	"log"

	_ "github.com/go-sql-driver/mysql"
	"github.com/isayme/go-amqp-reconnect/rabbitmq"
	"github.com/jmoiron/sqlx"
	"github.com/nilorg/eventbus"
	"github.com/nilorg/outbox"
)

var (
	db     *sqlx.DB
	bus    eventbus.EventBus
	engine outbox.Engine
)

func init() {
	var (
		err  error
		conn *rabbitmq.Connection
	)
	// 初始化 RabbitMQ
	conn, err = rabbitmq.Dial("amqp://root:test123@localhost:5672/")
	if err != nil {
		panic(err)
	}
	bus, err = eventbus.NewRabbitMQ(conn)
	if err != nil {
		panic(err)
	}

	// 初始化 Sqlx DB
	db, err = sqlx.Connect("mysql", "root:test123@tcp(127.0.0.1:3306)/outbox?charset=utf8&parseTime=True&loc=Local")
	if err != nil {
		panic(err)
	}

	// 创建 Outbox Engine (使用 Sqlx，表结构会自动创建)
	engine, err = outbox.New(outbox.EngineTypeSqlx, db, bus)
	if err != nil {
		panic(err)
	}
}

// User 示例
type User struct {
	ID   int64  `db:"id"`
	Name string `db:"name"`
	Age  int    `db:"age"`
}

func main() {
	ctx := context.Background()

	// 订阅消息
	group := eventbus.NewGroupIDContext(ctx, "user.service.group")
	err := engine.Subscribe(group, "user.created", func(ctx context.Context, msg *outbox.Message) error {
		fmt.Printf("Received user.created: %+v\n", msg)
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	// 方式1：使用事务发布消息
	testUserTransactionWithBegin()

	// 方式2：使用 Transaction 方法
	testUserTransactionMethod()
}

func testUserTransactionWithBegin() {
	ctx := context.Background()
	var err error

	// 开始事务
	tx, err := engine.Begin(ctx)
	if err != nil {
		log.Printf("begin tx error: %v", err)
		return
	}

	// 获取 sqlx.Tx
	txDB := tx.Session().(*sqlx.Tx)

	// 执行业务操作
	_, err = txDB.Exec("INSERT INTO users (name, age) VALUES (?, ?)", "Alice", 25)
	if err != nil {
		tx.Rollback(ctx)
		log.Printf("insert user error: %v", err)
		return
	}

	// 提交事务并发布消息
	err = tx.Commit(ctx, &outbox.CommitMessage{
		Topic: "user.created",
		Value: map[string]interface{}{
			"name": "Alice",
			"age":  25,
		},
	})
	if err != nil {
		log.Printf("commit error: %v", err)
		return
	}

	fmt.Println("User created and message published successfully")
}

func testUserTransactionMethod() {
	ctx := context.Background()

	err := engine.Transaction(ctx, func(ctx context.Context, db interface{}) error {
		txDB := db.(*sqlx.Tx)

		// 执行业务操作
		_, err := txDB.Exec("INSERT INTO users (name, age) VALUES (?, ?)", "Bob", 30)
		if err != nil {
			return err
		}

		return nil
	}, &outbox.CommitMessage{
		Topic: "user.created",
		Value: map[string]interface{}{
			"name": "Bob",
			"age":  30,
		},
	})

	if err != nil {
		log.Printf("transaction error: %v", err)
		return
	}

	fmt.Println("User created with Transaction method successfully")
}