package action

import (
	"context"
	"strconv"
	"time"

	"github.com/tidbcloud/fastflow/pkg/entity/run"
	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"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/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/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	maxPolicyControlAttempts  = 2
	policyControlPollInterval = 60 * time.Second
)

type EnableGKEPolicyControlAction struct {
	clusterRepo clusterrepo.Repo
	regionRepo  region.Repo
	regionalCli regionserver.Client
}

type PollUntilGKEPolicyControlEnabledAction struct {
	clusterRepo clusterrepo.Repo
	regionRepo  region.Repo
	regionalCli regionserver.Client
}

func NewEnableGKEPolicyControlAction(
	clusterRepo clusterrepo.Repo,
	regionRepo region.Repo,
	regionalCli regionserver.Client,
) *EnableGKEPolicyControlAction {
	return &EnableGKEPolicyControlAction{
		clusterRepo: clusterRepo,
		regionRepo:  regionRepo,
		regionalCli: regionalCli,
	}
}

func NewPollUntilGKEPolicyControlEnabledAction(
	clusterRepo clusterrepo.Repo,
	regionRepo region.Repo,
	regionalCli regionserver.Client,
) *PollUntilGKEPolicyControlEnabledAction {
	return &PollUntilGKEPolicyControlEnabledAction{
		clusterRepo: clusterRepo,
		regionRepo:  regionRepo,
		regionalCli: regionalCli,
	}
}

func (a *EnableGKEPolicyControlAction) Name() string {
	return EnableGKEPolicyControlActionName
}

func (a *PollUntilGKEPolicyControlEnabledAction) Name() string {
	return PollUntilGKEPolicyControlEnabledActionName
}

func (a *EnableGKEPolicyControlAction) ParameterNew() any {
	return &gkeHeterogeneousDeployActionParam{}
}

