package workflowimpl

import (
	"os"
	"strconv"
	"strings"
	"sync/atomic"

	"github.com/go-sql-driver/mysql"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/pkgv2/db"

	mgmtconfig "github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/utilities/idgenerator"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

var (
	globalWorkflow workflow.Workflow
	initialized    atomic.Bool
)

func InitWorkflow(cfg *mysql.Config, provider db.TransProvider, wfCfg *mgmtconfig.Workflow) (workflow.Workflow, error) {
	if initialized.Swap(true) {
		return globalWorkflow, nil
	}

	var id string
	if os.Getenv("POD_NAME") != "" {
		id = os.Getenv("POD_NAME")
	}
	if id == "" {
		id = "local"
	}
	workflowMode := workflow.ModeCompetitive
	if wfCfg != nil && strings.EqualFold(wfCfg.Mode, string(workflow.ModeDispatcher)) {
		workflowMode = workflow.ModeDispatcher
	}

	wf, err := workflow.NewWorkflow(
		workflow.ID(id),
		cfg,
		provider,
		newIDGenerator(),
		workflow.WithWorkerMode(workflowMode),
	)
	if err != nil {
		return nil, err
	}
	globalWorkflow = wf

	// start workflow
	if err := globalWorkflow.Start(); err != nil {
		return nil, err
	}

	globalWorkflow.RegisterMetrics(prometheus.DefaultRegisterer)

	return globalWorkflow, nil
}

func GetWorkflow() workflow.Workflow {
	return globalWorkflow
}

func newIDGenerator() idgenerator.IDGenerator {
	podName := os.Getenv("POD_NAME")
	podIP := os.Getenv(idgenerator.DefaultPodIPEnvKey)
	nid := idgenerator.AutoNodeID
	if podName != "" {
		i := strings.LastIndex(podName, "-")
		if i != -1 {
			node := podName[i+1:]
			nodeNum, err := strconv.ParseUint(node, 10, 64)
			if err == nil {
				nid = idgenerator.NodeID(nodeNum)
			}
		}
	}

	return idgenerator.NewSonyflakeID(nid, podIP)
}