package gke

import (
	"context"
	"fmt"
	"net/url"
	"path"
	"sort"
	"time"

	iamv1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/iam/v1alpha1"
	"github.com/crossplane-contrib/provider-gcp/apis/storage/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"golang.org/x/exp/slices"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

// Unlike the AWS provider grant permissions of shared bucket on IAM role, GCP grant permissions at resource-level.
// So we add an independent task to grant permission to SA from shared_bucket.

// SharedBucket task is responsible for granting proper permissions to IAM SAs.
type SharedBucket struct {
	*task.BaseTask

	gcpCli gcp.Client
}

func NewSharedBucket(ctx *task.Context, gcpCli gcp.Client) *SharedBucket {
	return &SharedBucket{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}
}

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

func (t *SharedBucket) Create(ctx context.Context) deploy.Result {
	if !t.Cluster.SharedBucketEnabled() {
		return deploy.Done()
	}

	t.BeginAt = util.Pointer(time.Now())
	u, err := url.Parse(t.Cluster.Spec.SharedBucket.BucketURL)
	if err != nil {
		return deploy.Error(fmt.Errorf("parse shared bucket URL: %w", err))
	}

	policy, err := t.ensureBucketPolicy(ctx, u)
	if err != nil {
		return deploy.Error(err)
	}
	if err := t.grantPermissionsToSAs(ctx, u, policy); err != nil {
		return deploy.Error(err).WithRequeuAfter(time.Minute)
	}

	return deploy.Done()
}

func (t *SharedBucket) Delete(ctx context.Context) deploy.Result {
	if !t.Cluster.SharedBucketEnabled() {
		return deploy.Done()
	}

	u, err := url.Parse(t.Cluster.Spec.SharedBucket.BucketURL)
	if err != nil {
		return deploy.Error(fmt.Errorf("parse shared bucket URL: %w", err))
	}
	if err := t.revokePermissionsFromSAs(ctx, u); err != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func (t *SharedBucket) grantPermissionsToSAs(ctx context.Context, bucketURL *url.URL, policy *v1alpha1.BucketPolicy) error {
	var (
		nodeSAEmail = t.GetGCPNodePoolServiceAccountEmail()
		roSAEmail   = namer.SharedBucketReadonlySAEmail(t.Cluster)
	)
	bucketName := bucketURL.Host
	prefix := t.buildPrefix(bucketURL)
	conditionTitle := t.buildConditionTitle()
	resourceRestrictExpression := t.buildSharedBucketPolicyCondition(bucketName, prefix)

	// Note:
	// 1. The limit of `Role bindings in an allow policy that include the same role and the same principal, but different condition expressions` is 20. So that
	//    we can't create more than 20 clusters using the same IAM SA (node SA or readonly SA) and the same shared bucket. For this situation,
	//    we can use different shared buckets for each cluster even if they are in the same data-plane. Refer to https://cloud.google.com/iam/quotas.

	// append bindings to policy
	bindings := []*iamv1alpha1.Binding{
		{
			Role: "roles/storage.objectUser",
			Members: []string{
				fmt.Sprintf("serviceAccount:%s", nodeSAEmail),
			},
			Condition: &iamv1alpha1.Expr{
				Expression:  resourceRestrictExpression,
				Title:       ptr.To(conditionTitle),
				Description: ptr.To("Restrict accessing shared bucket within cluster-level"),
			},
		},
		{
			Role: "roles/storage.objectViewer",
			Members: []string{
				fmt.Sprintf("serviceAccount:%s", roSAEmail),
			},
			Condition: &iamv1alpha1.Expr{
				Expression:  resourceRestrictExpression,
				Title:       ptr.To(conditionTitle),
				Description: ptr.To("Restrict accessing shared bucket within cluster-level"),
			},
		},
	}
	for _, binding := range bindings {
		old, ok := findIAMBindingByMembersAndConditionTitle(policy.Spec.ForProvider.Policy.Bindings, binding)
		if !ok {
			policy.Spec.ForProvider.Policy.Bindings = append(policy.Spec.ForProvider.Policy.Bindings, binding)
			continue
		}

		old.Role = binding.Role
		old.Condition = binding.Condition
	}

	if err := t.Local.Update(ctx, policy); err != nil {
		return fmt.Errorf("update BucketPolicy(%s): %w", bucketURL.Host, err)
	}
	return nil
}

func (t *SharedBucket) revokePermissionsFromSAs(ctx context.Context, bucketURL *url.URL) error {
	var (
		bucketName = bucketURL.Host
	)
	bp := &v1alpha1.BucketPolicy{}
	bp.Name = t.sharedBucketPolicyName(bucketName)
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(bp), bp); err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}

		return fmt.Errorf("get BucketPolicy(%s): %w", bucketName, err)
	}

	changed := make([]*iamv1alpha1.Binding, 0, len(bp.Spec.ForProvider.Policy.Bindings))
	for _, binding := range bp.Spec.ForProvider.Policy.Bindings {
		// If the condition is nil, it means this policy is not created by us, so skip it.
		if binding.Condition != nil && ptr.Deref(binding.Condition.Title, "") == t.buildConditionTitle() {
			continue
		}

		changed = append(changed, binding)
	}

	bp.Spec.ForProvider.Policy.Bindings = changed
	return t.Local.Update(ctx, bp)
}

