package integration

import (
	"context"
	"fmt"
	"os"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	apps "k8s.io/api/apps/v1"
	"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/client"

	apierrors "k8s.io/apimachinery/pkg/api/errors"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"

	corev1 "k8s.io/api/core/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/consts"
	gardencorev1beta1 "github.com/tidbcloud/infra-api/pkg/gardener/core/apis/v1beta1"
	pkgprovider "github.com/tidbcloud/infra-api/pkg/provider"
)

var _ = Describe("CDC controller test", func() {
	const (
		timeout           = time.Second * 120
		interval          = time.Second * 1
		TiCDCTmpFinalizer = "ticdc.tmp.cloud.pingcap.com/finalizer"
	)

	Context("When create cdc", func() {
		It("CDC config should be correct", func() {
			By("Prepare network")
			ctx := context.Background()
			nt, shootClient, gardenStatus := prepareNetwork("cdc", ctx)

			By("Create a basic cluster")
			c := newCluster(nt, "", "", "test")
			Expect(k8sClient.Create(ctx, c)).To(Succeed())
			tcKey := types.NamespacedName{
				Namespace: c.PrefixedId(),
				Name:      consts.ClusterPrefix,
			}
			tc := &v1alpha1.TidbCluster{}
			Eventually(func() error {
				return shootClient.Get(ctx, tcKey, tc)
			}, timeout, interval).Should(Succeed(), "TiDB cluster should be created")

			By("Create a cdc")
			cdc := newCDC(nt.Namespace, nt.Name, c.Name, "")
			Expect(k8sClient.Create(ctx, cdc)).To(Succeed())

			By("Observer ticdc owner references")
			Eventually(func() error {
				if err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: cdc.Namespace,
					Name:      cdc.Name,
				}, cdc); err != nil {
					return err
				}

				refs := cdc.GetOwnerReferences()
				for _, ref := range refs {
					if ref.Kind == "Cluster" && ref.Name == c.Name {
						return nil
					}
				}
				return fmt.Errorf("Waiting owner reference set to cdc object")
			}, timeout, interval).Should(Succeed(), "TiCDC should be have owner reference")

			By("Check tidbcluster have cdc spec in shoot cluster")
			Eventually(func() error {
				if err := shootClient.Get(ctx, tcKey, tc); err != nil {
					return err
				}
				if tc.Spec.TiCDC != nil {
					Expect(tc.Spec.TiCDC.Config.MP["data-dir"]).To(Equal("/var/lib/ticdc"))
					Expect(*tc.Spec.TiCDC.StorageVolumes[0].StorageClassName).To(Equal("ticdc-gp3"))
					Expect(*tc.Spec.TiCDC.HostNetwork).To(Equal(true))
				} else {
					return fmt.Errorf("Wait TiCDC ready")
				}
				return nil
			}, timeout, interval).Should(Succeed(), "TiCDC should be created")

			By("Ticdc status should be creating")
			Eventually(func() error {
				if err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: cdc.Namespace,
					Name:      cdc.Name,
				}, cdc); err != nil {
					return err
				}
				if cdc.Status.Phase == cloudv1.TiCDCPhaseCreating {
					return nil
				} else {
					return fmt.Errorf("Wait TiCDC status to creating")
				}
			}, timeout, interval).Should(Succeed(), "TiCDC should be creating")

			By("Check nodescaler create a nodegroup for ticdc")
			Eventually(func() error {
				if err := ensureCDCNodeGroupSize(gardenStatus.ShootName, upDeployedWgPrefix(c.Name), 1, 1); err != nil {
					return err
				}
				return nil
			}, timeout, interval).Should(Succeed(), "TiCDC node group should be created, and size is 1")

			By("Observer ticdc status to running")
			err := makeTiCDCReady(shootClient, tcKey)
			Expect(err).ToNot(HaveOccurred())

			Eventually(func() error {
				if err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: cdc.Namespace,
					Name:      cdc.Name,
				}, cdc); err != nil {
					return err
				}
				if cdc.Status.Phase == cloudv1.TiCDCPhaseRunning {
					return nil
				} else {
					return fmt.Errorf("Wait TiCDC status to running")
				}
			}, timeout, interval).Should(Succeed(), "TiCDC should be running")

			Eventually(func() error {
				if os.Getenv(pkgprovider.NodeGroupCreated) == "true" {
					return nil
				}
				return fmt.Errorf("Wait create nodegroup function called")
			}, timeout, interval).Should(Succeed(), "CreateNodeGroup function should be called")

			/*
				By("Scaling CDC replicas to 2")
				err = scaleCDCReplicas(k8sClient, types.NamespacedName{
					Namespace: cdc.Namespace,
					Name:      cdc.Name,
				}, 2)
				Expect(err).ToNot(HaveOccurred())

				By("Check nodescaler create two nodegroup for ticdc")
				Eventually(func() error {
					if err := ensureCDCNodeGroupSize(gardenStatus.ShootName, c.Name, 2); err != nil {
						return err
					}
					return nil
				}, timeout, interval).Should(Succeed(), "TiCDC node group should be created, and size is 2")

				Eventually(func() error {
					if err := k8sClient.Get(ctx, types.NamespacedName{
						Namespace: cdc.Namespace,
						Name:      cdc.Name,
					}, cdc); err != nil {
						return err
					}
					if cdc.Status.Phase == cloudv1.TiCDCPhaseScaling {
						return nil
					} else {
						return fmt.Errorf("Wait TiCDC status to scaling")
					}
				}, timeout, interval).Should(Succeed(), "TiCDC status should be scaling")

				By("Observer ticdc status to running")
				err = makeTiCDCReady(shootClient, tcKey)
				Expect(err).ToNot(HaveOccurred())

				Eventually(func() error {
					if err := k8sClient.Get(ctx, types.NamespacedName{
						Namespace: cdc.Namespace,
						Name:      cdc.Name,
					}, cdc); err != nil {
						return err
					}
					if cdc.Status.Phase == cloudv1.TiCDCPhaseRunning {
						return nil
					} else {
						return fmt.Errorf("Wait TiCDC status to running")
					}
				}, timeout, interval).Should(Succeed(), "TiCDC should be running")
			*/

			By("Delete cdc")
			err = k8sClient.Delete(ctx, cdc)
			Expect(err).ToNot(HaveOccurred())

			Eventually(func() error {
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: cdc.Namespace,
					Name:      cdc.Name,
				}, cdc)
				if apierrors.IsNotFound(err) {
					return nil
				}
				return fmt.Errorf("Wait TiCDC clean up")
			}, timeout, interval).Should(Succeed(), "TiCDC should be deleted")

			By("Check nodescaler scaling-in to 0")
			Eventually(func() error {
				if err := ensureCDCNodeGroupDeleted(gardenStatus.ShootName, upDeployedWgPrefix(c.Name)); err != nil {
					return err
				}
				return nil
			}, timeout, interval).Should(Succeed(), "TiCDC node group should be created, and size is 2")

			By("Check ticdc deleted in shoot")
			Eventually(func() error {
				if err := shootClient.Get(ctx, tcKey, tc); err != nil {
					return err
				}
				if tc.Spec.TiCDC == nil {
					return nil
				} else {
					return fmt.Errorf("Wait TiCDC ready")
				}
			}, timeout, interval).Should(Succeed(), "TiCDC should be deleted in shoot")

			By("Check ticdc component cert deleted in shoot")
			Eventually(func() error {
				err := shootClient.Get(ctx, types.NamespacedName{
					Namespace: c.Namespace,
					Name:      fmt.Sprintf("%s-%s-cluster-secret", c.Name, cloudv1.ComponentTiCDC),
				}, tc)
				if apierrors.IsNotFound(err) {
					return nil
				}
				return fmt.Errorf("Wait ticdc component cert deleted")
			}, timeout, interval).Should(Succeed(), "TiCDC component cert should be deleted in shoot")

			Eventually(func() error {
				if os.Getenv(pkgprovider.NodeGroupDeleted) == "true" {
					return nil
				}
				return fmt.Errorf("Wait delete nodegroup function called")
			}, timeout, interval).Should(Succeed(), "DeleteNodeGroup function should be called")
		})
	})

	Context("When create replication", func() {
		It("CDC config should be correct", func() {
			By("Prepare network")
			ctx := context.Background()
			upstreamNetwork, upstreamClient, _ := prepareNetwork("replication", ctx)
			downstreamNetwork, downstreamClient, gardenStatus := prepareNetwork("replication", ctx)

			By("Create a cluster in upstream network")
			upstreamCluster := newCluster(upstreamNetwork, "", "", "test")
			Expect(k8sClient.Create(ctx, upstreamCluster)).To(Succeed())
			downstreamCluster := newCluster(downstreamNetwork, "", "", "test")
			Expect(k8sClient.Create(ctx, downstreamCluster)).To(Succeed())

			By("Create a cdc")
			cdc := newCDC("replication", downstreamNetwork.Name, downstreamCluster.Name, upstreamCluster.Name)
			Expect(k8sClient.Create(ctx, cdc)).To(Succeed())
			Eventually(func() error {
				svc := &corev1.Service{}
				err := upstreamClient.Get(ctx, types.NamespacedName{
					Namespace: "kube-system",
					Name:      "coredns-lb",
				}, svc)
				if err != nil {
					return err
				}
				return nil
			}, timeout, interval).Should(Succeed(), "Should be create load labalance type svc for coredns in upstream")

			Eventually(func() error {
				tc := &v1alpha1.TidbCluster{}
				err := downstreamClient.Get(ctx, types.NamespacedName{
					Namespace: downstreamCluster.PrefixedId(),
					Name:      fmt.Sprintf("%s-%s", consts.ClusterPrefix, "cdc"),
				}, tc)
				if err != nil {
					return err
				}
				By("Cdc log should by print to STDOUT")
				Expect(tc.Spec.TiCDC.Config.Get("log-file").MustString()).Should(Equal(""))

				By("Cdc update strategy shoud be rolling update")
				Expect(tc.Spec.ConfigUpdateStrategy).Should(Equal(v1alpha1.ConfigUpdateStrategyRollingUpdate))

				By("Cdc pv reclaim policy should be delete")
				Expect(*tc.Spec.PVReclaimPolicy).Should(Equal(corev1.PersistentVolumeReclaimDelete))
				if tc.Spec.DNSConfig == nil {
					return fmt.Errorf("DNS config should be not empty")
				} else {
					Expect(tc.Spec.DNSConfig.Searches).Should(Equal([]string{
						"svc.cluster.local",
						"cluster.local",
						"ec2.internal",
					}))
					Expect(tc.Spec.DNSConfig.Options).Should(Equal([]corev1.PodDNSConfigOption{
						{
							Name:  "ndots",
							Value: pointer.String("5"),
						},
					}))
				}
				return nil
			}, timeout, interval).Should(Succeed(), "ticdc should be create in downstream cluster")

			By("Delete ticdc")
			ticdc := &cloudv1.TiCDC{}
			err := k8sClient.Get(ctx, types.NamespacedName{
				Namespace: cdc.Namespace,
				Name:      cdc.Name,
			}, ticdc)
			Expect(err).NotTo(HaveOccurred())
			err = k8sClient.Delete(ctx, ticdc)
			Expect(err).ToNot(HaveOccurred())

			By("Check coredns and db-cdc shoud be deleted in downstream cluster")
			Eventually(func() error {
				svc := &corev1.Service{}
				err := downstreamClient.Get(ctx, types.NamespacedName{
					Namespace: "kube-system",
					Name:      fmt.Sprintf("coredns-%s", downstreamCluster.Name),
				}, svc)

				if !apierrors.IsNotFound(err) {
					return fmt.Errorf("Wait coredns cleared")
				}

				tc := &v1alpha1.TidbCluster{}
				err = downstreamClient.Get(ctx, types.NamespacedName{
					Namespace: downstreamCluster.PrefixedId(),
					Name:      "db-cdc",
				}, tc)
				if !apierrors.IsNotFound(err) {
					return fmt.Errorf("wait db-cdc deleted")
				}
				return nil
			}, timeout, interval).Should(Succeed(), "Coredns and db-cdc should be deleted is downstream cluster")

			By("Check nodescaler scaling-in to 0")
			Eventually(func() error {
				if err := ensureCDCNodeGroupDeleted(gardenStatus.ShootName, downDeployedWgPrefix(downstreamCluster.Name)); err != nil {
					return err
				}
				return nil
			}, timeout, interval).Should(Succeed(), "TiCDC node group should be created, and size is 0")

			Eventually(func() error {
				err := k8sClient.Get(ctx, types.NamespacedName{
					Namespace: cdc.Namespace,
					Name:      cdc.Name,
				}, cdc)
				if apierrors.IsNotFound(err) {
					return nil
				}
				return fmt.Errorf("wait TiCDC clean up")
			}, timeout, interval).Should(Succeed(), "TiCDC should be deleted")
		})
	})
})

