package action

import (
	"context"
	"fmt"
	"strconv"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	ApplyPDMicroserviceComponentsActionName  = "apply-pd-microservice-components"
	DeletePDMicroserviceComponentsActionName = "delete-pd-microservice-components"
	PatchPDDeploymentModeActionName          = "patch-pd-deployment-mode"
	WaitPDMicroserviceComponentsReadyName    = "wait-pd-microservice-components-ready"
	WaitPDMicroserviceComponentsGoneName     = "wait-pd-microservice-components-gone"
	CommitPDDeploymentModeActionName         = "commit-pd-deployment-mode"
)

// PDMicroserviceComponentsAction applies or deletes the PD microservice
// component resources described by the target shared pool snapshot in DAG args.
// The apply variant is used before switching PDGroup to microservice mode; the
// delete variant is used after PDGroup has switched back to monolithic mode.
type PDMicroserviceComponentsAction struct {
	name                string
	needDelete          bool
	sharedPoolRepo      repo.SharedPoolRepo
	dataplaneManager    dataplane.Manager
	clusterConfigClient clusterconfig.Client
	dataplaneClient     dplanemgrv1.DataPlaneServiceClient
	templateVersionSvc  *templateversion.Service
	imageConfig         *config.ImageConfig
	meteringConfig      *config.MeteringConfig
}

// NewApplyPDMicroserviceComponentsAction creates PD_TSO, PD_SCHEDULING, and
// RESOURCE_MANAGER resources for a monolithic-to-microservice switch.
func NewApplyPDMicroserviceComponentsAction(
	sharedPoolRepo repo.SharedPoolRepo,
	dataplaneManager dataplane.Manager,
	clusterConfigClient clusterconfig.Client,
	dataplaneClient dplanemgrv1.DataPlaneServiceClient,
	templateVersionSvc *templateversion.Service,
	imageConfig *config.ImageConfig,
	meteringConfig *config.MeteringConfig,
) actions.Action {
	return &PDMicroserviceComponentsAction{
		name:                ApplyPDMicroserviceComponentsActionName,
		sharedPoolRepo:      sharedPoolRepo,
		dataplaneManager:    dataplaneManager,
		clusterConfigClient: clusterConfigClient,
		dataplaneClient:     dataplaneClient,
		templateVersionSvc:  templateVersionSvc,
		imageConfig:         imageConfig,
		meteringConfig:      meteringConfig,
	}
}

// NewDeletePDMicroserviceComponentsAction deletes PD_TSO, PD_SCHEDULING, and
// RESOURCE_MANAGER resources after switching PDGroup back to monolithic mode.
func NewDeletePDMicroserviceComponentsAction(
	sharedPoolRepo repo.SharedPoolRepo,
	dataplaneManager dataplane.Manager,
	clusterConfigClient clusterconfig.Client,
	dataplaneClient dplanemgrv1.DataPlaneServiceClient,
	templateVersionSvc *templateversion.Service,
	imageConfig *config.ImageConfig,
	meteringConfig *config.MeteringConfig,
) actions.Action {
	return &PDMicroserviceComponentsAction{
		name:                DeletePDMicroserviceComponentsActionName,
		needDelete:          true,
		sharedPoolRepo:      sharedPoolRepo,
		dataplaneManager:    dataplaneManager,
		clusterConfigClient: clusterConfigClient,
		dataplaneClient:     dataplaneClient,
		templateVersionSvc:  templateVersionSvc,
		imageConfig:         imageConfig,
		meteringConfig:      meteringConfig,
	}
}

func (a *PDMicroserviceComponentsAction) Name() string {
	return a.name
}

func (a *PDMicroserviceComponentsAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	sharedPool, err := a.loadPDDeploymentModeTempSharedPool(ctx, args)
	if err != nil {
		return err
	}
	if !a.needDelete {
		if err := a.clusterConfigClient.EnsurePDMicroserviceComponents(ctx, sharedPool); err != nil {
			tilog.Error(ctx, "failed to ensure pd microservice components in config center",
				zap.String("sharedPoolId", sharedPool.GetId()),
				zap.String("dataplaneId", sharedPool.GetDataplaneName()),
				zap.String("tenantSource", pdMicroserviceTenantSource(sharedPool)),
				zap.Bool("hasMetaProfile", sharedPool.HasConfigCenterMetaProfile()),
				zap.Error(err))
			return fmt.Errorf("failed to ensure pd microservice components in config center: %w", err)
		}
	}

	remoteCli, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}
	nodeVersionMap, err := getNodeVersions(ctx, a.dataplaneClient, sharedPool.GetDataplaneName())
	if err != nil {
		return fmt.Errorf("failed to get node versions for shared pool: %w", err)
	}

	for _, comp := range sharedPool.GetComponentGroups() {
		if !isPDMicroserviceComponent(comp) || !workflow.CheckIsTargetComponent(ctx, comp, args) {
			continue
		}
		if !a.needDelete && comp.GetPhase() == apiv1.Component_AVAILABLE {
			continue
		}

		if err := workflow.FullComponentConfigVersion(ctx, a.clusterConfigClient, a.meteringConfig, sharedPool, comp, ""); err != nil {
			return fmt.Errorf("failed to full component config: %w", err)
		}
		templArgs, err := buildCompTempArgs(a.imageConfig, comp, sharedPool, nodeVersionMap)
		if err != nil {
			return fmt.Errorf("failed to build component template args: %w", err)
		}
		if err := workflow.AddAzureDfsArgs(sharedPool, templArgs); err != nil {
			return fmt.Errorf("failed to add azure args: %w", err)
		}
		for _, storage := range comp.GetStorages() {
			if storage.GetName() != apiv1.Component_Storage_DATA {
				continue
			}
			templArgs["DataStorageSize"] = strconv.FormatUint(uint64(storage.GetSizeGi()), 10)
			templArgs["DataStorageClass"] = workflow.GenStorageClassName(sharedPool, storage)
			templArgs["DataVolumeAttributesClass"] = workflow.GenVolumeAttributesClassName(storage)
		}

		templateName, err := entity.GetComponentTemplateName(comp)
		if err != nil {
			return fmt.Errorf("failed to get component template name: %w", err)
		}
		if err := templateExecuteWithVersionResolver(
			ctx, remoteCli, templateName, comp.GetTemplateVersion(), templArgs, a.needDelete, a.templateVersionSvc,
		); err != nil {
			return fmt.Errorf("failed to execute template for component %s %s: %w", entity.GetComponentTypeName(comp), comp.GetId(), err)
		}
	}
	return nil
}

func pdMicroserviceTenantSource(sharedPool *entity.SharedPool) string {
	if sharedPool.GetProvider().GetTags()[schemav1.LabelUserTenantKey] != "" {
		return "dataplane-tag"
	}
	return "unavailable"
}

func (a *PDMicroserviceComponentsAction) loadPDDeploymentModeTempSharedPool(ctx context.Context, args actions.ActionArgs) (*entity.SharedPool, error) {
	// The DAG carries a frozen target topology while the repo keeps the current
	// shared pool metadata. Only overlay PD deployment components so regenerated
	// non-PD component data in the target snapshot cannot be persisted without
	// matching dataplane apply steps.
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return nil, fmt.Errorf("failed to get sharedPool id: %w", err)
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return nil, fmt.Errorf("failed to load sharedPool: %w", err)
	}

	sharedPoolStr, err := args.GetString(workflow.SharedPoolKey)
	if err != nil {
		return nil, fmt.Errorf("failed to get sharedPool: %w", err)
	}
	var tempSharedPool entity.SharedPool
	if err := protojson.Unmarshal([]byte(sharedPoolStr), &tempSharedPool); err != nil {
		return nil, fmt.Errorf("failed to unmarshal sharedPool: %w", err)
	}
	sharedPool.ComponentGroups = MergePDDeploymentModeComponents(sharedPool.GetComponentGroups(), tempSharedPool.GetComponentGroups())
	sharedPool.PdDeploymentMode = tempSharedPool.GetPdDeploymentMode()
	return sharedPool, nil
}

type pdDeploymentModeComponentKey struct {
	id            string
	componentType entity.ComponentType
}

