package main

import (
	"database/sql"
	"fmt"
	"reflect"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
)

const (
	DBName             = "test_tx"
	ProcessedTableName = "processed"
	StatusTableName    = "status"
	EventType          = "test-tx"
)

type Action string

const (
	Open  Action = "open"
	Close Action = "close"
)

type Status string

const (
	Active   Status = "active"
	Inactive Status = "inactive"
)

type EventData struct {
	ID int
	Action
}

func consumeAction(db *sql.DB, event *tieventbus.Event) (err error) {
	eventData, ok := event.Data.(*EventData)
	if !ok {
		return fmt.Errorf("cannot reflect data to %s", reflect.TypeOf(EventData{}).Name())
	}
	tx, err := db.Begin()
	if err != nil {
		return err
	}
	defer func() {
		if err == nil {
			err = tx.Commit()
		} else {
			_ = tx.Rollback()
		}
	}()
	fmt.Printf("Start Process event[ID=%s]\n", event.ID)
	row := tx.QueryRow(fmt.Sprintf("SELECT COUNT(1) FROM %s.%s WHERE event_id=? FOR UPDATE", DBName, ProcessedTableName), event.ID)
	var count int
	err = row.Scan(&count)
	if err != nil {
		return err
	}
	if count != 0 {
		// this event has been processed
		return nil
	}
	_, err = tx.Exec(fmt.Sprintf("INSERT INTO %s.%s (event_id) VALUES(?)", DBName, ProcessedTableName), event.ID)

	switch eventData.Action {
	case Open:
		_, err = tx.Exec(fmt.Sprintf("REPLACE INTO %s.%s (id, status) VALUES (?, ?)", DBName, StatusTableName), eventData.ID, Active)
	case Close:
		_, err = tx.Exec(fmt.Sprintf("REPLACE INTO %s.%s (id, status) VALUES (?, ?)", DBName, StatusTableName), eventData.ID, Inactive)
	}
	return nil
}

func main() {
	conf := mysql.NewConfig()
	conf.Net = "tcp"
	conf.User = "root"
	conf.Addr = "localhost"
	conf.DBName = DBName
	db, err := sql.Open("mysql", conf.FormatDSN())
	if err != nil {
		return
	}
	// Init Eventbus
	eventbus, err := tieventbus.NewEventbus(tieventbus.EventbusConfig{
		AppName: "app",
	})
	if err != nil {
		panic(err)
	}

	// Register EventHandleFunc
	eventbus.RegisterHandler(
		tieventbus.HandlerConfig{
			EventType: EventType,
			Group:     "consumer",
			HandleFunc: func(event *tieventbus.Event) error {
				return consumeAction(db, event)
			},
			DataStruct: EventData{},
		})
	err = eventbus.FixDLQ(time.Second * 15)
	if err != nil {
		panic(err)
	}
}