package integration

import (
	"context"
	"fmt"
	"time"

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

	"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"
	cdcv1 "github.com/tidbcloud/ticdc-service-ng/api/v1"
	"github.com/tidbcloud/ticdc-service-ng/pkg/cdccluster"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"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 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())

			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{
					SharedPoolID: clusterID,
					Replicas:     3,
					NodeSpec: cdcv1.NodeSpec{
						CpuCore:           2,
						MemoryGi:          8,
						KubernetesVersion: k8sVersion,
					},
					Storage: cdcv1.StorageSpec{
						Type:          "ssd",
						SizeGi:        100,
						Iops:          3000,
						ThroughputMbs: 250,
					},
					Configuration: &cdcv1.TiCDCConfiguration{
						Config:    "# empty config",
						BaseImage: "busybox",
						Tag:       "next-gen-master",
						Version:   "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))
		})
	})
})