package impl

import (
	"errors"
	"fmt"
	"os"

	"github.com/AlekSi/pointer"
	"github.com/go-sql-driver/mysql"
	"github.com/shiningrush/fastflow"
	mysqlKeeper "github.com/shiningrush/fastflow/keeper/mysql"
	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/shiningrush/fastflow/pkg/utils/data"
	mysqlStore "github.com/shiningrush/fastflow/store/mysql"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	service_config "github.com/tidbcloud/dataflow-service-ng/internal/config"
)

const watcherName = "dataflow-service-watcher"

type workflowImpl struct {
	// store the metadata interface of workflow
	store mod.Store
	// keeper the data interface of workflow
	keeper mod.Keeper
	// commander the command interface of workflow
	commander mod.Commander
}

func NewWorkflow(isWorker bool) (workflow.Workflow, error) {
	store, err := initStore(isWorker)
	if err != nil {
		return nil, err
	}

	keeper, err := initKeeper(isWorker)
	if err != nil {
		return nil, err
	}

	commander := initCommander()

	return &workflowImpl{
		store:     store,
		keeper:    keeper,
		commander: commander,
	}, nil
}

func initKeeper(isWorker bool) (mod.Keeper, error) {
	podName, err := queryPodName(isWorker)
	if err != nil {
		return nil, err
	}
	keeper := mysqlKeeper.NewKeeper(&mysqlKeeper.KeeperOption{
		Key: *podName,
		MySQLConfig: &mysql.Config{
			Addr:   fmt.Sprintf("%s:%d", service_config.Global().MysqlConfig.Host, service_config.Global().MysqlConfig.Port),
			User:   service_config.Global().MysqlConfig.User,
			Passwd: service_config.Global().MysqlConfig.Password,
			DBName: service_config.Global().MysqlConfig.Database,
		},
		MigrationSwitch: isWorker,
		WatcherFlag:     !isWorker,
	})
	if err := keeper.Init(); err != nil {
		return nil, err
	}
	mod.SetKeeper(keeper)
	return keeper, nil
}

func queryPodName(isWorker bool) (*string, error) {
	if !isWorker {
		return pointer.To(watcherName), nil
	}

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

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

func initStore(isWorker bool) (mod.Store, error) {
	store := mysqlStore.NewStore(&mysqlStore.StoreOption{
		MySQLConfig: &mysql.Config{
			Addr:   fmt.Sprintf("%s:%d", service_config.Global().MysqlConfig.Host, service_config.Global().MysqlConfig.Port),
			User:   service_config.Global().MysqlConfig.User,
			Passwd: service_config.Global().MysqlConfig.Password,
			DBName: service_config.Global().MysqlConfig.Database,
		},
		MigrationSwitch: isWorker,
	})
	if err := store.Init(); err != nil {
		return nil, err
	}
	mod.SetStore(store)
	return store, nil
}

func (impl *workflowImpl) Close() {
	mod.SetKeeper(nil)
	mod.SetStore(nil)
	mod.SetCommander(nil)
	impl.store.Close()
	impl.keeper.Close()
}

func (impl *workflowImpl) RegisterDag(dag *entity.Dag, actions []run.Action) error {
	fastflow.RegisterAction(actions)
	oldDag, err := mod.GetStore().GetDag(dag.ID)
	if errors.Is(err, data.ErrDataNotFound) {
		if err := mod.GetStore().CreateDag(dag); err != nil {
			return err
		}
	}
	if oldDag != nil {
		if err := mod.GetStore().UpdateDag(dag); err != nil {
			return err
		}
	}
	return nil
}

func (impl *workflowImpl) RunDag(dagId string, specVar map[string]string) (*entity.DagInstance, error) {
	return impl.commander.RunDag(dagId, specVar)
}

func (impl *workflowImpl) RunDagWithTags(
	dagId string, specVar map[string]string, tags map[string]string,
) (*entity.DagInstance, error) {
	return impl.commander.RunDagWithTags(dagId, specVar, tags)
}

func (impl *workflowImpl) RetryDagIns(dagInsId string, ops ...mod.CommandOptSetter) error {
	return impl.commander.RetryDagIns(dagInsId, ops...)
}

func (impl *workflowImpl) ListDagIns(input *mod.ListDagInstanceInput) ([]*entity.DagInstance, error) {
	return impl.store.ListDagInstance(input)
}

func (impl *workflowImpl) CancelDagIns(dagInsId string, ops ...mod.CommandOptSetter) error {
	return impl.commander.CancelDagIns(dagInsId, ops...)
}