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/tiflow/task/cluster"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/task/cluster/master"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const defaultStorageClass = "STANDARD"

type Bucket struct {
	*master.BaseTask
}

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

var _ deploy.Task = (*Bucket)(nil)

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

// Create implements deploy.Task interface.
func (b *Bucket) Create(ctx context.Context) deploy.Result {
	b.BeginAt = util.Pointer(time.Now())
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())
	tags := cluster.TagsForBucket(b.Network, name)

	serviceAccount := b.Network.Status.IAM.ServiceAccountEmail
	if serviceAccount == "" {
		return deploy.Error(fmt.Errorf("service account is empty"))
	}

	cluster.AddBucketAnnotationToMaster(b.TiflowClusterMaster, name)

	bucket := b.buildGCPBucket(name, tags)
	err := k8s.CreateIfNotExist(ctx, b.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 := b.buildBucketPolicyMember(bucket, serviceAccount)
	err = k8s.CreateIfNotExist(ctx, b.Local, policy)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

// PostCreate implements deploy.Task interface.
func (b *Bucket) PostCreate(ctx context.Context) deploy.Result {
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())

	bucket := &storagev1alpha3.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err := b.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 {
		cluster.UpdateBucketReadyForMaster(b.TiflowClusterMaster, false)
		return deploy.Error(fmt.Errorf("bucket not ready name: %s", name))
	}

	// check policy
	bucketPolicy := &v1alpha1.BucketPolicyMember{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err = b.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 {
		cluster.UpdateBucketReadyForMaster(b.TiflowClusterMaster, false)
		return deploy.Error(fmt.Errorf("bucket policy not ready: %s", name))
	}

	cluster.UpdateBucketReadyForMaster(b.TiflowClusterMaster, true)
	return deploy.Done()
}

// Delete implements deploy.Task interface.
func (b *Bucket) Delete(ctx context.Context) deploy.Result {
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())

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

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

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

	return deploy.Done()
}

func (b *Bucket) PostDelete(ctx context.Context) deploy.Result {
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())
	bucket := &storagev1alpha3.Bucket{ObjectMeta: metav1.ObjectMeta{
		Name: name,
	}}
	if err := k8s.GetDeletedObject(ctx, b.Local, client.ObjectKeyFromObject(bucket), bucket); err != nil {
		return deploy.Error(err)
	}

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

	return deploy.Done()
}

func (b *Bucket) emptyBucket(ctx context.Context, bucket string) error {
	if bucket == "" {
		return nil
	}
	err := b.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 (b *Bucket) buildGCPBucket(name string, labels map[string]string) *storagev1alpha3.Bucket {
	providerConfigRef := namer.UniqueIdentifier(b.Network.ObjectMeta)
	return &storagev1alpha3.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Spec: storagev1alpha3.BucketSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfigRef},
				WriteConnectionSecretToReference: &xpv1.SecretReference{
					Name:      name,
					Namespace: b.Network.GetNamespace(),
				},
			},
			BucketParameters: storagev1alpha3.BucketParameters{
				BucketSpecAttrs: storagev1alpha3.BucketSpecAttrs{
					Location: b.Network.Spec.Region,
					BucketUpdatableAttrs: storagev1alpha3.BucketUpdatableAttrs{
						Labels: labels,
					},
					StorageClass: defaultStorageClass,
				},
			},
		},
	}
}

func (b *Bucket) buildBucketPolicyMember(bucket *storagev1alpha3.Bucket, nodeServiceAccount string) *v1alpha1.BucketPolicyMember {
	// https://cloud.google.com/storage/docs/access-control/iam-roles?hl=zh-cn
	providerConfigRef := namer.UniqueIdentifier(b.Network.ObjectMeta)
	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,
			},
		},
	}
}