package aks

import (
	"context"
	"fmt"
	"time"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"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"
	storagev1beta1 "github.com/upbound/provider-azure/apis/storage/v1beta1"
	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"
)

const defaultRequeueInterval = time.Second * 30

type StorageAccount struct {
	*task.BaseTask
}

func NewStorageAccount(ctx *task.Context) *StorageAccount {
	return &StorageAccount{BaseTask: &task.BaseTask{Context: ctx}}
}

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

// Notice: still can reconcile after dataplane deleted
func (t *StorageAccount) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())

	dataplaneID := t.Bucket.Spec.DataplaneId

	// if dataplane is missing or deleting, we don't need to create storage account
	dataPlaneCR := &v1alpha1.DataPlane{}
	if err := t.DataPlaneR.Get(ctx, client.ObjectKey{
		Namespace: t.Config.Namespace,
		Name:      dataplaneID,
	}, dataPlaneCR); err != nil {
		if apierrors.IsNotFound(err) {
			return deploy.Done()
		}
		return deploy.Error(err)
	}
	if !dataPlaneCR.DeletionTimestamp.IsZero() {
		return deploy.Done()
	}

	// try to get existing storage account for dataplane
	account, err := getStorageAccountForDataplane(ctx, t.Local, dataplaneID)
	if err != nil {
		return deploy.Error(err)
	}
	if account != nil {
		return deploy.Done()
	}

	// create storage account
	resourceGroupName, err := getBucketResourceGroup(t.Bucket)
	if err != nil {
		return deploy.Error(err)
	}
	accountName, err := storageAccountName(t.Config.ResourceNamePrefix, dataplaneID)
	if err != nil {
		return deploy.Error(fmt.Errorf("generate storage account name error: %w", err))
	}
	labels, tags := getLabelsAndTags(map[string]string{consts.DataplaneLabelKey: dataplaneID}, dataPlaneCR.Labels, t.Config)
	applyAccount := &storagev1beta1.Account{
		ObjectMeta: metav1.ObjectMeta{
			Name:   dataplaneID,
			Labels: labels,
		},
		Spec: storagev1beta1.AccountSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.Bucket.Spec.AccountRef.ProviderConfigName,
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
			ForProvider: storagev1beta1.AccountParameters{
				AccountKind:            ptr.To("StorageV2"),
				AccountReplicationType: ptr.To("ZRS"),
				AccountTier:            ptr.To("Standard"),
				Location:               ptr.To(t.Bucket.Spec.Region),
				ResourceGroupName:      ptr.To(resourceGroupName),
				Tags:                   util.ToMapPointer(tags),
			},
		},
	}
	meta.SetExternalName(applyAccount, accountName)
	if err := k8s.Apply(ctx, t.Local, applyAccount); err != nil {
		return deploy.Error(fmt.Errorf("failed to apply storage account: %w", err))
	}

	return deploy.Done()
}

func (t *StorageAccount) PostCreate(ctx context.Context) deploy.Result {
	dataplaneID := t.Bucket.Spec.DataplaneId

	// if dataplane is missing or deleting, we don't need to create storage account
	dataPlaneCR := &v1alpha1.DataPlane{}
	if err := t.DataPlaneR.Get(ctx, client.ObjectKey{
		Namespace: t.Config.Namespace,
		Name:      dataplaneID,
	}, dataPlaneCR); err != nil {
		if apierrors.IsNotFound(err) {
			return deploy.Done()
		}
		return deploy.Error(err)
	}
	if !dataPlaneCR.DeletionTimestamp.IsZero() {
		return deploy.Done()
	}

	// try to get existing storage account for dataplane
	account, err := getStorageAccountForDataplane(ctx, t.Local, dataplaneID)
	if err != nil {
		return deploy.Error(err)
	}
	if account == nil {
		return deploy.Panic(fmt.Errorf("storage account not found")).WithRequeuAfter(defaultRequeueInterval)
	}
	if !controllers.IsResourceReady(account.Status.ResourceStatus) {
		return deploy.Panic(fmt.Errorf("storage account is not ready")).WithRequeuAfter(defaultRequeueInterval)
	}
	return deploy.Done()
}

func (t *StorageAccount) Delete(ctx context.Context) deploy.Result {
	stillHasOtherBuckets, err := t.stillHasOtherBuckets(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	if stillHasOtherBuckets {
		return deploy.Done()
	}

	hasContainers, err := t.checkIfStorageAccountHasContainers(ctx)
	if err != nil {
		return deploy.Error(fmt.Errorf("check if storage account has containers error: %w", err))
	}
	if hasContainers {
		return deploy.Error(fmt.Errorf("storage account has unknown containers")).WithRequeuAfter(defaultRequeueInterval)
	}

	dataplaneID := t.Bucket.Spec.DataplaneId
	account, err := getStorageAccountForDataplane(ctx, t.Local, dataplaneID)
	if err != nil {
		return deploy.Error(err)
	}
	if account == nil {
		return deploy.Done()
	}

	if err := controllers.DeleteUpboundResource(ctx, t.Local, account); err != nil {
		return deploy.Error(fmt.Errorf("failed to delete storage account: %w", err))
	}

	return deploy.Done()
}

func (t *StorageAccount) PostDelete(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *StorageAccount) stillHasOtherBuckets(ctx context.Context) (bool, error) {
	dataplaneID := t.Bucket.Spec.DataplaneId
	if dataplaneID == "" {
		return false, fmt.Errorf("dataplane id is missing for bucket %s", t.Bucket.Name)
	}

	// check if there are any buckets still not deleted for the same dataplane
	bucketList := &v1alpha1.BucketList{}
	listOpts := &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			consts.DataplaneLabelKey: dataplaneID,
		}),
	}
	if err := t.Local.List(ctx, bucketList, listOpts); err != nil {
		return false, fmt.Errorf("list buckets error: %w", err)
	}
	for _, b := range bucketList.Items {
		if b.Name == t.Bucket.Name {
			continue
		}
		if b.DeletionTimestamp.IsZero() {
			return true, nil
		}
	}
	return false, nil
}

// Notice: In azure storage account can be deleted even if it has containers in it.
// To prevent mistakenly delete storage account because of bugs/operations, add strict defensive logic here to avoid delete storage account when it still has containers.
func (t *StorageAccount) checkIfStorageAccountHasContainers(ctx context.Context) (bool, error) {
	// for the cases storage account is not created done yet, so it will not have containers
	account, err := getStorageAccountForDataplane(ctx, t.Local, t.Bucket.Spec.DataplaneId)
	if err != nil {
		return false, err
	}
	if account == nil {
		return false, nil
	}
	accountName := meta.GetExternalName(account)
	if accountName == "" {
		return false, nil
	}

	resourceGroupName, err := getBucketResourceGroup(t.Bucket)
	if err != nil {
		return false, err
	}

	subscriptionID := ""
	if t.Bucket.Spec.AccountRef != nil {
		subscriptionID = t.Bucket.Spec.AccountRef.AccountID
	}
	if subscriptionID == "" {
		return false, fmt.Errorf("azure subscription id is missing for bucket %s", t.Bucket.Name)
	}

	blobClient, err := t.AzureCli.StorageAccounts(subscriptionID)
	if err != nil {
		return false, fmt.Errorf("create storage accounts client error: %w", err)
	}
	hasContainers, err := blobClient.StorageAccountHasContainers(ctx, resourceGroupName, accountName)
	if err != nil {
		return false, fmt.Errorf("list storage containers error: %w", err)
	}
	return hasContainers, nil
}