package common

import (
	"context"
	"fmt"
	"strings"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/helm"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

const (
	// The configuration for Kong audit log is the same in all dataplanes in this region.
	// So we can use a fixed name for the ConfigMap in seed and dataplane clusters.
	kongAuditConfigMapNameInSeed      = "kong-audit-config"
	kongAuditConfigMapNameInDataplane = "kong-audit-config"
	kongAuditConfigKey                = "config.yaml"
)

type ClusterGateway struct {
	*task.BaseTask
	config *ClusterGatewayConfig
}

type ClusterGatewayConfig struct {
	config.InfraComponentConfig

	lbServiceTags      map[string]string
	appboxMeta         metav1.ObjectMeta
	resourceNamePrefix string
}

func (c *ClusterGatewayConfig) GetName() string {
	return "clustergateway"
}

func (c *ClusterGatewayConfig) ToHelmValues() map[string]interface{} {
	values := c.InfraComponentConfig.ToHelmValues()
	return values
}

func (c *ClusterGatewayConfig) IsAuditKongEnabled() bool {
	values := c.InfraComponentConfig.ToHelmValues()
	if _, ok := values["pluginServerNames"]; !ok {
		return false
	}
	pluginServerNames, ok := values["pluginServerNames"].([]interface{})
	if !ok {
		return false
	}
	for _, name := range pluginServerNames {
		if nameStr, ok := name.(string); ok && nameStr == "audit-kong" {
			return true
		}
	}
	return false
}

func NewClusterGateway(ctx *task.Context) *ClusterGateway {
	return &ClusterGateway{
		BaseTask: &task.BaseTask{Context: ctx},
		config:   &ClusterGatewayConfig{InfraComponentConfig: ctx.DedicatedTmplConfig.ClusterGateway},
	}
}

func (t *ClusterGateway) Name() string {
	return "ClusterGateway"
}

func (t *ClusterGateway) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())

	// If Kong audit log is enabled, we need to create the ConfigMap in dataplane cluster first.
	if t.config.IsAuditKongEnabled() {
		if err := t.createKongAuditConfigMapInDataplane(ctx); err != nil {
			return deploy.Error(err)
		}
	}

	// use the correct tags for loadbalancer
	tags := util.MergeMaps(
		controllers.ConvertToCloudTags(t.APPBox.Labels, cloudv1.ProviderTypeEKS),
		map[string]string{
			consts.TagKeyComponent: "cluster-gateway",
			consts.TagKeyUsedBy:    "dbaas-control-plane-shoot",
		})
	t.config.lbServiceTags = tags
	t.config.appboxMeta = t.Context.APPBox.ObjectMeta
	t.config.resourceNamePrefix = t.Context.Config.ResourceNamePrefix
	return deploy.ErrorOrDone(helm.Apply(ctx, t.Remote, t.config))
}

func (t *ClusterGateway) PostCreate(ctx context.Context) (res deploy.Result) {
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		return deploy.Panic(err)
	}
	var service corev1.Service
	if err := cli.Get(ctx, types.NamespacedName{Namespace: t.config.GetNamespace(), Name: fmt.Sprintf("%s-kong-proxy", t.config.GetName())}, &service); err != nil {
		return deploy.Error(err)
	}

	if t.config.IsAuditKongEnabled() {
		key := types.NamespacedName{
			Namespace: t.config.GetNamespace(),
			Name:      fmt.Sprintf("%s-kong", t.config.GetName()),
		}
		if err := controllers.EnsureChartStatefulSetReady(ctx, t.Remote, key); err != nil {
			return deploy.Error(err)
		}

		// Service proxy uses an ephemeral volumeClaimTemplate, so new PVCs can safely inherit
		// labels from the template. Kong runs as a StatefulSet, and its volumeClaimTemplates
		// are immutable after creation. We patch the PVC objects directly here to keep Helm
		// upgrades working for existing dataplanes while still allowing resourcetagger to
		// discover the desired tags from PVC metadata.
		if err := t.patchAuditPVCLabels(ctx); err != nil {
			return deploy.Error(err)
		}
		return deploy.Done()
	}

	return deploy.ErrorOrDone(controllers.EnsureChartDeploymentReady(ctx, t.Remote, types.NamespacedName{
		Namespace: t.config.GetNamespace(),
		Name:      fmt.Sprintf("%s-kong", t.config.GetName()),
	}))
}

