package procedure

import (
	"context"
	"crypto/md5"
	"fmt"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/arn"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/ec2"
	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
	"github.com/aws/aws-sdk-go/service/elbv2"
	"github.com/aws/aws-sdk-go/service/elbv2/elbv2iface"
	"github.com/aws/aws-sdk-go/service/lambda"
	"github.com/aws/aws-sdk-go/service/lambda/lambdaiface"
	cpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	ec2v1alpha1 "github.com/crossplane/provider-aws/apis/ec2/v1alpha1"
	cpv1beta1 "github.com/crossplane/provider-aws/apis/eks/v1beta1"
	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	ProcedureConfigmap = "procedure-configmap"
	ProcedureNamespace = "procedure-system"
)

// getEKSObject returns the EKS cluster object for the connection.
func getEKSObject(ctx context.Context, cli client.Reader, conn *eksv1alpha1.EKSConnection) (*cpv1beta1.Cluster, error) {
	key := types.NamespacedName{
		Name: conn.Spec.Target.Name,
	}
	eks := &cpv1beta1.Cluster{}
	err := cli.Get(ctx, key, eks)
	if err != nil {
		return nil, err
	}
	return eks, nil
}

// GetProcedureInfo returns the cluster info which identifies the target EKS.
func GetProcedureInfo(ctx context.Context, cli client.Client, conn *eksv1alpha1.EKSConnection, namespace string) (ProcedureInfo, error) {
	eks, err := getEKSObject(ctx, cli, conn)
	if err != nil {
		return ProcedureInfo{}, err
	}
	if namespace == "" {
		namespace = ProcedureNamespace
	}
	if eks.Status.AtProvider.ResourcesVpcConfig.VpcID == "" {
		return ProcedureInfo{}, fmt.Errorf("eks cluster %s doesn't have vpc id", eks.Name)
	}
	return ProcedureInfo{
		Region:            aws.StringValue(eks.Spec.ForProvider.Region),
		Cluster:           eks.Name,
		ProviderConfigRef: conn.Spec.ProviderConfigRef,
		VpcID:             eks.Status.AtProvider.ResourcesVpcConfig.VpcID,
		SubnetIDs:         eks.Spec.ForProvider.ResourcesVpcConfig.SubnetIDs,
		SecurityGroupIDs:  eks.Spec.ForProvider.ResourcesVpcConfig.SecurityGroupIDs,
		eksConnection:     conn,
		namespace:         namespace,
		assumeRole:        conn.Spec.AssumeRole,
		externalID:        conn.Spec.ExternalID,
	}, nil
}

func stringHash(s string) string {
	return fmt.Sprintf("%x", md5.Sum([]byte(s)))
}

// ResourceKey returns the namespaced name for a procedure.
func ResourceKey(info ProcedureInfo, procedure string) types.NamespacedName {
	name := fmt.Sprintf("res-%s-%s-%s-%s", info.eksConnection.Name, info.Region, info.ProviderConfigRef, procedure)
	// During the joint tests we failed with the name longer than 63.
	if len(name) > 63 {
		name = fmt.Sprintf("res-%s-%s", procedure, stringHash(name))
	}
	return types.NamespacedName{
		// TODO: set dedicated namespace for namespaced resources
		Namespace: info.namespace,
		Name:      name,
	}
}

// getConfigKey returns the object key for the procedures configmap.
func getConfigKey(info ProcedureInfo) types.NamespacedName {
	return types.NamespacedName{
		Namespace: info.namespace,
		Name:      ProcedureConfigmap,
	}
}

// GetConfigWithKey returns the value for corresponding key in the procedure configmap.
func GetConfigWithKey(ctx context.Context, cli client.Reader, info ProcedureInfo, key string) (string, error) {
	cm := &corev1.ConfigMap{}
	err := cli.Get(ctx, getConfigKey(info), cm)
	if err != nil {
		return "", err
	}
	errNotExist := fmt.Errorf("key %q doesn't exist in procedure configmap", key)
	if cm.Data != nil {
		if value, exist := cm.Data[key]; !exist {
			return "", errNotExist
		} else {
			return value, nil
		}
	}
	return "", errNotExist
}

