package gke

import (
	"context"
	"fmt"
	"time"

	"cloud.google.com/go/storage"
	"github.com/crossplane-contrib/provider-gcp/apis/storage/v1alpha1"
	storagev1alpha3 "github.com/crossplane-contrib/provider-gcp/apis/storage/v1alpha3"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const defaultStorageClass = "STANDARD"

type Bucket struct {
	*task.BaseTask
}

// NewBucket create Bucket instance.
func NewBucket(ctx *task.Context) *Bucket {
	return &Bucket{BaseTask: &task.BaseTask{Context: ctx}}
}

// Name implements deploy.Task.Name interface.
func (t *Bucket) Name() string {
	return "Bucket"
}

// Create implements deploy.Task interface.
func (t *Bucket) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())
	name := getBucketName(t.ProjectID, t.GetRegion(), t.GetMasterName())

	labels := t.GetLables()
	labels["dm_job_name"] = t.DMJob.Name

	serviceAccount, err := getGCPNodePoolServiceAccountEmail(t.BaseTask)
	if err != nil {
		return deploy.Error(err)
	} else if serviceAccount == "" {
		return deploy.Error(fmt.Errorf("service account is empty"))
	}

	bucket := t.buildGCPBucket(name, labels)
	err = k8s.CreateIfNotExist(ctx, t.Local, bucket)
	if err != nil {
		return deploy.Error(err)
	}

	// TODO(zhengjiajin): remove this when provider-aws can get reference ID
	if meta.GetExternalName(bucket) == "" {
		return deploy.Error(fmt.Errorf("bucket externalName is empty: %s", name))
	}

	// allow service account access this bucket.
	policy := t.buildBucketPolicyMember(bucket, serviceAccount)
	err = k8s.CreateIfNotExist(ctx, t.Local, policy)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

// PostCreate implements deploy.Task interface.
func (t *Bucket) PostCreate(ctx context.Context) deploy.Result {
	name := getBucketName(t.ProjectID, t.GetRegion(), t.GetMasterName())

	bucket := &storagev1alpha3.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err := t.Local.Get(ctx, client.ObjectKeyFromObject(bucket), bucket)
	if err != nil {
		return deploy.Error(err)
	}

	ready := bucket.GetCondition(xpv1.TypeReady)
	synced := bucket.GetCondition(xpv1.TypeSynced)

	if ready.Status != corev1.ConditionTrue && synced.Status != corev1.ConditionTrue {
		return deploy.Error(fmt.Errorf("bucket not ready name: %s", name))
	}

	// check policy
	bucketPolicy := &v1alpha1.BucketPolicyMember{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err = t.Local.Get(ctx, client.ObjectKeyFromObject(bucketPolicy), bucketPolicy)
	if err != nil {
		return deploy.Error(err)
	}
	ready = bucketPolicy.GetCondition(xpv1.TypeReady)
	synced = bucketPolicy.GetCondition(xpv1.TypeSynced)
	if ready.Status != corev1.ConditionTrue && synced.Status != corev1.ConditionTrue {
		return deploy.Error(fmt.Errorf("bucket policy not ready: %s", name))
	}

	t.DMJob.Status.BlobContainerName = name
	return deploy.Done()
}

// Delete implements deploy.Task interface.
func (t *Bucket) Delete(ctx context.Context) deploy.Result {
	name := getBucketName(t.ProjectID, t.GetRegion(), t.GetMasterName())

	// GCS support delete non-empty bucket with forceDelete, but not implemented in cross-plane
	// https://githut.com/crossplane-contrib/provider-gcp/issues/295
	// cross-plane will use the name of the bucket CR as the GCS bucket name
	if err := t.emptyBucket(ctx, name); err != nil {
		return deploy.Error(err)
	}

	bucketPolicy := &v1alpha1.BucketPolicyMember{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err := k8s.DeleteObject(ctx, t.Local, bucketPolicy)
	if err != nil {
		return deploy.Error(err)
	}

	bucket := &storagev1alpha3.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err = k8s.DeleteObject(ctx, t.Local, bucket)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *Bucket) PostDelete(ctx context.Context) deploy.Result {
	name := getBucketName(t.ProjectID, t.GetRegion(), t.GetMasterName())
	bucket := &storagev1alpha3.Bucket{ObjectMeta: metav1.ObjectMeta{
		Name: name,
	}}
	if err := k8s.GetDeletedObject(ctx, t.Local, client.ObjectKeyFromObject(bucket), bucket); err != nil {
		return deploy.Error(err)
	}

	policyMember := &v1alpha1.BucketPolicyMember{ObjectMeta: metav1.ObjectMeta{
		Name: name,
	}}
	if err := k8s.GetDeletedObject(ctx, t.Local, client.ObjectKeyFromObject(policyMember), policyMember); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *Bucket) emptyBucket(ctx context.Context, bucket string) error {
	if bucket == "" {
		return nil
	}
	err := t.GCP.EmptyBucket(ctx, bucket)
	// we wait until the bucket CR has finished deletion, so might meet this error if we reconcile again
	if err == storage.ErrBucketNotExist {
		return nil
	}
	return err
}

func (t *Bucket) buildGCPBucket(name string, labels map[string]string) *storagev1alpha3.Bucket {
	providerConfigRef := t.GetProviderConfigName()
	return &storagev1alpha3.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: storagev1alpha3.BucketSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfigRef},
				WriteConnectionSecretToReference: &xpv1.SecretReference{
					Name:      name,
					Namespace: t.DMJob.Namespace,
				},
			},
			BucketParameters: storagev1alpha3.BucketParameters{
				BucketSpecAttrs: storagev1alpha3.BucketSpecAttrs{
					Location: t.GetRegion(),
					BucketUpdatableAttrs: storagev1alpha3.BucketUpdatableAttrs{
						Labels: t.GetPlainResourceTag(),
					},
					StorageClass: defaultStorageClass,
				},
			},
		},
	}
}

func (t *Bucket) buildBucketPolicyMember(bucket *storagev1alpha3.Bucket, nodeServiceAccount string) *v1alpha1.BucketPolicyMember {
	// https://cloud.google.com/storage/docs/access-control/iam-roles?hl=zh-cn
	providerConfigRef := t.GetProviderConfigName()
	bucketExternalName := meta.GetExternalName(bucket)
	// give higher role to node service account to 'list'/'delete' object
	roleName := "roles/storage.admin"
	member := fmt.Sprintf("serviceAccount:%s", nodeServiceAccount)

	return &v1alpha1.BucketPolicyMember{
		ObjectMeta: metav1.ObjectMeta{
			Name: bucket.Name,
		},
		Spec: v1alpha1.BucketPolicyMemberSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfigRef},
			},
			ForProvider: v1alpha1.BucketPolicyMemberParameters{
				Bucket: &bucketExternalName,
				Role:   roleName,
				Member: &member,
			},
		},
	}
}

func getBucketName(project string, region string, master string) string {
	const (
		labelLengthLimit = 63
		hashSize         = 8
		resultLimit      = labelLengthLimit - hashSize - 1
	)

	result := fmt.Sprintf("dm-%s-%s-%s", project, region, util.FnvHash(master))
	if len(result) > resultLimit {
		result = result[:resultLimit]
	}

	return result
}