func (t *ClusterGateway) patchAuditPVCLabels(ctx context.Context) error {
	labels := persistentVolumeClaimLabelsForAppBox(t.APPBox, "cluster-gateway")
	if len(labels) == 0 {
		return nil
	}

	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		return err
	}

	pvcs := &corev1.PersistentVolumeClaimList{}
	if err := cli.List(
		ctx,
		pvcs,
		client.InNamespace(t.config.GetNamespace()),
		client.MatchingLabels(map[string]string{
			consts.ComponentLabelKey: "app",
			consts.InstanceLabelKey:  t.config.GetName(),
		}),
	); err != nil {
		return fmt.Errorf("list kong audit pvcs: %w", err)
	}

	for i := range pvcs.Items {
		pvc := &pvcs.Items[i]
		if !isKongAuditPVC(pvc, t.config.GetName()) {
			continue
		}
		origin := pvc.DeepCopy()
		if pvc.Labels == nil {
			pvc.Labels = map[string]string{}
		}
		changed := false
		for key, value := range labels {
			if pvc.Labels[key] == value {
				continue
			}
			pvc.Labels[key] = value
			changed = true
		}
		if !changed {
			continue
		}
		if err := cli.Patch(ctx, pvc, client.MergeFrom(origin)); err != nil {
			return fmt.Errorf("patch kong audit pvc %s: %w", pvc.Name, err)
		}
	}

	return nil
}

func isKongAuditPVC(pvc *corev1.PersistentVolumeClaim, releaseName string) bool {
	if pvc == nil {
		return false
	}

	expectedPrefix := fmt.Sprintf("audit-%s-kong-", releaseName)
	if !strings.HasPrefix(pvc.Name, expectedPrefix) {
		return false
	}

	expectedOwner := fmt.Sprintf("%s-kong", releaseName)
	for _, ownerRef := range pvc.OwnerReferences {
		if ownerRef.Controller != nil && *ownerRef.Controller &&
			ownerRef.Kind == "StatefulSet" && ownerRef.Name == expectedOwner {
			return true
		}
	}

	return false
}

// Delete removes the proxy service, must use deploy.Panic() if error is not null.
// Because we want to remove the service before other tasks(eks,nodegroup,vpc).
func (t *ClusterGateway) Delete(ctx context.Context) deploy.Result {
	if deleted, ok := t.APPBox.Annotations[task.AnnotationKeyGatewayVolumeDeleted]; ok && deleted == "true" {
		return deploy.Done()
	}
	if _, ok := t.Remote.(*k8s.EmptyGetter); ok {
		t.Logger.Info("remote client is empty getter, skip deleting")
		return deploy.Done()
	}
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		return deploy.Panic(err)
	}

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.config.GetNamespace(),
			Name:      "clustergateway-kong-proxy",
		},
	}

	if err := cli.Get(ctx, client.ObjectKeyFromObject(svc), svc); err != nil {
		if client.IgnoreNotFound(err) != nil {
			return deploy.Panic(fmt.Errorf("get service %s: %w", svc.Name, err))
		}
		// continue to delete the sts if needed
	} else {
		if _, ok := svc.Annotations["service.beta.kubernetes.io/aws-load-balancer-attributes"]; ok {
			origin := svc.DeepCopy()
			svc.Annotations["service.beta.kubernetes.io/aws-load-balancer-attributes"] = "load_balancing.cross_zone.enabled=true,deletion_protection.enabled=false"
			if err := cli.Patch(ctx, svc, client.MergeFrom(origin)); err != nil {
				return deploy.Panic(fmt.Errorf("patch service %s: %w", svc.Name, err))
			}
		}
		if err := k8s.DeleteObjectCompleted(ctx, cli, svc); err != nil {
			return deploy.Panic(fmt.Errorf("delete service %s: %w", svc.Name, err))
		}
	}

	// check if we are using `StatefulSet` instead of `Deployment` for Kong
	// if we're using StatefulSet, we need to delete it to release PVCs with `persistentVolumeClaimRetentionPolicy`
	if t.config.IsAuditKongEnabled() {
		sts := &appsv1.StatefulSet{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: t.config.GetNamespace(),
				Name:      fmt.Sprintf("%s-kong", t.config.GetName()),
			},
		}
		if err := cli.Get(ctx, client.ObjectKeyFromObject(sts), sts); client.IgnoreNotFound(err) != nil {
			return deploy.Panic(fmt.Errorf("get statefulset %s: %w", sts.Name, err))
		}
		t.Logger.Info("deleting statefulset", "name", sts.Name)
		if err := k8s.DeleteObjectCompleted(ctx, cli, sts); err != nil {
			return deploy.Panic(fmt.Errorf("delete statefulset %s: %w", sts.Name, err))
		}

		// check if PVCs are released with the following labels
		// when reaching here, the sts should already be deleted,
		// so we can't get the labels from the sts
		labels := map[string]string{
			consts.ComponentLabelKey: "app", // always "app" for kong
			consts.InstanceLabelKey:  t.config.GetName(),
		}
		pvcs := &corev1.PersistentVolumeClaimList{}
		if err := cli.List(ctx, pvcs, client.InNamespace(t.config.GetNamespace()), client.MatchingLabels(labels)); err != nil {
			return deploy.Panic(fmt.Errorf("list pvcs: %w", err))
		}
		t.Logger.Info("listing pvcs", "labels", labels, "count", len(pvcs.Items))
		if len(pvcs.Items) > 0 {
			return deploy.Panic(fmt.Errorf("waiting for %d pvcs to be released", len(pvcs.Items)))
		}

		// we also need to ensure the PV is deleted,
		// otherwise, if the CSI driver (or EKS nodes running the CSI driver) is deleted first,
		// then the EBS will be leaked
		pvs := &corev1.PersistentVolumeList{}
		if err := cli.List(ctx, pvs); err != nil {
			return deploy.Panic(fmt.Errorf("list pvs: %w", err))
		}
		kongPVCount := 0
		for _, pv := range pvs.Items {
			if pv.Spec.ClaimRef != nil &&
				pv.Spec.ClaimRef.Namespace == t.config.GetNamespace() &&
				strings.Contains(pv.Spec.ClaimRef.Name, sts.Name) {
				kongPVCount++
				t.Logger.Info("found kong pv not released", "pv", pv.Name, "status", pv.Status.Phase)
			}
		}
		if kongPVCount > 0 {
			return deploy.Panic(fmt.Errorf("waiting for %d pvs to be released", kongPVCount))
		}
	}
	t.APPBox.Annotations[task.AnnotationKeyGatewayVolumeDeleted] = "true"

	return deploy.Done()
}

