package main

import (
	"database/sql"
	"fmt"
	"math/rand"
	"reflect"
	"strconv"
	"time"

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

const (
	DBName             = "test_tx"
	ProcessedTableName = "processed"
	ActionTableName    = "action"
	StatusTableName    = "status"
	TxPubTable         = "msg_store"
	EventType          = "test-tx"
	RandomFailedFactor = 0.5
)

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)
	if rand.Float64() < RandomFailedFactor {
		return fmt.Errorf("random failed")
	}
	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 pubAction(db *sql.DB, publisher tieventbus.Eventbus, eventID string, opID int, action Action) (err error) {
	tx, err := db.Begin()
	if err != nil {
		return err
	}
	defer func() {
		if err == nil {
			err = tx.Commit()
		} else {
			_ = tx.Rollback()
		}
	}()
	event := &tieventbus.Event{
		ID:   eventID,
		Type: EventType,
		Data: &EventData{opID, action},
	}
	_, err = tx.Exec(fmt.Sprintf("INSERT INTO %s (`event_id`, `op_id`,`action`) VALUES (?, ?, ?)", ActionTableName), eventID, opID, action)
	if err != nil {
		return err
	}
	fmt.Println(fmt.Sprintf("Pub event[ID=%s, Type=%s opID=%d action=%s]", eventID, EventType, opID, action))
	return publisher.PublishWithDBTx(tx, event)
}

func main() {
	conf := mysql.NewConfig()
	conf.Net = "tcp"
	conf.User = "root"
	conf.Addr = "localhost"
	db, err := sql.Open("mysql", conf.FormatDSN())
	if err != nil {
		return
	}
	_, _ = db.Exec("DROP DATABASE " + DBName)
	_, err = db.Exec("CREATE DATABASE " + DBName)
	if err != nil {
		panic(err)
	}
	conf.DBName = DBName
	_ = db.Close()
	db, err = sql.Open("mysql", conf.FormatDSN())
	if err != nil {
		panic(err)
	}
	_, err = db.Exec(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY AUTO_INCREMENT, event_id VARCHAR(128) UNIQUE)", DBName, ProcessedTableName))
	if err != nil {
		panic(err)
	}
	_, err = db.Exec(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY AUTO_INCREMENT, event_id VARCHAR(128), op_id int, `action` VARCHAR(128))", DBName, ActionTableName))
	if err != nil {
		panic(err)
	}
	_, err = db.Exec(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY, `status` VARCHAR(128))", DBName, StatusTableName))
	if err != nil {
		panic(err)
	}
	// Init Eventbus
	eventbus, err := tieventbus.NewEventbus(tieventbus.EventbusConfig{
		AppName:    "app",
		DB:         db,
		TxPubTable: TxPubTable,
	})
	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.Run()
	if err != nil {
		panic(err)
	}
	finished := make(chan struct{})
	// Start to publish
	go func() {
		defer func() { close(finished) }()
		for i := 0; i != 100; i++ {
			ID := i / 2
			var action Action
			if i%2 == 0 {
				action = Open
			} else {
				action = Close
			}
			err := pubAction(db, eventbus, strconv.Itoa(i), ID, action)
			if err != nil {
				fmt.Println("Error occurred: " + err.Error())
				break
			}
			time.Sleep(time.Second)
		}
	}()
	<-finished
}