package action

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"

	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
	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/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"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	"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"
	appsv1 "k8s.io/api/apps/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// Component label keys read by this action; written by the SetCassandraProxy
// handler before sharedPoolRepo.Save. host-id is a uuid generated at create
// time so it cannot be derived from the component proto. node-spec carries
// the selector derived from NodeQuota.NodeSpec at SetCassandraProxy request
// time, for example cpu=4 and memory_gi=8 becomes "4C8G".
const (
	cassandraProxyHostIdLabel   = "cassandra-proxy/host-id"
	cassandraProxyNodeSpecLabel = "cassandra-proxy/node-spec"
)

// PartitionDefaultSize is frozen at 40 to match prod/02-configmap.yaml.
const cassandraProxyPartitionDefaultSize = "40"

type ApplyCassandraProxyAction struct {
	dataplaneManager   dataplane.Manager
	sharedPoolRepo     repo.SharedPoolRepo
	cassandraStateRepo repo.CassandraProxyStateRepo
	tidbClusterRepo    repo.TiDBClusterRepo
	templateVersionSvc *templateversion.Service
	imageConfig        *config.ImageConfig
	dataplaneClient    dplanemgrv1.DataPlaneServiceClient
}

const ApplyCassandraProxyActionName = "apply-cassandra-proxy"

func NewApplyCassandraProxyAction(
	sharedPoolRepo repo.SharedPoolRepo,
	cassandraStateRepo repo.CassandraProxyStateRepo,
	tidbClusterRepo repo.TiDBClusterRepo,
	dataplaneManager dataplane.Manager,
	dataplaneClient dplanemgrv1.DataPlaneServiceClient,
	templateVersionSvc *templateversion.Service,
	imageConfig *config.ImageConfig,
) actions.Action {
	return &ApplyCassandraProxyAction{
		sharedPoolRepo:     sharedPoolRepo,
		cassandraStateRepo: cassandraStateRepo,
		tidbClusterRepo:    tidbClusterRepo,
		dataplaneManager:   dataplaneManager,
		dataplaneClient:    dataplaneClient,
		templateVersionSvc: templateVersionSvc,
		imageConfig:        imageConfig,
	}
}

func (a *ApplyCassandraProxyAction) Name() string {
	return ApplyCassandraProxyActionName
}

