package cloud

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

	compute "cloud.google.com/go/compute/apiv1"
	"cloud.google.com/go/compute/apiv1/computepb"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/elliotchance/pie/v2"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/scheme"
	"github.com/tidbcloud/infra-provider/pkg/util"
	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	ticdc_consts "github.com/tidbcloud/ticdc-service/pkg/consts"
	computev1beta1 "github.com/upbound/provider-gcp/apis/compute/v1beta1"
	"github.com/upbound/provider-gcp/apis/container/v1beta1"
	dnsv1beta1 "github.com/upbound/provider-gcp/apis/dns/v1beta1"
	gcpv1beta1 "github.com/upbound/provider-gcp/apis/v1beta1"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/selection"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	gkeClusterURLFormat = "projects/%s/locations/%s/clusters/%s"
	vpcURLFormat        = "projects/%s/global/networks/%s"
	subnetURLFormat     = "projects/%s/regions/%s/subnetworks/%s"

	gkeExecConfigAPIVersion = "client.authentication.k8s.io/v1beta1"
)

var initForwardingRulesCliFunc func() (*compute.ForwardingRulesClient, error) = sync.OnceValues(
	func() (*compute.ForwardingRulesClient, error) {
		return compute.NewForwardingRulesRESTClient(context.Background())
	})

type gcp struct {
	appbox                *regionalpb.APPBox
	ticdcServiceCfg       *config.Config
	forwardingRulesClient *compute.ForwardingRulesClient
}

func NewGcpProvider(appbox *regionalpb.APPBox,
	ticdcServiceCfg *config.Config) (Provider, error) {
	forwardingRulesClient, err := initForwardingRulesCliFunc()
	if err != nil {
		return nil, errors.Wrap(err, "fail to init forwarding rules client")
	}
	return &gcp{
		appbox:                appbox,
		ticdcServiceCfg:       ticdcServiceCfg,
		forwardingRulesClient: forwardingRulesClient,
	}, nil
}

func (g *gcp) GetKubeClient(ctx context.Context) (client.Client, error) {
	opts := client.Options{
		Scheme: scheme.Scheme,
	}

	restCfg, err := g.getConfig()
	if err != nil {
		return nil, errors.Trace(err)
	}
	return client.New(restCfg, opts)
}

func (g *gcp) GetKubernetesName(ctx context.Context) (string, error) {
	return g.appbox.Kubernetes.Name, nil
}

func (g *gcp) GetKubernetesVersion(ctx context.Context) (string, error) {
	return g.appbox.Kubernetes.Version, nil
}

func (g *gcp) GetKubernetesID(ctx context.Context) (string, error) {
	return g.appbox.Kubernetes.Name, nil
}

func (g *gcp) GetNodeZones(ctx context.Context) ([]string, error) {
	return g.appbox.Kubernetes.GetGcp().GetNodeZones(), nil
}

func (g *gcp) GetNodeSubnets(ctx context.Context) ([]string, error) {
	var subnets = make([]string, 0, len(g.appbox.Kubernetes.GetGcp().GetNodeSubnets()))
	for _, subNet := range g.appbox.Kubernetes.GetGcp().GetNodeSubnets() {
		subnets = append(subnets, subNet.Id)
	}
	return subnets, nil
}

func (g *gcp) GetPodSubnets(ctx context.Context) ([]string, error) {
	var zones = make([]string, 0, len(g.appbox.Kubernetes.GetGcp().GetNodeSubnets()))
	for _, subNet := range g.appbox.Kubernetes.GetGcp().GetNodeSubnets() {
		zones = append(zones, subNet.Id)
	}
	return zones, nil
}

func (g *gcp) GetSecondaryNodeSubnets(context.Context) ([]string, error) {
	return []string{}, nil
}

