package aks

import (
	"context"
	"fmt"
	"strings"

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

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/scheme"
)

type ProviderConfig2 struct {
	task.BaseTask
}

func NewProviderConfig2(ctx *task.Context) *ProviderConfig2 {
	return &ProviderConfig2{BaseTask: task.BaseTask{Context: ctx}}
}

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

func (t *ProviderConfig2) Create(ctx context.Context) deploy.Result {
	cl := cloudv1.ClusterList{}
	if err := t.Local.List(ctx, &cl, client.MatchingFields{
		"spec.kubernetes.name": t.Cluster.Spec.Kubernetes.Name,
	}); err != nil {
		return deploy.Error(err)
	}

	anno := map[string]string{}
	for _, c := range cl.Items {
		if !c.DeletionTimestamp.IsZero() {
			continue
		}

		anno[annoKeyPrefixUsedByCluster+c.Name] = annoValInUse
	}

	config := &v1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name:        t.GetAKSDataplaneProviderConfigName(),
			Annotations: anno,
			Labels: map[string]string{
				"tidbcloud-namespace": t.Cluster.Namespace,
			},
		},
		Spec: v1beta1.ProviderConfigSpec{
			Credentials: v1beta1.ProviderCredentials{
				Source: "OIDCTokenFile",
			},
			SubscriptionID: ptr.To(t.Cluster.Spec.Kubernetes.Account),
			ClientID:       ptr.To(t.Config.ClusterController.Azure.CrossplaneAzureProviderClientID),
			TenantID:       ptr.To(t.Config.ClusterController.Azure.CrossplaneAzureProviderTenantID),
		},
	}

	if err := k8s.Apply(ctx, t.Local, config); err != nil {
		return deploy.Panic(err)
	}

	if !config.DeletionTimestamp.IsZero() {
		return deploy.Panic(fmt.Errorf("provider config for aks %s is deleting, wait for it being deleted", t.Cluster.Spec.Kubernetes.Name))
	}

	return deploy.Done()
}

func (t *ProviderConfig2) Delete(ctx context.Context) deploy.Result {
	config := &v1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.GetAKSDataplaneProviderConfigName(),
		},
	}

	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(config), config); err != nil {
		if !apierrors.IsNotFound(err) {
			return deploy.Error(err)
		}

		return deploy.Done()
	}

	if _, ok := config.Annotations[annoKeyPrefixUsedByCluster+t.Cluster.Name]; ok {
		if err := k8s.PatchObject(ctx, t.Local, config, removeAnnoPatch(annoKeyPrefixUsedByCluster+t.Cluster.Name)); err != nil {
			return deploy.Error(err)
		}
	}

	for k := range config.Annotations {
		// there is still at least one cluster, skip providerconfig deletion
		if strings.HasPrefix(k, annoKeyPrefixUsedByCluster) {
			return deploy.Done()
		}
	}

	if err := t.deleteAll(ctx); err != nil {
		return deploy.Error(err)
	}

	return deploy.ErrorOrDone(k8s.DeleteObjectCompleted(ctx, t.Local, config))
}

func (t *ProviderConfig2) deleteAll(ctx context.Context) error {
	errList := []error{}
	usl := resource.MustCreateObject(v1beta1.ProviderConfigUsageListGroupVersionKind, scheme.Scheme).(resource.ProviderConfigUsageList)
	if err := t.Local.List(ctx, usl, client.MatchingLabels{
		xpv1.LabelKeyProviderName: t.GetAKSDataplaneProviderConfigName(),
	}); err != nil {
		errList = append(errList, err)
	}

	for _, item := range usl.GetItems() {
		ref := item.GetResourceReference()
		obj := resource.MustCreateObject(ref.GroupVersionKind(), scheme.Scheme).(client.Object)
		obj.SetName(ref.Name)
		if err := t.Local.Delete(ctx, obj); err != nil {
			errList = append(errList, err)
		}
	}

	if len(errList) != 0 {
		return errors.NewAggregate(errList)
	}

	return nil
}