package cloud

import (
	"context"
	"encoding/base64"
	"fmt"
	"os"
	"strings"

	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork/v6"
	pcav1beta1 "github.com/cert-manager/aws-privateca-issuer/pkg/api/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/elliotchance/pie/v2"
	certmanagerv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	opv2alpha1 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	asts "github.com/tidbcloud/infra-provider/pkg/pingcap/apis/v1"
	"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/consts"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	"github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	netv1beta1 "github.com/upbound/provider-azure/apis/network/v1beta1"
	azv1beta1 "github.com/upbound/provider-azure/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/runtime"
	"k8s.io/apimachinery/pkg/types"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"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 (
	crossplaneExternalName = "crossplane.io/external-name"
)

var (
	storageSkuMap = map[string]string{
		"standard":  "Standard_LRS",
		"premiumv2": "PremiumV2_LRS",
		"ultrassd":  "UltraSSD_LRS",
	}

	// infra-provider/pkg/scheme registers both controlplane and dataplane scheme
	// TODO: We should create scheme from scratch and only register application specific scheme
	// scheme = runtime.NewScheme() // dataplane scheme
)

type addToSchemeFunc func(s *runtime.Scheme) error

func init() {
	schemeFuncs := []addToSchemeFunc{
		clientgoscheme.AddToScheme,
		certmanagerv1.AddToScheme,
		pcav1beta1.AddToScheme,
		asts.AddToScheme,
		opv1alpha1.AddToScheme,
		opv2alpha1.Install,
	}
	for _, f := range schemeFuncs {
		utilruntime.Must(f(scheme.Scheme))
	}
}

type azure struct {
	azTenantID             string
	azSubscriptionID       string
	azClientID             string
	regionalCli            regionalservice.Client
	kubeCli                client.Client
	appbox                 *regionalpb.APPBox
	privateEndpointsClient *armnetwork.PrivateEndpointsClient
}

func NewAzureProvider(
	ctx context.Context,
	azTenantID string,
	azClientID string,
	regionalCli regionalservice.Client,
	kubeCli client.Client,
	appbox *regionalpb.APPBox,
) (Provider, error) {
	cred, err := azidentity.NewDefaultAzureCredential(nil)
	if err != nil {
		return nil, err
	}
	azSubscriptionID := appbox.Kubernetes.AccountId
	networkClientFactory, err := armnetwork.NewClientFactory(azSubscriptionID, cred, nil)
	if err != nil {
		return nil, err
	}
	privateEndpointsClient := networkClientFactory.NewPrivateEndpointsClient()
	return &azure{
		azTenantID:             azTenantID,
		azSubscriptionID:       appbox.Kubernetes.AccountId,
		azClientID:             azClientID,
		regionalCli:            regionalCli,
		kubeCli:                kubeCli,
		appbox:                 appbox,
		privateEndpointsClient: privateEndpointsClient,
	}, nil
}

