package loadbalancer

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	regionalapiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"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/ownership"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
)

type LoadBalancerService struct {
	regionalapiv1.UnimplementedLoadBalancerServiceServer

	loadbalancerrw repository.LoadBalancerReaderWriter
	dataPlaneR     repository.DataPlaneReader
	rootCfg        config.RootConfig
}

func NewLoadBalancerService(
	loadbalancerrw repository.LoadBalancerReaderWriter,
	dataPlaneR repository.DataPlaneReader,
	rootCfg config.RootConfig,
) *LoadBalancerService {
	return &LoadBalancerService{
		loadbalancerrw: loadbalancerrw,
		dataPlaneR:     dataPlaneR,
		rootCfg:        rootCfg,
	}
}

func (s *LoadBalancerService) CreateLoadBalancer(ctx context.Context,
	req *regionalapiv1.CreateLoadBalancerRequest) (*regionalapiv1.LoadBalancer, error) {
	if err := validateLoadBalancerPorts(req.Loadbalancer); err != nil {
		return nil, err
	}
	tenantID := req.Loadbalancer.Tags[consts.TenantLabelKey]
	if tenantID == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "missing required tag: tags.tidbcloud.com/tenant")
	}
	dataPlaneObj, err := s.getDataPlane(ctx, req.DataPlaneId)
	if err != nil {
		return nil, errors.Trace(err)
	}
	lb := &v1alpha1.LoadBalancer{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: dataPlaneObj.Namespace,
			Labels:    loadBalancerLabels(req.DataPlaneId, req.Loadbalancer.Tags),
		},
		Spec: v1alpha1.LoadBalancerSpec{
			DataPlane: v1alpha1.Reference{
				Name:      req.DataPlaneId,
				Namespace: s.rootCfg.Namespace,
			},
			Identity:                 req.Loadbalancer.Identity,
			SubIdentity:              req.Loadbalancer.SubIdentity,
			Namespace:                req.Loadbalancer.Namespace,
			Tags:                     req.Loadbalancer.Tags,
			Port:                     req.Loadbalancer.Port,
			Ports:                    pbPorts2CR(req.Loadbalancer.Ports),
			Annotations:              req.Loadbalancer.Annotations,
			Selector:                 req.Loadbalancer.Selectors,
			IpWhiteList:              req.Loadbalancer.IpWhiteList,
			GracefulUnhealthyEnabled: req.Loadbalancer.GracefulUnhealthyEnabled,
			Type:                     pbType2CRType(req.Loadbalancer.Type),
			PreserveClientIP:         req.Loadbalancer.PreserveClientIp,
			Ownership:                ownership.FromRequest(req.OwnershipContext),
		},
	}
	if err := s.loadbalancerrw.Create(ctx, lb); err != nil {
		if apierrors.IsAlreadyExists(err) {
			log.Info("loadbalancer already exists", zap.String("id", req.Id))
			existing := &v1alpha1.LoadBalancer{}
			if getErr := s.loadbalancerrw.Get(ctx, client.ObjectKeyFromObject(lb), existing); getErr != nil {
				return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get existing loadbalancer", getErr)
			}
			if matchErr := ownership.Match(req.OwnershipContext, existing.Spec.Ownership); matchErr != nil {
				return nil, matchErr
			}
			return req.Loadbalancer, bizerr.New(bizerr.ResourceConflict, "loadbalancer already exists")
		}
		log.Error("failed to create loadbalancer", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to create loadbalancer")
	}

	req.Loadbalancer.Id = req.Id
	req.Loadbalancer.Phase = regionalapiv1.LoadBalancer_CREATING
	return req.Loadbalancer, nil
}