func (t *SharedBucket) ensureBucketPolicy(ctx context.Context, bucketURL *url.URL) (*v1alpha1.BucketPolicy, error) {
	policy := &v1alpha1.BucketPolicy{}
	err := t.Local.Get(ctx, client.ObjectKey{Name: t.sharedBucketPolicyName(bucketURL.Host)}, policy)
	if err == nil {
		return policy, nil
	} else if err != nil && !apierrors.IsNotFound(err) {
		return nil, fmt.Errorf("get BucketPolicy(%s): %w", bucketURL.Host, err)
	}

	// build initial bindings
	var bindings []*iamv1alpha1.Binding
	bucket := t.gcpCli.StorageCli().Bucket(bucketURL.Host).IAM()
	iamPolicy, err := t.gcpCli.GetIAMPolicy3(ctx, bucket)
	if err != nil {
		return nil, fmt.Errorf("get policy: %w", err)
	}
	log.FromContext(ctx).Info("get policy", "iamPolicy", iamPolicy)
	for _, binding := range iamPolicy.Bindings {
		each := &iamv1alpha1.Binding{
			Role:    binding.Role,
			Members: binding.Members,
		}
		if binding.Condition != nil {
			each.Condition = &iamv1alpha1.Expr{
				Expression:  binding.Condition.Expression,
				Title:       ptr.To(binding.Condition.Title),
				Description: ptr.To(binding.Condition.Description),
			}
		}
		bindings = append(bindings, each)
	}

	// build policy
	bucketName := bucketURL.Host
	policy = &v1alpha1.BucketPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.sharedBucketPolicyName(bucketName),
		},
		Spec: v1alpha1.BucketPolicySpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionOrphan,
				ProviderConfigReference: &xpv1.Reference{Name: t.providerConfigNameForBucketPolicy()},
			},
			ForProvider: v1alpha1.BucketPolicyParameters{
				Bucket: ptr.To(bucketURL.Host),
				Policy: iamv1alpha1.Policy{
					Bindings: bindings,
				},
			},
		},
	}

	if err := t.Local.Create(ctx, policy); err != nil {
		return nil, fmt.Errorf("create BucketPolicy(%s): %w", bucketURL.Host, err)
	}
	return policy, nil
}

func (t *SharedBucket) buildSharedBucketPolicyCondition(bucketName string, prefix string) string {
	// Refer to https://stackoverflow.com/questions/38575445/how-to-set-access-permissions-of-google-cloud-storage-bucket-folder
	// and https://cloud.google.com/storage/docs/access-control/iam#conditions.
	// Since the storage.objects.list permission is granted at the bucket level, you cannot use the resource.name
	// condition attribute to restrict object listing access to a subset of objects in the bucket.
	// So that I hack a condition `(resource.type == "storage.googleapis.com/Bucket")` to allow list in bucket level.
	return fmt.Sprintf(
		`( resource.type == "storage.googleapis.com/Bucket" && resource.name == "projects/_/buckets/%s" ) ||
  ( resource.service == "storage.googleapis.com" && resource.name.startsWith("projects/_/buckets/%s") )`,
		bucketName, prefix)
}

func (t *SharedBucket) buildConditionTitle() string {
	return fmt.Sprintf("ForCluster-%s", t.Cluster.Name)
}

func (t *SharedBucket) buildPrefix(bucketURL *url.URL) string {
	// Note: the prefix in Condition Expression has an additional middle string 'objects'
	// For example, if the bucket URl is 'gcs://chenfei-test-dev/60205/3199728/10911990867609419212/',
	// the generated prefix should be 'chenfei-test-dev/objects/60205/3199728/10911990867609419212/'
	return path.Join(bucketURL.Host, "objects", bucketURL.Path) + "/"
}

func (t *SharedBucket) sharedBucketPolicyName(bucketName string) string {
	return bucketName
}

// providerConfigNameForBucketPolicy returns the name of ProviderConfig.
// We can't use the ProviderConfig created by cluster-controller as its lifecycle is same as the data-plane rather the region.
// But the BucketPolicy's lifecycle is same as the shared bucket which is long lived as the region.
// Note: this ProviderConfig is created by the CD of region-server, defined in [here](https://github.com/tidbcloud/app-delivery/blob/main/dedicated-regional-server/regional-server/envs/gcp-dev-us-west1/account/providerconfig.yaml#L13).
func (t *SharedBucket) providerConfigNameForBucketPolicy() string {
	return t.EnvProvider.GetAccount()
}

func findIAMBindingByMembersAndConditionTitle(bindings []*iamv1alpha1.Binding, target *iamv1alpha1.Binding) (_ *iamv1alpha1.Binding, ok bool) {
	sort.Strings(target.Members)
	for _, each := range bindings {
		sort.Strings(each.Members)
	}

	for _, each := range bindings {
		var title, titleInTarget string
		if each.Condition != nil {
			title = ptr.Deref(each.Condition.Title, "")
		}
		if target.Condition != nil {
			titleInTarget = ptr.Deref(target.Condition.Title, "")
		}

		if slices.Equal(each.Members, target.Members) && title == titleInTarget {
			return each, true
		}
	}
	return nil, false
}