package privateendpoints

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/go-logr/logr"

	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	azerrors "github.com/Azure/azure-sdk-for-go-extensions/pkg/errors"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/metrics"
	"github.com/tidbcloud/observability/pkg/common/namer"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/azure"
)

// PrivateEndpointReconciler reconciles a PrivateEndpoint object
type PrivateEndpointReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

const (
	privateEndpointfinalizerName = "observability.pingcap.com/private-endpoint"
	errNotPrivateEndpoint        = "not PrivateEndpoint resource"
	pollInterval                 = 5 * time.Second
)

func NewPrivateEndpointReconciler(mgr ctrl.Manager, log logr.Logger, serviceInfo structs.ServiceInfo) *PrivateEndpointReconciler {
	return &PrivateEndpointReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: privateLinkEndpointConnector{
				cli:         mgr.GetClient(),
				serviceInfo: serviceInfo,
			},
			NewResource: func() client.Object {
				return &observabilitypingcapcomv1.PrivateEndpoint{}
			},
		},
	}
}

type privateLinkEndpointConnector struct {
	cli         client.Client
	serviceInfo structs.ServiceInfo
}

func (p privateLinkEndpointConnector) getAWSClient() (providers.PrivateEndpointClient, error) {
	return providers.GetAWSPrivateEndpointClient(), nil
}
func (p privateLinkEndpointConnector) getGCPClient() (providers.PSCClient, error) {
	return providers.GetGCPPSCClient(), nil
}

func (p privateLinkEndpointConnector) getAzureClient(pe *observabilitypingcapcomv1.PrivateEndpoint) (providers.PLSEndpointClient, error) {
	cred, err := azidentity.NewDefaultAzureCredential(nil)
	if err != nil {
		return nil, fmt.Errorf("failed to create default Azure credentials: %w", err)
	}
	return azure.NewPrivateEndpointClient(pe.Spec.Azure.SubscriptionID, pe.Spec.Azure.ResourceGroup, cred)
}

func (p privateLinkEndpointConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	peCli := &PrivateEndpoint{
		cli:         p.cli,
		serviceInfo: p.serviceInfo,
	}

	pe, ok := obj.(*observabilitypingcapcomv1.PrivateEndpoint)
	if !ok {
		return nil, errors.New(errNotPrivateEndpoint)
	}
	switch p.serviceInfo.Vendor {
	case string(constants.AWS):
		awsclient, err := p.getAWSClient()
		if err != nil {
			return nil, err
		}
		peCli.awscli = awsclient
	case string(constants.GCP):
		gcpclient, err := p.getGCPClient()
		if err != nil {
			return nil, err
		}
		peCli.gcpcli = gcpclient
	case string(constants.Azure):
		if pe.Spec.Azure == nil {
			return nil, fmt.Errorf("missing azure meta")
		}
		azurecli, err := p.getAzureClient(pe)
		if err != nil {
			return nil, err
		}
		peCli.azurecli = azurecli
	default:
		return nil, fmt.Errorf("not supported provider")
	}
	return peCli, nil
}

type PrivateEndpoint struct {
	awscli      providers.PrivateEndpointClient
	gcpcli      providers.PSCClient
	azurecli    providers.PLSEndpointClient
	cli         client.Client
	serviceInfo structs.ServiceInfo
}