func ensureCDCNodeGroupSize(shootName, wgPrefixName string, expectNodeGroupSize, workerSize int) error {
	shoot := new(gardencorev1beta1.Shoot)
	err := k8sClient.Get(context.Background(), types.NamespacedName{
		Namespace: "garden-test",
		Name:      shootName,
	}, shoot)
	if err != nil {
		return err
	}

	workers := shoot.Spec.Provider.Workers
	tiCDCNodeGroupCount := 0

	for _, worker := range workers {
		if worker.Labels["gardener.wg.prefix.name"] == wgPrefixName {
			tiCDCNodeGroupCount++
			Expect(worker.Maximum).Should(Equal(int32(workerSize)))
			Expect(worker.Minimum).Should(Equal(int32(workerSize)))
			Expect(worker.Machine.Type).Should(Equal("r5.2xlarge"))
		}
	}
	if tiCDCNodeGroupCount != expectNodeGroupSize {
		return fmt.Errorf("Wait ticdc node group created")
	}
	return nil
}

func ensureCDCNodeGroupDeleted(shootName, wgPrefixName string) error {
	shoot := new(gardencorev1beta1.Shoot)
	err := k8sClient.Get(context.Background(), types.NamespacedName{
		Namespace: "garden-test",
		Name:      shootName,
	}, shoot)
	if err != nil {
		return err
	}

	workers := shoot.Spec.Provider.Workers
	for _, worker := range workers {
		if worker.Labels["gardener.wg.prefix.name"] == wgPrefixName {
			return fmt.Errorf("wait ticdc node group deleted")
		}
	}
	return nil
}