func (a *PollUntilGKEPolicyControlEnabledAction) ParameterNew() any {
	return &gkeHeterogeneousDeployActionParam{}
}

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

	tilog.Info(cctx, "start to enable gke policy control")
	defer func() {
		tilog.Info(cctx, "end to enable gke policy control")
	}()

	req, err := parseApplyGKEHeterogeneousDeployRequest(cctx, params)
	if err != nil {
		tilog.Error(cctx, "parse apply gke heterogeneous deploy request failed", zap.Error(err))
		return err
	}

	skip, err := skipIfClusterDeleted(cctx, a.clusterRepo, req.GetClusterId())
	if err != nil {
		tilog.Error(cctx, "check cluster deleted failed", zap.Uint64("cluster_id", req.GetClusterId()), zap.Error(err))
		return err
	}
	if skip {
		return nil
	}

	regionEn, err := a.prepare(cctx, req)
	if err != nil {
		tilog.Error(cctx, "prepare gke policy control enable failed", zap.Uint64("cluster_id", req.GetClusterId()), zap.Error(err))
		return err
	}

	clusterID := strconv.FormatUint(req.GetClusterId(), 10)
	if _, err := a.regionalCli.EnableGKEPolicyControl(
		cctx,
		regionEn,
		&regionaldedicatedv1.EnableGKEPolicyControlRequest{ClusterId: clusterID},
	); err != nil {
		tilog.Error(cctx, "enable gke policy control failed", zap.Uint64("cluster_id", req.GetClusterId()), zap.Error(err))
		return err
	}
	return nil
}

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

	tilog.Info(cctx, "start to poll until gke policy control enabled")
	defer func() {
		tilog.Info(cctx, "end to poll until gke policy control enabled")
	}()

	req, err := parseApplyGKEHeterogeneousDeployRequest(cctx, params)
	if err != nil {
		tilog.Error(cctx, "parse apply gke heterogeneous deploy request failed", zap.Error(err))
		return err
	}

	skip, err := skipIfClusterDeleted(cctx, a.clusterRepo, req.GetClusterId())
	if err != nil {
		tilog.Error(cctx, "check cluster deleted failed", zap.Uint64("cluster_id", req.GetClusterId()), zap.Error(err))
		return err
	}
	if skip {
		return nil
	}

	regionEn, err := a.prepare(cctx, req)
	if err != nil {
		tilog.Error(cctx, "prepare gke policy control poll failed", zap.Uint64("cluster_id", req.GetClusterId()), zap.Error(err))
		return err
	}

	clusterID := strconv.FormatUint(req.GetClusterId(), 10)

	// The first enable is performed by the preceding EnableGKEPolicyControlAction
	// in the DAG. This poll action sleeps, then gets the policy control state.
	// If the state is not ready and not DEGRADED, it re-enables and polls again.
	var lastErr error
	for attempt := range maxPolicyControlAttempts {
		if attempt > 0 {
			tilog.Info(cctx, "re-enable gke policy control",
				zap.Int("attempt", attempt+1),
				zap.Uint64("cluster_id", req.GetClusterId()))
			if _, berr := a.regionalCli.EnableGKEPolicyControl(
				cctx, regionEn,
				&regionaldedicatedv1.EnableGKEPolicyControlRequest{ClusterId: clusterID},
			); berr != nil {
				tilog.Error(cctx, "re-enable gke policy control failed",
					zap.Error(berr),
					zap.Uint64("cluster_id", req.GetClusterId()))
				return berr
			}
		}

		select {
		case <-cctx.Done():
			return cctx.Err()
		case <-time.After(policyControlPollInterval):
		}

		policyControl, err := a.regionalCli.GetGKEPolicyControl(
			cctx, regionEn,
			&regionaldedicatedv1.GetGKEPolicyControlRequest{ClusterId: clusterID},
		)
		if err != nil {
			tilog.Error(cctx, "get gke policy control failed", zap.Uint64("cluster_id", req.GetClusterId()), zap.Error(err))
			return err
		}
		if berr := checkGKEPolicyControlEnabled(policyControl); berr != nil {
			lastErr = berr
			if policyControl.GetState() == regionaldedicatedv1.GKEPolicyControlState_GKE_POLICY_CONTROL_STATE_DEGRADED {
				tilog.Error(cctx, "gke policy control is degraded, stop retry",
					zap.Uint64("cluster_id", req.GetClusterId()),
					zap.Error(berr))
				return berr
			}
			tilog.Warn(cctx, "gke policy control is not ready, will retry",
				zap.Uint64("cluster_id", req.GetClusterId()),
				zap.Error(berr))
			continue
		}
		return nil
	}

	return lastErr
}

func (a *EnableGKEPolicyControlAction) prepare(
	ctx context.Context,
	req *globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest,
) (*entity.Region, bizerr.BizError) {
	return prepareGKEHeterogeneousApplyTarget(ctx, a.clusterRepo, a.regionRepo, req)
}

func (a *PollUntilGKEPolicyControlEnabledAction) prepare(
	ctx context.Context,
	req *globaldedicatedv1.ApplyGKEHeterogeneousDeployRequest,
) (*entity.Region, bizerr.BizError) {
	return prepareGKEHeterogeneousApplyTarget(ctx, a.clusterRepo, a.regionRepo, req)
}

func checkGKEPolicyControlEnabled(policyControl *regionaldedicatedv1.GKEPolicyControl) bizerr.BizError {
	if policyControl == nil {
		return bizerr.New(bizerr.ErrCodeRegionalAPI, "empty gke policy control response from regional-server")
	}

	switch policyControl.GetState() {
	case regionaldedicatedv1.GKEPolicyControlState_GKE_POLICY_CONTROL_STATE_ENABLED:
		return nil
	case regionaldedicatedv1.GKEPolicyControlState_GKE_POLICY_CONTROL_STATE_DEGRADED:
		return bizerr.New(
			bizerr.ErrCodeRegionalAPI,
			"gke policy control is degraded; step=%s; message=%s",
			policyControl.GetCurrentStep(),
			policyControl.GetMessage(),
		)
	default:
		return bizerr.New(
			bizerr.ErrCodeUnknown,
			"gke policy control is not ready; state=%s; step=%s; message=%s",
			policyControl.GetState().String(),
			policyControl.GetCurrentStep(),
			policyControl.GetMessage(),
		)
	}
}