// MergePDDeploymentModeComponents overlays only PD deployment mode components
// from the target snapshot onto the current repo component list.
func MergePDDeploymentModeComponents(currentComponents, targetComponents []*entity.Component) []*entity.Component {
	targetPDComponents := make(map[pdDeploymentModeComponentKey]*entity.Component)
	for _, comp := range targetComponents {
		if isPDDeploymentModeComponent(comp) {
			targetPDComponents[pdDeploymentModeComponentKeyFor(comp)] = comp
		}
	}

	mergedComponents := make([]*entity.Component, 0, len(currentComponents)+len(targetPDComponents))
	mergedPDComponents := make(map[pdDeploymentModeComponentKey]struct{}, len(targetPDComponents))
	for _, comp := range currentComponents {
		key := pdDeploymentModeComponentKeyFor(comp)
		if targetComp, ok := targetPDComponents[key]; ok {
			mergedComponents = append(mergedComponents, targetComp)
			mergedPDComponents[key] = struct{}{}
			continue
		}
		if isPDDeploymentModeComponent(comp) {
			continue
		}
		mergedComponents = append(mergedComponents, comp)
	}

	for _, comp := range targetComponents {
		if !isPDDeploymentModeComponent(comp) {
			continue
		}
		key := pdDeploymentModeComponentKeyFor(comp)
		if _, ok := mergedPDComponents[key]; ok {
			continue
		}
		mergedComponents = append(mergedComponents, comp)
	}
	return mergedComponents
}

func pdDeploymentModeComponentKeyFor(comp *entity.Component) pdDeploymentModeComponentKey {
	return pdDeploymentModeComponentKey{
		id:            comp.GetId(),
		componentType: comp.GetComponentType(),
	}
}

// CommitPDDeploymentModeAction persists the final shared pool metadata after
// all dataplane-side mode switch steps have completed successfully.
type CommitPDDeploymentModeAction struct {
	sharedPoolRepo repo.SharedPoolRepo
}

// NewCommitPDDeploymentModeAction creates the final metadata commit step for a
// PD deployment mode switch DAG.
func NewCommitPDDeploymentModeAction(sharedPoolRepo repo.SharedPoolRepo) actions.Action {
	return &CommitPDDeploymentModeAction{
		sharedPoolRepo: sharedPoolRepo,
	}
}

func (a *CommitPDDeploymentModeAction) Name() string {
	return CommitPDDeploymentModeActionName
}

func (a *CommitPDDeploymentModeAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	loader := &PDMicroserviceComponentsAction{sharedPoolRepo: a.sharedPoolRepo}
	sharedPool, err := loader.loadPDDeploymentModeTempSharedPool(ctx, args)
	if err != nil {
		return err
	}

	targetModeStr, err := args.GetString(workflow.PDDeploymentModeKey)
	if err != nil {
		return fmt.Errorf("failed to get pd deployment mode: %w", err)
	}
	targetMode, err := parsePDDeploymentMode(targetModeStr)
	if err != nil {
		return err
	}

	FinalizePDDeploymentModeSharedPool(sharedPool, targetMode)
	if err := a.sharedPoolRepo.Save(ctx, sharedPool); err != nil {
		return fmt.Errorf("failed to save sharedPool meta: %w", err)
	}
	return nil
}

// FinalizePDDeploymentModeSharedPool normalizes the PD-related repo topology to
// the stable target mode. Microservice mode keeps the generated PD microservice
// components; monolithic mode removes them after their resources are gone.
func FinalizePDDeploymentModeSharedPool(sharedPool *entity.SharedPool, targetMode entity.PDDeploymentMode) {
	sharedPool.PdDeploymentMode = targetMode
	sharedPool.Phase = apiv1.SharedPool_AVAILABLE
	sharedPool.PdDeploymentStatus = &entity.PDDeploymentStatus{
		CurrentMode: targetMode,
		Phase:       apiv1.PDDeploymentStatusPhase_PD_DEPLOYMENT_STATUS_PHASE_STABLE,
	}

	switch targetMode {
	case apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MICROSERVICE:
		for _, comp := range sharedPool.GetComponentGroups() {
			if comp.GetComponentType() == apiv1.ComponentType_PD || isPDMicroserviceComponent(comp) {
				comp.Phase = apiv1.Component_AVAILABLE
			}
		}
	case apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC:
		componentGroups := sharedPool.GetComponentGroups()
		sharedPool.ComponentGroups = componentGroups[:0]
		for _, comp := range componentGroups {
			if isPDMicroserviceComponent(comp) {
				continue
			}
			if comp.GetComponentType() == apiv1.ComponentType_PD {
				comp.Phase = apiv1.Component_AVAILABLE
			}
			sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, comp)
		}
	}
}

