package k8supgrade

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"

	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	eksv1beta1 "github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	gkev1beta2 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta2"
	"github.com/pingcap/log"
	"github.com/samber/lo"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	clustercontrollerv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/acscli"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/ticdccli"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/appbox"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/approute"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/vpc"
)

const defaultConcurrentWorker = 10

var cloudProviders = map[string]string{
	consts.ProviderTypeEKS: "aws",
	consts.ProviderTypeGKE: "gcp",
	consts.ProviderTypeAKS: "azure",
}

type Service struct {
	appboxReader   appbox.APPBoxReader
	approuteReader approute.APPRouteReader
	vpcReader      vpc.VPCReader
	ccli           clustercli.ClusterClient
	ticdcCli       ticdccli.Client
	acsCli         acscli.Interface
	local          client.Client
	gcpCli         gcp.Client
	config         rsconfig.RootConfig
	workers        int
}

func NewService(cli client.Client, config rsconfig.RootConfig) *Service {
	worker := defaultConcurrentWorker
	if config.K8sUpgradeConfig.ConcurrentWorker > 0 {
		worker = config.K8sUpgradeConfig.ConcurrentWorker
	}

	return &Service{
		local:          cli,
		appboxReader:   appbox.NewAPPBoxReader(cli),
		approuteReader: approute.NewAPPRouteReader(cli),
		vpcReader:      vpc.NewVPCReader(cli),
		ccli:           clustercli.NewClusterCli(config.ClusterControlServiceEndpoint),
		ticdcCli:       ticdccli.NewClient(config.GetTiCDCServiceEndpoint()),
		acsCli:         acscli.NewClient(config.GetACSServiceEndpoint()),
		config:         config,
		workers:        worker,
	}
}

func (s *Service) WithGCPClient(cli gcp.Client) {
	s.gcpCli = cli
}

func (s *Service) ListK8sInformation(ctx context.Context, req *regionalpb.ListK8SInformationRequest) (regionalpb.K8SInformationList, error) {
	var list v1alpha1.APPBoxList
	if err := s.appboxReader.List(ctx, &list, client.MatchingFieldsSelector{
		Selector: fields.OneTermEqualSelector(controllers.KubernetesVersionField, req.Version),
	}); err != nil {
		log.Error("failed to list appbox", zap.Error(err), zap.String("version", req.Version))
		return regionalpb.K8SInformationList{}, err
	}
	log.Info("list appbox", zap.Int("count", len(list.Items)))

	// Use fan-out pattern to get k8s information from appbox CRs
	inputs := make(chan *v1alpha1.APPBox, len(list.Items))
	results := make(chan *regionalpb.K8SInformation, len(list.Items))

	for i := range list.Items {
		inputs <- &list.Items[i]
	}
	close(inputs)

	var wg sync.WaitGroup
	for i := 0; i < s.workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for item := range inputs {
				out, err := s.appbox2K8sInfo(ctx, item, false)
				if err != nil {
					log.Error("failed to convert appbox to k8s information", zap.Error(err), zap.String("appbox", item.GetNamespace()+"/"+item.GetName()))
				} else {
					results <- out
				}
			}
		}()
	}

	// Start a goroutine to close the results channel after all other goroutines finished
	go func() {
		wg.Wait()
		close(results)
	}()

	k8sInfoList := make([]*regionalpb.K8SInformation, 0, len(list.Items))
	for k8sInfo := range results {
		k8sInfoList = append(k8sInfoList, k8sInfo)
	}

	return regionalpb.K8SInformationList{K8SInfos: k8sInfoList, Total: uint32(len(k8sInfoList))}, nil
}

func (s *Service) GetK8sInformation(ctx context.Context, req *regionalpb.GetK8SInformationRequest) (*regionalpb.K8SInformation, error) {
	var name, ns string

	switch s.config.Provider {
	case consts.ProviderTypeEKS:
		var eks eksv1beta1.Cluster
		if err := s.local.Get(ctx, client.ObjectKey{Name: req.Name}, &eks); err != nil {
			log.Error("failed to get eks cluster", zap.Error(err))
			return &regionalpb.K8SInformation{}, err
		}
		name, ns = eks.Labels[namer.AppBoxSelectorKey], eks.Labels[namer.LegacyNetworkNamespaceSelectorKey]

	case consts.ProviderTypeGKE:
		var gke gkev1beta2.Cluster
		if err := s.local.Get(ctx, client.ObjectKey{Name: req.Name}, &gke); err != nil {
			log.Error("failed to get gke cluster", zap.Error(err))
			return &regionalpb.K8SInformation{}, err
		}
		name, ns = gke.Labels[namer.AppBoxSelectorKey], gke.Labels[namer.LegacyNetworkNamespaceSelectorKey]
	}

	var box v1alpha1.APPBox
	if err := s.appboxReader.Get(ctx, client.ObjectKey{
		Name:      name,
		Namespace: ns,
	}, &box); err != nil {
		log.Error("failed to get appbox", zap.Error(err), zap.String("name", name), zap.String("namespace", ns))
		return &regionalpb.K8SInformation{}, err
	}

	return s.appbox2K8sInfo(ctx, &box, true)
}

