package gke

import (
	"context"
	"fmt"

	iamv1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/iam/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

// GCPServiceAccount task create gcp sa for all node pool in gke cluster.
// https://doc.crds.dev/github.com/crossplane/provider-gcp/iam.gcp.crossplane.io/ServiceAccount/v1alpha1@v0.17.1
type GCPServiceAccount struct {
	*task.BaseTask
}

func NewGCPServiceAccount(ctx *task.Context) *GCPServiceAccount {
	return &GCPServiceAccount{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *GCPServiceAccount) Name() string {
	return "GCPServiceAccount"
}

func (t *GCPServiceAccount) Create(ctx context.Context) deploy.Result {
	if err := t.createNodeSA(ctx); err != nil {
		return deploy.Error(err)
	}

	// Unlike AWS did iam_role.go, we don't need to create a DB audit log writer role for GCP because we
	// always use the node SA to write logs wherever the bucket is managed by PingCAP or the customer.
	if t.Cluster.Status.SharedBucket == nil {
		t.Cluster.Status.SharedBucket = &cloudv1.SharedBucketStatus{}
	}
	if t.Cluster.Status.SharedBucket.GCP == nil {
		t.Cluster.Status.SharedBucket.GCP = &cloudv1.GCPSharedBucketStatus{}
	}
	if t.Cluster.SharedBucketEnabled() {
		if err := t.createSharedBucketReadonlySA(ctx); err != nil {
			return deploy.Error(err)
		}
	}

	return deploy.Done()
}

func (t *GCPServiceAccount) createNodeSA(ctx context.Context) error {
	// TODO: also do it for migrated cluster
	if t.IsLegacy() {
		// Cluster.Status.NodeIdentity is updated in k8s_info_saver task
		return nil
	}

	sa := t.NewServiceAccount()
	key := t.NewServiceAccountKey()
	policy := t.NewServiceAccountPolicy()

	if err := k8s.Apply(ctx, t.Local, sa); err != nil {
		return fmt.Errorf("apply IAM SA: %w", err)
	}
	t.Cluster.Status.NodeIdentity = &cloudv1.NodeIdentity{
		GCP: &cloudv1.GCPIdentity{
			ServiceAccount: t.GetGCPNodePoolServiceAccountEmail(),
		},
	}

	if err := k8s.Apply(ctx, t.Local, key); err != nil {
		return fmt.Errorf("apply IAM SA key: %w", err)
	}
	if err := k8s.Apply(ctx, t.Local, policy); err != nil {
		return fmt.Errorf("apply IAM SA policy: %w", err)
	}

	return nil
}

func (t *GCPServiceAccount) Delete(ctx context.Context) deploy.Result {
	if err := t.deleteNodeSA(ctx); err != nil {
		return deploy.Error(err)
	}

	if t.Cluster.SharedBucketEnabled() {
		if err := t.deleteSharedBucketReadonlySA(ctx); err != nil {
			return deploy.Error(err)
		}
	}
	return deploy.Done()
}

// TODO: implement deletion of this resource
func (t *GCPServiceAccount) deleteNodeSA(ctx context.Context) error {
	return nil
}

func (t *GCPServiceAccount) deleteSharedBucketReadonlySA(ctx context.Context) error {
	return nil
}

func (t *GCPServiceAccount) NewServiceAccount() *iamv1alpha1.ServiceAccount {
	saName := t.GetGCPNodePoolServiceAccountName()
	return &iamv1alpha1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name: saName,
		},
		Spec: iamv1alpha1.ServiceAccountSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.GetGCPProviderConfigName(),
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountParameters{
				DisplayName: pointer.String(saName),
				Description: pointer.String("The service account for nodepool"),
			},
		},
	}
}

func (t *GCPServiceAccount) NewServiceAccountKey() *iamv1alpha1.ServiceAccountKey {
	saName := t.GetGCPNodePoolServiceAccountName()
	return &iamv1alpha1.ServiceAccountKey{
		ObjectMeta: metav1.ObjectMeta{
			Name: saName,
		},
		Spec: iamv1alpha1.ServiceAccountKeySpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.GetGCPProviderConfigName(),
				},
				WriteConnectionSecretToReference: &xpv1.SecretReference{
					Namespace: t.Cluster.Namespace,
					Name:      saName,
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountKeyParameters{
				ServiceAccountReferer: iamv1alpha1.ServiceAccountReferer{
					ServiceAccount: pointer.String(t.getServiceAccountReference()),
				},
			},
		},
	}
}