// PatchPDDeploymentModeAction patches PDGroup.spec.template.spec.mode and waits
// until TiDB Operator reports the target mode and mode switching is finished.
type PatchPDDeploymentModeAction struct {
	sharedPoolRepo   repo.SharedPoolRepo
	dataplaneManager dataplane.Manager
}

// NewPatchPDDeploymentModeAction creates the Operator-side PDGroup mode patch
// step for both switch directions.
func NewPatchPDDeploymentModeAction(sharedPoolRepo repo.SharedPoolRepo, dataplaneManager dataplane.Manager) actions.AsyncAction {
	return &PatchPDDeploymentModeAction{
		sharedPoolRepo:   sharedPoolRepo,
		dataplaneManager: dataplaneManager,
	}
}

func (a *PatchPDDeploymentModeAction) Name() string {
	return PatchPDDeploymentModeActionName
}

func (a *PatchPDDeploymentModeAction) Start(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	sharedPool, targetMode, err := a.loadSharedPoolAndTargetMode(ctx, args)
	if err != nil {
		return err
	}
	remoteCli, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}
	if err := tioperator.UpdatePDGroupMode(ctx, remoteCli, sharedPool.GetId(), defaultPDGroupID(sharedPool), tioperator.APIToPDMode(targetMode)); err != nil {
		return err
	}
	return nil
}

func (a *PatchPDDeploymentModeAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	ctx := utils.GenerateContext(args)
	sharedPool, targetMode, err := a.loadSharedPoolAndTargetMode(ctx, args)
	if err != nil {
		return false, err
	}
	remoteCli, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return false, fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}
	pdGroup, err := tioperator.GetPDGroupCR(ctx, remoteCli, sharedPool.GetId(), defaultPDGroupID(sharedPool))
	if err != nil {
		return false, err
	}
	if tioperator.IsPDModeSwitchingTrue(tioperator.GetPDModeSwitchingCondition(pdGroup)) {
		return false, nil
	}
	targetPDMode := tioperator.APIToPDMode(targetMode)
	return pdGroup.Spec.Template.Spec.Mode == targetPDMode && pdGroup.Status.Mode == targetPDMode, nil
}

func (a *PatchPDDeploymentModeAction) loadSharedPoolAndTargetMode(ctx context.Context, args actions.ActionArgs) (*entity.SharedPool, entity.PDDeploymentMode, error) {
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return nil, apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED, fmt.Errorf("failed to get sharedPool id: %w", err)
	}
	targetModeStr, err := args.GetString(workflow.PDDeploymentModeKey)
	if err != nil {
		return nil, apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED, fmt.Errorf("failed to get pd deployment mode: %w", err)
	}
	targetMode, err := parsePDDeploymentMode(targetModeStr)
	if err != nil {
		return nil, apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED, err
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return nil, apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED, fmt.Errorf("failed to load sharedPool: %w", err)
	}
	return sharedPool, targetMode, nil
}

// WaitPDMicroserviceComponentsAction waits for PD microservice resources to
// either become available or be fully deleted, depending on the switch direction.
type WaitPDMicroserviceComponentsAction struct {
	name             string
	waitGone         bool
	sharedPoolRepo   repo.SharedPoolRepo
	dataplaneManager dataplane.Manager
}

// NewWaitPDMicroserviceComponentsReadyAction creates the wait step used after
// rendering PD microservice resources during monolithic-to-microservice switch.
func NewWaitPDMicroserviceComponentsReadyAction(sharedPoolRepo repo.SharedPoolRepo, dataplaneManager dataplane.Manager) actions.AsyncAction {
	return &WaitPDMicroserviceComponentsAction{
		name:             WaitPDMicroserviceComponentsReadyName,
		sharedPoolRepo:   sharedPoolRepo,
		dataplaneManager: dataplaneManager,
	}
}