func (a *ApplyCassandraProxyAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "apply cassandra proxy group action")

	if _, err := args.GetString(workflow.ComponentIdsKey); err != nil {
		return fmt.Errorf("target component id is required, but not found")
	}

	sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return fmt.Errorf("failed to get sharedPool id: %w", err)
	}

	// Override component groups from the frozen workflow.SharedPoolKey snapshot
	// captured at DAG submit time. This is the desired-state freeze pattern
	// (see apply_tiproxy_group.go:88-105).
	sharedPoolStr, err := args.GetString(workflow.SharedPoolKey)
	if err != nil {
		return fmt.Errorf("failed to get sharedPool: %w", err)
	}
	var tempSharedPool entity.SharedPool
	if err := protojson.Unmarshal([]byte(sharedPoolStr), &tempSharedPool); err != nil {
		tilog.Error(ctx, "failed to unmarshal sharedPool", zap.Error(err))
		return fmt.Errorf("failed to unmarshal sharedPool: %w", err)
	}
	targetComponents := targetCassandraProxyComponents(ctx, args, tempSharedPool.GetComponentGroups())
	if len(targetComponents) == 0 {
		tilog.Info(ctx, "no target cassandra-proxy component, skip apply")
		return nil
	}
	markTargetsUnavailable := func() {
		a.markTargetCassandraProxyUnavailable(ctx, args, &tempSharedPool)
	}
	hasNonDeletingTarget := hasNonDeletingCassandraProxyTarget(targetComponents)

	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		markTargetsUnavailable()
		return fmt.Errorf("failed to load sharedPool: %w", err)
	}
	sharedPool.ComponentGroups = tempSharedPool.GetComponentGroups()
	if sharedPool.GetVersion() == "" {
		sharedPool.Version = tempSharedPool.GetVersion()
	}

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

	// Pre-loop fetches: nodeVersionMap and dp depend only on dataplaneName, not
	// on individual components. Doing one RPC each before the loop matches the
	// pattern in apply_tiproxy_group.go and keeps RPC count O(1) regardless of
	// how many cassandra-proxy components the SP grows to in the future.
	// These stay nil when every target is deleting; the component loop returns
	// from the delete branch before using create/update-only template inputs.
	var nodeVersionMap map[string]string
	if hasNonDeletingTarget {
		nodeVersionMap, err = getNodeVersions(ctx, a.dataplaneClient, sharedPool.GetDataplaneName())
		if err != nil {
			tilog.Error(ctx, "failed to get node versions", zap.String("dataplaneName", sharedPool.GetDataplaneName()), zap.Error(err))
			markTargetsUnavailable()
			return fmt.Errorf("failed to get node versions: %w", err)
		}
	}

	// dp is needed by SelectCustomerFacingIssuer (PCA vs self-signed) and
	// BuildCassandraProxyCertDNSNames (PrivateLink wildcard via dp.Domain).
	var dp *dplanemgrv1.DataPlane
	if hasNonDeletingTarget {
		dp, err = a.dataplaneClient.GetDataPlane(ctx, &dplanemgrv1.GetDataPlaneRequest{
			DataPlaneId: sharedPool.GetDataplaneName(),
		})
		if err != nil {
			tilog.Error(ctx, "failed to get dataplane", zap.String("dataplaneName", sharedPool.GetDataplaneName()), zap.Error(err))
			markTargetsUnavailable()
			return fmt.Errorf("failed to get dataplane: %w", err)
		}
	}

	for _, comp := range targetComponents {
		if comp.GetPhase() == apiv1.Component_AVAILABLE {
			tilog.Info(ctx, "cassandra-proxy is already available, skip",
				zap.String("componentId", comp.GetId()))
			continue
		}

		hostId := comp.GetLabels()[cassandraProxyHostIdLabel]
		if hostId == "" {
			return status.Errorf(codes.FailedPrecondition,
				"cassandra-proxy component %s missing label %q (handler must set this before DAG submit)",
				comp.GetId(), cassandraProxyHostIdLabel)
		}
		needDelete := comp.GetPhase() == apiv1.Component_DELETING

		if err := func() error {
			if needDelete {
				if err := deleteCassandraProxyResources(ctx, remoteCli, comp.GetId(), entity.GetDataPlaneNamespace(sharedPool.GetId())); err != nil {
					return err
				}
				if err := a.cassandraStateRepo.Delete(ctx, comp.GetId(), hostId); err != nil {
					return fmt.Errorf("failed to delete cassandra-proxy state: %w", err)
				}
				return nil
			}

			nodeSpec := comp.GetLabels()[cassandraProxyNodeSpecLabel]
			if nodeSpec == "" {
				return status.Errorf(codes.FailedPrecondition,
					"cassandra-proxy component %s missing label %q (handler must set this before DAG submit)",
					comp.GetId(), cassandraProxyNodeSpecLabel)
			}
			// Defensive: SetCassandraProxy validates these before saving the
			// component, but a stray persisted record could still arrive here.
			if comp.GetNodeQuota() == nil || comp.GetNodeQuota().GetNodeSpec() == nil {
				return status.Errorf(codes.FailedPrecondition,
					"cassandra-proxy component %s has nil NodeQuota or NodeSpec", comp.GetId())
			}
			state, err := a.cassandraStateRepo.LoadByClusterID(ctx, comp.GetId(), false)
			if err != nil {
				return fmt.Errorf("failed to load cassandra-proxy state: %w", err)
			}
			if state == nil {
				return status.Errorf(codes.FailedPrecondition, "cassandra-proxy state for component %s is nil", comp.GetId())
			}
			if state.HostID != hostId {
				return status.Errorf(codes.FailedPrecondition,
					"cassandra-proxy component %s host-id mismatch: component label has %q, state has %q",
					comp.GetId(), hostId, state.HostID)
			}
			image := resolveCassandraProxyStateImage(state, a.imageConfig)
			if image == "" {
				return status.Errorf(codes.FailedPrecondition,
					"cassandra-proxy image is not configured (state image is empty and imageConfig.cassandraProxyImage is empty)")
			}
			// Load tidbCluster (1:1 with comp.Id). GatewayIdentity lives here, NOT on dp.
			tidbCluster, err := a.tidbClusterRepo.Load(ctx, comp.GetId(), false)
			if err != nil {
				tilog.Error(ctx, "failed to load tidb cluster", zap.String("tidbClusterId", comp.GetId()), zap.Error(err))
				return fmt.Errorf("failed to load tidb cluster: %w", err)
			}

			// nodeVersion lookup uses the shared helper (apply_component_group.go).
			nodeVersion, err := getNodeVersion(nodeVersionMap, comp)
			if err != nil {
				tilog.Error(ctx, "failed to get node version", zap.String("componentId", comp.GetId()), zap.Error(err))
				return fmt.Errorf("failed to get node version: %w", err)
			}

			templArgs, err := buildCassandraProxyTempArgs(image, comp, sharedPool, tidbCluster, dp, hostId, nodeSpec, nodeVersion)
			if err != nil {
				return fmt.Errorf("failed to build cassandra-proxy template args: %w", err)
			}

			// templateExecuteWithVersionResolver -> GenerateUnstructuredFromYaml uses
			// sigs.k8s.io/yaml.YAMLToJSON, which only decodes the first YAML document
			// in a multi-doc file and silently drops the rest. Render each kind from
			// its own single-doc .templ file (matches apply_tici.go convention).
			deploymentTemplateName, err := entity.GetComponentTemplateName(comp)
			if err != nil {
				return fmt.Errorf("failed to get component template name: %w", err)
			}
			for _, tmpl := range []struct {
				name      string
				tmplName  string
				templVers string
			}{
				{"cassandra_proxy_cert", "cassandra_proxy_cert", comp.GetTemplateVersion()},
				{"cassandra_proxy_configmap", "cassandra_proxy_configmap", comp.GetTemplateVersion()},
				{"cassandra_proxy", deploymentTemplateName, comp.GetTemplateVersion()},
			} {
				if err := templateExecuteWithVersionResolver(
					ctx, remoteCli, tmpl.tmplName, tmpl.templVers, templArgs,
					false, a.templateVersionSvc,
				); err != nil {
					tilog.Error(ctx, "failed to execute cassandra-proxy template",
						zap.String("componentId", comp.GetId()),
						zap.String("template", tmpl.name),
						zap.Error(err))
					return fmt.Errorf("failed to execute %s template: %w", tmpl.name, err)
				}
			}

			// Render HPA (separate file, separate single-doc template).
			hpaTemplateName, err := entity.GetComponentHPATemplateName(comp)
			if err != nil {
				return fmt.Errorf("failed to get component hpa template name: %w", err)
			}
			if err := templateExecuteWithVersionResolver(
				ctx, remoteCli, hpaTemplateName, comp.GetTemplateVersion(), templArgs,
				false, a.templateVersionSvc,
			); err != nil {
				tilog.Error(ctx, "failed to execute cassandra_proxy_hpa template", zap.String("componentId", comp.GetId()), zap.Error(err))
				return fmt.Errorf("failed to execute cassandra_proxy_hpa template: %w", err)
			}
			return nil
		}(); err != nil {
			if !needDelete {
				if markErr := a.cassandraStateRepo.UpdatePhase(ctx, comp.GetId(), hostId, apiv1.Component_UNAVAILABLE); markErr != nil {
					tilog.Warn(ctx, "failed to mark cassandra-proxy state unavailable after apply failure",
						zap.String("componentId", comp.GetId()),
						zap.String("hostId", hostId),
						zap.Error(markErr))
				}
			}
			return err
		}
	}

	return nil
}