func (s *Service) appbox2K8sInfo(ctx context.Context, box *v1alpha1.APPBox, getNodesInfo bool) (*regionalpb.K8SInformation, error) {
	tenantID := box.Labels[v1alpha1.TagTenantKey]
	projectID := box.Labels[v1alpha1.TagProjectKey]
	info := &regionalpb.K8SInformation{
		CloudProvider:       cloudProviders[box.Spec.Provider],
		Region:              box.Spec.Region,
		TenantId:            tenantID,
		ProjectId:           projectID,
		K8SName:             box.Status.KubernetesClusterName,
		ControlPlaneVersion: box.Spec.KubernetesConfig.Version,
		AppboxName:          box.Name,
		AppboxNamespace:     box.Namespace,
	}
	acct, err := getCloudAccountFromAppBox(ctx, s.vpcReader, box)
	if err != nil {
		log.Error("failed to get cloud account", zap.Error(err))
	} else {
		info.CloudAccount = acct
	}

	var (
		sharedTiDBNodes *regionalpb.NodeGroup
		compNodesMap    = make(map[string]map[string]*regionalpb.NodeGroup)
		notes           []string
	)
	if getNodesInfo {
		// List admin nodes
		switch box.Spec.Provider {
		case consts.ProviderTypeEKS:
			var ngList manualv1alpha1.NodeGroupList
			if err := s.local.List(ctx, &ngList, utils.GetAdminNodesLabelSelector(box)); err != nil {
				log.Error("failed to list admin nodes' nodegroups", zap.Error(err))
				return nil, err
			}
			info.AdminNodes = managedNodeGroups2PB(ngList)

		case consts.ProviderTypeGKE:
			var gke gkev1beta2.Cluster
			if err := s.local.Get(ctx, client.ObjectKey{
				Name: box.Status.KubernetesClusterName,
			}, &gke); err != nil {
				log.Error("failed to get gke cluster CR", zap.Error(err))
				return nil, err
			}
			info.AdminNodes, sharedTiDBNodes, compNodesMap, notes = getNodesInfoFromGKECR(ctx, s.gcpCli, &gke)
		}

		if err := s.populateCDCNodeVersion(ctx, tenantID, projectID, compNodesMap, &notes); err != nil {
			log.Error("failed to populate CDC node versions", zap.Error(err))
			notes = append(notes, "failed to populate CDC node versions")
		}

		if err := s.populateO11yNodeVersion(ctx, tenantID, projectID, compNodesMap, &notes); err != nil {
			log.Error("failed to populate o11y node versions", zap.Error(err))
			notes = append(notes, "failed to populate o11y node versions")
		}
		if m := compNodesMap[string(v1alpha1.O11yK8sNodes)]; m != nil {
			for _, ng := range m {
				info.O11YK8SNodes = append(info.O11YK8SNodes, ng)
			}
		}
	}

	// List clusters in the appbox
	var approuteList v1alpha1.APPRouteList
	if err := s.approuteReader.List(ctx, &approuteList, client.MatchingLabelsSelector{
		Selector: labels.SelectorFromSet(labels.Set{
			v1alpha1.APPBoxNameKey: box.GetName(),
		}),
	}); err != nil {
		log.Error("failed to list approute", zap.Error(err))
	}
	info.NumTidbClusters = 0
	info.TidbClusters = make([]*regionalpb.TiDBClusterInfraInformation, 0, len(approuteList.Items))
	for _, route := range approuteList.Items {
		if route.Status.Phase != v1alpha1.APPRoutePhaseActive {
			log.Info("skip inactive approute", zap.String("name", route.GetName()), zap.String("phase", string(route.Status.Phase)))
			continue
		}
		info.NumTidbClusters++

		if getNodesInfo {
			clusterInfraInfo, err := s.ccli.GetInfraInformation(ctx, &clustercontrollerv1.GetInfraInformationRequest{
				ClusterId: route.Name,
			})
			if err != nil {
				log.Error("failed to get cluster infra information", zap.Error(err), zap.String("clusterId", route.Name))
				continue
			}
			tmp := infraInfo2PB(clusterInfraInfo)
			tmp.Notes = append(tmp.Notes, notes...)

			if sharedTiDBNodes != nil {
				tmp.ComponentNodes = append(tmp.ComponentNodes, sharedTiDBNodes)
			}

			if m := compNodesMap[string(cloudv1.CDCNodes)]; m != nil {
				if cdcNodes := m[route.Name]; cdcNodes != nil {
					tmp.ComponentNodes = append(tmp.ComponentNodes, cdcNodes)
				}
			}
			if m := compNodesMap[string(cloudv1.DMNodes)]; m != nil {
				if dmNodes := m[route.Name]; dmNodes != nil {
					tmp.ComponentNodes = append(tmp.ComponentNodes, dmNodes)
				}
			}
			if m := compNodesMap[string(v1alpha1.O11yTiDBNodes)]; m != nil {
				if o11yTiDBNodes := m[route.Name]; o11yTiDBNodes != nil {
					tmp.ComponentNodes = append(tmp.ComponentNodes, o11yTiDBNodes)
				}
			}
			info.TidbClusters = append(info.TidbClusters, tmp)
		}
	}

	sort.Slice(info.TidbClusters, func(i, j int) bool {
		return info.TidbClusters[i].ClusterId < info.TidbClusters[j].ClusterId
	})

	return info, nil
}

