package action

import (
	"context"
	"fmt"
	"path"
	"strings"

	v2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	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/cert"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"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/protobuf/encoding/protojson"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type ApplyDMClusterAction struct {
	dataplaneManager    dataplane.Manager
	sharedPoolRepo      repo.SharedPoolRepo
	dmClusterRepo       repo.DMClusterRepo
	clusterConfigClient clusterconfig.Client
	imageConfig         *config.ImageConfig
	dataplaneClient     dplanemgrv1.DataPlaneServiceClient
}

const ApplyDMClusterActionName = "apply-dm-cluster-action"

func NewApplyDMClusterAction(
	sharedPoolRepo repo.SharedPoolRepo,
	dmClusterRepo repo.DMClusterRepo,
	dataplaneManager dataplane.Manager,
	dataplaneClient dplanemgrv1.DataPlaneServiceClient,
	imageConfig *config.ImageConfig,
	clusterConfigClient clusterconfig.Client,
) actions.Action {
	return &ApplyDMClusterAction{
		sharedPoolRepo:      sharedPoolRepo,
		dmClusterRepo:       dmClusterRepo,
		clusterConfigClient: clusterConfigClient,
		dataplaneManager:    dataplaneManager,
		dataplaneClient:     dataplaneClient,
		imageConfig:         imageConfig,
	}
}

func (a *ApplyDMClusterAction) Name() string {
	return ApplyDMClusterActionName
}

func (a *ApplyDMClusterAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "apply dm cluster action start")

	// check if need delete
	needDelete := false
	deleteStr, err := args.GetString(workflow.DMDeleteKey)
	if err == nil && deleteStr == workflow.DMDeleteValue {
		needDelete = true
	}

	dmClusterId, err := args.GetString(workflow.DMClusterIdKey)
	if err != nil {
		return fmt.Errorf("dm cluster id is required, but not found: %w", err)
	}

	// Load DM cluster
	dmCluster, err := a.dmClusterRepo.Load(ctx, dmClusterId, needDelete)
	if err != nil {
		return fmt.Errorf("failed to load dm cluster: %w", err)
	}

	dmClusterStr, err := args.GetString(workflow.DMClusterKey)
	if err != nil {
		return fmt.Errorf("failed to get dmCluster: %w", err)
	}

	var tempDMCluster entity.DMCluster
	err = protojson.Unmarshal([]byte(dmClusterStr), &tempDMCluster)
	if err != nil {
		tilog.Error(ctx, "failed to unmarshal dmCluster", zap.Error(err))
		return fmt.Errorf("failed to unmarshal dmCluster: %w", err)
	}

	dmCluster.MasterNodeQuota = tempDMCluster.MasterNodeQuota
	dmCluster.WorkerNodeQuota = tempDMCluster.WorkerNodeQuota

	updateImage, _ := args.GetBool(workflow.DMUpdateImageKey)

	// Get remote client
	remoteCli, err := a.dataplaneManager.GetClient(dmCluster.GetDataplaneName())
	if err != nil {
		tilog.Error(ctx, "failed to get remote client for dm cluster",
			zap.String("dataplaneName", dmCluster.GetDataplaneName()),
			zap.Error(err))
		return fmt.Errorf("failed to get remote client for dm cluster: %w", err)
	}

	return a.applyDMClusterV2(ctx, remoteCli, dmCluster, needDelete, updateImage)
}

func (a *ApplyDMClusterAction) applyDMClusterV2(
	ctx context.Context,
	remoteCli client.Client,
	dmCluster *entity.DMCluster,
	needDelete bool,
	updateImage bool,
) error {
	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
	if needDelete {
		if err := tioperator.DeleteDMClusterV2(ctx, remoteCli, dmCluster); err != nil {
			return fmt.Errorf("failed to delete dm cluster v2 resources: %w", err)
		}
		if err := deleteDMCertsV2(ctx, remoteCli, dmCluster.GetId(), namespace); err != nil {
			return err
		}
		return nil
	}

	nodeVersionMap, err := getNodeVersions(ctx, a.dataplaneClient, dmCluster.GetDataplaneName())
	if err != nil {
		tilog.Error(ctx, "failed to get node versions for dm cluster v2",
			zap.String("dmClusterId", dmCluster.GetId()),
			zap.Error(err))
		return fmt.Errorf("failed to get node versions for dm cluster v2: %w", err)
	}
	if err := applyNodeVersionsToDMCluster(dmCluster, nodeVersionMap); err != nil {
		return fmt.Errorf("failed to resolve node versions for dm cluster v2: %w", err)
	}

	resolveImage, err := needDMClusterV2ImageFallback(ctx, remoteCli, dmCluster, namespace, updateImage)
	if err != nil {
		return fmt.Errorf("failed to check dm cluster v2 image and version: %w", err)
	}

	var dmVersion, dmImage string
	if resolveImage {
		dmVersion, dmImage, err = a.resolveDMComponentImage(ctx, dmCluster)
		if err != nil {
			return fmt.Errorf("failed to resolve dm component image: %w", err)
		}
	}

	if err := k8s.EnsureNamespace(ctx, remoteCli, namespace, ""); err != nil {
		return fmt.Errorf("failed to ensure dataplane namespace %s: %w", namespace, err)
	}

	if err := createDMCertsV2(ctx, remoteCli, dmCluster.GetId(), namespace); err != nil {
		return err
	}

	if err := tioperator.ApplyDMClusterV2(ctx, remoteCli, dmCluster, dmVersion, dmImage, updateImage); err != nil {
		return fmt.Errorf("failed to apply dm cluster v2 resources: %w", err)
	}
	return nil
}