// SetConfigKeyValue sets key value pair in configmap.
func SetConfigWithKey(ctx context.Context, cli client.Client, info ProcedureInfo, key, value string) error {
	cm, objkey := &corev1.ConfigMap{}, getConfigKey(info)
	err := cli.Get(ctx, objkey, cm)
	if err != nil && !errors.IsNotFound(err) {
		return err
	}
	// create configmap if not existed
	if err != nil {
		cm = &corev1.ConfigMap{}
		cm.Name = objkey.Name
		cm.Namespace = objkey.Namespace
		cm.Data = map[string]string{key: value}
		err = cli.Create(ctx, cm)
		if err != nil {
			return err
		}
		return nil
	}

	if cm.Data == nil {
		cm.Data = make(map[string]string)
	}
	cm.Data[key] = value
	return cli.Update(ctx, cm)
}

// GetCrossplaneProvider returns the crossplane provider for the tenant.
func GetCrossplaneProvider(info ProcedureInfo) *cpv1.Reference {
	return &cpv1.Reference{
		Name: info.ProviderConfigRef,
	}
}

// IsCrossplaneObjReady returns true if the crossplane object is ready.
func IsCrossplaneObjReady(status cpv1.ConditionedStatus) bool {
	return status.GetCondition(cpv1.TypeReady).Status == corev1.ConditionTrue
}

// setOwnerRef adds EKSConnection to the object's owner reference list
func setOwnerRef(obj *metav1.ObjectMeta, info ProcedureInfo) {
	obj.OwnerReferences = append(obj.OwnerReferences, metav1.OwnerReference{
		APIVersion: info.eksConnection.APIVersion,
		Kind:       info.eksConnection.Kind,
		Name:       info.eksConnection.Name,
		UID:        info.eksConnection.UID,
		Controller: aws.Bool(true),
	})
}

// getEndpointServiceID returns the endpoint service ID for the EKS info.
func getEndpointServiceID(ctx context.Context, cli client.Reader, info ProcedureInfo, epSvcName string) (string, error) {
	epSvc := &ec2v1alpha1.VPCEndpointServiceConfiguration{}
	epSvcKey := ResourceKey(info, frpsEpSvcName)
	epSvcKey.Namespace = ""
	err := cli.Get(ctx, epSvcKey, epSvc)
	if err != nil {
		return "", err
	}
	// endpoint service ID in form "vpce-svc-07d8722e7e7887f5c"
	return epSvc.Annotations[crossplaneExternalName], nil
}

// ConditionEqual checks if the condition is equal to the given status except for the lastTransitionTime.
func ConditionEqual(cond1, cond2 eksv1alpha1.Condition) bool {
	return cond1.Type == cond2.Type &&
		cond1.Status == cond2.Status &&
		cond1.Reason == cond2.Reason &&
		cond1.Message == cond2.Message
}

// privateLinkMngr provides utility functions for managing private links.
type privateLinkMngr interface {
	getEndpointServiceName(ctx context.Context, cli client.Reader, info ProcedureInfo, epSvcName string) (string, error)
	getEndpointDNS(ctx context.Context, cli client.Reader, epName string) (string, error)
	acceptPrivateLink(ctx context.Context, endpointSvcID, endpointID string) error
}

// plMngr returns the private link manager for the EKS info.
type plMngr struct {
	aws AWSService
}

func (p *plMngr) getEndpointService(ctx context.Context, cli client.Reader, info ProcedureInfo, epSvcName string) (*ec2.ServiceConfiguration, error) {
	epSvcID, err := getEndpointServiceID(ctx, cli, info, epSvcName)
	if err != nil {
		return nil, err
	}
	epSvcs, err := p.aws.EC2API(nil).DescribeVpcEndpointServiceConfigurationsWithContext(ctx, &ec2.DescribeVpcEndpointServiceConfigurationsInput{
		ServiceIds: []*string{aws.String(epSvcID)},
	})
	if err != nil {
		return nil, err
	}
	if len(epSvcs.ServiceConfigurations) == 0 {
		return nil, fmt.Errorf("endpoint service %q not found", epSvcID)
	}
	return epSvcs.ServiceConfigurations[0], nil
}

func (p *plMngr) getEndpointServiceName(ctx context.Context, cli client.Reader, info ProcedureInfo, epSvcName string) (string, error) {
	epSvc, err := p.getEndpointService(ctx, cli, info, epSvcName)
	if err != nil {
		return "", err
	}
	return aws.StringValue(epSvc.ServiceName), nil
}