/*
func scaleCDCReplicas(cli client.Client, ts types.NamespacedName, replicas int) error {
	ctx := context.TODO()
	cdc := &cloudv1.TiCDC{}
	if err := cli.Get(ctx, ts, cdc); err != nil {
		return err
	}
	cdc.Spec.Replicas = 2

	return cli.Update(context.TODO(), cdc)
}
*/

func makeTiCDCReady(shootClient client.Client, ts types.NamespacedName) error {
	ctx := context.TODO()
	tc := &v1alpha1.TidbCluster{}
	if err := shootClient.Get(ctx, ts, tc); err != nil {
		return err
	}
	tc.Status.TiCDC.Phase = v1alpha1.NormalPhase
	tc.Status.TiCDC.StatefulSet = &apps.StatefulSetStatus{ReadyReplicas: tc.Spec.TiCDC.Replicas}
	// Operator do not use status subresource
	return shootClient.Update(context.TODO(), tc)
}

func newCDC(namespace, networkName, clusterName string, upstreamCluster string) *cloudv1.TiCDC {
	return &cloudv1.TiCDC{
		ObjectMeta: metav1.ObjectMeta{
			GenerateName: "sample-cdc",
			Namespace:    namespace,
		},
		Spec: cloudv1.TiCDCSpec{
			NetworkName:         networkName,
			ClusterName:         clusterName,
			StorageSize:         resource.MustParse("1Ti"),
			Replicas:            1,
			InstanceType:        "r5.2xlarge",
			UpstreamClusterName: upstreamCluster,
		},
	}
}

func upDeployedWgPrefix(clusterName string) string {
	return fmt.Sprintf("tidb%s-ticdc", clusterName)
}

func downDeployedWgPrefix(clusterName string) string {
	return fmt.Sprintf("tidb%s-cdc-ticdc", clusterName)
}