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"
)

type Container struct {
	*task.BaseTask
}

func NewContainer(ctx *task.Context) *Container {
	return &Container{BaseTask: &task.BaseTask{Context: ctx}}
}

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

func (t *Container) Create(ctx context.Context) deploy.Result {
	t.BeginAt = util.Pointer(time.Now())

	dataplaneID := t.Bucket.Spec.DataplaneId
	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)
	}

	containerName, err := t.ensureContainer(ctx, account)
	if err != nil {
		return deploy.Error(err)
	}

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

	return deploy.Done()
}

func (t *Container) PostCreate(ctx context.Context) deploy.Result {
	if t.Bucket.Status.Bucket == nil || t.Bucket.Status.Bucket.Name == "" {
		return deploy.Panic(fmt.Errorf("container name not set")).WithRequeuAfter(defaultRequeueInterval)
	}
	container := &storagev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.Bucket.Status.Bucket.Name,
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(container), container); err != nil {
		return deploy.Error(fmt.Errorf("failed to get storage container: %w", err))
	}
	if !controllers.IsResourceReady(container.Status.ResourceStatus) {
		return deploy.Panic(fmt.Errorf("storage container is not ready")).WithRequeuAfter(defaultRequeueInterval)
	}

	return deploy.Done()
}

// Notice: use panic here because the next task depends on the container to be deleted
func (t *Container) Delete(ctx context.Context) deploy.Result {
	shouldDeleteContainer, err := t.shouldDeleteContainer(ctx)
	if err != nil {
		return deploy.Panic(err)
	}
	if !shouldDeleteContainer {
		return deploy.Panic(fmt.Errorf("container is not ready to be deleted")).WithRequeuAfter(defaultRequeueInterval)
	}

	if t.Bucket.Status.Bucket != nil && t.Bucket.Status.Bucket.Name != "" {
		container := &storagev1beta1.Container{
			ObjectMeta: metav1.ObjectMeta{
				Name: t.Bucket.Status.Bucket.Name,
			},
		}
		if err := controllers.DeleteUpboundResource(ctx, t.Local, container); err != nil {
			return deploy.Panic(fmt.Errorf("failed to delete container: %w", err))
		}
	}
	return deploy.Done()
}

// Notice:
// 1. since container deletion is async, we need to check if container is really deleted
// 2. use panic here because the next task depends on the container to be deleted
func (t *Container) PostDelete(ctx context.Context) deploy.Result {
	if t.Bucket.Status.Bucket != nil && t.Bucket.Status.Bucket.Name != "" {
		container := &storagev1beta1.Container{
			ObjectMeta: metav1.ObjectMeta{
				Name: t.Bucket.Status.Bucket.Name,
			},
		}
		if err := k8s.GetDeletedObject(ctx, t.Local, client.ObjectKeyFromObject(container), container); err != nil {
			return deploy.Panic(err)
		}
		account, err := getStorageAccountForDataplane(ctx, t.Local, t.Bucket.Spec.DataplaneId)
		if err != nil {
			return deploy.Panic(err)
		}
		if account == nil {
			return deploy.Done()
		}
		accountName := meta.GetExternalName(account)
		if accountName == "" {
			return deploy.Done()
		}

		subscriptionID := ""
		if t.Bucket.Spec.AccountRef != nil {
			subscriptionID = t.Bucket.Spec.AccountRef.AccountID
		}
		if subscriptionID == "" {
			return deploy.Panic(fmt.Errorf("azure subscription id is missing for bucket %s", t.Bucket.Name))
		}
		resourceGroupName, err := getBucketResourceGroup(t.Bucket)
		if err != nil {
			return deploy.Panic(fmt.Errorf("get bucket resource group error: %w", err))
		}

		blobClient, err := t.AzureCli.StorageAccounts(subscriptionID)
		if err != nil {
			return deploy.Panic(fmt.Errorf("create blob containers client error: %w", err))
		}
		exists, err := blobClient.StorageAccountContainerExists(ctx, resourceGroupName, accountName, container.Name)
		if err != nil {
			return deploy.Panic(fmt.Errorf("check container exists error: %w", err))
		}
		if exists {
			return deploy.Panic(fmt.Errorf("container still exists in storage account")).WithRequeuAfter(defaultRequeueInterval)
		}
	}
	return deploy.Done()
}

func (t *Container) ensureContainer(ctx context.Context, account *storagev1beta1.Account) (string, error) {
	containerList := &storagev1beta1.ContainerList{}
	listOpts := &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			consts.BucketLabelKey: t.Bucket.Spec.Id,
		}),
	}
	if err := t.Local.List(ctx, containerList, listOpts); err != nil {
		return "", fmt.Errorf("failed to list storage containers: %w", err)
	}

	var name string
	if len(containerList.Items) == 0 {
		name = containerName(t.Bucket.Spec.Id)
	} else {
		name = containerList.Items[0].Name
	}

	toAddLabels := map[string]string{
		consts.BucketLabelKey: t.Bucket.Spec.Id,
	}
	labels, _ := getLabelsAndTags(toAddLabels, t.Bucket.Labels, t.Config)

	container := &storagev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: storagev1beta1.ContainerSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: t.Bucket.Spec.AccountRef.ProviderConfigName,
				},
				ManagementPolicies: controllers.GetResourceManagementPolicies(),
			},
			ForProvider: storagev1beta1.ContainerParameters{
				ContainerAccessType: ptr.To("private"),
				StorageAccountNameRef: &xpv1.Reference{
					Name: account.Name,
				},
			},
		},
	}
	if err := k8s.Apply(ctx, t.Local, container); err != nil {
		return "", fmt.Errorf("failed to apply storage container: %w", err)
	}

	return name, nil
}

func (t *Container) shouldDeleteContainer(ctx context.Context) (bool, error) {
	// handle the cases container is not created done yet
	if t.Bucket.Status.Bucket == nil || t.Bucket.Status.Bucket.Name == "" {
		return true, nil
	}
	container := &storagev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.Bucket.Status.Bucket.Name,
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(container), container); err != nil {
		if apierrors.IsNotFound(err) {
			return true, nil
		}
		return false, fmt.Errorf("get storage container error: %w", err)
	}

	accountName := ptr.Deref(container.Spec.ForProvider.StorageAccountName, "")
	if accountName == "" {
		return false, fmt.Errorf("storage account name is missing for container %s", container.Name)
	}

	blobClient, err := t.AzureCli.BlobContainers(accountName)
	if err != nil {
		return false, fmt.Errorf("create blob containers client error: %w", err)
	}
	hasBlobs, err := blobClient.ContainerHasBlobs(ctx, accountName, container.Name)
	if err != nil {
		return false, fmt.Errorf("check container blobs error: %w", err)
	}
	return !hasBlobs, nil
}