package workflow

import (
	"github.com/tidbcloud/fastflow"
	"github.com/tidbcloud/fastflow/pkg/entity/run"
	"github.com/tidbcloud/pkgv2/db"

	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dataflow"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dedicatedglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/remoteencryptor"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/serverlessglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow/action"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow/dag"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

// Global action instances - each action is created exactly once
type actionInstances struct {
	createDedicatedClusterAction             run.Action
	dedicatedFinalBackupAction               run.Action
	deleteDedicatedClusterAction             run.Action
	restoreDedicatedClusterAction            run.Action
	createDataSyncSinkEndpointAction         run.Action
	approveDataSyncSinkEndpointAction        run.Action
	waitDataSyncSinkEndpointActiveAction     run.Action
	createDataSyncChangefeedAction           run.Action
	waitDataSyncChangefeedRunningAction      run.Action
	createNextGenClusterAction               run.Action
	nextGenFinalBackupAction                 run.Action
	deleteNextGenClusterAction               run.Action
	deleteNextGenResourcePoolIfEmptyAction   run.Action
	stopNextGenClusterRestoreIfExistsAction  run.Action
	waitNextGenClusterReadyAction            run.Action
	restoreNextGenClusterAction              run.Action
	clearNextGenClusterRestoreStatusAction   run.Action
	emitNextGenClusterRestoredEventAction    run.Action
	createServerlessClusterAction            run.Action
	deleteServerlessClusterAction            run.Action
	waitServerlessClusterReadyAction         run.Action
	ensureZeroClusterSpendingLimitAction     run.Action
	createServerlessDataMigrationTaskAction  run.Action
	checkServerlessDataMigrationTaskAction   run.Action
	createMigrationClusterAction             run.Action
	pollDedicatedClusterUntilAvailableAction run.Action
	createDedicatedNodeGroupAction           run.Action
	syncDedicatedSystemVariablesAction       run.Action
	syncDedicatedPdOnlineConfigAction        run.Action
	setMigrationClusterStateAction           run.Action
	shutdownUpgradeClusterAction             run.Action
	restoreToExistingClusterAction           run.Action
	pollRestoreClusterFinishedAction         run.Action

	// Project-migration actions (dynamic DAG)
	ensureProjectMigrationTargetsAction      run.Action
	migrateProjectClustersAction             run.Action
	completeProjectMigrationTaskAction       run.Action
	runDedicatedToPremiumPrecheckItemAction  run.Action
	finalizeDedicatedToPremiumPrecheckAction run.Action
}

type DedicatedToPremiumPrecheckOptions struct {
	SupportedChangefeedSinkTypes []string
}

func RegisterDags(
	wflow workflow.Workflow,
	transaction db.Transactioner,
	customStatusRepo repository.CustomStatusRepository,
	migrationClusterRepo repository.MigrationClusterRepo,
	clusterRepo repository.ClusterRepository,
	dataSyncRepo repository.DataSynchronizationRepo,
	projectMigrationTaskRepo repository.ProjectMigrationTaskRepo,
	dedicatedToPremiumPrecheckRepo repository.DedicatedToPremiumPrecheckRepo,
	serverlessCli serverlessglobal.ServerlessGlobal,
	clusterMover action.BatchClusterMover,
	accountCli account.Client,
	encryptor remoteencryptor.Client,
	dedicatedCli dedicatedglobal.DedicatedGlobal,
	dataflowCli dataflow.DataFlow,
	nextgenCli nextgenglobal.NextGenGlobal,
	eventCommitter eventbus.WALCommitter,
	precheckOptions ...DedicatedToPremiumPrecheckOptions,
) error {
	options := DedicatedToPremiumPrecheckOptions{SupportedChangefeedSinkTypes: []string{"KAFKA", "MYSQL"}}
	if len(precheckOptions) > 0 && len(precheckOptions[0].SupportedChangefeedSinkTypes) > 0 {
		options = precheckOptions[0]
	}
	// Initialize all action instances exactly once
	actions := &actionInstances{
		// Dedicated cluster actions
		createDedicatedClusterAction:      action.NewCreateDedicatedClusterAction(dedicatedCli, encryptor),
		dedicatedFinalBackupAction:        action.NewDedicatedFinalBackupAction(dedicatedCli, dataflowCli, customStatusRepo),
		deleteDedicatedClusterAction:      action.NewDeleteDedicatedClusterAction(dedicatedCli),
		restoreDedicatedClusterAction:     action.NewRestoreDedicatedClusterAction(dedicatedCli, encryptor, accountCli),
		createDataSyncSinkEndpointAction:  action.NewCreateDataSyncSinkEndpointAction(dataSyncRepo, clusterRepo, dedicatedCli, dataflowCli),
		approveDataSyncSinkEndpointAction: action.NewApproveDataSyncSinkEndpointAction(dataSyncRepo, clusterRepo, dedicatedCli, dataflowCli),
		waitDataSyncSinkEndpointActiveAction: action.NewWaitDataSyncSinkEndpointActiveAction(
			dataSyncRepo,
			clusterRepo,
			dedicatedCli,
			dataflowCli,
		),
		createDataSyncChangefeedAction: action.NewCreateDataSyncChangefeedAction(
			dataSyncRepo,
			clusterRepo,
			dedicatedCli,
			dataflowCli,
			encryptor,
		),
		waitDataSyncChangefeedRunningAction:      action.NewWaitDataSyncChangefeedRunningAction(dataSyncRepo, clusterRepo, dataflowCli),
		createMigrationClusterAction:             action.NewCreateMigrationClusterAction(migrationClusterRepo, dedicatedCli),
		pollDedicatedClusterUntilAvailableAction: action.NewPollDedicatedClusterUntilAvailableAction(dedicatedCli),
		createDedicatedNodeGroupAction:           action.NewCreateDedicatedNodeGroupAction(dedicatedCli),
		syncDedicatedSystemVariablesAction:       action.NewSyncDedicatedSystemVariablesAction(migrationClusterRepo, dedicatedCli),
		syncDedicatedPdOnlineConfigAction:        action.NewSyncDedicatedPdOnlineConfigAction(migrationClusterRepo, dedicatedCli),
		setMigrationClusterStateAction:           action.NewSetMigrationClusterStateAction(migrationClusterRepo),
		shutdownUpgradeClusterAction:             action.NewShutdownUpgradeClusterAction(migrationClusterRepo, dedicatedCli),
		restoreToExistingClusterAction:           action.NewRestoreToExistingClusterAction(migrationClusterRepo, clusterRepo, dataflowCli),
		pollRestoreClusterFinishedAction:         action.NewPollRestoreClusterFinishedAction(dataflowCli),

		// NextGen cluster actions
		createNextGenClusterAction: action.NewCreateNextGenClusterAction(
			nextgenCli,
			encryptor,
			clusterRepo,
		),
		nextGenFinalBackupAction:                action.NewNextGenFinalBackupAction(nextgenCli),
		deleteNextGenClusterAction:              action.NewDeleteNextGenClusterAction(nextgenCli),
		deleteNextGenResourcePoolIfEmptyAction:  action.NewDeleteNextGenResourcePoolIfEmptyAction(nextgenCli),
		stopNextGenClusterRestoreIfExistsAction: action.NewStopNextGenClusterRestoreIfExistsAction(nextgenCli),
		waitNextGenClusterReadyAction:           action.NewWaitNextGenClusterReadyAction(nextgenCli),
		restoreNextGenClusterAction:             action.NewRestoreNextGenClusterAction(nextgenCli),
		clearNextGenClusterRestoreStatusAction:  action.NewClearNextGenClusterRestoreStatusAction(nextgenCli, customStatusRepo),
		emitNextGenClusterRestoredEventAction:   action.NewEmitNextGenClusterRestoredEventAction(clusterRepo, eventCommitter),

		// Serverless cluster actions
		createServerlessClusterAction:    action.NewCreateServerlessClusterAction(serverlessCli, encryptor),
		deleteServerlessClusterAction:    action.NewDeleteServerlessClusterAction(serverlessCli),
		waitServerlessClusterReadyAction: action.NewWaitServerlessClusterReadyAction(serverlessCli),
		ensureZeroClusterSpendingLimitAction: action.NewEnsureZeroClusterSpendingLimitAction(
			serverlessCli,
		),
		createServerlessDataMigrationTaskAction: action.NewCreateServerlessDataMigrationTaskAction(
			serverlessCli,
			clusterRepo,
			encryptor,
		),
		checkServerlessDataMigrationTaskAction: action.NewCheckServerlessDataMigrationTaskAction(
			serverlessCli,
			clusterRepo,
		),

		// Project-migration actions (dynamic DAG)
		ensureProjectMigrationTargetsAction: action.NewEnsureProjectMigrationTargetsAction(projectMigrationTaskRepo, accountCli),
		migrateProjectClustersAction:        action.NewMigrateProjectClustersAction(projectMigrationTaskRepo, clusterMover),
		completeProjectMigrationTaskAction:  action.NewCompleteProjectMigrationTaskAction(projectMigrationTaskRepo),
		runDedicatedToPremiumPrecheckItemAction: action.NewRunDedicatedToPremiumPrecheckItemAction(
			transaction, dedicatedToPremiumPrecheckRepo, dedicatedCli, accountCli, dataflowCli, options.SupportedChangefeedSinkTypes,
		),
		finalizeDedicatedToPremiumPrecheckAction: action.NewFinalizeDedicatedToPremiumPrecheckAction(dedicatedToPremiumPrecheckRepo),
	}

	// Build action slices for each DAG by referencing the shared instances
	// Register Dedicated cluster DAGs
	createDedicatedClusterActions := []run.Action{actions.createDedicatedClusterAction}
	if err := wflow.RegisterDag(dag.CreateDedicatedClusterDag(), createDedicatedClusterActions); err != nil {
		return bizerr.WrapError(err, "register create dedicated cluster dag failed")
	}

	deleteDedicatedClusterActions := []run.Action{actions.dedicatedFinalBackupAction, actions.deleteDedicatedClusterAction}
	if err := wflow.RegisterDag(dag.DeleteDedicatedClusterDag(), deleteDedicatedClusterActions); err != nil {
		return bizerr.WrapError(err, "register delete dedicated cluster dag failed")
	}

	restoreDedicatedClusterActions := []run.Action{actions.restoreDedicatedClusterAction}
	if err := wflow.RegisterDag(dag.RestoreDedicatedClusterDag(), restoreDedicatedClusterActions); err != nil {
		return bizerr.WrapError(err, "register restore dedicated cluster dag failed")
	}

	// Register NextGen cluster DAGs
	createNextGenClusterActions := []run.Action{actions.createNextGenClusterAction}
	if err := wflow.RegisterDag(dag.CreateNextGenClusterDag(), createNextGenClusterActions); err != nil {
		return bizerr.WrapError(err, "failed to register create nextgen cluster dag")
	}

	deleteNextGenClusterActions := []run.Action{
		actions.nextGenFinalBackupAction,
		actions.stopNextGenClusterRestoreIfExistsAction,
		actions.deleteNextGenClusterAction,
		actions.deleteNextGenResourcePoolIfEmptyAction,
	}
	if err := wflow.RegisterDag(dag.DeleteNextGenClusterDag(), deleteNextGenClusterActions); err != nil {
		return bizerr.WrapError(err, "failed to register delete nextgen cluster dag")
	}

	restoreNextGenClusterActions := []run.Action{
		actions.waitNextGenClusterReadyAction,
		actions.restoreNextGenClusterAction,
		actions.clearNextGenClusterRestoreStatusAction,
		actions.emitNextGenClusterRestoredEventAction,
	}
	if err := wflow.RegisterDag(dag.RestoreNextGenClusterDag(), restoreNextGenClusterActions); err != nil {
		return bizerr.WrapError(err, "failed to register restore nextgen cluster dag")
	}

	// Register Serverless cluster DAGs
	createServerlessClusterActions := []run.Action{actions.createServerlessClusterAction}
	if err := wflow.RegisterDag(dag.CreateServerlessClusterDag(), createServerlessClusterActions); err != nil {
		return bizerr.WrapError(err, "failed to register create serverless cluster dag")
	}

	deleteServerlessClusterActions := []run.Action{actions.deleteServerlessClusterAction}
	if err := wflow.RegisterDag(dag.DeleteServerlessClusterDag(), deleteServerlessClusterActions); err != nil {
		return bizerr.WrapError(err, "failed to register delete serverless cluster dag")
	}

	claimServerlessClusterFromZeroActions := []run.Action{
		actions.createServerlessClusterAction,
		actions.waitServerlessClusterReadyAction,
		actions.ensureZeroClusterSpendingLimitAction,
		actions.createServerlessDataMigrationTaskAction,
		actions.checkServerlessDataMigrationTaskAction,
	}
	if err := wflow.RegisterDag(dag.ClaimServerlessClusterFromZeroDag(), claimServerlessClusterFromZeroActions); err != nil {
		return bizerr.WrapError(err, "failed to register claim serverless cluster from zero dag")
	}

	// Register actions for dynamic DAGs
	fastflow.RegisterAction(
		[]run.Action{
			actions.createDataSyncSinkEndpointAction,
			actions.approveDataSyncSinkEndpointAction,
			actions.waitDataSyncSinkEndpointActiveAction,
			actions.createDataSyncChangefeedAction,
			actions.waitDataSyncChangefeedRunningAction,
			actions.createMigrationClusterAction,
			actions.pollDedicatedClusterUntilAvailableAction,
			actions.createDedicatedNodeGroupAction,
			actions.syncDedicatedSystemVariablesAction,
			actions.syncDedicatedPdOnlineConfigAction,
			actions.setMigrationClusterStateAction,
			actions.shutdownUpgradeClusterAction,
			actions.restoreToExistingClusterAction,
			actions.pollRestoreClusterFinishedAction,
			// Project-migration dynamic DAG actions
			actions.ensureProjectMigrationTargetsAction,
			actions.migrateProjectClustersAction,
			actions.completeProjectMigrationTaskAction,
			actions.runDedicatedToPremiumPrecheckItemAction,
			actions.finalizeDedicatedToPremiumPrecheckAction,
		},
	)

	return nil
}