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,
			},
		}
		networkName := "network-" + cloudv1.RandomString(8)
		network = &cloudv1.Network{
			ObjectMeta: metav1.ObjectMeta{
				Name:      networkName,
				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.String("gcr.io/pingcap-public/dbaas/tidb-lightning"),
					Tag:       pointer.String("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",
				Labels:    map[string]string{},
				Annotations: map[string]string{
					"import.tidbcloud.com/network-name": networkName,
				},
			},
		}
	})

	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.Labels[cloudv1.ProviderTypeLabel] = "aws"
				importObj.Labels[cloudv1.RegionLabel] = "us-west-2"
				importObj.Spec = cloudv1.ImportSpec{
					ClusterName:    cluster.Name,
					StorageRequest: "100Mi",
					LightningConfig: `[checkpoint]
					driver = "file"
					dsn = "/etc/endpoint/cp.pb"
					enable = true

					[cron]
					log-progress = "5m"
					switch-mode = "5m"

					[lightning]
					check-requirements = false
					file = ""
					index-concurrency = 4
					level = "info"
					pprof-port = 8289
					table-concurrency = 6

					[mydumper]
					data-source-dir = "fakeURL"
					no-schema = false
					strict-format = false
					[mydumper.csv]
					  backslash-escape = true
					  delimiter = "\""
					  header = true
					  not-null = false
					  null = "\\N"
					  separator = ","

					[post-restore]
					analyze = true
					checksum = true

					[security]
					ca-path = "/var/lib/lightning-tls/ca.crt"
					cert-path = "/var/lib/lightning-tls/tls.crt"
					key-path = "/var/lib/lightning-tls/tls.key"

					[tidb]
					build-stats-concurrency = 4
					distsql-scan-concurrency = 15
					host = "fakeName-tidb"
					index-serial-scan-concurrency = 2
					log-level = "error"
					password = ""
					pd-addr = "fakeName-pd:2379"
					port = 4000
					status-port = 10080
					tls = "false"
					user = ""

					[tikv-importer]
					backend = "local"
					sorted-kv-dir = "/etc/endpoint"
					`,
				}
			})
			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
					}
					if cluster.Status.Phase == cloudv1.ClusterPhaseRunning {
						importObj.Annotations["import.tidbcloud.com/cluster-namespace"] = clusterStatus.Kubernetes.Namespace
						return true
					} else {
						return false
					}
				}, 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("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
}