func (p *azure) BuildNodePool(providerConfig string, name types.NamespacedName, np *NodePool) (client.Object, error) {
	taints := []*string{}
	for _, t := range np.Taints {
		taint := fmt.Sprintf("%s=%s:%s", t.Key, t.Value, t.Effect)
		taints = append(taints, ptr.To(taint))
	}
	if len(np.NodeSubnets) < 1 || len(np.PodSubnets) < 1 {
		return nil, fmt.Errorf("missing node or pod subnet")
	}
	zones := make([]string, len(np.Zones))
	for i, z := range np.Zones {
		s := strings.Split(z, "-")
		if len(s) != 2 {
			return nil, fmt.Errorf("invalid zone %s", z)
		}
		zones[i] = s[len(s)-1]
	}
	diskSize := ptr.To(float64(defaultRootDiskSizeGB))
	if np.RootDiskSizeGB != 0 {
		diskSize = ptr.To(float64(np.RootDiskSizeGB))
	}
	kcnp := &v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   shortName(np.Name),
			Labels: appendAppboxIDLabel(np.Labels, p.appbox.Name),
		},
		Spec: v1beta1.KubernetesClusterNodePoolSpec{
			ForProvider: v1beta1.KubernetesClusterNodePoolParameters{
				VMSize:              ptr.To(np.InstanceType),
				OsSku:               ptr.To("AzureLinux"),
				OsDiskSizeGb:        diskSize,
				Zones:               util.ToSlicePointer(zones),
				Tags:                util.ToMapPointer(np.Tags),
				NodeLabels:          util.ToMapPointer(np.Labels),
				NodeTaints:          taints,
				KubernetesClusterID: ptr.To(np.KubeClusterID),
				VnetSubnetID:        ptr.To(np.NodeSubnets[0]),
				PodSubnetID:         ptr.To(np.PodSubnets[0]),
				// if it is not set, the default value is 10%, and aks will be frequently update by crossplane
				// because it is not set by crossplane lateinitialize
				// https://github.com/crossplane-contrib/provider-upjet-azure/issues/686
				UpgradeSettings: []v1beta1.KubernetesClusterNodePoolUpgradeSettingsParameters{
					{
						MaxSurge: ptr.To("10%"),
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfig,
				},
			},
		},
	}
	if np.NodeCount != 0 {
		kcnp.Spec.ForProvider.NodeCount = ptr.To(float64(np.NodeCount))
	} else {
		kcnp.Spec.ForProvider.EnableAutoScaling = ptr.To(true)
		kcnp.Spec.ForProvider.MinCount = ptr.To(float64(0))
		kcnp.Spec.ForProvider.MaxCount = ptr.To(float64(maxNodeCount))
	}

	return kcnp, nil
}

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

func (p *azure) DeleteNodePool(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	np := v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: shortName(name.Name),
		},
	}
	return k8s.DeleteObject(ctx, cli, &np)
}

func (p *azure) BuildProviderConfig(name types.NamespacedName) (client.Object, error) {
	return &azv1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name.Name,
			Labels: appendAppboxIDLabel(nil, p.appbox.Name),
		},
		Spec: azv1beta1.ProviderConfigSpec{
			Credentials: azv1beta1.ProviderCredentials{
				Source: "OIDCTokenFile",
			},
			TenantID:       ptr.To(p.azTenantID),
			SubscriptionID: ptr.To(p.azSubscriptionID),
			ClientID:       ptr.To(p.azClientID),
		},
	}, nil
}

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

func (p *azure) BuildStorageClass(name types.NamespacedName, storageClass *StorageClass) (client.Object, error) {
	sc := &storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
		Provisioner:          "disk.csi.azure.com",
		VolumeBindingMode:    ptr.To(storagev1.VolumeBindingWaitForFirstConsumer),
		AllowVolumeExpansion: ptr.To(true),
		ReclaimPolicy:        ptr.To(corev1.PersistentVolumeReclaimDelete),
		Parameters: map[string]string{
			"csi.storage.k8s.io/fstype": "ext4",
		},
		MountOptions: []string{"nodelalloc,noatime"},
	}
	if iops := storageClass.Iops; iops != 0 {
		sc.Parameters["DiskIOPSReadWrite"] = fmt.Sprint(iops)
	}
	if throughput := storageClass.ThroughputMbs; throughput != 0 {
		sc.Parameters["DiskMBpsReadWrite"] = fmt.Sprint(throughput)
	}
	for typ, sku := range storageSkuMap {
		if typ == storageClass.Type {
			sc.Parameters["skuName"] = sku
		}
	}
	return sc, nil
}

func (p *azure) 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)
}

func aksWorkloadIdentityEnabled() bool {
	return os.Getenv("AZURE_FEDERATED_TOKEN_FILE") != ""
}

func overrideExecArgs(execConfig *clientcmdapi.ExecConfig) {
	if execConfig.Command != "kubelogin" {
		return
	}
	args := execConfig.Args
	for idx, arg := range execConfig.Args {
		if arg == "workloadidentity" {
			args[idx] = "azurecli"
		}
	}
	execConfig.Args = args
}