func (t *GCPServiceAccount) NewServiceAccountPolicy() *iamv1alpha1.ServiceAccountPolicy {
	saName := t.GetGCPNodePoolServiceAccountName()
	nodeMember := fmt.Sprintf("serviceAccount:%s@%s.iam.gserviceaccount.com", saName, t.EnvProvider.GetAccount())
	adminNodeMember := fmt.Sprintf("serviceAccount:%s", t.Cluster.Spec.Kubernetes.Cloud.GCP.AdminNodeSa)
	tokenCreatorMember := []string{
		nodeMember,
		adminNodeMember,
	}
	return &iamv1alpha1.ServiceAccountPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: saName,
		},
		Spec: iamv1alpha1.ServiceAccountPolicySpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.GetGCPProviderConfigName(),
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountPolicyParameters{
				ServiceAccountReferer: iamv1alpha1.ServiceAccountReferer{
					ServiceAccount: pointer.String(t.getServiceAccountReference()),
				},
				Policy: iamv1alpha1.Policy{
					Bindings: []*iamv1alpha1.Binding{
						{
							Role:    "roles/iam.serviceAccountTokenCreator",
							Members: tokenCreatorMember,
						},
						// Permissions for read-only actions that do not affect state, such as viewing (but not modifying) existing resources or data.
						{
							Role: "roles/viewer",
							Members: []string{
								nodeMember,
							},
						},
					},
				},
			},
		},
	}
}

func (t *GCPServiceAccount) getServiceAccountReference() string {
	return fmt.Sprintf("projects/%s/serviceAccounts/%s", t.EnvProvider.GetAccount(), t.GetGCPNodePoolServiceAccountEmail())
}

// createSharedBucketReadonlySA creates a readonly SA for accessing shared bucket.
// TODO(chenfei): since this SA is created per data-plane and we grant permissions at project level,
// maybe we can move this task into region-server.
func (t *GCPServiceAccount) createSharedBucketReadonlySA(ctx context.Context) error {
	// build and create SA
	saName := namer.SharedBucketReadonlySAName(t.Cluster)
	sa := &iamv1alpha1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name: saName,
		},
		Spec: iamv1alpha1.ServiceAccountSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.GetGCPProviderConfigName(),
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountParameters{
				DisplayName: pointer.String(saName),
				Description: pointer.String("The service account for accessing shared bucket in readonly"),
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, sa); err != nil {
		return fmt.Errorf("apply IAM SA %s: %w", sa.Name, err)
	}

	if sa.Status.AtProvider.Name == "" {
		return fmt.Errorf("readonly SA(%s) is not ready", sa.Name)
	}

	// build and create SA policy
	clusterServerMember := fmt.Sprintf("serviceAccount:%s", namer.ClusterServerIAMSAEmail(t.Config.FullManage.GCPClusterServerSAName, t.Config.FullManage.GCPProjectID))
	// currently, only the cluster-server will assume it.
	tokenCreatorMember := []string{
		clusterServerMember,
	}
	policy := &iamv1alpha1.ServiceAccountPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: saName,
		},
		Spec: iamv1alpha1.ServiceAccountPolicySpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.GetGCPProviderConfigName(),
				},
			},
			ForProvider: iamv1alpha1.ServiceAccountPolicyParameters{
				ServiceAccountReferer: iamv1alpha1.ServiceAccountReferer{
					ServiceAccount: pointer.String(sa.Status.AtProvider.Name),
				},
				Policy: iamv1alpha1.Policy{
					Bindings: []*iamv1alpha1.Binding{
						{
							Role:    "roles/iam.serviceAccountTokenCreator",
							Members: tokenCreatorMember,
						},
					},
				},
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, policy); err != nil {
		return fmt.Errorf("apply IAM SA policy %s: %w", policy.Name, err)
	}

	t.Cluster.Status.SharedBucket.GCP.ReadonlySA = sa.Status.AtProvider.Email
	return nil
}