func (s *LoadBalancerService) UpdateLoadBalancer(ctx context.Context,
	req *regionalapiv1.UpdateLoadBalancerRequest) (*regionalapiv1.LoadBalancer, error) {
	if err := validateLoadBalancerPorts(req.Loadbalancer); err != nil {
		return nil, err
	}
	lb := &v1alpha1.LoadBalancer{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: s.rootCfg.Namespace,
		},
	}
	if err := s.loadbalancerrw.Get(ctx, client.ObjectKeyFromObject(lb), lb); err != nil {
		log.Error("failed to get loadbalancer", zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get loadbalancer", err)
	}
	if err := ownership.Match(req.OwnershipContext, lb.Spec.Ownership); err != nil {
		return nil, err
	}
	lb.Spec.Identity = req.Loadbalancer.Identity
	lb.Spec.SubIdentity = req.Loadbalancer.SubIdentity
	lb.Spec.IpWhiteList = req.Loadbalancer.IpWhiteList
	lb.Spec.Namespace = req.Loadbalancer.Namespace
	lb.Spec.Tags = req.Loadbalancer.Tags
	lb.Spec.Port = req.Loadbalancer.Port
	// Update must persist the multi-port and annotations fields so clients
	// can drift-fix them after the CR is created; otherwise successive
	// Update calls silently discard the new fields.
	lb.Spec.Ports = pbPorts2CR(req.Loadbalancer.Ports)
	lb.Spec.Annotations = req.Loadbalancer.Annotations
	lb.Spec.Selector = req.Loadbalancer.Selectors
	lb.Spec.GracefulUnhealthyEnabled = req.Loadbalancer.GracefulUnhealthyEnabled
	lb.Spec.Type = pbType2CRType(req.Loadbalancer.Type)
	lb.Spec.PreserveClientIP = req.Loadbalancer.PreserveClientIp
	lb.Labels = loadBalancerLabels(lb.Spec.DataPlane.Name, req.Loadbalancer.Tags)
	if err := s.loadbalancerrw.Update(ctx, lb); err != nil {
		log.Error("failed to update loadbalancer", zap.String("id", req.Id), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to update loadbalancer", err)
	}

	req.Loadbalancer.Id = req.Id
	return req.Loadbalancer, nil
}

func loadBalancerLabels(dataPlaneID string, tags map[string]string) map[string]string {
	labels := make(map[string]string, len(tags)+1)
	for k, v := range tags {
		labels[k] = v
	}
	labels[v1alpha1.DataPlaneIDKey] = dataPlaneID
	return labels
}

func (s *LoadBalancerService) GetLoadBalancer(ctx context.Context, req *regionalapiv1.GetLoadBalancerRequest) (*regionalapiv1.LoadBalancer, error) {
	lb := &v1alpha1.LoadBalancer{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: s.rootCfg.Namespace,
		},
	}
	if err := s.loadbalancerrw.Get(ctx, client.ObjectKeyFromObject(lb), lb); err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error("failed to get loadbalancer", zap.Error(err))
			return nil, bizerr.New(bizerr.K8SClientError, "failed to get loadbalancer")
		}
		log.Error("loadbalancer not found", zap.String("id", req.Id))
		return nil, bizerr.New(bizerr.ResourceNotFound, "loadbalancer not found")
	}
	if err := ownership.Match(req.OwnershipContext, lb.Spec.Ownership); err != nil {
		return nil, err
	}
	return cr2PB(lb), nil
}

func (s *LoadBalancerService) DeleteLoadBalancer(ctx context.Context, req *regionalapiv1.DeleteLoadBalancerRequest) (*emptypb.Empty, error) {
	lb := &v1alpha1.LoadBalancer{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.Id,
			Namespace: s.rootCfg.Namespace,
		},
	}
	if err := s.loadbalancerrw.Get(ctx, client.ObjectKeyFromObject(lb), lb); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("loadbalancer not found", zap.String("id", req.Id))
			return new(emptypb.Empty), nil
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get loadbalancer", err)
	}
	if err := ownership.Match(req.OwnershipContext, lb.Spec.Ownership); err != nil {
		return nil, err
	}
	if err := s.loadbalancerrw.Delete(ctx, lb); err != nil {
		if apierrors.IsNotFound(err) {
			log.Info("loadbalancer not found", zap.String("id", req.Id))
			return new(emptypb.Empty), nil
		}
		log.Error("failed to delete loadbalancer", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, "failed to delete loadbalancer")
	}
	return new(emptypb.Empty), nil
}

func pbType2CRType(balancerType regionalapiv1.LoadBalancer_Type) v1alpha1.LoadBalancerType {
	switch balancerType {
	case regionalapiv1.LoadBalancer_PUBLIC:
		return v1alpha1.PublicLoadBalancerType
	case regionalapiv1.LoadBalancer_PRIVATE:
		return v1alpha1.PrivateLoadBalancerType
	default:
		return v1alpha1.PublicLoadBalancerType
	}
}

func crType2PBType(balancerType v1alpha1.LoadBalancerType) regionalapiv1.LoadBalancer_Type {
	switch balancerType {
	case v1alpha1.PublicLoadBalancerType:
		return regionalapiv1.LoadBalancer_PUBLIC
	case v1alpha1.PrivateLoadBalancerType:
		return regionalapiv1.LoadBalancer_PRIVATE
	default:
		return regionalapiv1.LoadBalancer_PRIVATE
	}
}

func crPhase2PBPhase(balancerPhase v1alpha1.LoadBalancerPhase) regionalapiv1.LoadBalancer_Phase {
	switch balancerPhase {
	case v1alpha1.LoadBalancerPhaseActive:
		return regionalapiv1.LoadBalancer_ACTIVE
	case v1alpha1.LoadBalancerPhaseCreating:
		return regionalapiv1.LoadBalancer_CREATING
	default:
		return regionalapiv1.LoadBalancer_CREATING
	}
}