func (p *PrivateEndpoint) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	log := log.FromContext(ctx).WithName("Observe")
	pe, ok := obj.(*observabilitypingcapcomv1.PrivateEndpoint)
	if !ok {
		p.reconcilerRepoter(obj.GetName(), "", errNotPrivateEndpoint)
		return reconciler.ExternalObservation{Droped: true}, errors.New(errNotPrivateEndpoint)
	}
	log.V(1).Info("PrivateEndpoint crd info", "crd", fmt.Sprintf("%v", pe))
	if pe.Status.Phase == constants.CREATION_FAILED {
		// final state, nothing to do
		p.reconcilerRepoter(pe.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}

	if pe.Status.EndpointID == "" {
		//if endpointID is empty, it means the private endpoint should be created
		return reconciler.ExternalObservation{ResourceExists: false}, nil
	}
	if pe.Status.Phase == constants.ACTIVE {
		// we should update the state, but now we just return the state
		p.reconcilerRepoter(pe.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}
	// exists but Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (p *PrivateEndpoint) waitPrivateEndpointService(ctx context.Context, pe *observabilitypingcapcomv1.PrivateEndpoint) (bool, error) {
	// if not check private endpoint service, we should return true
	if pe.Spec.PrivateEndpointServiceName == "" {
		return true, nil
	}
	// check private endpoint service is ready or not
	// get private endpoint service
	var svc observabilitypingcapcomv1.EndpointService
	err := p.cli.Get(ctx, client.ObjectKey{Namespace: pe.Namespace, Name: pe.Spec.PrivateEndpointServiceName}, &svc)
	if err != nil {
		return false, err
	}
	if svc.Status.Phase != constants.ACTIVE {
		return false, nil
	}
	return true, nil
}

func (p *PrivateEndpoint) CreateAWS(ctx context.Context, pe *observabilitypingcapcomv1.PrivateEndpoint) error {
	var awscli providers.PrivateEndpointClient
	if pe.Spec.AWS.AccountID != "" {
		// shloud assume role and create role in this account
		bizType := pe.GetLabels()[constants.LabelBizType]
		if _, ok := p.serviceInfo.BizTypeConfigs[bizType]; !ok {
			return fmt.Errorf("not supported biz type %s", bizType)
		}
		managerole := fmt.Sprintf("arn:aws:iam::%s:role/%s", pe.Spec.AWS.AccountID, p.serviceInfo.BizTypeConfigs[bizType].DataplaneRole)
		awscli = p.awscli.AssumeRole(ctx, managerole)
	} else {
		awscli = p.awscli
	}
	// describe vpc endpoint by name
	//for merge
	endpointName := pe.GetEndpointName()
	vpcEndpoint, err := awscli.DescribeVpcEndpointsByName(ctx, endpointName)
	if err != nil {
		return err
	}
	if len(vpcEndpoint.VpcEndpoints) > 0 {
		// if the vpc endpoint exists, we should update the state
		pe.Status.EndpointID = *vpcEndpoint.VpcEndpoints[0].VpcEndpointId
		pe.Status.Phase = constants.ACTIVE
		if err := p.cli.Status().Update(ctx, pe); err != nil {
			return err
		}
		return nil
	}
	// get vpc endpoint service
	endpointServiceOut, err := awscli.DescribeVpcEndpointServices(ctx, []string{pe.Spec.AWS.ServiceName})
	if err != nil {
		return fmt.Errorf("failed to describe endpoint services: %v", err)
	}
	if len(endpointServiceOut.ServiceDetails) != 1 || len(endpointServiceOut.ServiceDetails[0].AvailabilityZones) == 0 {
		return fmt.Errorf("no common az between endpoint service and current dataplane: %v", err)
	}
	epSvcAZ := endpointServiceOut.ServiceDetails[0].AvailabilityZones[0]

	// Get the node subnet that is in the same availability zone as the endpoint service
	subnetOut, err := awscli.DescribeSubnets(ctx, pe.Spec.AWS.SubnetIDs)
	if err != nil {
		return fmt.Errorf("failed to describe node subnets: %v", err)
	}
	pickedSubnet := findSubnetIdsByZones(subnetOut.Subnets, []string{epSvcAZ})
	// create vpc endpoint
	vpcEndpointOut, err := awscli.CreateVpcEndpoint(ctx, endpointName, pe.Spec.AWS.VPCID, pe.Spec.AWS.ServiceName, pickedSubnet, string(pe.Spec.AWS.VPCEndpointType), pe.Spec.AWS.SecurityGroupIDs, pe.Spec.AWS.TagSpecifications)
	if err != nil {
		return fmt.Errorf("failed to create vpc endpoint: %v", err)
	}
	pe.Status.EndpointID = *vpcEndpointOut.VpcEndpoint.VpcEndpointId
	pe.Status.Phase = constants.ACTIVE
	if err := p.cli.Status().Update(ctx, pe); err != nil {
		return err
	}
	return nil
}

func (p *PrivateEndpoint) CreateGCP(ctx context.Context, pe *observabilitypingcapcomv1.PrivateEndpoint) error {
	var gcpcli providers.PSCClient
	if pe.Spec.GCP.ProjectID != "" {
		gcpcli = p.gcpcli.AssumeProject(ctx, pe.Spec.GCP.ProjectID)
	} else {
		gcpcli = p.gcpcli
	}
	endpointServiceName := pe.Spec.GCP.ServiceName
	serviceDnsName := pe.Spec.GCP.ServiceDNSName
	region := pe.Spec.GCP.Region
	subnet := pe.Spec.GCP.Subnet

	network := pe.Spec.GCP.Network
	//for merge
	name := pe.GetEndpointName()
	namespace := namer.DomainNamespace(serviceDnsName)

	// Compatibility issue with projects migrated from cloud-native arch
	// In cloud-native arch, the network associates to the servicedirectory namespace with the name format of `namer.DomainNamespace(serviceDnsName)`
	// In 3-layers arch, the network associates to the servicedirectory namespace with the name format of `namer.DomainNamespace(serviceDnsName)-{env}`
	// If we create a new endpoint with a network that associates to another namespace, the endpoint dns name could not be resolved.
	associated, err := gcpcli.CheckNamespaceNetworkAssociation(ctx, region, network, namespace)
	if err != nil {
		return fmt.Errorf("check if network %s associated with namespace %s: %w", network, namespace, err)
	}
	if !associated {
		namespace = fmt.Sprintf("%s-%s", namer.DomainNamespace(serviceDnsName), p.serviceInfo.Env)
	}
	tags := pe.Labels

	//describe psc endpoint by name
	pscEndpoint, err := gcpcli.GetPublishedServicePscEndpoint(ctx, pe.Spec.GCP.Region, name)
	if err != nil {
		return err
	}
	if pscEndpoint != nil {
		pe.Status.EndpointID = *pscEndpoint.Name
		pe.Status.Phase = constants.ACTIVE
		if err := p.cli.Status().Update(ctx, pe); err != nil {
			return err
		}
		return nil
	}
	//create psc endpoint
	err = gcpcli.CreatePublishedServicePscEndpoint(ctx, region, network, subnet, endpointServiceName, name, namespace, tags)
	if err != nil {
		return fmt.Errorf("failed to create psc endpoint: %v", err)
	}
	pe.Status.EndpointID = name
	pe.Status.Phase = constants.ACTIVE
	if err := p.cli.Status().Update(ctx, pe); err != nil {
		return err
	}
	return nil
}

func (p *PrivateEndpoint) CreateAzure(ctx context.Context, pe *observabilitypingcapcomv1.PrivateEndpoint) error {
	endpointServiceName := pe.Spec.Azure.ServiceName
	subnet := pe.Spec.Azure.Subnet
	network := pe.Spec.Azure.Network

	//for merge
	name := pe.GetEndpointName()
	tags := pe.Labels

	//describe psc endpoint by name
	plsEndpoint, err := p.azurecli.Get(ctx, name)
	if err != nil && !azerrors.IsNotFoundErr(err) {
		return err
	}
	if plsEndpoint != nil {
		pe.Status.EndpointID = *plsEndpoint.Name
		pe.Status.Phase = constants.ACTIVE
		ip, err := p.azurecli.GetPrivateIP(ctx, name)
		if err != nil {
			return err
		}
		if ip == nil {
			return fmt.Errorf("private IP is not ready")
		}
		pe.Status.PrivateIP = *ip
		if err := p.cli.Status().Update(ctx, pe); err != nil {
			return err
		}
		return nil
	}
	//create pls endpoint
	_, err = p.azurecli.Create(ctx, name, network, subnet, endpointServiceName, pe.Spec.Azure.Region, tags)
	if err != nil {
		return fmt.Errorf("failed to create azure private endpoint: %w", err)
	}
	ip, err := p.azurecli.GetPrivateIP(ctx, name)
	if err != nil {
		return err
	}
	if ip == nil {
		return fmt.Errorf("private IP is not ready")
	}

	pe.Status.EndpointID = name
	pe.Status.PrivateIP = *ip
	pe.Status.Phase = constants.ACTIVE
	return p.cli.Status().Update(ctx, pe)
}

func (p *PrivateEndpoint) Create(ctx context.Context, obj client.Object) (errResult error) {
	pe, ok := obj.(*observabilitypingcapcomv1.PrivateEndpoint)
	if !ok {
		p.reconcilerRepoter(obj.GetName(), "", errNotPrivateEndpoint)
		return errors.New(errNotPrivateEndpoint)
	}
	//wait
	ready, err := p.waitPrivateEndpointService(ctx, pe)
	if err != nil {
		p.reconcilerRepoter(pe.Name, "", err.Error())
		return err
	}
	if !ready {
		return errors.New("private endpoint service not ready")
	}
	//update state and then create privatelink endpoint
	pe.Status.Phase = constants.CREATING
	if err := p.cli.Status().Update(ctx, pe); err != nil {
		p.reconcilerRepoter(pe.Name, "", err.Error())
		return err
	}
	switch p.serviceInfo.Vendor {
	case string(constants.AWS):
		err = p.CreateAWS(ctx, pe)
	case string(constants.GCP):
		err = p.CreateGCP(ctx, pe)
	case string(constants.Azure):
		err = p.CreateAzure(ctx, pe)
	}
	if err != nil {
		p.reconcilerRepoter(pe.Name, "", err.Error())
		return err
	}
	p.reconcilerRepoter(pe.Name, "", "")
	return nil
}
func (p *PrivateEndpoint) DeleteAWS(ctx context.Context, phase string, pe *observabilitypingcapcomv1.PrivateEndpoint) error {
	var awscli providers.PrivateEndpointClient
	if pe.Spec.AWS.AccountID != "" {
		// shloud assume role and create role in this account
		bizType := pe.GetLabels()[constants.LabelBizType]
		if _, ok := p.serviceInfo.BizTypeConfigs[bizType]; !ok {
			return fmt.Errorf("not supported biz type %s", bizType)
		}
		managerole := fmt.Sprintf("arn:aws:iam::%s:role/%s", pe.Spec.AWS.AccountID, p.serviceInfo.BizTypeConfigs[bizType].DataplaneRole)
		awscli = p.awscli.AssumeRole(ctx, managerole)
	} else {
		awscli = p.awscli
	}
	// delete vpc endpoint
	if phase == "" && pe.Status.EndpointID == "" {
		//if phase and endpointID is empty, it means the private endpoint not be created
		return nil
	}
	if pe.Status.EndpointID != "" {
		_, err := awscli.DeleteVpcEndpoints(ctx, []string{pe.Status.EndpointID})
		if err != nil {
			return err
		}
		return nil
	}
	// describe vpc endpoint by name and delete it
	//for merge
	endpointName := pe.GetEndpointName()
	vpcEndpoint, err := awscli.DescribeVpcEndpointsByName(ctx, endpointName)
	if err != nil {
		return err
	}
	if len(vpcEndpoint.VpcEndpoints) > 0 {
		// if the vpc endpoint exists, we should delete it
		endpointID := *vpcEndpoint.VpcEndpoints[0].VpcEndpointId
		_, err := awscli.DeleteVpcEndpoints(ctx, []string{endpointID})
		if err != nil {
			return err
		}
	}
	return nil
}
func (iam *PrivateEndpoint) DeleteGCP(ctx context.Context, phase string, pe *observabilitypingcapcomv1.PrivateEndpoint) error {
	var gcpcli providers.PSCClient
	if pe.Spec.GCP.ProjectID != "" {
		gcpcli = iam.gcpcli.AssumeProject(ctx, pe.Spec.GCP.ProjectID)
	} else {
		gcpcli = iam.gcpcli
	}
	// delete vpc endpoint
	if phase == "" && pe.Status.EndpointID == "" {
		//if phase and endpointID is empty, it means the private endpoint not be created
		return nil
	}
	if pe.Status.EndpointID != "" {
		err := gcpcli.DeletePublishedServicePscEndpoint(ctx, pe.Spec.GCP.Region, pe.Status.EndpointID)
		if err != nil {
			return err
		}
		return nil
	}
	// describe psc endpoint by name and delete it
	//for merge
	name := pe.GetEndpointName()
	pscEndpoint, err := gcpcli.GetPublishedServicePscEndpoint(ctx, pe.Spec.GCP.Region, name)
	if err != nil {
		return err
	}
	if pscEndpoint != nil {
		// if the psc endpoint exists, we should delete it
		err := gcpcli.DeletePublishedServicePscEndpoint(ctx, pe.Spec.GCP.Region, name)
		if err != nil {
			return err
		}
	}
	return nil
}

func (p *PrivateEndpoint) DeleteAzure(ctx context.Context, phase string, pe *observabilitypingcapcomv1.PrivateEndpoint) error {
	// delete vpc endpoint
	if phase == "" && pe.Status.EndpointID == "" {
		//if phase and endpointID is empty, it means the private endpoint not be created
		return nil
	}
	return p.azurecli.Delete(ctx, pe.Status.EndpointID)
}

func (p *PrivateEndpoint) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	//private endpoint delete path
	pe, ok := obj.(*observabilitypingcapcomv1.PrivateEndpoint)
	if !ok {
		p.reconcilerRepoter(obj.GetName(), "", errNotPrivateEndpoint)
		return reconciler.ExternalDeletion{}, errors.New(errNotPrivateEndpoint)
	}
	oldphase := pe.Status.Phase
	pe.Status.Phase = constants.DELETING
	if err := p.cli.Status().Update(ctx, pe); err != nil {
		return reconciler.ExternalDeletion{}, err
	}
	var err error
	switch p.serviceInfo.Vendor {
	case string(constants.AWS):
		err = p.DeleteAWS(ctx, string(oldphase), pe)
	case string(constants.GCP):
		err = p.DeleteGCP(ctx, string(oldphase), pe)
	case string(constants.Azure):
		err = p.DeleteAzure(ctx, string(oldphase), pe)
	}
	if err != nil {
		p.reconcilerRepoter(pe.Name, "", err.Error())
		return reconciler.ExternalDeletion{}, err
	}
	p.reconcilerRepoter(pe.Name, "", "")
	return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
}

