package workflow

import (
	"fmt"
	"os"
	"strings"

	"github.com/go-sql-driver/mysql"
	"github.com/shiningrush/fastflow"
	"github.com/tidbcloud/metrics-integration/global/internal/infrastructure/config"

	mysqlKeeper "github.com/shiningrush/fastflow/keeper/mysql"
	"github.com/shiningrush/fastflow/pkg/mod"
	mysqlStore "github.com/shiningrush/fastflow/store/mysql"
)

const watcherName = "metrics-integration-watcher"

type WorkflowConfig struct {
	IsWorker      bool
	InstanceKey   string // Unique key for the workflow instance, e.g., "metrics-integration"
	MySQLAddr     string
	MySQLPort     uint16
	MySQLUser     string
	MySQLPassword string
	MySQLDBName   string
	DagDir        string
}

func (cfg *WorkflowConfig) initStore() *mysqlStore.Store {
	st := mysqlStore.NewStore(&mysqlStore.StoreOption{
		MySQLConfig: &mysql.Config{
			Addr:   fmt.Sprintf("%s:%d", cfg.MySQLAddr, cfg.MySQLPort),
			User:   cfg.MySQLUser,
			Passwd: cfg.MySQLPassword,
			DBName: cfg.MySQLDBName,
		},
		MigrationSwitch: true,
	})
	if err := st.Init(); err != nil {
		panic(fmt.Sprintf("init store failed: %s", err))
	}
	mod.SetStore(st)
	return st
}

func (cfg *WorkflowConfig) NewWorkflowKeeper() *mysqlKeeper.Keeper {
	keeper := mysqlKeeper.NewKeeper(&mysqlKeeper.KeeperOption{
		Key: cfg.InstanceKey,
		MySQLConfig: &mysql.Config{
			Addr:   fmt.Sprintf("%s:%d", cfg.MySQLAddr, cfg.MySQLPort),
			User:   cfg.MySQLUser,
			Passwd: cfg.MySQLPassword,
			DBName: cfg.MySQLDBName,
		},
		MigrationSwitch: cfg.IsWorker,
		WatcherFlag:     !cfg.IsWorker,
	})
	if err := keeper.Init(); err != nil {
		panic(fmt.Sprintf("init keeper failed: %s", err))
	}
	mod.SetKeeper(keeper)
	return keeper
}

func initCommander() {
	commander := &mod.DefCommander{}
	mod.SetCommander(commander)
}

func NewWorkflowInitialOption(config *config.GlobalConfig) *fastflow.InitialOption {
	workflowConfig := &WorkflowConfig{
		MySQLAddr:     config.MysqlConfig.Host,
		MySQLPort:     config.MysqlConfig.Port,
		MySQLUser:     config.MysqlConfig.User,
		MySQLPassword: config.MysqlConfig.Password,
		MySQLDBName:   config.WorkflowConfig.Database,
		DagDir:        config.WorkflowConfig.DAGDir,
		IsWorker:      checkIsWorker(),
		InstanceKey:   queryPodName(),
	}

	store := workflowConfig.initStore()

	keeper := workflowConfig.NewWorkflowKeeper()

	initCommander()

	return &fastflow.InitialOption{
		Keeper:         keeper,
		Store:          store,
		ReadDagFromDir: config.WorkflowConfig.DAGDir,
	}
}

func checkIsWorker() bool {
	podName := os.Getenv("POD_NAME")
	// Assuming workflow pods have "workflow" in their name
	return strings.Contains(podName, "workflow")
}

func queryPodName() string {
	isWorker := checkIsWorker()
	if !isWorker {
		return watcherName
	}

	podName := os.Getenv("POD_NAME")
	if podName == "" {
		panic(fmt.Errorf("POD_NAME environment variable is empty"))
	}
	return podName
}