package integration

import (
	"context"
	"fmt"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"

	certv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1"
	"github.com/pelletier/go-toml"
	opv2alpha1 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/pkg/cdccluster"
	ticdc_consts "github.com/tidbcloud/ticdc-service/pkg/consts"
	azv1beta1 "github.com/upbound/provider-azure/apis/v1beta1"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	k8sVersion = "1.30"
)

var _ = Describe("TiCDC Controller", func() {
	Context("When reconciling a resource", func() {
		It("Should create a TiCDC cluster without error", func() {
			const (
				resourceName = "test-ticdc"
				clusterID    = "123"
			)

			timeout := 10 * time.Second
			interval := 1 * time.Second
			ctx := context.Background()

			// dataplane k8s namespace does not exist due to no real tidb cluster
			// so we need to prepare this namespace for other dataplane resources to be created
			// such as certificate, tidbcluster
			remoteNS := fmt.Sprintf("tidb%s", clusterID)
			Expect(k8sClient.Create(ctx, &corev1.Namespace{
				ObjectMeta: metav1.ObjectMeta{Name: remoteNS},
			})).To(Succeed())

			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      clusterID,
					Namespace: "default",
					Annotations: map[string]string{
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseTiDBOperatorV2:                  "false",
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseSelfSignedCertForTiDBComponents: "true",
					},
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeAKS,
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						PD:   cloudv1.PD{Replicas: 3},
						TiKV: cloudv1.TiKV{Replicas: 3},
					},
					SqlEndpoints: cloudv1.SqlEndpoints{
						Public: cloudv1.Service{
							Enabled: ptr.To(true),
						},
					},
				},
			}
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())

			ticdc := &cdcv1.TiCDC{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: "default",
					Labels: map[string]string{
						cloudv1.TenantTagKey:  "999",
						cloudv1.ProjectTagKey: "888",
						cloudv1.ClusterTagKey: clusterID,
					},
				},
				Spec: cdcv1.TiCDCSpec{
					ClusterID:    clusterID,
					Replicas:     3,
					NodeVersion:  k8sVersion,
					InstanceType: "standard-8c",
					Storage: cdcv1.StorageSpec{
						Type:          "ssd",
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 250,
					},
					Configuration: &cdcv1.TiCDCConfiguration{
						Config: "# empty config",
					},
				},
			}
			Expect(k8sClient.Create(ctx, ticdc)).To(Succeed())

			// controller will create providerconfig, nodepool, storageclass
			pc := &azv1beta1.ProviderConfig{
				ObjectMeta: metav1.ObjectMeta{
					Name: fmt.Sprintf("cdc-%s", k8sName),
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(pc), pc)).To(Succeed())
			}, timeout, interval).Should(Succeed())

			// controller will create cluster cert
			clusterCert := &certv1.Certificate{
				ObjectMeta: metav1.ObjectMeta{
					Name:      fmt.Sprintf("%s-ticdc-cluster-secret", resourceName),
					Namespace: remoteNS,
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clusterCert), clusterCert)).To(Succeed())
			}, timeout, interval).Should(Succeed())
			// controller will create client cert
			clientCert := &certv1.Certificate{
				ObjectMeta: metav1.ObjectMeta{
					Name:      fmt.Sprintf("%s-cluster-client-secret", resourceName),
					Namespace: remoteNS,
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clientCert), clientCert)).To(Succeed())
			}, timeout, interval).Should(Succeed())

			// controller will create TidbCluster CR
			tc := &opv1alpha1.TidbCluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: remoteNS,
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(tc), tc)).To(Succeed())
			}, timeout, interval).Should(Succeed())
			Expect(tc.Spec.TiCDC.Config.Get("cluster-id").MustString()).To(Equal(resourceName))
			Expect(tc.Spec.TiCDC.Config.Get("data-dir").MustString()).To(Equal(cdccluster.TiCDCDefaultDirPath))
		})

		It("Should create a TiCDC cluster using operator v2 without error", func() {
			const (
				resourceName = "test-v2-ticdc"
				clusterID    = "123v2"
			)

			timeout := 10 * time.Second
			interval := 1 * time.Second
			ctx := context.Background()

			// dataplane k8s namespace does not exist due to no real tidb cluster
			// so we need to prepare this namespace for other dataplane resources to be created
			// such as certificate, tidbcluster
			remoteNS := fmt.Sprintf("tidb%s", clusterID)
			Expect(k8sClient.Create(ctx, &corev1.Namespace{
				ObjectMeta: metav1.ObjectMeta{Name: remoteNS},
			})).To(Succeed())

			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      clusterID,
					Namespace: "default",
					Annotations: map[string]string{
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseTiDBOperatorV2:                  "true",
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseSelfSignedCertForTiDBComponents: "true",
					},
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeAKS,
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						PD:   cloudv1.PD{Replicas: 3},
						TiKV: cloudv1.TiKV{Replicas: 3},
					},
					SqlEndpoints: cloudv1.SqlEndpoints{
						Public: cloudv1.Service{
							Enabled: ptr.To(true),
						},
					},
				},
			}
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())

			ticdc := &cdcv1.TiCDC{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: "default",
					Labels: map[string]string{
						cloudv1.TenantTagKey:  "999",
						cloudv1.ProjectTagKey: "888",
						cloudv1.ClusterTagKey: clusterID,
					},
				},
				Spec: cdcv1.TiCDCSpec{
					ClusterID:    clusterID,
					Replicas:     3,
					NodeVersion:  k8sVersion,
					InstanceType: "standard-8c",
					Storage: cdcv1.StorageSpec{
						Type:          "ssd",
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 250,
					},
					Configuration: &cdcv1.TiCDCConfiguration{
						Config:    "# empty config",
						BaseImage: "busybox",
						Tag:       "v8.5.2",
					},
				},
			}
			Expect(k8sClient.Create(ctx, ticdc)).To(Succeed())

			// controller will create TidbCluster CR
			cdcGroup := &opv2alpha1.TiCDCGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: remoteNS,
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cdcGroup), cdcGroup)).To(Succeed())
			}, timeout, interval).Should(Succeed())
			cdcConf := opv1alpha1.NewCDCConfig()
			Expect(toml.Unmarshal([]byte(cdcGroup.Spec.Template.Spec.Config), &cdcConf.GenericConfig.MP)).To(Succeed())
			Expect(cdcConf.Get("cluster-id").MustString()).To(Equal(resourceName))
			Expect(cdcConf.Get("data-dir").MustString()).To(Equal(cdccluster.TiCDCDefaultDirPath))
		})

		It("Should create overprovision pods when ticdc scale vertically", func() {
			const (
				resourceName = "test-ticdc-vertical-scale"
				clusterID    = "123vertscale"
			)

			timeout := 10 * time.Second
			interval := 1 * time.Second
			ctx := context.Background()

			// dataplane k8s namespace does not exist due to no real tidb cluster
			// so we need to prepare this namespace for other dataplane resources to be created
			// such as certificate, tidbcluster
			remoteNS := fmt.Sprintf("tidb%s", clusterID)
			Expect(k8sClient.Create(ctx, &corev1.Namespace{
				ObjectMeta: metav1.ObjectMeta{Name: remoteNS},
			})).To(Succeed())

			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      clusterID,
					Namespace: "default",
					Annotations: map[string]string{
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseTiDBOperatorV2:                  "true",
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseSelfSignedCertForTiDBComponents: "true",
					},
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeAKS,
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						PD:   cloudv1.PD{Replicas: 3},
						TiKV: cloudv1.TiKV{Replicas: 3},
					},
					SqlEndpoints: cloudv1.SqlEndpoints{
						Public: cloudv1.Service{
							Enabled: ptr.To(true),
						},
					},
				},
			}
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())

			ticdc := &cdcv1.TiCDC{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: "default",
					Labels: map[string]string{
						cloudv1.TenantTagKey:  "999",
						cloudv1.ProjectTagKey: "888",
						cloudv1.ClusterTagKey: clusterID,
					},
				},
				Spec: cdcv1.TiCDCSpec{
					ClusterID:    clusterID,
					Replicas:     3,
					NodeVersion:  k8sVersion,
					InstanceType: "standard-8c",
					Storage: cdcv1.StorageSpec{
						Type:          "ssd",
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 250,
					},
					Configuration: &cdcv1.TiCDCConfiguration{
						Config:    "# empty config",
						BaseImage: "busybox",
						Tag:       "v8.5.2",
					},
				},
			}
			Expect(k8sClient.Create(ctx, ticdc)).To(Succeed())

			ticdc.Spec.InstanceType = "standard-16c"
			Expect(k8sClient.Update(ctx, ticdc)).To(Succeed())

			// controller will create overprovision deployment
			deploy := &appsv1.Deployment{
				ObjectMeta: metav1.ObjectMeta{
					Name:      overscaleName,
					Namespace: overscaleNS,
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(deploy), deploy)).To(Succeed())
			}, timeout, interval).Should(Succeed())

			// TiCDCGroup instance type should not be changed since new nodes not ready
			cdcGroup := &opv2alpha1.TiCDCGroup{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: remoteNS,
				},
			}
			Consistently(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cdcGroup), cdcGroup)).To(Succeed())
				nodeSelector := cdcGroup.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector
				g.Expect(nodeSelector[consts.InstanceTypeLabelKey]).To(Equal("standard-8c"))
			}, 3*timeout, interval).Should(Succeed())

			// when nodes ready, TiCDCGroup instance type would be changed
			fakeScaler.ready = true
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(cdcGroup), cdcGroup)).To(Succeed())
				nodeSelector := cdcGroup.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector
				g.Expect(nodeSelector[consts.InstanceTypeLabelKey]).To(Equal("standard-8c"))
			}, timeout, interval).Should(Succeed())
		})

		It("should restart ticdc cluster by operator v1", func() {
			const (
				resourceName = "test-ticdc-restart-operator-v1"
				clusterID    = "123ticdc-restart-operator-v1"
			)

			timeout := 10 * time.Second
			interval := 1 * time.Second
			ctx := context.Background()

			// dataplane k8s namespace does not exist due to no real tidb cluster
			// so we need to prepare this namespace for other dataplane resources to be created
			// such as certificate, tidbcluster
			remoteNS := fmt.Sprintf("tidb%s", clusterID)
			Expect(k8sClient.Create(ctx, &corev1.Namespace{
				ObjectMeta: metav1.ObjectMeta{Name: remoteNS},
			})).To(Succeed())

			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      clusterID,
					Namespace: "default",
					Annotations: map[string]string{
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseTiDBOperatorV2:                  "false",
						cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseSelfSignedCertForTiDBComponents: "true",
					},
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeAKS,
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						PD:   cloudv1.PD{Replicas: 3},
						TiKV: cloudv1.TiKV{Replicas: 3},
					},
					SqlEndpoints: cloudv1.SqlEndpoints{
						Public: cloudv1.Service{
							Enabled: ptr.To(true),
						},
					},
				},
			}
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())

			ticdc := &cdcv1.TiCDC{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: "default",
					Labels: map[string]string{
						cloudv1.TenantTagKey:  "999",
						cloudv1.ProjectTagKey: "888",
						cloudv1.ClusterTagKey: clusterID,
					},
				},
				Spec: cdcv1.TiCDCSpec{
					ClusterID:    clusterID,
					Replicas:     3,
					NodeVersion:  k8sVersion,
					InstanceType: "standard-8c",
					Storage: cdcv1.StorageSpec{
						Type:          "ssd",
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 250,
					},
					Configuration: &cdcv1.TiCDCConfiguration{
						Config:    "# empty config",
						BaseImage: "busybox",
						Tag:       "v8.5.2",
					},
				},
			}
			Expect(k8sClient.Create(ctx, ticdc)).To(Succeed())

			// controller will create TidbCluster CR
			tc := &opv1alpha1.TidbCluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      resourceName,
					Namespace: remoteNS,
				},
			}
			Eventually(func(g Gomega) {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(tc), tc)).To(Succeed())
			}, timeout, interval).Should(Succeed())
			Expect(tc.Spec.TiCDC).NotTo(BeNil())

			// add restart annotation
			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(ticdc), ticdc)).To(Succeed())
			restartedAt := time.Now().Format(time.RFC3339)
			if len(ticdc.Annotations) == 0 {
				ticdc.Annotations = make(map[string]string)
			}
			ticdc.Annotations[ticdc_consts.RestartTiCDCAnnotation] = restartedAt
			Expect(k8sClient.Update(ctx, ticdc)).To(Succeed())
			Eventually(func(g Gomega) string {
				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(tc), tc)).To(Succeed())
				g.Expect(tc.Spec.TiCDC.Annotations).NotTo(BeEmpty())
				return tc.Spec.TiCDC.Annotations[ticdc_consts.RestartTiCDCOperatorV1Annotation]
			}, timeout, interval).Should(Equal(restartedAt))
		})
	})
})