func needDMClusterV2ImageFallback(
	ctx context.Context,
	remoteCli client.Client,
	dmCluster *entity.DMCluster,
	namespace string,
	updateImage bool,
) (bool, error) {
	if updateImage {
		return true, nil
	}

	groupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	if err != nil {
		return false, err
	}

	if dmCluster.GetMasterNodeQuota() != nil {
		needFallback, err := needDMGroupV2ImageFallback(ctx, remoteCli, namespace, groupName)
		if err != nil || needFallback {
			return needFallback, err
		}
	}
	if dmCluster.GetWorkerNodeQuota() != nil {
		return needDMWorkerGroupV2ImageFallback(ctx, remoteCli, namespace, groupName)
	}
	return false, nil
}

func needDMGroupV2ImageFallback(ctx context.Context, remoteCli client.Client, namespace, name string) (bool, error) {
	var group v2.DMGroup
	if err := remoteCli.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, &group); err != nil {
		if apierrors.IsNotFound(err) {
			return true, nil
		}
		return false, err
	}
	return group.Spec.Template.Spec.Version == "" ||
		group.Spec.Template.Spec.Image == nil ||
		*group.Spec.Template.Spec.Image == "", nil
}

func needDMWorkerGroupV2ImageFallback(ctx context.Context, remoteCli client.Client, namespace, name string) (bool, error) {
	var group v2.DMWorkerGroup
	if err := remoteCli.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, &group); err != nil {
		if apierrors.IsNotFound(err) {
			return true, nil
		}
		return false, err
	}
	return group.Spec.Template.Spec.Version == "" ||
		group.Spec.Template.Spec.Image == nil ||
		*group.Spec.Template.Spec.Image == "", nil
}

func (a *ApplyDMClusterAction) resolveDMComponentImage(ctx context.Context, dmCluster *entity.DMCluster) (string, string, error) {
	if dmCluster == nil {
		return "", "", fmt.Errorf("dm cluster is nil")
	}
	if a.sharedPoolRepo == nil {
		return "", "", fmt.Errorf("shared pool repo is nil")
	}
	if a.clusterConfigClient == nil {
		return "", "", fmt.Errorf("cluster config client is nil")
	}

	sharedPool, err := a.sharedPoolRepo.Load(ctx, dmCluster.GetSharedPoolId(), false)
	if err != nil {
		return "", "", fmt.Errorf("failed to load shared pool: %w", err)
	}
	if sharedPool == nil {
		return "", "", fmt.Errorf("shared pool is nil")
	}
	version, cfg, err := a.clusterConfigClient.GetDMComponentVersionAndConfig(ctx, sharedPool)
	if err != nil {
		return a.fallbackDMImage(ctx, dmCluster, sharedPool, version, err)
	}
	image, err := buildDMImageFromComponentConfig(a.imageConfig, cfg)
	if err != nil {
		return a.fallbackDMImage(ctx, dmCluster, sharedPool, version, err)
	}
	return version, image, nil
}

func (a *ApplyDMClusterAction) fallbackDMImage(
	ctx context.Context,
	dmCluster *entity.DMCluster,
	sharedPool *entity.SharedPool,
	version string,
	resolveErr error,
) (string, string, error) {
	if a.imageConfig == nil || strings.TrimSpace(a.imageConfig.DmImage) == "" {
		return "", "", resolveErr
	}
	if strings.TrimSpace(version) == "" {
		version = sharedPool.GetVersion()
	}
	if strings.TrimSpace(version) == "" {
		return "", "", resolveErr
	}

	tilog.Warn(ctx, "failed to resolve dm component image from config center, fallback to static dm image",
		zap.String("dmClusterId", dmCluster.GetId()),
		zap.String("sharedPoolId", sharedPool.GetId()),
		zap.Error(resolveErr))
	return version, strings.TrimSpace(a.imageConfig.DmImage), nil
}

