package ack

import (
	"context"
	"fmt"
	"strings"

	ecsv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/ecs/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"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/loadbalancer/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

const (
	// sgRefLabelKey is a label used for grouping SecurityGroupRule belong to a specific SecurityGroup
	sgRefLabelKey = "sg-reference"
)

type securityGroup struct {
	*task.BaseTask
}

func NewSecurityGroup(ctx *task.Context) deploy.Task {
	return &securityGroup{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *securityGroup) Name() string {
	return "SecurityGroup"
}

func (t *securityGroup) Create(ctx context.Context) deploy.Result {
	sgName := t.getSecurityGroupName()
	tags := controllers.ConvertToCloudTags(t.LoadBalancer.Labels, consts.ProviderTypeACK)
	labels := t.LoadBalancer.Labels
	sg := &ecsv1alpha1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:   sgName,
			Labels: labels,
		},
		Spec: ecsv1alpha1.SecurityGroupSpec{
			ForProvider: ecsv1alpha1.SecurityGroupParameters{
				SecurityGroupName: ptr.To(sgName),
				Description:       ptr.To("Security group for tidb"),
				Region:            ptr.To(t.AppBox.Spec.Region),
				VPCID:             ptr.To(t.VPC.Status.VPC.VPCID),
				Tags:              util.ToMapPointer(tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: t.ProviderConfigName()},
				ManagementPolicies:      controllers.GetResourceManagementPolicies(),
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, sg); err != nil {
		return deploy.Panic(err)
	}

	// Reconcile security group rules using the "apply first, cleanup later" pattern
	// to prevent service interruption during configuration updates:
	//   - Desired rules are applied first to ensure new configurations are active
	//   - Stale rules are cleaned up after to remove obsolete configurations
	// This ensures continuous service availability when IpWhiteList configuration
	// is updated (e.g., IPs are added or removed).
	desiredRules := t.buildSecurityGroupRules()
	desiredNames := make(map[string]bool, len(desiredRules))
	for _, rule := range desiredRules {
		desiredNames[rule.Name] = true
		if err := k8s.Apply(ctx, t.Local, &rule); err != nil {
			return deploy.Error(err)
		}
	}

	// Cleanup stale rules after desired rules are applied
	if err := t.reconcileSecurityGroupRules(ctx, desiredNames); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *securityGroup) PostCreate(ctx context.Context) deploy.Result {
	sgName := t.getSecurityGroupName()
	sg := &ecsv1alpha1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{Name: sgName},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(sg), sg); err != nil {
		return deploy.Error(err)
	}
	if !controllers.IsResourceReady(sg.Status.ResourceStatus) {
		return deploy.Error(fmt.Errorf("tidb security group not ready: %s", sgName))
	}
	ruleList := &ecsv1alpha1.SecurityGroupRuleList{}
	matchLabels := client.MatchingLabels{sgRefLabelKey: sgName}
	if err := t.Local.List(ctx, ruleList, matchLabels); err != nil {
		return deploy.Error(err)
	}
	for _, rule := range ruleList.Items {
		if !controllers.IsResourceReady(rule.Status.ResourceStatus) {
			return deploy.Error(fmt.Errorf("security group rule not ready: %s", rule.Name))
		}
	}
	return deploy.Done()
}

func (t *securityGroup) Delete(ctx context.Context) deploy.Result {
	sgName := t.getSecurityGroupName()
	ruleList := &ecsv1alpha1.SecurityGroupRuleList{}
	matchLabels := client.MatchingLabels{sgRefLabelKey: sgName}
	if err := t.Local.List(ctx, ruleList, matchLabels); err != nil {
		return deploy.Error(err)
	}
	for _, rule := range ruleList.Items {
		if err := controllers.DeleteUpboundResource(ctx, t.Local, &rule); err != nil {
			return deploy.Error(err)
		}
	}
	sg := &ecsv1alpha1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{Name: sgName},
	}
	return deploy.ErrorOrDone(controllers.DeleteUpboundResource(ctx, t.Local, sg))
}

func (t *securityGroup) getSecurityGroupName() string {
	return controllers.UniqueIdentifier(t.LoadBalancer.ObjectMeta, namer.WithPrefix(t.Config.ResourceNamePrefix+"-tidb"))
}

func (t *securityGroup) getSecurityGroupRuleName(cidr string, port int32) string {
	ip := strings.Split(cidr, "/")[0] // remove the /xx suffix if exists, it'sip always /32 or /0
	sip := strings.ReplaceAll(ip, ".", "-")
	prefix := fmt.Sprintf("%s-tidb-%s-%d", t.Config.ResourceNamePrefix, sip, port)
	return controllers.UniqueIdentifier(t.LoadBalancer.ObjectMeta, namer.WithPrefix(prefix))
}

