package impl

import (
	"context"
	"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/app/workflow"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
	"gorm.io/gorm"
)

const watcherName = "dataflow-service-watcher"

const businessIDTagKey = "business_id"

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
	// db is a gorm handle for the same MySQL database that fastflow's store
	// writes to. It powers selective-first tag queries that bypass fastflow's
	// (key,value) IN + GROUP BY HAVING aggregation. May be nil in legacy
	// callers that have not migrated yet — methods that require it must
	// nil-check.
	db *gorm.DB
}

func NewWorkflow(isWorker bool, db *gorm.DB) (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,
		db:        db,
	}, 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...)
}

func (impl *workflowImpl) FindDagInstancesByBusinessID(
	ctx context.Context,
	businessID string,
	dagIDs []string,
	statuses []entity.DagInstanceStatus,
	limit int,
) ([]*entity.DagInstance, error) {
	if impl.db == nil {
		return nil, errors.New("workflowImpl: db handle is not configured")
	}
	if businessID == "" {
		return nil, errors.New("workflowImpl: businessID is required")
	}

	// Step 1: high-selectivity lookup. `(key, value)` is indexed by
	// `idx_dag_instance_tags_key_value`, and businessID is unique enough
	// that this returns O(1) rows regardless of how many backups exist.
	var dagInsIDs []string
	if err := impl.db.WithContext(ctx).
		Table("dag_instance_tags").
		Where("`key` = ? AND `value` = ?", businessIDTagKey, businessID).
		Pluck("dag_ins_id", &dagInsIDs).Error; err != nil {
		return nil, fmt.Errorf("query dag_instance_tags by business_id %q: %w", businessID, err)
	}
	if len(dagInsIDs) == 0 {
		return nil, nil
	}

	// Step 2: PK + status + dag_id filter on the small candidate set.
	tx := impl.db.WithContext(ctx).
		Table("dag_instances").
		Where("id IN ?", dagInsIDs)
	if len(dagIDs) > 0 {
		tx = tx.Where("dag_id IN ?", dagIDs)
	}
	if len(statuses) > 0 {
		tx = tx.Where("status IN ?", statuses)
	}
	if limit > 0 {
		tx = tx.Limit(limit)
	}

	var ret []*entity.DagInstance
	if err := tx.Find(&ret).Error; err != nil {
		return nil, fmt.Errorf("query dag_instances for business_id %q: %w", businessID, err)
	}
	return ret, nil
}