func (p *plMngr) getEndpointDNS(ctx context.Context, cli client.Reader, epName string) (string, error) {
	ep := &ec2v1alpha1.VPCEndpoint{}
	epKey := types.NamespacedName{
		Name: epName,
	}
	err := cli.Get(ctx, epKey, ep)
	if err != nil {
		return "", err
	}
	if len(ep.Status.AtProvider.VPCEndpoint.DNSEntries) == 0 {
		return "", fmt.Errorf("empty DNS entries of endpoint %q", epName)
	}
	// Assuming the first DNS entry is the one for all AZs.
	// It should in the form "vpce-0430449365e0683e9-w49plspa.vpce-svc-08d4bd8696f2bf724.us-west-2.vpce.amazonaws.com".
	// If it's AZ specific DNS entry (in the form "vpce-0430449365e0683e9-w49plspa-us-west-2a.vpce-svc-08d4bd8696f2bf724.us-west-2.vpce.amazonaws.com"),
	// then it would only be resolved to ENI in only one AZ.
	return aws.StringValue(ep.Status.AtProvider.VPCEndpoint.DNSEntries[0].DNSName), nil
}

func (p *plMngr) acceptPrivateLink(ctx context.Context, endpointSvcID, endpointID string) error {
	_, err := p.aws.EC2API(nil).AcceptVpcEndpointConnectionsWithContext(ctx, &ec2.AcceptVpcEndpointConnectionsInput{
		ServiceId:      aws.String(endpointSvcID),
		VpcEndpointIds: aws.StringSlice([]string{endpointID}),
	})
	return err
}

func NewPlMngr(aws AWSService) *plMngr {
	return &plMngr{aws: aws}
}

var _ privateLinkMngr = &plMngr{}

// getRootPrincipal returns the root principal for the given tenant role ARN.
func getRootPrincipal(arnStr string) (string, error) {
	parsedArn, err := arn.Parse(arnStr)
	if err != nil {
		return "", err
	}
	parsedArn.Resource = "root"
	return parsedArn.String(), nil
}

func (info *ProcedureInfo) getAssumeRole() *assumeRole {
	return &assumeRole{
		roleARN:    info.assumeRole,
		externalID: info.externalID,
	}
}

// assumeRole specifies the assumed role for issuing AWS requests.
type assumeRole struct {
	roleARN    string
	externalID string
}

func assumeRoleCred(sess *session.Session, role *assumeRole) *credentials.Credentials {
	return stscreds.NewCredentials(sess, role.roleARN, func(provider *stscreds.AssumeRoleProvider) {
		provider.ExternalID = aws.String(role.externalID)
	})
}

// AWSService returns the AWS service client.
type AWSService interface {
	EC2API(role *assumeRole) ec2iface.EC2API
	ELBV2API(role *assumeRole) elbv2iface.ELBV2API
	LambdaAPI(role *assumeRole) lambdaiface.LambdaAPI
}

type awsSvc struct {
	sess *session.Session
}

func (a *awsSvc) EC2API(role *assumeRole) ec2iface.EC2API {
	if role == nil {
		return ec2.New(a.sess)
	}
	return ec2.New(a.sess, &aws.Config{
		Credentials: assumeRoleCred(a.sess, role),
	})
}

func (a *awsSvc) ELBV2API(role *assumeRole) elbv2iface.ELBV2API {
	if role == nil {
		return elbv2.New(a.sess)
	}
	return elbv2.New(a.sess, &aws.Config{
		Credentials: assumeRoleCred(a.sess, role),
	})
}

func (a *awsSvc) LambdaAPI(role *assumeRole) lambdaiface.LambdaAPI {
	if role == nil {
		return lambda.New(a.sess)
	}
	return lambda.New(a.sess, &aws.Config{
		Credentials: assumeRoleCred(a.sess, role),
	})
}

func NewAWSService(sess *session.Session) *awsSvc {
	return &awsSvc{sess: sess}
}

var _ AWSService = &awsSvc{}

// getReader drops k8s client privilege and returns the reader.
func getReader(cli client.Client) client.Reader {
	return cli.(client.Reader)
}