func (t *securityGroup) buildSecurityGroupRules() []ecsv1alpha1.SecurityGroupRule {
	sgName := t.getSecurityGroupName()
	labels := util.MergeMaps(t.LoadBalancer.Labels, map[string]string{sgRefLabelKey: sgName})
	rules := []ecsv1alpha1.SecurityGroupRule{
		{ // default deny all with lower priority
			ObjectMeta: metav1.ObjectMeta{
				Name:   fmt.Sprintf("%s-deny", t.getSecurityGroupName()),
				Labels: labels,
			},
			Spec: ecsv1alpha1.SecurityGroupRuleSpec{
				ForProvider: ecsv1alpha1.SecurityGroupRuleParameters{
					SecurityGroupIDRef: &xpv1.Reference{Name: t.getSecurityGroupName()},
					CidrIP:             ptr.To("0.0.0.0/0"),
					Priority:           ptr.To(float64(100)),
					IPProtocol:         ptr.To("all"),
					PortRange:          ptr.To("-1/-1"),
					Policy:             ptr.To("drop"),
					Region:             ptr.To(t.AppBox.Spec.Region),
					Type:               ptr.To("ingress"),
				},
				ResourceSpec: xpv1.ResourceSpec{
					ProviderConfigReference: &xpv1.Reference{Name: t.ProviderConfigName()},
					ManagementPolicies:      controllers.GetResourceManagementPolicies(),
				},
			},
		},
	}
	// One SG rule per (cidr, port) combination. Multi-port LBs expand to
	// len(IpWhiteList) * len(ports); legacy single-port LBs still get the
	// historical one-rule-per-cidr layout.
	for _, ap := range t.allowedPorts() {
		for _, cidr := range t.LoadBalancer.Spec.IpWhiteList {
			rule := ecsv1alpha1.SecurityGroupRule{
				ObjectMeta: metav1.ObjectMeta{
					Name:   t.getSecurityGroupRuleName(cidr, ap.Port),
					Labels: labels,
				},
				Spec: ecsv1alpha1.SecurityGroupRuleSpec{
					ForProvider: ecsv1alpha1.SecurityGroupRuleParameters{
						SecurityGroupIDRef: &xpv1.Reference{Name: t.getSecurityGroupName()},
						CidrIP:             ptr.To(cidr),
						IPProtocol:         ptr.To(ap.Protocol),
						PortRange:          ptr.To(fmt.Sprintf("%d/%d", ap.Port, ap.Port)),
						Region:             ptr.To(t.AppBox.Spec.Region),
						Type:               ptr.To("ingress"),
					},
					ResourceSpec: xpv1.ResourceSpec{
						ProviderConfigReference: &xpv1.Reference{Name: t.ProviderConfigName()},
						ManagementPolicies:      controllers.GetResourceManagementPolicies(),
					},
				},
			}
			rules = append(rules, rule)
		}
	}
	return rules
}

// allowedPort is a (port, IPProtocol) pair used to materialize ACK SG rules.
type allowedPort struct {
	Port     int32
	Protocol string // ACK SecurityGroupRule.IPProtocol literal: "tcp" or "udp".
}

// allowedPorts returns the listener (port, protocol) pairs that should be
// opened in the SG. Mirrors the multi-port contract of BuildBasicService(),
// dedupes on Port (BuildServicePorts already rejects duplicate ports across
// protocols), and maps the spec protocol to ACK's IPProtocol literal so a
// UDP listener does not silently open TCP.
func (t *securityGroup) allowedPorts() []allowedPort {
	spec := &t.LoadBalancer.Spec
	if len(spec.Ports) > 0 {
		seen := make(map[int32]struct{}, len(spec.Ports))
		ports := make([]allowedPort, 0, len(spec.Ports))
		for _, p := range spec.Ports {
			if p.Port == 0 {
				continue
			}
			if _, ok := seen[p.Port]; ok {
				continue
			}
			seen[p.Port] = struct{}{}
			ports = append(ports, allowedPort{Port: p.Port, Protocol: aliCloudIPProtocol(p.Protocol)})
		}
		return ports
	}
	if spec.Port != 0 {
		return []allowedPort{{Port: spec.Port, Protocol: "tcp"}}
	}
	return nil
}

// aliCloudIPProtocol maps a LoadBalancerSpec.Ports[].Protocol string to the
// IPProtocol literal accepted by ACK SecurityGroupRule. Unknown values fall
// back to "tcp" to match buildServicePorts' default normalization (TLS and
// the empty string both collapse to TCP at the K8s ServicePort layer).
func aliCloudIPProtocol(proto string) string {
	switch strings.ToUpper(proto) {
	case "UDP":
		return "udp"
	default:
		return "tcp"
	}
}

// reconcileSecurityGroupRules removes security group rules not in the desired set.
func (t *securityGroup) reconcileSecurityGroupRules(ctx context.Context, desiredNames map[string]bool) error {
	sgName := t.getSecurityGroupName()
	ruleList := &ecsv1alpha1.SecurityGroupRuleList{}
	opts := k8s.ListOpts[*ecsv1alpha1.SecurityGroupRule]{
		ObjectList: ruleList,
		Unpack: func() []*ecsv1alpha1.SecurityGroupRule {
			return k8s.MapPtr(ruleList.Items)
		},
	}

	return opts.ReconcileByDesiredNamesWithDelete(
		ctx,
		t.Local,
		desiredNames,
		controllers.DeleteUpboundResource,
		client.MatchingLabels{sgRefLabelKey: sgName},
	)
}