func (p *PrivateEndpoint) Update(ctx context.Context, obj client.Object) (errResult error) {
	pe, ok := obj.(*observabilitypingcapcomv1.PrivateEndpoint)
	if !ok {
		p.reconcilerRepoter(obj.GetName(), "", errNotPrivateEndpoint)
		return errors.New(errNotPrivateEndpoint)
	}
	if pe.Status.Phase != constants.ACTIVE {
		pe.Status.Phase = constants.ACTIVE
		if err := p.cli.Status().Update(ctx, pe); err != nil {
			return err
		}
	}
	return nil
}
func (p *PrivateEndpoint) reconcilerRepoter(name string, interval string, err string) {
	metrics.HandleRegionalControllerRunsMetrics(p.serviceInfo.Vendor, p.serviceInfo.Region, "PrivateEndpointController", name, interval, err)
}

// FindSubnetIdsByZones returns the subnet IDs for the given zones.
func findSubnetIdsByZones(subnets []ec2types.Subnet, zones []string) []string {
	var exceptSubnetIds []string
	for _, subnet := range subnets {
		for _, zone := range zones {
			if *subnet.AvailabilityZone == zone {
				exceptSubnetIds = append(exceptSubnetIds, *subnet.SubnetId)
			}
		}
	}
	return exceptSubnetIds
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the IAM object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.2/pkg/reconcile
func (r *PrivateEndpointReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := r.Log.WithValues("PrivateEndpoint", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)
	return r.reconciler.Reconcile(ctx, req, privateEndpointfinalizerName)
}

// SetupWithManager sets up the controller with the Manager.
func (r *PrivateEndpointReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&observabilitypingcapcomv1.PrivateEndpoint{}).
		Complete(r)
}