package gardener

import (
	"context"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	iamv1alpha1 "github.com/crossplane/provider-gcp/apis/iam/v1alpha1"
	"github.com/pingcap/errors"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/utils/pointer"
)

func init() {
	utilruntime.Must(iamv1alpha1.SchemeBuilder.AddToScheme(scheme.Scheme))
}

func (r *NetworkReconciler) ensureIAM(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network, info *contextInfo) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureIAM")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	switch provider.Spec.Type {
	case consts.ProviderTypeAWS, consts.ProviderTypeAWSFreeTier,
		consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
	case consts.ProviderTypeGCP:
		if err := r.ensureGCPServiceAccount(spanCtx, provider, network, info); err != nil {
			return err
		}
		return r.ensureGCPServiceAccountKey(spanCtx, provider, network, info)
	default:
		return errors.Errorf("unknown provider type: %s", provider.Spec.Type)
	}
	return nil
}

func (r *NetworkReconciler) ensureGCPServiceAccount(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network, info *contextInfo) (errResult error) {
	ctx, span := tracer.Start(ctx, "ensureGCPServiceAccount")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return err
	}
	saName := getGCPAccessServiceAccountName(network.Spec.Region, network.GetProjectId())
	sa := &iamv1alpha1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: network.Namespace,
			Name:      saName,
		},
		Spec: iamv1alpha1.ServiceAccountSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: spec.GCP.CrossplaneProvider},
				WriteConnectionSecretToReference: &xpv1.SecretReference{
					Namespace: network.Namespace,
					Name:      saName,
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountParameters{
				DisplayName: pointer.String(saName),
				Description: pointer.String("The service account that import and backup jobs will use to access customer's data."),
			},
		},
	}
	_, err = k8s.CreateIgnoreExists(ctx, r.Client, sa)
	if err != nil {
		return err
	}

	saNamespacedName := types.NamespacedName{Namespace: network.Namespace, Name: saName}
	var created iamv1alpha1.ServiceAccount
	if err = r.Client.Get(ctx, saNamespacedName, &created); err != nil {
		return k8s.RequeueIf(err, apierrors.IsNotFound, defaultRequeueAfter)
	}
	var ready bool
	for _, cond := range created.Status.Conditions {
		if cond.Status == corev1.ConditionTrue && cond.Type == xpv1.TypeReady {
			ready = true
			break
		}
	}
	if !ready {
		// TODO(tangent): Re-enable after we figure out the permission issue.
		// return fmt.Errorf("service account is not ready yet")
		return nil
	}
	if created.Status.AtProvider.Email != "" {
		network.Status.IAM.ServiceAccountEmail = created.Status.AtProvider.Email
	} else {
		network.Status.IAM.ServiceAccountEmail = getGCPAccessServiceAccountEmail(saName, spec.GCP.ProjectID)
	}
	err = r.Status().Update(ctx, network)
	if err != nil {
		return err
	}

	// ref: https://cloud.google.com/iam/docs/impersonating-service-accounts#impersonate-sa-level
	// Make nodeSA can impersonate network.Status.IAM.ServiceAccountEmail
	shootName, err := shootName(network, provider.Spec.Type, network.GetProjectId())
	if err != nil {
		return err
	}
	nodeSA, err := getShootWorkerIamIdentity(consts.ProviderTypeGCP, r.env.ProjectName(), shootName, spec.GCP.ProjectID)
	if err != nil {
		return err
	}
	err = r.gcpapi.AddBidingToServiceAccount(
		ctx,
		spec.GCP.ProjectID,
		network.Status.IAM.ServiceAccountEmail,
		"roles/iam.serviceAccountTokenCreator",
		"serviceAccount:"+nodeSA,
	)
	if err != nil {
		return err
	}

	log.FromContext(ctx).Info("ensure GCP service account")
	return nil
}

func (r *NetworkReconciler) ensureGCPServiceAccountKey(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network, info *contextInfo) (errResult error) {
	ctx, span := tracer.Start(ctx, "ensureGCPServiceAccountKey")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return err
	}
	saName := getGCPAccessServiceAccountName(network.Spec.Region, network.GetProjectId())
	keyName := getGCPAccessServiceAccountKeyName(saName)
	key := &iamv1alpha1.ServiceAccountKey{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: network.Namespace,
			Name:      keyName,
		},
		Spec: iamv1alpha1.ServiceAccountKeySpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: spec.GCP.CrossplaneProvider},
				WriteConnectionSecretToReference: &xpv1.SecretReference{
					Namespace: network.Namespace,
					Name:      keyName,
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountKeyParameters{
				ServiceAccountReferer: iamv1alpha1.ServiceAccountReferer{
					ServiceAccountRef: &xpv1.Reference{
						Name: saName,
					},
				},
			},
		},
	}
	_, err = k8s.CreateIgnoreExists(ctx, r.Client, key)
	if err != nil {
		return err
	}

	keyNamespacedName := types.NamespacedName{Namespace: network.Namespace, Name: keyName}
	var created iamv1alpha1.ServiceAccountKey
	if err = r.Client.Get(ctx, keyNamespacedName, &created); err != nil {
		return k8s.RequeueIf(err, apierrors.IsNotFound, defaultRequeueAfter)
	}
	var ready bool
	for _, cond := range created.Status.Conditions {
		if cond.Status == corev1.ConditionTrue && cond.Type == xpv1.TypeReady {
			ready = true
			break
		}
	}
	if !ready {
		// TODO(tangent): Re-enable after we figure out the permission issue.
		// return fmt.Errorf("service account key is not ready yet")
		return nil
	}
	log.FromContext(ctx).Info("ensure GCP service account key")
	return nil
}

func (r *NetworkReconciler) cleanUpIAM(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network, info *contextInfo) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "cleanUpIAM")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	switch provider.Spec.Type {
	case consts.ProviderTypeAWS, consts.ProviderTypeAWSFreeTier,
		consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
	case consts.ProviderTypeGCP:
		// Crossplane provider does not clean up the service account key when deleting a service account.
		// So delete the service account key explicitly.
		saName := getGCPAccessServiceAccountName(network.Spec.Region, network.GetProjectId())
		keyName := getGCPAccessServiceAccountKeyName(saName)
		saKey := &iamv1alpha1.ServiceAccountKey{
			ObjectMeta: metav1.ObjectMeta{
				Name: keyName,
			},
		}
		err := r.Client.Delete(spanCtx, saKey)
		if client.IgnoreNotFound(err) != nil {
			return err
		}
		keyNamespacedName := types.NamespacedName{Namespace: network.Namespace, Name: keyName}
		var keyResource iamv1alpha1.ServiceAccountKey
		err = r.Client.Get(spanCtx, keyNamespacedName, &keyResource)
		if !apierrors.IsNotFound(err) {
			return errors.New("wait for the GCP access service acccount key to be deleted")
		}
		sa := &iamv1alpha1.ServiceAccount{
			ObjectMeta: metav1.ObjectMeta{
				Name: saName,
			},
		}
		err = r.Client.Delete(spanCtx, sa)
		if client.IgnoreNotFound(err) != nil {
			return err
		}
		saNamespacedName := types.NamespacedName{Namespace: network.Namespace, Name: saName}
		var saResource iamv1alpha1.ServiceAccount
		err = r.Client.Get(spanCtx, saNamespacedName, &saResource)
		if !apierrors.IsNotFound(err) {
			return errors.New("wait for the GCP access service acccount to be deleted")
		}
	default:
		return errors.Errorf("unknown provider type: %s", provider.Spec.Type)
	}
	return nil
}