func buildDMImageFromComponentConfig(imageConfig *config.ImageConfig, cfg *entity.ComponentConfiguration) (string, error) {
	if cfg == nil {
		return "", fmt.Errorf("dm component config is nil")
	}
	imageRepo := strings.TrimSpace(cfg.GetImageRepo())
	imageTag := strings.TrimSpace(cfg.GetImageTag())
	if imageRepo == "" || imageTag == "" {
		return "", fmt.Errorf("dm component image repo or tag is empty")
	}

	if imageConfig != nil && imageConfig.Registry != "" && imageConfig.RepoPrefix != "" {
		return fmt.Sprintf("%s/%s/%s:%s", imageConfig.Registry, imageConfig.RepoPrefix, path.Base(imageRepo), imageTag), nil
	}
	return fmt.Sprintf("%s:%s", imageRepo, imageTag), nil
}

func createDMCertsV2(ctx context.Context, remoteCli client.Client, dmClusterID, namespace string) error {
	certs, err := cert.DMClusterCertsV2(dmClusterID, namespace)
	if err != nil {
		return fmt.Errorf("failed to build dm cluster certs: %w", err)
	}
	for _, certCR := range certs {
		if err := k8s.CreateIfNotExist(ctx, remoteCli, certCR); err != nil {
			tilog.Error(ctx, "failed to create dm cluster cert",
				zap.String("certName", certCR.Name),
				zap.String("namespace", certCR.Namespace),
				zap.Error(err))
			return fmt.Errorf("failed to create dm cluster cert %s: %w", certCR.Name, err)
		}
		tilog.Info(ctx, "dm cluster cert created",
			zap.String("certName", certCR.Name),
			zap.String("namespace", certCR.Namespace))
	}
	return nil
}

func deleteDMCertsV2(ctx context.Context, remoteCli client.Client, dmClusterID, namespace string) error {
	certs, err := cert.DMClusterCertsV2(dmClusterID, namespace)
	if err != nil {
		return fmt.Errorf("failed to build dm cluster certs: %w", err)
	}
	for _, certCR := range certs {
		if err := k8s.DeleteObject(ctx, remoteCli, certCR); err != nil {
			tilog.Error(ctx, "failed to delete dm cluster cert",
				zap.String("certName", certCR.Name),
				zap.String("namespace", certCR.Namespace),
				zap.Error(err))
			return fmt.Errorf("failed to delete dm cluster cert %s: %w", certCR.Name, err)
		}
		tilog.Info(ctx, "dm cluster cert deleted",
			zap.String("certName", certCR.Name),
			zap.String("namespace", certCR.Namespace))

		certSecret := &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      certCR.Name,
				Namespace: namespace,
			},
		}
		if err := k8s.DeleteObject(ctx, remoteCli, certSecret); err != nil {
			tilog.Error(ctx, "failed to delete dm cluster cert secret",
				zap.String("secretName", certSecret.Name),
				zap.String("namespace", certSecret.Namespace),
				zap.Error(err))
			return fmt.Errorf("failed to delete dm cluster cert secret %s: %w", certSecret.Name, err)
		}
		tilog.Info(ctx, "dm cluster cert secret deleted",
			zap.String("secretName", certSecret.Name),
			zap.String("namespace", certSecret.Namespace))
	}
	return nil
}

func applyNodeVersionsToDMCluster(dmCluster *entity.DMCluster, nodeVersionMap map[string]string) error {
	if err := applyNodeVersionToNodeQuota(dmCluster.GetMasterNodeQuota(), nodeVersionMap); err != nil {
		return fmt.Errorf("failed to resolve master node version: %w", err)
	}
	if err := applyNodeVersionToNodeQuota(dmCluster.GetWorkerNodeQuota(), nodeVersionMap); err != nil {
		return fmt.Errorf("failed to resolve worker node version: %w", err)
	}
	return nil
}

func applyNodeVersionToNodeQuota(nodeQuota *entity.NodeQuota, nodeVersionMap map[string]string) error {
	if nodeQuota == nil {
		return nil
	}
	if nodeQuota.GetNodeSpec() == nil {
		return fmt.Errorf("node spec is nil")
	}

	nodeSpec := entity.GetComponentNodeSpec(nodeQuota.GetNodeSpec())
	nodeVersion, ok := nodeVersionMap[nodeSpec]
	if !ok {
		return fmt.Errorf("node version not found for node spec: %s", nodeSpec)
	}
	if nodeVersion == "" {
		return fmt.Errorf("get node version is empty")
	}

	nodeQuota.NodeSpec.NodeVersion = nodeVersion
	return nil
}