package action

import (
	"context"
	"strconv"

	"github.com/tidbcloud/fastflow/pkg/entity/run"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/regionserver"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	clusterrepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cluster"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	ResumeCluster = "resume_cluster"
)

type ResumeClusterAction struct {
	regionalCli   regionserver.Client
	clusterRepo   clusterrepo.Repo
	regionRepo    region.Repo
	pausePlanRepo clusterrepo.PausePlanRepo
	committer     eventbus.WALCommitter
	// This action also write back to db, so it needs a transaction.
	transaction db.Transactioner
}

func NewResumeClusterAction(
	regionalCli regionserver.Client,
	clusterRepo clusterrepo.Repo,
	regionRepo region.Repo,
	pausePlanRepo clusterrepo.PausePlanRepo,
	committer eventbus.WALCommitter,
	transaction db.Transactioner,
) *ResumeClusterAction {
	return &ResumeClusterAction{
		regionalCli:   regionalCli,
		clusterRepo:   clusterRepo,
		regionRepo:    regionRepo,
		committer:     committer,
		transaction:   transaction,
		pausePlanRepo: pausePlanRepo,
	}
}

func (a *ResumeClusterAction) Name() string {
	return ResumeCluster
}

func (a *ResumeClusterAction) ParameterNew() any {
	return &common.ClusterIdParameter{}
}

func (a *ResumeClusterAction) Run(ctx run.ExecuteContext, params any) error {
	cctx, span := workflow.GenerateContext(ctx)

	clusterId, err := common.GetClusterId(params)
	if err != nil {
		tilog.Error(cctx, "failed to get cluster id", zap.Error(err))
		return err
	}

	tilog.Info(cctx, "start to resume cluster", zap.Uint64("cluster_id", clusterId))
	defer func() {
		tilog.Info(cctx, "end to resume cluster", zap.Uint64("cluster_id", clusterId))
		span.End()
	}()

	cluster, err := a.clusterRepo.Get(cctx, clusterId, false)
	if err != nil {
		tilog.Error(cctx, "failed to get cluster", zap.Error(err))
		return err
	}

	regionEntity, err := a.regionRepo.SelectRegionById(cctx, cluster.RegionId, cluster.GetOrganizationId())
	if err != nil {
		tilog.Error(cctx, "failed to find region", zap.Error(err))
		return err
	}

	err = a.transaction.Transaction(cctx, func(_ context.Context) error {
		// set cluster status to resuming
		cluster.State = globaldedicatedv1.Cluster_RESUMING

		// Save cluster record to db.
		if err := a.clusterRepo.Save(cctx, cluster); err != nil {
			return err
		}

		// delete cluster pause plan
		if err := a.pausePlanRepo.DeleteByClusterId(cctx, cluster.ClusterId); err != nil {
			return err
		}

		resumeReq := &regionaldedicatedv1.ResumeClusterRequest{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
		}

		if _, err := a.regionalCli.ResumeCluster(cctx, regionEntity, resumeReq); err != nil {
			return err
		}

		// send event
		plan, ok := ctx.GetVar(common.KeyClusterPausePlan)
		if !ok {
			plan = ""
		}
		ev := entity.NewClusterEventBuilder(cctx, cluster, clusterevent.EventTypeResuming).
			SetPausePlanType(plan).
			Build()
		if err := a.committer.Write(cctx, clusterevent.TopicClusterEvent, ev); err != nil {
			return err
		}

		return nil
	})

	if err != nil {
		tilog.Error(cctx, "failed to resume cluster", zap.Error(err))
		return err
	}

	return nil
}