func (p *azure) GetKubeClient(ctx context.Context) (client.Client, error) {
	opts := client.Options{
		Scheme: scheme.Scheme,
	}
	data := p.appbox.Kubernetes.GetKubeconfig()
	var kubeconfig []byte
	kubeconfig, err := base64.StdEncoding.DecodeString(data)
	if err != nil {
		return nil, err
	}

	if !aksWorkloadIdentityEnabled() {
		// override when running without workload identity
		cfg, err := clientcmd.Load(kubeconfig)
		if err != nil {
			return nil, err
		}

		cc := cfg.CurrentContext
		var user string
		for name, ctx := range cfg.Contexts {
			if name == cc {
				user = ctx.AuthInfo
			}
		}
		for name, auth := range cfg.AuthInfos {
			if name == user {
				if auth.Exec != nil {
					overrideExecArgs(auth.Exec)
					kubeconfig, err = clientcmd.Write(*cfg)
					if err != nil {
						return nil, err
					}
				}
			}
		}
	}

	restcfg, err := clientcmd.RESTConfigFromKubeConfig(kubeconfig)
	if err != nil {
		return nil, err
	}
	return client.New(restcfg, opts)
}

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

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

func (p *azure) GetKubernetesID(ctx context.Context) (string, error) {
	az := p.appbox.Kubernetes.GetAzure()
	if az == nil {
		return "", fmt.Errorf("can't get azure info from appbox")
	}
	kid := fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.ContainerService/managedClusters/%s", p.azSubscriptionID, az.ResourceGroup, p.appbox.Kubernetes.Name)
	return kid, nil
}

func (p *azure) GetNodeZones(ctx context.Context) ([]string, error) {
	return p.appbox.Kubernetes.GetAzure().NodeZones, nil
}

func (p *azure) GetNodeSubnets(ctx context.Context) ([]string, error) {
	subnets := p.appbox.Kubernetes.GetAzure().GetNodeSubnets()
	nets := make([]string, len(subnets))
	for idx, subnet := range subnets {
		nets[idx] = subnet.Id
	}
	return nets, nil
}

func (p *azure) GetPodSubnets(ctx context.Context) ([]string, error) {
	subnets := p.appbox.Kubernetes.GetAzure().GetPodSubnets()
	nets := make([]string, len(subnets))
	for idx, subnet := range subnets {
		nets[idx] = subnet.Id
	}
	return nets, nil
}

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

// BuildSinkEndpoint builds a private endpoint for Azure using crossplane
func (p *azure) BuildSinkEndpoint(providerConfig string,
	name types.NamespacedName,
	endpoint *EndpointRequest) ([]client.Object, error) {
	azureEndpoint := &netv1beta1.PrivateEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
		Spec: netv1beta1.PrivateEndpointSpec{
			ForProvider: netv1beta1.PrivateEndpointParameters{
				Location:          ptr.To(endpoint.Region),
				ResourceGroupName: ptr.To(p.appbox.Kubernetes.GetAzure().ResourceGroup),
				SubnetID:          ptr.To(p.appbox.Kubernetes.GetAzure().NodeSubnets[0].Id),
				Tags:              util.ToMapPointer(endpoint.Tags),
				PrivateServiceConnection: []netv1beta1.PrivateServiceConnectionParameters{
					{
						IsManualConnection:             ptr.To(true),
						Name:                           ptr.To(endpoint.CloudEndpointName),
						PrivateConnectionResourceAlias: ptr.To(endpoint.EndpointServiceName),
						SubresourceNames:               p.getSubresourceNames(endpoint.EndpointServiceName),
						RequestMessage:                 p.getRequestMessage(endpoint.EndpointServiceName),
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfig,
				},
			},
		},
	}
	return []client.Object{azureEndpoint}, nil
}

// DeleteSinkEndpoint deletes a private endpoint for Azure using crossplane
func (p *azure) DeleteSinkEndpoint(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	endpoint := netv1beta1.PrivateEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}
	return k8s.DeleteObject(ctx, cli, &endpoint)
}