// We only *create* the Kong audit log ConfigMap in dataplane cluster here.
// For *update*, it should be handled by the dataplane-manager *server*.
// As the client needs to know the result directly instead of waiting for the next reconciliation loop.
// For *delete*, it will be deleted automatically when the dataplane cluster is deleted.
func (t *ClusterGateway) createKongAuditConfigMapInDataplane(ctx context.Context) error {
	cli, err := t.Remote.GetClient(ctx)
	if err != nil {
		return err
	}
	cmInDP := &corev1.ConfigMap{}
	err = cli.Get(ctx, types.NamespacedName{
		Namespace: t.config.GetNamespace(),
		Name:      kongAuditConfigMapNameInDataplane},
		cmInDP)
	if err == nil {
		// already exists
		return nil
	}
	if client.IgnoreNotFound(err) != nil {
		return fmt.Errorf("get kong audit configmap in dataplane: %w", err)
	}

	// check if we have the ConfigMap in seed cluster
	cmInSeed := &corev1.ConfigMap{}
	err = t.Local.Get(ctx, types.NamespacedName{
		Namespace: t.Config.Namespace, // `Config` not `config`! "dataplane-metadata"
		Name:      kongAuditConfigMapNameInSeed},
		cmInSeed)
	if err == nil {
		cmInDP = &corev1.ConfigMap{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: t.config.GetNamespace(),
				Name:      kongAuditConfigMapNameInDataplane,
			},
			Data: cmInSeed.Data, // copy the data from seed
		}
		// we need to append the suffix to the bucketURL in the config
		if originalConfig, ok := cmInDP.Data[kongAuditConfigKey]; ok {
			cmInDP.Data[kongAuditConfigKey] = t.appendBucketURLSuffixIfNeeded(originalConfig)
		}
	} else if client.IgnoreNotFound(err) == nil {
		cmInDP = t.defaultKongAuditConfigMap()
	} else {
		return fmt.Errorf("get kong audit configmap in seed: %w", err)
	}
	if err := cli.Create(ctx, cmInDP); err != nil {
		return fmt.Errorf("create kong audit configmap in dataplane: %w", err)
	}
	return nil
}

// If no configuration items are provided, we still create an empty ConfigMap for Kong.
func (t *ClusterGateway) defaultKongAuditConfigMap() *corev1.ConfigMap {
	return &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      kongAuditConfigMapNameInDataplane,
			Namespace: t.config.GetNamespace(),
		},
		Data: map[string]string{
			kongAuditConfigKey: `
enableLog: false
logDir: /var/log/auditlog
auditLogFileName: ops-audit.log
maxSize: 10
maxLifetime: 86400
bucketURL: ""
region: ""
aws: {}
gracefulStopWaitSeconds: 20
producerPort: 9080
`,
		},
	}
}

// As all dataplanes in this region share the same S3 bucket for Kong audit log,
// we use the dataplane ID or APPBox name as the suffix to avoid conflicts.
func (t *ClusterGateway) appendBucketURLSuffixIfNeeded(originConfig string) string {
	suffix := t.APPBox.Labels[v1alpha1.DataPlaneIDKey]
	if suffix == "" {
		suffix = t.APPBox.Name
	}

	lines := strings.Split(originConfig, "\n")
	for i, line := range lines {
		if strings.HasPrefix(line, "bucketURL:") {
			parts := strings.SplitN(line, ":", 2)
			if len(parts) == 2 {
				bucketURL := strings.TrimSpace(parts[1])
				if bucketURL != "" {
					if !strings.HasSuffix(bucketURL, suffix) {
						if strings.HasSuffix(bucketURL, "/") {
							bucketURL = fmt.Sprintf("%s%s", bucketURL, suffix)
						} else {
							bucketURL = fmt.Sprintf("%s/%s", bucketURL, suffix)
						}
					}
					lines[i] = fmt.Sprintf("bucketURL: %s", bucketURL)
				}
			}
			break
		}
	}
	return strings.Join(lines, "\n")
}