func hasNonDeletingCassandraProxyTarget(components []*entity.Component) bool {
	for _, comp := range components {
		if comp.GetPhase() != apiv1.Component_DELETING {
			return true
		}
	}
	return false
}

func resolveCassandraProxyStateImage(state *entity.CassandraProxyState, imageConfig *config.ImageConfig) string {
	image := strings.TrimSpace(state.Image)
	if image != "" {
		return image
	}
	if imageConfig == nil {
		return ""
	}
	return strings.TrimSpace(imageConfig.CassandraProxyImage)
}

func deleteCassandraProxyResources(ctx context.Context, remoteCli client.Writer, clusterID, namespace string) error {
	name := entity.GetCassandraProxyResourceName(clusterID)
	resources := []struct {
		kind string
		obj  client.Object
	}{
		{kind: "Certificate", obj: &certmanagerv1.Certificate{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}}},
		{kind: "ConfigMap", obj: &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}}},
		{kind: "Deployment", obj: &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}}},
		{kind: "HorizontalPodAutoscaler", obj: &autoscalingv2.HorizontalPodAutoscaler{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}}},
		{kind: "Secret", obj: &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}}},
	}
	for _, resource := range resources {
		if err := k8s.DeleteObject(ctx, remoteCli, resource.obj); err != nil {
			return fmt.Errorf("failed to delete cassandra-proxy %s %s/%s: %w", resource.kind, namespace, name, err)
		}
	}
	return nil
}