// NewWaitPDMicroserviceComponentsGoneAction creates the wait step used after
// deleting PD microservice resources during microservice-to-monolithic switch.
func NewWaitPDMicroserviceComponentsGoneAction(sharedPoolRepo repo.SharedPoolRepo, dataplaneManager dataplane.Manager) actions.AsyncAction {
	return &WaitPDMicroserviceComponentsAction{
		name:             WaitPDMicroserviceComponentsGoneName,
		waitGone:         true,
		sharedPoolRepo:   sharedPoolRepo,
		dataplaneManager: dataplaneManager,
	}
}

func (a *WaitPDMicroserviceComponentsAction) Name() string {
	return a.name
}

func (a *WaitPDMicroserviceComponentsAction) Start(args actions.ActionArgs) error {
	return nil
}

func (a *WaitPDMicroserviceComponentsAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	ctx := utils.GenerateContext(args)
	loader := &PDMicroserviceComponentsAction{
		sharedPoolRepo:   a.sharedPoolRepo,
		dataplaneManager: a.dataplaneManager,
	}
	sharedPool, err := loader.loadPDDeploymentModeTempSharedPool(ctx, args)
	if err != nil {
		return false, err
	}
	remoteCli, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return false, fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}

	for _, comp := range sharedPool.GetComponentGroups() {
		if !isPDMicroserviceComponent(comp) || !workflow.CheckIsTargetComponent(ctx, comp, args) {
			continue
		}
		available, err := isPDMicroserviceComponentAvailable(ctx, remoteCli, sharedPool.GetId(), comp)
		if a.waitGone {
			if status.Code(err) == codes.NotFound {
				continue
			}
			if err != nil {
				return false, err
			}
			return false, nil
		}
		if err != nil {
			return false, err
		}
		if !available {
			tilog.Info(ctx, "pd microservice component is not ready",
				zap.String("componentId", comp.GetId()),
				zap.String("componentType", entity.GetComponentTypeName(comp)))
			return false, nil
		}
	}
	return true, nil
}

func isPDMicroserviceComponent(comp *entity.Component) bool {
	switch comp.GetComponentType() {
	case apiv1.ComponentType_PD_TSO, apiv1.ComponentType_PD_SCHEDULING, apiv1.ComponentType_RESOURCE_MANAGER:
		return true
	default:
		return false
	}
}

func isPDDeploymentModeComponent(comp *entity.Component) bool {
	return comp.GetComponentType() == apiv1.ComponentType_PD || isPDMicroserviceComponent(comp)
}

func isPDMicroserviceComponentAvailable(ctx context.Context, remoteCli client.Client, sharedPoolID string, comp *entity.Component) (bool, error) {
	switch comp.GetComponentType() {
	case apiv1.ComponentType_PD_TSO:
		tsoGroup, err := tioperator.GetTsoGroup(ctx, remoteCli, sharedPoolID, comp.GetId())
		if err != nil {
			return false, err
		}
		return tsoGroup.GetPhase() == apiv1.Component_AVAILABLE, err
	case apiv1.ComponentType_PD_SCHEDULING:
		schedulingGroup, err := tioperator.GetSchedulingGroup(ctx, remoteCli, sharedPoolID, comp.GetId())
		if err != nil {
			return false, err
		}
		return schedulingGroup.GetPhase() == apiv1.Component_AVAILABLE, err
	case apiv1.ComponentType_RESOURCE_MANAGER:
		rmGroup, err := tioperator.GetResourceManagerGroup(ctx, remoteCli, sharedPoolID, comp.GetId())
		if err != nil {
			return false, err
		}
		return rmGroup.GetPhase() == apiv1.Component_AVAILABLE, err
	default:
		return true, nil
	}
}

func parsePDDeploymentMode(modeStr string) (entity.PDDeploymentMode, error) {
	switch modeStr {
	case apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MICROSERVICE.String():
		return apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MICROSERVICE, nil
	case apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC.String():
		return apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC, nil
	default:
		return apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED, fmt.Errorf("unsupported pd deployment mode %q", modeStr)
	}
}

func defaultPDGroupID(sharedPool *entity.SharedPool) string {
	if defaultComponentID := sharedPool.GetDefaultComponenId(); defaultComponentID != "" {
		return defaultComponentID
	}
	return utils.FnvHash(sharedPool.GetId())
}