package eks

import (
	"context"
	"fmt"
	"strings"
	"time"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	s3v1beta2 "github.com/upbound/provider-aws/apis/s3/v1beta2"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

type Bucket struct {
	*task.BaseTask
}

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

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

func (t *Bucket) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())
	effectiveTags, err := t.EffectiveCloudTags(ctx, consts.ProviderTypeEKS)
	if err != nil {
		return deploy.Error(err)
	}
	tags := util.ToMapPointer(effectiveTags)
	providerConfig := t.Bucket.Spec.AccountRef.ProviderConfigName
	// three scenarios:
	// 1. bucket does not exist, we need to create a new one
	// 2. bucket already exists, and the name is globally unique, we just need to update the cr
	// 3. bucket exists, but the name is not globally unique, we need to regenerate the name, delete the old one and create a new one
	bucketList := &s3v1beta2.BucketList{}
	listOpts := &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			consts.DataplaneLabelKey:       t.Bucket.Spec.DataplaneId,
			"tags.tidbcloud.com/bucket-id": t.Bucket.Spec.Id,
		}),
	}
	if err := t.Local.List(ctx, bucketList, listOpts); err != nil {
		return deploy.Error(fmt.Errorf("failed to list Buckets: %w", err))
	}
	if len(bucketList.Items) > 1 {
		return deploy.Error(fmt.Errorf("found multiple Buckets for dataplane %q and bucket id %q: %s",
			t.Bucket.Spec.DataplaneId, t.Bucket.Spec.Id, strings.Join(bucketNames(bucketList.Items), ", ")))
	}

	var bucketName string

	if len(bucketList.Items) == 0 {
		// scenario 1
		var err error
		bucketName, err = t.GetCurrentOrSubResourceName()
		if err != nil {
			return deploy.Error(fmt.Errorf("failed to generate bucket name: %w", err))
		}
		if err := t.applyBucket(ctx, bucketName, tags, providerConfig); err != nil {
			return deploy.Error(fmt.Errorf("failed to create bucket: %w", err))
		}
	} else {
		bucket := bucketList.Items[0]
		if t.isBucketNameConflict(&bucket) {
			// scenario 3
			if err := t.Local.Delete(ctx, &bucket); err != nil {
				return deploy.Error(fmt.Errorf("failed to delete existing bucket: %w", err))
			}
			var err error
			bucketName, err = t.GetSubResourceNameAfterConflict(bucket.Name)
			if err != nil {
				return deploy.Error(fmt.Errorf("failed to generate bucket name: %w", err))
			}
			if err := t.applyBucket(ctx, bucketName, tags, providerConfig); err != nil {
				return deploy.Error(fmt.Errorf("failed to recreate bucket: %w", err))
			}
		} else {
			// scenario 2
			bucketName = bucket.Name
			if err := t.applyBucket(ctx, bucketName, tags, providerConfig); err != nil {
				return deploy.Error(fmt.Errorf("failed to update bucket: %w", err))
			}
		}
	}

	t.Bucket.Status.Bucket = &v1alpha1.Reference{
		Name: bucketName,
	}

	return deploy.Done()
}

func (t *Bucket) PostCreate(ctx context.Context) deploy.Result {
	bucket := &s3v1beta2.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.Bucket.Status.Bucket.Name,
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(bucket), bucket); err != nil {
		return deploy.Error(fmt.Errorf("failed to get Bucket resource: %w", err))
	}
	if t.isBucketNameConflict(bucket) {
		// the bucket name should be globally unique, if it already exists, we need to requeue immediately without executing next task
		return deploy.Panic(nil).WithRequeuAfter(0)
	}
	if controllers.IsResourceReady(bucket.Status.ResourceStatus) {
		return deploy.Done()
	}
	// Use Panic here because the bucket policy depends on the bucket to be ready,
	// we need to requeue immediately without executing next task.
	return deploy.Panic(fmt.Errorf("bucket is not ready"))
}

func (t *Bucket) Delete(ctx context.Context) deploy.Result {
	bucket := &s3v1beta2.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.Bucket.Status.Bucket.Name,
		},
	}
	if err := controllers.DeleteUpboundResource(ctx, t.Local, bucket); err != nil {
		return deploy.Error(fmt.Errorf("failed to delete Bucket resource: %w", err))
	}
	return deploy.Done()
}

// S3 bucket deletion take some time to complete
// we need to ensure the s3 bucket CR is cleaned before marking the task done
func (t *Bucket) PostDelete(ctx context.Context) deploy.Result {
	bucket := &s3v1beta2.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.Bucket.Status.Bucket.Name,
		},
	}
	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, t.Local, client.ObjectKeyFromObject(bucket), bucket))
}

func (t *Bucket) isBucketNameConflict(bucket *s3v1beta2.Bucket) bool {
	syncedCondition := bucket.Status.GetCondition(xpv1.TypeSynced)
	message := syncedCondition.Message
	// When upbound tries to create a bucket with a name used by other account bucket, aws API will return 409 (BucketAlreadyExist) error.
	// Then upbound tries to observe the bucket, but the HeadBucket API call would fail due to the bucket is in other aws account.
	// So we can't get the 409 error from upbound Bucket CR instead we get 403 error. We can't use 403 error in this case.
	// TODO: The following code doesn't work actually. We have to handle this case manually. We need to find a better solution.
	return syncedCondition.Status == corev1.ConditionFalse && strings.Contains(message, "BucketAlreadyExists")
}

func bucketNames(buckets []s3v1beta2.Bucket) []string {
	names := make([]string, 0, len(buckets))
	for _, bucket := range buckets {
		names = append(names, bucket.Name)
	}
	return names
}

func (t *Bucket) applyBucket(ctx context.Context, name string, tags map[string]*string, providerConfig string) error {
	tags["Name"] = ptr.To(name)
	labels := map[string]string{
		consts.DataplaneLabelKey:       t.Bucket.Spec.DataplaneId,
		"tags.tidbcloud.com/bucket-id": t.Bucket.Spec.Id,
	}
	labels = util.MergeMaps(labels, t.Bucket.Labels)
	bucket := &s3v1beta2.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: s3v1beta2.BucketSpec{
			ForProvider: s3v1beta2.BucketParameters{
				Region: &t.Bucket.Spec.Region,
				Tags:   tags,
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfig,
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
		},
	}
	return k8s.Apply(ctx, t.Local, bucket)
}