package integration

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	batchv1 "k8s.io/api/batch/v1"
	v1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/imports"
	gardencorev1beta1 "github.com/tidbcloud/infra-api/pkg/gardener/core/apis/v1beta1"
	"github.com/tidbcloud/infra-api/pkg/nodegroup"
	"github.com/tidbcloud/infra-api/pkg/util"
)

var _ = Describe("Import controller", func() {

	// Define utility constants for object names and testing timeouts/durations and intervals.
	const (
		timeout  = time.Second * 120
		interval = time.Second * 1
	)

	var (
		ctx           context.Context
		testNamespace string

		ns            *v1.Namespace
		network       *cloudv1.Network
		networkStatus *cloudv1.GardenerNetworkStatus
		cluster       *cloudv1.Cluster
		clusterConfig *cloudv1.ClusterConfig
		importObj     *cloudv1.Import
	)

	BeforeEach(func() {
		ctx = context.Background()
		testNamespace = cloudv1.RandomString(5)
		ns = &v1.Namespace{
			ObjectMeta: metav1.ObjectMeta{
				Name: testNamespace,
			},
		}
		network = &cloudv1.Network{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "network-" + cloudv1.RandomString(8),
				Namespace: "default",
				Labels: map[string]string{
					cloudv1.ProjectLabel: cloudv1.RandomString(5),
				},
			},
		}
		cluster = &cloudv1.Cluster{
			ObjectMeta: metav1.ObjectMeta{
				Name:      fmt.Sprintf("%d", rand.Int()),
				Namespace: "default",
				Labels: map[string]string{
					"tags.tidbcloud.com/billing-project-id": "1",
					"tags.tidbcloud.com/tenant":             "666666",
					"tags.tidbcloud.com/cluster":            "888999",
				},
			},
			Spec: cloudv1.ClusterSpec{
				NetworkName: network.Name,
				Cluster: cloudv1.ClusterTopology{
					TiDB: cloudv1.TiDB{
						Replicas:     2,
						InstanceType: "test",
					},
					TiKV: cloudv1.TiKV{
						Replicas:     3,
						InstanceType: "test",
						StorageSize:  resource.MustParse("100Gi"),
					},
					PD: cloudv1.PD{
						Replicas:     3,
						InstanceType: "test",
						StorageSize:  resource.MustParse("1Gi"),
					},
				},
				SqlEndpoints: cloudv1.SqlEndpoints{
					Public: cloudv1.Service{
						Enabled: pointer.Bool(true),
					},
				},
			},
		}
		clusterConfig = &cloudv1.ClusterConfig{
			ObjectMeta: metav1.ObjectMeta{
				Name:      fmt.Sprintf("%d", rand.Int()),
				Namespace: "default",
			},
			Spec: cloudv1.ClusterConfigSpec{
				Selector: &metav1.LabelSelector{
					MatchLabels: map[string]string{
						"tags.tidbcloud.com/tenant":  "666666",
						"tags.tidbcloud.com/cluster": "888999",
					},
				},
				Lightning: &cloudv1.Configuration{
					BaseImage: pointer.StringPtr("gcr.io/pingcap-public/dbaas/tidb-lightning"),
					Tag:       pointer.StringPtr("v5.3.0-20211123"),
					Config: &cloudv1.TomlConfig{
						MP: map[string]interface{}{"mydumper": map[string]string{"strict-format": "false"}},
					},
				},
			},
		}
		importObj = &cloudv1.Import{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "import" + cloudv1.RandomString(8),
				Namespace: "default",
			},
		}
	})

	JustBeforeEach(func() {
		Expect(k8sClient.Create(ctx, ns)).To(Succeed())
		Expect(k8sClient.Create(ctx, network)).To(Succeed())
		Eventually(func() error {
			var err error
			networkStatus, err = networkNormal(ctx, network, true)
			return err
		}, timeout, interval).Should(Succeed())
		Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
		Expect(k8sClient.Create(ctx, clusterConfig)).To(Succeed())
	})
	AfterEach(func() {
	})

	Context("When reconcile import", func() {
		Context("With provider AWS", func() {
			var (
				clusterStatus *cloudv1.AwsClusterStatus
			)
			BeforeEach(func() {
				network.Spec = cloudv1.NetworkSpec{
					Provider: "aws",
					Region:   "us-east-2",
					CIDR:     "10.251.0.0/16",
				}
				importObj.Spec = cloudv1.ImportSpec{
					ClusterName:    cluster.Name,
					StorageRequest: "100Mi",
					Source: cloudv1.Source{
						Format:   cloudv1.DataFormatAuroraParquet,
						UserName: "test",
						Password: "test",
						Storage: cloudv1.Storage{
							S3: &cloudv1.S3{
								URL:    "s3://test",
								Region: "us-west-2",
							},
						},
					},
				}
			})
			JustBeforeEach(func() {
				Eventually(func() bool {
					err := k8sClient.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
					if err != nil {
						return false
					}
					clusterStatus, err = cloudv1.AwsClusterStatusFromCluster(cluster)
					if err != nil {
						return false
					}
					return cluster.Status.Phase == cloudv1.ClusterPhaseRunning
				}, timeout, interval).Should(BeTrue(), "cluster phase should be running")
				Expect(k8sClient.Create(ctx, importObj)).To(Succeed())
			})
			It("Should reconcile AWS import", func() {
				Eventually(func() bool {
					if err := k8sClient.Get(ctx, types.NamespacedName{Name: importObj.Name, Namespace: importObj.Namespace}, importObj); err != nil {
						return false
					}
					return controllerutil.ContainsFinalizer(importObj, imports.ImportFinalizer)
				}, timeout, interval).Should(BeTrue(), "should add finalizer to Import")

				Eventually(func() bool {
					var job batchv1.Job
					return k8sClient.Get(ctx, types.NamespacedName{Name: importObj.Name, Namespace: clusterStatus.Kubernetes.Namespace}, &job) == nil
				}, timeout, interval).Should(BeTrue(), "should create import job")

				Eventually(func() bool {
					var svc v1.Service
					return k8sClient.Get(ctx, types.NamespacedName{Name: importObj.Name, Namespace: clusterStatus.Kubernetes.Namespace}, &svc) == nil
				}, timeout, interval).Should(BeTrue(), "should create import service")

				Eventually(func() bool {
					return hasImportWorker(ctx, networkStatus.ShootName, importObj)
				}, timeout, interval).Should(BeTrue(), "should container import worker")

				Expect(k8sClient.Delete(ctx, importObj)).Should(Succeed())

				Eventually(func() bool {
					var svc v1.Service
					err := k8sClient.Get(ctx, types.NamespacedName{Name: importObj.Name, Namespace: clusterStatus.Kubernetes.Namespace}, &svc)
					return kerrors.IsNotFound(err)
				}, timeout, interval).Should(BeTrue(), "should delete import service")

				Eventually(func() bool {
					return !hasImportWorker(ctx, networkStatus.ShootName, importObj)
				}, timeout, interval).Should(BeTrue(), "should delete import worker")
			})

			Context("For precheck", func() {
				BeforeEach(func() {
					importObj.Spec.DryRun = true
					importObj.Spec.AccessConfig = cloudv1.AccessConfig{
						RoleArn: "role-arn-default",
					}
				})
				It("Should reconcile AWS import precheck", func() {
					Eventually(func() bool {
						if err := k8sClient.Get(ctx, types.NamespacedName{Name: importObj.Name, Namespace: importObj.Namespace}, importObj); err != nil {
							return false
						}
						return controllerutil.ContainsFinalizer(importObj, imports.ImportFinalizer)
					}, timeout, interval).Should(BeTrue(), "should add finalizer to Import")

					precheckName := importObj.Name + "-precheck"
					key := types.NamespacedName{Name: precheckName, Namespace: clusterStatus.Kubernetes.Namespace}
					Eventually(func() bool {
						return k8sClient.Get(ctx, key, &batchv1.Job{}) == nil
					}, timeout, interval).Should(BeTrue(), "should create import precheck job")

					Expect(k8sClient.Delete(ctx, importObj)).Should(Succeed())

					Eventually(func() bool {
						var job batchv1.Job
						err := k8sClient.Get(ctx, key, &job)
						return apierrors.IsNotFound(err) || !job.DeletionTimestamp.IsZero()
					}, timeout, interval).Should(BeTrue(), "should delete import precheck job")

				})

			})

			Context("With storage request", func() {
				var storageQuantity = "100G"
				BeforeEach(func() {
					importObj.Spec.StorageRequest = storageQuantity
				})
				It("Should reconcile corresponding PVC", func() {
					key := types.NamespacedName{Name: importObj.Name, Namespace: clusterStatus.Kubernetes.Namespace}
					var pvc v1.PersistentVolumeClaim
					Eventually(func() bool {
						return k8sClient.Get(ctx, key, &pvc) == nil
					}, timeout, interval).Should(BeTrue(), "should create pvc")
					Expect(pvc.Spec.Resources.Requests[v1.ResourceStorage]).To(Equal(resource.MustParse(storageQuantity)))
					Expect(k8sClient.Delete(ctx, importObj)).Should(Succeed())
					Eventually(func() bool {
						var deletePVC v1.PersistentVolumeClaim
						err := k8sClient.Get(ctx, key, &deletePVC)
						return apierrors.IsNotFound(err) || !deletePVC.DeletionTimestamp.IsZero()
					}, timeout, interval).Should(BeTrue(), "should delete import pvc")
				})
			})
		})
	})
})

func hasImportWorker(ctx context.Context, shootName string, i *cloudv1.Import) bool {
	var shoot gardencorev1beta1.Shoot
	err := k8sClient.Get(ctx, types.NamespacedName{Name: shootName, Namespace: gardenerEnv.GetShootNamespace()}, &shoot)
	if err != nil {
		return false
	}

	for _, w := range shoot.Spec.Provider.Workers {
		if w.Labels[nodegroup.GardenerWorkerPrefixLabelName] == nodegroup.NodeGroupNamePrefix(util.FnvHash(fmt.Sprintf("%s-%s", i.Spec.ClusterName, i.Name)), imports.TiDBLightningComName) {
			return true
		}
	}

	return false
}