func (a *ApplyCassandraProxyAction) markTargetCassandraProxyUnavailable(ctx context.Context, args actions.ActionArgs, sharedPool *entity.SharedPool) {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() != apiv1.ComponentType_CASSANDRA_PROXY {
			continue
		}
		if !workflow.CheckIsTargetComponent(ctx, comp, args) {
			continue
		}
		switch comp.GetPhase() {
		case apiv1.Component_CREATING, apiv1.Component_MODIFYING:
		default:
			continue
		}
		hostID := comp.GetLabels()[cassandraProxyHostIdLabel]
		if hostID == "" {
			continue
		}
		if err := a.cassandraStateRepo.UpdatePhase(ctx, comp.GetId(), hostID, apiv1.Component_UNAVAILABLE); err != nil {
			tilog.Warn(ctx, "failed to mark cassandra-proxy state unavailable after apply failure",
				zap.String("componentId", comp.GetId()),
				zap.String("hostId", hostID),
				zap.Error(err))
		}
	}
}

func targetCassandraProxyComponents(ctx context.Context, args actions.ActionArgs, components []*entity.Component) []*entity.Component {
	targetComponents := make([]*entity.Component, 0)
	for _, comp := range components {
		if comp.GetComponentType() != apiv1.ComponentType_CASSANDRA_PROXY {
			continue
		}
		if !workflow.CheckIsTargetComponent(ctx, comp, args) {
			tilog.Info(ctx, "skip apply cassandra-proxy component",
				zap.String("componentId", comp.GetId()))
			continue
		}
		targetComponents = append(targetComponents, comp)
	}
	return targetComponents
}

// buildCassandraProxyTempArgs assembles the full template-args map per the
// frozen shared contract (00-overview.md "Template args" table). Every key is
// required; missing values are caller bugs and surface here as errors.
func buildCassandraProxyTempArgs(
	image string,
	comp *entity.Component,
	sharedPool *entity.SharedPool,
	tidbCluster *entity.TiDBCluster,
	dp *dplanemgrv1.DataPlane,
	hostId, nodeSpec, nodeVersion string,
) (map[string]string, error) {
	tidbClusterId := tidbCluster.GetId()
	namespace := entity.GetDataPlaneNamespace(sharedPool.GetId())

	// MaxNodeCount fallback so HPA has a sane upper bound (mirrors
	// buildCompTempArgs default-fill).
	if comp.GetNodeQuota().GetMaxNodeCount() == 0 {
		comp.GetNodeQuota().MaxNodeCount = comp.GetNodeQuota().GetNodeCount()
	}

	resName := entity.GetCassandraProxyResourceName(tidbClusterId)
	templArgs := map[string]string{
		"Namespace":            namespace,
		"ClusterId":            tidbClusterId,
		"DeploymentName":       resName,
		"ConfigMapName":        resName,
		"CertSecretName":       resName,
		"HpaName":              resName,
		"Image":                image,
		"Version":              comp.GetVersion(),
		"MinReplicas":          strconv.FormatUint(uint64(comp.GetNodeQuota().GetMinNodeCount()), 10),
		"MaxReplicas":          strconv.FormatUint(uint64(comp.GetNodeQuota().GetMaxNodeCount()), 10),
		"NodeSpec":             nodeSpec,
		"NodeVersion":          nodeVersion,
		"HostId":               hostId,
		"TiProxyHost":          fmt.Sprintf("db-%s-tiproxy.%s.svc.cluster.local:4000", tidbClusterId, namespace),
		"PartitionDefaultSize": cassandraProxyPartitionDefaultSize,
	}

	if err := setCassandraProxyResourceArgs(comp, templArgs); err != nil {
		return nil, fmt.Errorf("failed to set resource requirements: %w", err)
	}

	// Issuer: dev/staging -> per-namespace self-signed Issuer; prod -> AWS PCA
	// cluster issuer. Mirrors `pkg/cert.TiProxyServerCerts` so cassandra-proxy
	// and tiproxy share the same trust chain on each environment.
	issuerName, issuerKind, issuerGroup := entity.SelectCustomerFacingIssuer(dp)
	templArgs["CertIssuerName"] = issuerName
	templArgs["CertIssuerKind"] = issuerKind
	templArgs["CertIssuerGroup"] = issuerGroup

	// Cert SAN list: 4 in-cluster service-local names plus the
	// `*.{gatewayIdentity}.{domain}` wildcard that covers the customer-facing
	// PrivateLink endpoint (same shape as tiproxy).
	templArgs["CertDNSNamesYAML"] = renderYAMLDashList(entity.BuildCassandraProxyCertDNSNames(tidbCluster, sharedPool, dp))

	// Cost tag map. The render engine prefixes each continuation line with
	// the same leading whitespace it found before {{!Var}}, so we render
	// continuation lines unindented and the engine handles alignment.
	tags := entity.BuildCassandraProxyTags(comp, tidbCluster, sharedPool)
	templArgs["CostTagsYAML"] = renderYAMLLabelMap(tags)

	return templArgs, nil
}