func cr2PB(lb *v1alpha1.LoadBalancer) *regionalapiv1.LoadBalancer {
	lbPB := &regionalapiv1.LoadBalancer{
		Id:                       lb.Name,
		Identity:                 lb.Spec.Identity,
		SubIdentity:              lb.Spec.SubIdentity,
		Namespace:                lb.Spec.Namespace,
		Port:                     lb.Spec.Port,
		Ports:                    crPorts2PB(lb.Spec.Ports),
		Annotations:              lb.Spec.Annotations,
		Selectors:                lb.Spec.Selector,
		Tags:                     lb.Spec.Tags,
		IpWhiteList:              lb.Spec.IpWhiteList,
		GracefulUnhealthyEnabled: lb.Spec.GracefulUnhealthyEnabled,
		Type:                     crType2PBType(lb.Spec.Type),
		PreserveClientIp:         lb.Spec.PreserveClientIP,
		Dns:                      lb.Status.DNS,
		Phase:                    crPhase2PBPhase(lb.Status.Phase),
		PrivateIps:               lb.Status.PrivateIPs,
	}
	return lbPB
}

// validateLoadBalancerPorts rejects requests where neither the legacy Port
// nor the new Ports field carries a valid listener port. Proto-level required
// was dropped from Port to allow multi-port clients, so this check moves to
// the handler.
func validateLoadBalancerPorts(lb *regionalapiv1.LoadBalancer) error {
	if lb == nil {
		return bizerr.New(bizerr.ParameterInvalid, "loadbalancer is required")
	}
	seenPorts := make(map[int32]struct{}, len(lb.Ports))
	seenNames := make(map[string]struct{}, len(lb.Ports))
	multiPort := len(lb.Ports) > 1
	for i, p := range lb.Ports {
		if p == nil {
			return bizerr.New(bizerr.ParameterInvalid,
				fmt.Sprintf("loadbalancer.ports[%d] is nil", i))
		}
		if p.Port <= 0 {
			return bizerr.New(bizerr.ParameterInvalid,
				fmt.Sprintf("loadbalancer.ports[%d].port must be > 0", i))
		}
		if _, dup := seenPorts[p.Port]; dup {
			return bizerr.New(bizerr.ParameterInvalid,
				fmt.Sprintf("loadbalancer.ports contains duplicate port %d", p.Port))
		}
		seenPorts[p.Port] = struct{}{}
		// Kubernetes requires every ServicePort in a multi-port Service to
		// carry a non-empty unique name. Single-port Services accept an
		// empty name.
		if multiPort {
			if p.Name == "" {
				return bizerr.New(bizerr.ParameterInvalid,
					fmt.Sprintf("loadbalancer.ports[%d].name is required for multi-port LoadBalancers", i))
			}
			if _, dup := seenNames[p.Name]; dup {
				return bizerr.New(bizerr.ParameterInvalid,
					fmt.Sprintf("loadbalancer.ports contains duplicate name %q", p.Name))
			}
			seenNames[p.Name] = struct{}{}
		}
	}
	if lb.Port == 0 && len(lb.Ports) == 0 {
		return bizerr.New(bizerr.ParameterInvalid, "loadbalancer requires port or ports")
	}
	return nil
}

func pbPorts2CR(in []*regionalapiv1.LoadBalancer_Port) []v1alpha1.LoadBalancerPort {
	if len(in) == 0 {
		return nil
	}
	out := make([]v1alpha1.LoadBalancerPort, 0, len(in))
	for _, p := range in {
		if p == nil {
			continue
		}
		out = append(out, v1alpha1.LoadBalancerPort{
			Name:       p.Name,
			Port:       p.Port,
			TargetPort: p.TargetPort,
			Protocol:   p.Protocol,
		})
	}
	return out
}

func crPorts2PB(in []v1alpha1.LoadBalancerPort) []*regionalapiv1.LoadBalancer_Port {
	if len(in) == 0 {
		return nil
	}
	out := make([]*regionalapiv1.LoadBalancer_Port, 0, len(in))
	for _, p := range in {
		out = append(out, &regionalapiv1.LoadBalancer_Port{
			Name:       p.Name,
			Port:       p.Port,
			TargetPort: p.TargetPort,
			Protocol:   p.Protocol,
		})
	}
	return out
}

func (s *LoadBalancerService) getDataPlane(ctx context.Context, dataplaneId string) (*v1alpha1.DataPlane, error) {
	dataPlaneObj := &v1alpha1.DataPlane{}
	if err := s.dataPlaneR.Get(ctx,
		client.ObjectKey{Name: dataplaneId,
			Namespace: s.rootCfg.Namespace},
		dataPlaneObj); err != nil {
		log.Error("failed to get dataplane", zap.Error(err))
		return nil, bizerr.New(bizerr.K8SClientError, fmt.Sprintf("failed get dataplane:%s", dataplaneId))
	}
	return dataPlaneObj, nil
}