// GetSinkEndpointDetail gets the private endpoint detail for Azure using crossplane
func (p *azure) GetSinkEndpointDetail(ctx context.Context, c client.Client, name types.NamespacedName) (*SinkEndpoint, error) {
	pe := netv1beta1.PrivateEndpoint{}
	if err := c.Get(ctx, name, &pe); err != nil {
		return nil, err
	}
	detail := &SinkEndpoint{}
	var endpointName string
	if pe.Status.AtProvider.ID != nil {
		detail.EndpointID = *pe.Status.AtProvider.ID
	}
	if len(pe.Status.AtProvider.PrivateServiceConnection) > 0 {
		if pe.Status.AtProvider.PrivateServiceConnection[0].PrivateIPAddress != nil {
			detail.Address = *pe.Status.AtProvider.PrivateServiceConnection[0].PrivateIPAddress
		}
		if pe.Status.AtProvider.PrivateServiceConnection[0].Name != nil {
			endpointName = *pe.Status.AtProvider.PrivateServiceConnection[0].Name
		}
	}
	if isResourceReady(pe.Status.ResourceStatus) && detail.EndpointID != "" &&
		detail.Address != "" {
		resp, err := p.privateEndpointsClient.Get(ctx, p.appbox.Kubernetes.GetAzure().ResourceGroup, endpointName, nil)
		if err != nil {
			return nil, err
		}
		if resp.Properties != nil || len(resp.Properties.ManualPrivateLinkServiceConnections) != 0 ||
			resp.Properties.ManualPrivateLinkServiceConnections[0].Properties != nil ||
			resp.Properties.ManualPrivateLinkServiceConnections[0].Properties.PrivateLinkServiceConnectionState != nil {
			detail.Phase = azurePSCStatus2SinkEndpointPhase(*resp.Properties.ManualPrivateLinkServiceConnections[0].Properties.PrivateLinkServiceConnectionState.Status)
		}
	}
	return detail, nil
}

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

func (p *azure) BuildDNS(providerConfig string, dns *DNSRequest) ([]client.Object, error) {
	return []client.Object{
		// add dns zone
		&netv1beta1.PrivateDNSZone{
			ObjectMeta: metav1.ObjectMeta{
				Annotations: map[string]string{
					// the external name is the domain name on azure, we specify it here manually to avoid the crossplane auto generated name
					crossplaneExternalName: dns.RootDomainName,
				},
				Name: dns.Name,
			},
			Spec: netv1beta1.PrivateDNSZoneSpec{
				ForProvider: netv1beta1.PrivateDNSZoneParameters{
					ResourceGroupName: ptr.To(p.appbox.Kubernetes.GetAzure().ResourceGroup),
					Tags:              util.ToMapPointer(dns.Tags),
				},
				ResourceSpec: xpv1.ResourceSpec{
					ProviderConfigReference: &xpv1.Reference{
						Name: providerConfig,
					},
				},
			},
		},
		// link dns zone to vnet
		&netv1beta1.PrivateDNSZoneVirtualNetworkLink{
			ObjectMeta: metav1.ObjectMeta{
				Name: dns.Name,
			},
			Spec: netv1beta1.PrivateDNSZoneVirtualNetworkLinkSpec{
				ForProvider: netv1beta1.PrivateDNSZoneVirtualNetworkLinkParameters{
					ResourceGroupName: ptr.To(p.appbox.Kubernetes.GetAzure().ResourceGroup),
					VirtualNetworkID: ptr.To(fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Network/virtualNetworks/%s",
						p.azSubscriptionID,
						p.appbox.Kubernetes.GetAzure().ResourceGroup,
						p.appbox.Kubernetes.Vpc)),
					PrivateDNSZoneName: ptr.To(dns.RootDomainName),
					Tags:               util.ToMapPointer(dns.Tags),
				},
				ResourceSpec: xpv1.ResourceSpec{
					ProviderConfigReference: &xpv1.Reference{
						Name: providerConfig,
					},
				},
			},
		},
		// add one A record
		&netv1beta1.PrivateDNSARecord{
			ObjectMeta: metav1.ObjectMeta{
				Annotations: map[string]string{
					crossplaneExternalName: "*",
				},
				// the record name is not unique in the same zone, so we use the same name as the zone
				Name: dns.Name,
			},
			Spec: netv1beta1.PrivateDNSARecordSpec{
				ForProvider: netv1beta1.PrivateDNSARecordParameters{
					TTL: ptr.To(float64(3600)),
					Records: []*string{
						ptr.To(dns.Endpoint.Address),
					},
					ZoneName:          ptr.To(dns.RootDomainName),
					ResourceGroupName: ptr.To(p.appbox.Kubernetes.GetAzure().ResourceGroup),
				},
				ResourceSpec: xpv1.ResourceSpec{
					ProviderConfigReference: &xpv1.Reference{
						Name: providerConfig,
					},
				},
			},
		},
	}, nil
}