// populateCDCNodeVersion calls the ticdc service to get the CDC clusters for
// the given tidb cluster and writes their nodeVersion into compNodesMap.
func (s *Service) populateCDCNodeVersion(ctx context.Context, tenantID, projectID string, compNodesMap map[string]map[string]*regionalpb.NodeGroup, notes *[]string) error {
	cdcClusters, err := s.ticdcCli.ListTiCDCClusters(ctx, tenantID, projectID, "", nil)
	if err != nil {
		log.Warn("failed to list ticdc clusters", zap.Error(err), zap.String("tenantID", tenantID), zap.String("projectID", projectID))
		return err
	}
	nodeKind := string(cloudv1.CDCNodes)
	versionSet := make(map[string]map[string]struct{})
	for _, cdc := range cdcClusters.TicdcClusters {
		if cdc.NodeVersion == "" {
			continue
		}
		tidbClusterID := cdc.ClusterId
		if versionSet[tidbClusterID] == nil {
			versionSet[tidbClusterID] = make(map[string]struct{})
		}
		versionSet[tidbClusterID][cdc.NodeVersion] = struct{}{}
		if compNodesMap[nodeKind] == nil {
			compNodesMap[nodeKind] = make(map[string]*regionalpb.NodeGroup)
		}
		if existing, ok := compNodesMap[nodeKind][tidbClusterID]; !ok {
			compNodesMap[nodeKind][tidbClusterID] = &regionalpb.NodeGroup{
				Kind:           nodeKind,
				Version:        cdc.NodeVersion,
				Count:          cdc.NodeCount,
				MaxSkewVersion: managedNodeGroupMaxSkewVersion,
			}
		} else if existing.Version > cdc.NodeVersion {
			existing.Version = cdc.NodeVersion
		}
	}
	for clusterID, versions := range versionSet {
		if len(versions) > 1 {
			versions := lo.Keys(versions)
			sort.Strings(versions)
			*notes = append(*notes, fmt.Sprintf("multiple CDC node versions detected in tidb cluster %s: %s", clusterID, strings.Join(versions, ",")))
		}
	}
	return nil
}

// populateO11yNodeVersion calls the ACS service to get o11y nodegroup versions
// and writes them into compNodesMap.
func (s *Service) populateO11yNodeVersion(ctx context.Context, tenantID, projectID string, compNodesMap map[string]map[string]*regionalpb.NodeGroup, notes *[]string) error {
	resp, err := s.acsCli.ListNodeGroups(ctx, tenantID, projectID)
	if err != nil {
		log.Warn("failed to list o11y node groups", zap.Error(err), zap.String("tenantID", tenantID), zap.String("projectID", projectID))
		return err
	}
	versionSet := make(map[string]map[string]struct{})
	for _, ng := range resp.Data.NodeGroups {
		if ng.Version == "" {
			continue
		}
		clusterKey := ng.AgentName
		if versionSet[ng.Kind] == nil {
			versionSet[ng.Kind] = make(map[string]struct{})
		}
		versionSet[ng.Kind][ng.Version] = struct{}{}
		if compNodesMap[ng.Kind] == nil {
			compNodesMap[ng.Kind] = make(map[string]*regionalpb.NodeGroup)
		}
		if existing, ok := compNodesMap[ng.Kind][clusterKey]; !ok {
			compNodesMap[ng.Kind][clusterKey] = &regionalpb.NodeGroup{
				Kind:           ng.Kind,
				Version:        ng.Version,
				Count:          uint32(ng.Count),
				MaxSkewVersion: managedNodeGroupMaxSkewVersion,
			}
		} else if existing.Version > ng.Version {
			existing.Version = ng.Version
		}
	}
	for kind, versions := range versionSet {
		if len(versions) > 1 {
			versions := lo.Keys(versions)
			sort.Strings(versions)
			*notes = append(*notes, fmt.Sprintf("multiple o11y node versions detected for %s: %s", kind, strings.Join(versions, ",")))
		}
	}
	return nil
}

