package ack

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

	ossv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/oss/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	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"
)

const defaultPolicy = `{
	"Version": "1",
	"Statement": [{
		"Effect": "Allow",
		"Action": [
			"oss:*"
		],
		"Principal": [
			"%s"
		],
		"Resource": [
			"acs:oss:*:*:%s",
			"acs:oss:*:*:%s/*"
		]
	}]
}`

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.ProviderTypeACK)
	if err != nil {
		return deploy.Error(err)
	}
	tags := util.ToMapPointer(effectiveTags)

	dataPlaneCR := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.Config.Namespace,
			Name:      t.Bucket.Spec.DataplaneId,
		},
	}
	if err := t.DataPlaneR.Get(ctx, client.ObjectKeyFromObject(dataPlaneCR), dataPlaneCR); err != nil {
		if apierrors.IsNotFound(err) {
			return deploy.Error(fmt.Errorf("dataplane %s not found", t.Bucket.Spec.DataplaneId))
		}
		return deploy.Error(err)
	}
	providerConfig := dataPlaneCR.Spec.AccountRef.ProviderConfigName

	appBoxCR := &v1alpha1.APPBox{}
	if err := t.Local.Get(ctx, client.ObjectKey{
		Namespace: t.Config.Namespace,
		Name:      dataPlaneCR.Status.AppBox.Name,
	}, appBoxCR); err != nil {
		return deploy.Error(fmt.Errorf("get appbox '%s' error: %w",
			dataPlaneCR.Status.AppBox.Name, err))
	}
	principal := fmt.Sprintf("arn:sts::%s:assumed-role/%s/*", dataPlaneCR.Spec.AccountRef.AccountID, appBoxCR.Status.IAM.NodeRoleName)

	// 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 := &ossv1alpha1.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 OSS Buckets: %w", err))
	}
	if len(bucketList.Items) > 1 {
		return deploy.Error(fmt.Errorf("found multiple OSS 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, principal); 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, principal); 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, principal); 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 := &ossv1alpha1.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 OSS 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 := &ossv1alpha1.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: %w", err))
	}
	return deploy.Done()
}

// OSS bucket deletion take some time to complete
// we need to ensure the OSS bucket CR is cleaned before marking the task done
func (t *Bucket) PostDelete(ctx context.Context) deploy.Result {
	bucket := &ossv1alpha1.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 *ossv1alpha1.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, eks 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 eks 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 []ossv1alpha1.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, principal 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 := &ossv1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: ossv1alpha1.BucketSpec{
			ForProvider: ossv1alpha1.BucketParameters{
				Bucket: ptr.To(name),
				Region: &t.Bucket.Spec.Region,
				Tags:   tags,
			},
			// Only when forProvider.policy is unset and initProvider.policy set, the generated terraform will mark
			// the `policy` as ignore_changes field
			// we'll override the policy in BucketPolicy CR
			InitProvider: ossv1alpha1.BucketInitParameters{
				Policy:         ptr.To(fmt.Sprintf(defaultPolicy, principal, name, name)),
				RedundancyType: ptr.To("ZRS"),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfig,
				},
				// Remove `Delete` to avoid bucket gets deleted accidentally
				// Remove `LateInitialize` to avoid `policy` field being late initialized in spec
				// Otherwise, it'll cause `policy` not marked as ignore_changes field
				ManagementPolicies: xpv1.ManagementPolicies{
					xpv1.ManagementActionCreate,
					xpv1.ManagementActionUpdate,
					xpv1.ManagementActionObserve,
				},
			},
		},
	}
	return k8s.Apply(ctx, t.Local, bucket)
}