func (g *gcp) BuildSinkEndpoint(providerConfig string, name types.NamespacedName, request *EndpointRequest) ([]client.Object, error) {
	ip := &computev1beta1.Address{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
		Spec: computev1beta1.AddressSpec{
			ForProvider: computev1beta1.AddressParameters{
				AddressType: ptr.To("INTERNAL"),
				Project:     ptr.To(g.appbox.Kubernetes.AccountId),
				Region:      ptr.To(request.Region),
				Subnetwork: ptr.To(fmt.Sprintf(subnetURLFormat,
					g.appbox.Kubernetes.AccountId,
					request.Region,
					g.appbox.Kubernetes.GetGcp().NodeSubnets[0].Id)),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	pslc := &computev1beta1.ForwardingRule{
		ObjectMeta: metav1.ObjectMeta{
			Name:   ip.Name,
			Labels: request.Tags,
		},
		Spec: computev1beta1.ForwardingRuleSpec{
			ForProvider: computev1beta1.ForwardingRuleParameters{
				Target:  ptr.To(request.EndpointServiceName),
				Region:  ptr.To(request.Region),
				Project: ptr.To(g.appbox.Kubernetes.AccountId),
				Network: ptr.To(fmt.Sprintf(vpcURLFormat,
					g.appbox.Kubernetes.AccountId,
					g.appbox.Kubernetes.Vpc)),
				Subnetwork: ptr.To(fmt.Sprintf(subnetURLFormat,
					g.appbox.Kubernetes.AccountId,
					request.Region,
					g.appbox.Kubernetes.GetGcp().NodeSubnets[0].Id)),
				IPAddressRef: &xpv1.Reference{
					Name: ip.Name,
				},
				Labels:              util.ToMapPointer(request.Tags),
				LoadBalancingScheme: ptr.To(""),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	return []client.Object{ip, pslc}, nil
}

func (g *gcp) DeleteSinkEndpoint(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	if err := k8s.DeleteObject(ctx, cli, &computev1beta1.Address{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete Address failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &computev1beta1.ForwardingRule{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete ForwardingRule failed")
	}
	return nil
}

func (g *gcp) GetSinkEndpointDetail(ctx context.Context, cli client.Client,
	name types.NamespacedName) (*SinkEndpoint, error) {
	forwardingRule := &computev1beta1.ForwardingRule{}
	if err := cli.Get(ctx, name, forwardingRule); err != nil {
		return nil, errors.Trace(err)
	}
	detail := &SinkEndpoint{}
	detail.Phase = cdcv1.EndpointResourceStatePendingAcceptance
	if forwardingRule.Status.AtProvider.ID != nil {
		detail.EndpointID = *forwardingRule.Status.AtProvider.ID
	}
	if forwardingRule.Status.AtProvider.PscConnectionStatus != nil {
		detail.Phase = gcpForwardingRuleStatus2SinkEndpointPhase(*forwardingRule.Status.AtProvider.PscConnectionStatus)
	}
	if forwardingRule.Status.AtProvider.IPAddress != nil {
		detail.Address = *forwardingRule.Status.AtProvider.IPAddress
	}

	// the sink endpoint is ready, but the upbound doest not update the status, try to use aws SDK to get the detail
	if isResourceReady(forwardingRule.Status.ResourceStatus) && detail.EndpointID != "" &&
		detail.Address != "" &&
		detail.Phase != cdcv1.EndpointResourceStateEndpointAvailable {
		gcpConnection, err := g.forwardingRulesClient.Get(ctx, &computepb.GetForwardingRuleRequest{
			Project:        g.appbox.Kubernetes.AccountId,
			Region:         g.ticdcServiceCfg.Region,
			ForwardingRule: forwardingRule.GetName(),
		})
		if err != nil {
			return nil, errors.Annotate(err, "failed to get forwarding rule")
		}
		detail.Phase = gcpForwardingRuleStatus2SinkEndpointPhase(*gcpConnection.PscConnectionStatus)
	}
	return detail, nil
}

func (g *gcp) ValidateEndpointZones(ctx context.Context, endpointZones []string, serviceName string) error {
	return nil
}

func (g *gcp) BuildDNS(providerConfig string, dns *DNSRequest) ([]client.Object, error) {
	managedZone := &dnsv1beta1.ManagedZone{
		ObjectMeta: metav1.ObjectMeta{
			Name: dns.Name,
		},
		Spec: dnsv1beta1.ManagedZoneSpec{
			ForProvider: dnsv1beta1.ManagedZoneParameters{
				DNSName: ptr.To(dns.RootDomainName),
				PrivateVisibilityConfig: []dnsv1beta1.PrivateVisibilityConfigParameters{
					{
						Networks: []dnsv1beta1.NetworksParameters{
							{
								NetworkURL: ptr.To(fmt.Sprintf(vpcURLFormat,
									g.appbox.Kubernetes.AccountId,
									g.appbox.Kubernetes.Vpc)),
							},
						},
					},
				},
				Labels:     util.ToMapPointer(dns.Tags),
				Project:    ptr.To(g.appbox.Kubernetes.GetAccountId()),
				Visibility: ptr.To("private"),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	domainRecord := &dnsv1beta1.RecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: g.getDomainDNSRecordName(dns.Name),
		},
		Spec: dnsv1beta1.RecordSetSpec{
			ForProvider: dnsv1beta1.RecordSetParameters{
				ManagedZoneRef: &xpv1.Reference{Name: dns.Name},
				Name:           ptr.To(dns.RootDomainName),
				Project:        ptr.To(g.appbox.Kubernetes.GetAccountId()),
				Rrdatas: []*string{
					ptr.To(dns.Endpoint.Address),
				},
				TTL:  ptr.To(float64(300)),
				Type: ptr.To("A"),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	subDomainRecord := &dnsv1beta1.RecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: dns.Name,
		},
		Spec: dnsv1beta1.RecordSetSpec{
			ForProvider: dnsv1beta1.RecordSetParameters{
				ManagedZoneRef: &xpv1.Reference{Name: dns.Name},
				Name:           ptr.To(fmt.Sprintf("*.%s", dns.RootDomainName)),
				Project:        ptr.To(g.appbox.Kubernetes.GetAccountId()),
				Rrdatas: []*string{
					ptr.To(dns.Endpoint.Address),
				},
				TTL:  ptr.To(float64(300)),
				Type: ptr.To("A"),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	return []client.Object{managedZone, domainRecord, subDomainRecord}, nil
}

func (g *gcp) getDomainDNSRecordName(dnsZoneName string) string {
	return fmt.Sprintf("%s-domain", dnsZoneName)
}

func (g *gcp) DeleteDNS(ctx context.Context, cli client.Client, name string) error {
	if err := k8s.DeleteObject(ctx, cli, &dnsv1beta1.RecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: g.getDomainDNSRecordName(name),
		},
	}); err != nil {
		return errors.Annotate(err, "delete domain A record failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &dnsv1beta1.RecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete sub domain A record failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &dnsv1beta1.ManagedZone{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete managed zone failed")
	}
	return nil
}

func (g *gcp) BuildNodePool(providerConfig string, name types.NamespacedName, np *NodePool) (client.Object, error) {
	if len(np.NodeSubnets) < 1 {
		return nil, fmt.Errorf("missing node subnet")
	}
	var taints = make([]v1beta1.NodePoolNodeConfigTaintParameters, 0, len(np.NodeSubnets))
	for _, item := range np.Taints {
		taints = append(taints, v1beta1.NodePoolNodeConfigTaintParameters{
			Effect: ptr.To(taintMap[item.Effect]),
			Key:    ptr.To(item.Key),
			Value:  ptr.To(item.Value),
		})
	}

	if np.Labels == nil {
		np.Labels = make(map[string]string)
	}
	np.Labels[crOwnerRefLabel] = shortName(np.Name)
	return &v1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Labels: appendAppboxIDLabel(np.Labels, g.appbox.Name),
			Name:   shortName(np.Name),
		},
		Spec: v1beta1.NodePoolSpec{
			ForProvider: v1beta1.NodePoolParameters_2{
				Autoscaling: []v1beta1.NodePoolAutoscalingParameters{
					{
						MinNodeCount: ptr.To(float64(0)),
						MaxNodeCount: ptr.To(float64(maxNodeCount)),
					},
				},
				Project:  ptr.To(g.appbox.Kubernetes.GetAccountId()),
				Cluster:  ptr.To(g.appbox.Kubernetes.Name),
				Location: ptr.To(np.Region),
				NodeConfig: []v1beta1.NodePoolNodeConfigParameters_2{
					{
						DiskSizeGb:     ptr.To(float64(50)),
						DiskType:       ptr.To("pd-standard"),
						ImageType:      ptr.To("cos_containerd"),
						MachineType:    ptr.To(np.InstanceType),
						Labels:         util.ToMapPointer(np.Labels),
						ResourceLabels: util.ToMapPointer(np.Tags),
						OAuthScopes: []*string{
							ptr.To("https://www.googleapis.com/auth/cloud-platform"),
						},
						Taint:          taints,
						ServiceAccount: ptr.To(np.Identity),
					},
				},
				NodeLocations: util.ToSlicePointer(np.Zones),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfig,
				},
			},
		},
	}, nil
}

func (g *gcp) PostCreateNodePool(ctx context.Context, cli client.Client, resourceTags map[string]string) error {
	return nil
}

func (g *gcp) DeleteNodePool(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	sc := v1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: shortName(name.Name),
		},
	}
	return k8s.DeleteObject(ctx, cli, &sc)
}

func (g *gcp) BuildProviderConfig(name types.NamespacedName) (client.Object, error) {
	return &gcpv1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
		Spec: gcpv1beta1.ProviderConfigSpec{
			ProjectID: g.appbox.Kubernetes.AccountId,
			Credentials: gcpv1beta1.ProviderCredentials{
				Source: xpv1.CredentialsSourceInjectedIdentity,
			},
		},
	}, nil
}

func (g *gcp) DeleteProviderConfig(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	pc := gcpv1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}
	return k8s.DeleteObject(ctx, cli, &pc)
}

func (g *gcp) BuildStorageClass(name types.NamespacedName, class *StorageClass) (client.Object, error) {
	params := map[string]string{
		"csi.storage.k8s.io/fstype": "ext4",
		"type":                      consts.GCPVolumeTypePDBalanced,
	}
	return &storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
		VolumeBindingMode:    ptr.To(storagev1.VolumeBindingWaitForFirstConsumer),
		AllowVolumeExpansion: ptr.To(true),
		ReclaimPolicy:        ptr.To(corev1.PersistentVolumeReclaimDelete),
		Provisioner:          "pd.csi.storage.gke.io",
		Parameters:           params,
		MountOptions:         []string{"nodelalloc,noatime"},
	}, nil
}

func (g *gcp) DeleteStorageClass(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	sc := storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}
	return k8s.DeleteObject(ctx, cli, &sc)
}

var gcpStatusMap = map[string]cdcv1.SinkEndpointPhase{
	"pending":  cdcv1.EndpointResourceStatePendingAcceptance,
	"accepted": cdcv1.EndpointResourceStateEndpointAvailable,
	"closed":   cdcv1.EndpointResourceStateFailed,
	"rejected": cdcv1.EndpointResourceStateRejected,
}

func (g *gcp) getConfig() (*rest.Config, error) {
	if g.ticdcServiceCfg.Debug.Enable && g.ticdcServiceCfg.Debug.DataPlaneKubeConfig != "" {
		return clientcmd.BuildConfigFromFlags("", g.ticdcServiceCfg.Debug.DataPlaneKubeConfig)
	}

	cfg := clientcmdapi.NewConfig()

	clusterName := g.appbox.Kubernetes.Name
	cfgctx := clientcmdapi.NewContext()
	cfgctx.Cluster = clusterName
	cfgctx.AuthInfo = clusterName

	cluster := clientcmdapi.NewCluster()
	// TODO: get ca data from gcp
	cluster.InsecureSkipTLSVerify = true
	// cluster.CertificateAuthorityData = []byte(gke.Status.AtProvider.CertificateAuthorityData)
	cluster.Server = g.appbox.Kubernetes.GetEndpoint()

	auth := clientcmdapi.NewAuthInfo()
	auth.Exec = &clientcmdapi.ExecConfig{
		APIVersion:      gkeExecConfigAPIVersion,
		InteractiveMode: clientcmdapi.NeverExecInteractiveMode,
		Command:         "gke-gcloud-auth-plugin",
		Args:            []string{"--use_application_default_credentials"},
	}

	cfg.Contexts[clusterName] = cfgctx
	cfg.Clusters[clusterName] = cluster
	cfg.AuthInfos[clusterName] = auth
	cfg.CurrentContext = clusterName

	return clientcmd.NewDefaultClientConfig(*cfg, nil).ClientConfig()
}

func gcpForwardingRuleStatus2SinkEndpointPhase(status string) cdcv1.SinkEndpointPhase {
	if phase, ok := gcpStatusMap[strings.ToLower(status)]; ok {
		return phase
	}
	return cdcv1.EndpointResourceStatePendingAcceptance
}

type gcpGCProvider struct {
	cli client.Client
}

func NewGcpGCProvider(cli client.Client) GCProvider {
	return &gcpGCProvider{
		cli: cli,
	}
}

func (g *gcpGCProvider) ListAppbox(ctx context.Context) ([]string, error) {
	var ngs v1beta1.NodePoolList
	listOptions, err := resourceWithLabel(ticdc_consts.AppboxIDLabel)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if err := g.cli.List(ctx, &ngs, listOptions); err != nil {
		return nil, errors.Trace(err)
	}
	var appboxIDs = make(map[string]struct{})
	for _, ng := range ngs.Items {
		if ng.Labels != nil {
			appboxID := ng.Labels[ticdc_consts.AppboxIDLabel]
			if appboxID == "" {
				log.Warn("no appbox ID label found", zap.Any("node group", ng))
				continue
			}
			appboxIDs[appboxID] = struct{}{}
		}
	}
	return pie.Keys(appboxIDs), nil
}

func (g *gcpGCProvider) OnAppboxDeleting(ctx context.Context, appboxId string) error {
	var ngs v1beta1.NodePoolList
	if err := g.cli.List(ctx, &ngs, client.MatchingLabels{
		ticdc_consts.AppboxIDLabel: appboxId,
	}); err != nil {
		return errors.Trace(err)
	}
	for _, ng := range ngs.Items {
		if err := k8s.DeleteObject(ctx, g.cli, &ng); err != nil {
			log.Warn("delete node group failed",
				zap.Any("node group", ng), zap.Error(err))
			return errors.Trace(err)
		}
		log.Info("node group deleted", zap.Any("node group", ng))
	}
	return nil
}

// resourceWithLabel returns a ListOptions that check if a label exists
func resourceWithLabel(label string) (*client.ListOptions, error) {
	requirement, err := labels.NewRequirement(label, selection.Exists, []string{})
	if err != nil {
		return nil, errors.Trace(err)
	}
	selector := labels.NewSelector().Add(*requirement)
	return &client.ListOptions{
		LabelSelector: selector,
	}, nil
}