func (s *Service) CreateOrUpdateUpgradeOperation(ctx context.Context, reqUO *regionalpb.UpgradeOperation) error {
	desiredUO, err := pbUpgradeOperation2UpgradeOperation(reqUO)
	if err != nil {
		log.Error("failed to convert PB upgrade operation to upgrade operation CR", zap.Error(err))
		return err
	}
	uo := &v1alpha1.UpgradeOperation{
		ObjectMeta: metav1.ObjectMeta{
			Name:      desiredUO.Name,
			Namespace: utils.NewNamer("", "", "", s.config.RegionConfig.Name, s.config.Env).GetNamespace(),
		},
	}
	res, err := controllerutil.CreateOrUpdate(ctx, s.local, uo, func() error {
		uo.Spec = *desiredUO.Spec.DeepCopy()
		if len(desiredUO.Annotations) > 0 {
			uo.Annotations = mergeAnnotations(uo.Annotations, desiredUO.Annotations)
		}
		return nil
	})
	if err != nil {
		log.Error("failed to create or update upgrade operation CR", zap.Error(err))
		return err
	}
	log.Info("create or update upgrade operation CR", zap.String("result", string(res)), zap.String("name", uo.Name), zap.String("namespace", uo.Namespace))
	return nil
}

func (s *Service) GetUpgradeOperation(ctx context.Context, req *regionalpb.GetUpgradeOperationRequest) (*regionalpb.UpgradeOperation, error) {
	ns := utils.NewNamer("", "", "", s.config.RegionConfig.Name, s.config.Env).GetNamespace()
	uo := &v1alpha1.UpgradeOperation{}
	if err := s.local.Get(ctx, client.ObjectKey{
		Name:      req.Name,
		Namespace: ns,
	}, uo); err != nil {
		log.Error("failed to get upgrade operation CR", zap.Error(err))
		return nil, err
	}
	return upgradeOperation2PB(uo), nil
}

func (s *Service) DeleteUpgradeOperation(ctx context.Context, req *regionalpb.DeleteUpgradeOperationRequest) error {
	ns := utils.NewNamer("", "", "", s.config.RegionConfig.Name, s.config.Env).GetNamespace()
	uo := &v1alpha1.UpgradeOperation{}
	if err := s.local.Get(ctx, client.ObjectKey{
		Name:      req.Name,
		Namespace: ns,
	}, uo); err != nil {
		if client.IgnoreNotFound(err) != nil {
			log.Error("failed to get upgrade operation CR", zap.Error(err))
			return err
		}
		log.Info("upgrade operation CR not found", zap.String("name", req.Name), zap.String("namespace", ns))
		return nil
	}

	if err := s.local.Delete(ctx, uo); err != nil {
		return fmt.Errorf("failed to delete upgrade operation CR: %w", err)
	}
	return nil
}

func (s *Service) PatchUpgradeOperationAnnotations(ctx context.Context, req *regionalpb.PatchUpgradeOperationAnnotationsRequest) error {
	if req == nil || req.Name == "" {
		return fmt.Errorf("upgrade operation name is required")
	}
	if len(req.Annotations) == 0 {
		return nil
	}

	ns := utils.NewNamer("", "", "", s.config.RegionConfig.Name, s.config.Env).GetNamespace()
	uo := &v1alpha1.UpgradeOperation{}
	if err := s.local.Get(ctx, client.ObjectKey{
		Name:      req.Name,
		Namespace: ns,
	}, uo); err != nil {
		log.Error("failed to get upgrade operation CR", zap.Error(err))
		return err
	}

	patch := uo.DeepCopy()
	patch.Annotations = mergeAnnotations(patch.Annotations, req.Annotations)
	if err := s.local.Patch(ctx, patch, client.MergeFrom(uo)); err != nil {
		log.Error("failed to patch upgrade operation annotations", zap.Error(err), zap.String("name", req.Name))
		return fmt.Errorf("failed to patch upgrade operation annotations: %w", err)
	}
	log.Info("patched upgrade operation annotations", zap.String("name", req.Name), zap.String("namespace", ns))
	return nil
}