func setCassandraProxyResourceArgs(comp *entity.Component, templArgs map[string]string) error {
	nodeSpec := comp.GetNodeQuota().GetNodeSpec()
	if nodeSpec.GetCpu() < 2 {
		return fmt.Errorf("cpu must be >= 2")
	}
	if nodeSpec.GetMemoryGi() < 4 {
		return fmt.Errorf("memory_gi must be >= 4")
	}

	limitCPU := nodeSpec.GetCpu()
	limitMemoryGi := nodeSpec.GetMemoryGi()
	requestCPU := nodeSpec.GetCpu() / 2
	requestMemoryGi := nodeSpec.GetMemoryGi() / 2

	templArgs["CPU"] = strconv.FormatUint(uint64(limitCPU), 10)
	templArgs["RequestCPU"] = strconv.FormatUint(uint64(requestCPU), 10)
	templArgs["Memory"] = fmt.Sprintf("%dGi", limitMemoryGi)
	templArgs["RequestMemory"] = fmt.Sprintf("%dGi", requestMemoryGi)
	templArgs["SetResource"] = "true"
	return nil
}

// renderYAMLDashList renders ["a","b","c"] as:
//
//   - a
//   - b
//   - c
//
// The render engine prefixes each continuation line with the leading
// whitespace it finds before the substitution placeholder, so this output
// gets indented correctly at the call site.
func renderYAMLDashList(items []string) string {
	if len(items) == 0 {
		return ""
	}
	parts := make([]string, 0, len(items))
	for _, it := range items {
		// Quote items that start with a YAML reserved indicator (e.g.
		// wildcard SANs `*.foo.bar` would otherwise be parsed as an anchor
		// alias).
		if needsYAMLQuoting(it) {
			parts = append(parts, fmt.Sprintf("- %q", it))
		} else {
			parts = append(parts, fmt.Sprintf("- %s", it))
		}
	}
	return strings.Join(parts, "\n")
}

func needsYAMLQuoting(s string) bool {
	if s == "" {
		return false
	}
	switch s[0] {
	case '*', '&', '!', '|', '>', '%', '@', '`', '#', '?', ':':
		return true
	}
	return false
}

// renderYAMLLabelMap renders {"k": "v"} as:
//
//	k: "v"
//	k2: "v2"
//
// Keys are sorted to keep output deterministic. Continuation indentation
// is added by the render engine based on the placeholder's leading
// whitespace at the substitution site.
func renderYAMLLabelMap(m map[string]string) string {
	if len(m) == 0 {
		return ""
	}
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	parts := make([]string, 0, len(keys))
	for _, k := range keys {
		parts = append(parts, fmt.Sprintf("%s: %q", k, m[k]))
	}
	return strings.Join(parts, "\n")
}