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"
	"github.com/nilorg/sqlxplus"
)

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 (使用 SqlxPlus，表结构会自动创建)
	engine, err = outbox.New(outbox.EngineTypeSqlxPlus, 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：使用 SqlxPlus 的 ExecTransaction 方法
	testUserWithSqlxPlusTransaction()

	// 方式2：使用 engine.Transaction 方法 (推荐)
	testUserWithEngineTransaction()

	// 方式3：使用 engine.Begin 手动管理事务
	testUserWithBegin()
}

// testUserWithSqlxPlusTransaction 演示使用 sqlxplus.ExecTransaction
func testUserWithSqlxPlusTransaction() {
	ctx := context.Background()

	// 使用 sqlxplus.ExecTransaction 进行事务管理
	err := sqlxplus.ExecTransaction(ctx, func(txCtx context.Context) error {
		conn, err := sqlxplus.FromSqlxContext(txCtx)
		if err != nil {
			return err
		}

		// 执行业务操作
		_, err = sqlxplus.ExecContext(txCtx, conn.Adapter(), "INSERT INTO users (name, age) VALUES (?, ?)", "Alice", 25)
		if err != nil {
			return err
		}

		// 发布消息（也会在同一个事务中记录）
		return engine.Publish(txCtx, "user.created", map[string]interface{}{
			"name": "Alice",
			"age":  25,
		})
	}, sqlxplus.WithExecTransactionDB(db))

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

	fmt.Println("User created with sqlxplus.ExecTransaction successfully")
}

// testUserWithEngineTransaction 演示使用 engine.Transaction (推荐)
func testUserWithEngineTransaction() {
	ctx := context.Background()

	err := engine.Transaction(ctx, func(txCtx context.Context, db interface{}) error {
		// 通过 context 获取 DBConnect (推荐方式)
		conn, err := sqlxplus.FromSqlxContext(txCtx)
		if err != nil {
			return err
		}

		// 执行业务操作
		_, err = sqlxplus.ExecContext(txCtx, conn.Adapter(), "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 engine.Transaction successfully")
}

// testUserWithBegin 演示使用 engine.Begin 手动管理事务
func testUserWithBegin() {
	ctx := context.Background()

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

	// 通过 Session() 获取 *sqlxplus.Tx
	txDB := tx.Session().(*sqlxplus.Tx)

	// 执行业务操作（使用 sqlxplus.ExecContext）
	_, err = sqlxplus.ExecContext(ctx, txDB, "INSERT INTO users (name, age) VALUES (?, ?)", "Charlie", 35)
	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": "Charlie",
			"age":  35,
		},
	})
	if err != nil {
		log.Printf("commit error: %v", err)
		return
	}

	fmt.Println("User created with engine.Begin successfully")
}