// DeleteDNS deletes a private dns zone for Azure using crossplane
// PrivateZone,  PrivateDNSZoneVirtualNetworkLink and PrivateDNSARecord use the same name
func (p *azure) DeleteDNS(ctx context.Context, cli client.Client, name string) error {
	if err := k8s.DeleteObject(ctx, cli, &netv1beta1.PrivateDNSARecord{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete A record failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &netv1beta1.PrivateDNSZoneVirtualNetworkLink{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete vnet link failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &netv1beta1.PrivateDNSZone{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete private dns zone failed")
	}
	return nil
}

type azureGCProvider struct {
	cli client.Client
}

func NewAzureGCProvider(cli client.Client) GCProvider {
	return &azureGCProvider{
		cli: cli,
	}
}

func (a *azureGCProvider) ListAppbox(ctx context.Context) ([]string, error) {
	var ngs v1beta1.KubernetesClusterNodePoolList
	listOptions, err := resourceWithLabel(consts.AppboxIDLabel)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if err := a.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, ok := ng.Labels[consts.AppboxIDLabel]
			if !ok {
				log.Warn("no appbox ID label found", zap.Any("node group", ng))
				continue
			}
			appboxIDs[appboxID] = struct{}{}
		}
	}
	return pie.Keys(appboxIDs), nil
}

func (a *azureGCProvider) OnAppboxDeleting(ctx context.Context, appboxId string) error {
	var ngs v1beta1.KubernetesClusterNodePoolList
	if err := a.cli.List(ctx, &ngs, client.MatchingLabels{
		consts.AppboxIDLabel: appboxId,
	}); err != nil {
		return errors.Trace(err)
	}
	for _, ng := range ngs.Items {
		if err := k8s.DeleteObject(ctx, a.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
}

// getSubresourceNames converts subresource name to pointer array format
// Returns nil if subresourceName is empty, as Private Link Service connections
// via alias typically don't require subresource specification
func (p *azure) getSubresourceNames(serviceName string) []*string {
	serviceLower := strings.ToLower(serviceName)
	// if mysql resource return mysqlServer
	// Example: /subscriptions/.../providers/Microsoft.DBforMySQL/servers/...
	if strings.Contains(serviceLower, "microsoft.dbformysql") ||
		strings.Contains(serviceLower, "/mysqlservers/") {
		return []*string{ptr.To("mysqlServer")}
	}

	// if storage account resource return blob
	// Example: /subscriptions/.../providers/Microsoft.Storage/storageAccounts/...
	if strings.Contains(serviceLower, "microsoft.storage/storageaccounts") {
		return []*string{ptr.To("blob")}
	}

	return nil
}

// getRequestMessage returns the request message based on service type
func (p *azure) getRequestMessage(serviceName string) *string {
	serviceLower := strings.ToLower(serviceName)

	// MySQL resource
	if strings.Contains(serviceLower, "microsoft.dbformysql") ||
		strings.Contains(serviceLower, "/mysqlservers/") {
		return ptr.To("approve the connection to use dm on tidbcloud")
	}

	// Storage account resource
	if strings.Contains(serviceLower, "microsoft.storage/storageaccounts") {
		return ptr.To("approve the connection to use import on tidbcloud")
	}

	// Default for CDC Private Link Service
	return ptr.To("approve the connection to use changefeed on tidbcloud")
}

const (
	connectionApproved = "approved"
	connectionRejected = "rejected"
	connectionPending  = "pending"
	connectionRemoved  = "removed"
)

func azurePSCStatus2SinkEndpointPhase(status string) cdcv1.SinkEndpointPhase {
	switch strings.ToLower(status) {
	case connectionRejected:
		return cdcv1.EndpointResourceStateRejected
	case connectionApproved:
		return cdcv1.EndpointResourceStateEndpointAvailable
	case connectionPending:
		return cdcv1.EndpointResourceStatePendingAcceptance
	case connectionRemoved:
		return cdcv1.EndpointResourceStateDeleted
	default:
		return cdcv1.EndpointResourceStateFailed
	}
}