// Package aks add by blacknc <blacknc@163.com> at 2024/08/28 23:50:01
package aks

import (
	"context"
	"errors"
	"fmt"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	azurestoragev1beta1 "github.com/upbound/provider-azure/apis/storage/v1beta1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

// BlobContainer creates or delete blob containers for dm-master.
type BlobContainer struct {
	*task.BaseTask
}

func NewBlobContainer(ctx *task.Context) *BlobContainer {
	return &BlobContainer{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

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

func (t *BlobContainer) getMasterName() string {
	return fmt.Sprintf("%s-%s-%s", t.Namer.GetClusterName(), t.DMJob.ObjectMeta.Name, "master")
}

func (t *BlobContainer) Create(ctx context.Context) deploy.Result {
	if err := t.isStorageAccountReady(ctx); err != nil {
		return deploy.Error(err)
	}

	name := getBucketName(t.ProjectID, t.GetRegion(), t.getMasterName())

	storageAccountName := GetStorageAccountName(t.ProjectID, t.GetRegion(), t.GetClusterScopeName())
	providerConfigName := t.GetProviderConfigName()

	labels := t.GetLables()
	labels["dm_job_name"] = t.DMJob.Name

	container := &azurestoragev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: azurestoragev1beta1.ContainerSpec{
			ForProvider: azurestoragev1beta1.ContainerParameters{
				ContainerAccessType: ptr.To("private"),
				StorageAccountName:  ptr.To(storageAccountName),
				StorageAccountNameRef: &xpv1.Reference{
					Name: storageAccountName,
				},
				Metadata: t.GetResourceTag(),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfigName,
				},
			},
		},
	}

	client := t.Local

	if err := client.Create(ctx, container); err != nil {
		if !apierrors.IsAlreadyExists(err) {
			t.Logger.Info("create container error", "error", err)
			return deploy.Error(err)
		}
	}
	return deploy.Done()
}

func (t *BlobContainer) PostCreate(ctx context.Context) deploy.Result {

	name := getBucketName(t.ProjectID, t.GetRegion(), t.getMasterName())
	container := &azurestoragev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(container), container); err != nil {
		t.Logger.Error(err, "failed to get azure blob container")
		return deploy.Error(err)
	}
	var isReady bool
	for _, condition := range container.Status.Conditions {
		if condition.Type == xpv1.TypeReady && condition.Status == corev1.ConditionTrue {
			isReady = true
		}
	}
	if !isReady {
		t.Logger.Info("azure blob container is not ready")
		return deploy.Error(errors.New("azure blob container is not ready"))
	}
	t.DMJob.Status.BlobContainerName = name

	return deploy.Done()
}

func (t *BlobContainer) Delete(ctx context.Context) deploy.Result {

	name := getBucketName(t.ProjectID, t.GetRegion(), t.getMasterName())
	container := &azurestoragev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	if err := t.Local.Delete(ctx, container); err != nil {
		if !apierrors.IsNotFound(err) {
			t.Logger.Error(err, "failed to delete blob container", "name", name)
			return deploy.Error(err)
		}
	}
	return deploy.Done()
}

func (t *BlobContainer) PostDelete(ctx context.Context) deploy.Result {

	name := getBucketName(t.ProjectID, t.GetRegion(), t.getMasterName())
	container := &azurestoragev1beta1.Container{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(container), container); err == nil {
		err = errors.New("blob container still exists:" + name)
		t.Logger.Error(nil, "blob container still exists", "name", name)
		return deploy.Error(err)
	} else if !apierrors.IsNotFound(err) {
		t.Logger.Error(err, "failed to get blob container", "name", name)
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *BlobContainer) isStorageAccountReady(ctx context.Context) error {

	storageAccount := &azurestoragev1beta1.Account{
		ObjectMeta: metav1.ObjectMeta{
			Name:   GetStorageAccountName(t.ProjectID, t.GetRegion(), t.GetClusterScopeName()),
			Labels: t.GetCommonLabels4Azure(),
		},
	}

	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(storageAccount), storageAccount); err != nil {
		t.Logger.Error(err, "failed to get storage account error", "account", storageAccount.GetName())
		return err
	}

	var isReady bool
	for _, condition := range storageAccount.Status.Conditions {
		if condition.Type == xpv1.TypeReady && condition.Status == corev1.ConditionTrue {
			isReady = true
			break
		}
	}
	if !isReady {
		t.Logger.Info("storage account is not ready")
		return errors.New("storage account is not ready")
	}
	return nil
}

func (t *BlobContainer) GetResourceTag() map[string]*string {
	resourceTags := t.BaseTask.GetResourceTag()
	resourceTags["cluster_scope"] = ptr.To(t.GetClusterScopeName())
	resourceTags["dmjob_id"] = ptr.To(t.DMJob.Name)
	return resourceTags
}

func getBucketName(project string, region string, master string) string {
	const (
		labelLengthLimit = 63
		hashSize         = 8
		resultLimit      = labelLengthLimit - hashSize - 1
	)

	result := fmt.Sprintf("dm-%s-%s-%s", project, region, util.FnvHash(master))
	if len(result) > resultLimit {
		result = result[:resultLimit]
	}

	return result
}