package integration

import (
	"context"
	"fmt"
	"reflect"
	"time"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	gitopsv2 "github.com/tidbcloud/infra-api/pkg/gitops-v2"
)

const (
	timeout  = time.Second * 120
	interval = time.Second * 2
)

var _ = Describe("FClusterConfig controller", func() {
	var (
		ctx           context.Context
		testNamespace string

		network   *cloudv1.Network
		base      *cloudv1.ClusterConfig
		instance  *cloudv1.ClusterConfig
		tenant    *cloudv1.ClusterConfig
		clustercc *cloudv1.ClusterConfig
		dryruncc  *cloudv1.ClusterConfig
	)

	BeforeEach(func() {
		ctx = context.Background()
		testNamespace = cloudv1.RandomString(5)
		network, _, _ = prepareNetwork(testNamespace, ctx)
		base = &cloudv1.ClusterConfig{}
		Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "base"}, base)).To(Succeed())
		instance = &cloudv1.ClusterConfig{}
		Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "base--awsi32xlarge"}, instance)).To(Succeed())
		tenant = &cloudv1.ClusterConfig{}
		Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "base--t1362657880846635008"}, tenant)).To(Succeed())
		clustercc = &cloudv1.ClusterConfig{}
		Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "base--t1362657880846635008-c123"}, clustercc)).To(Succeed())
		dryruncc = &cloudv1.ClusterConfig{}
		Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "base--t123"}, dryruncc)).To(Succeed())
	})

	AfterEach(func() {
		// cleanup clusters
		cs := &cloudv1.ClusterList{}
		Expect(k8sClient.List(ctx, cs, client.InNamespace(testNamespace))).To(Succeed())
		for _, c := range cs.Items {
			Expect(k8sClient.Delete(context.TODO(), &c)).To(Succeed())
		}
	})

	Context("when clusters are created", func() {
		It("Should use base clusterConfig", func() {
			By("Create a cluster")
			cluster := newCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			Expect(cluster.Spec.ConfigRevisionRef).ToNot(BeNil())
			rev, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster))
			Expect(err).To(BeNil())
			_, existt := rev.Labels[cloudv1.TenantLabel]
			_, existc := rev.Labels[cloudv1.ClusterLabel]
			Expect(rev.Labels[cloudv1.VersionLabel] == base.Spec.Version && !existt && !existc).
				To(BeTrue(), "expected %s cluster's revision inherited from base only", cluster.Name)
		})

		It("Should ref the same configRevision when creating two cluster", func() {
			By("Create two clusters")
			cluster1 := newCluster(network, "1", "12234", "i3.2xlarge")
			cluster2 := newCluster(network, "1", "13342", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster1)).To(Succeed())
			Expect(k8sClient.Create(ctx, cluster2)).To(Succeed())
			By("Check using the same configRevision")
			rev1, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster1))
			Expect(err).To(BeNil())
			rev2, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster2))
			Expect(err).To(BeNil())
			Expect(reflect.DeepEqual(rev1, rev2)).To(BeTrue(), "expected the same, but got cluster1: %+v, cluster2: %+v", rev1, rev2)
		})

		It("Should use tenant clusterconfig", func() {
			By("Create a cluster")
			cluster := newCluster(network, tenant.Labels[cloudv1.TenantLabel], "", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			By("Check using tenant-level configRevision")
			rev, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(cluster))
			Expect(err).To(BeNil())
			_, existc := rev.Labels[cloudv1.ClusterLabel]
			Expect(rev.Labels[cloudv1.TenantLabel] == tenant.Labels[cloudv1.TenantLabel] && !existc).
				To(BeTrue(), "expected %s cluster's revision inherited from tenant", cluster.Name)
		})
	})

	Context("when clusterconfig are updated", func() {
		var (
			// will finally rollback clusterconfig after each node via copies.
			baseCp      *cloudv1.ClusterConfig
			instanceCp  *cloudv1.ClusterConfig
			tenantCp    *cloudv1.ClusterConfig
			clusterccCp *cloudv1.ClusterConfig
			dryrunccCp  *cloudv1.ClusterConfig
		)
		JustBeforeEach(func() {
			baseCp = base.DeepCopy()
			instanceCp = instance.DeepCopy()
			tenantCp = tenant.DeepCopy()
			clusterccCp = clustercc.DeepCopy()
			dryrunccCp = dryruncc.DeepCopy()
		})

		rollback := func(cc *cloudv1.ClusterConfig) error {
			updated := &cloudv1.ClusterConfig{}
			err := k8sClient.Get(ctx, client.ObjectKeyFromObject(cc), updated)
			if err != nil {
				return err
			}
			cc.ObjectMeta.ResourceVersion = updated.ObjectMeta.ResourceVersion
			return k8sClient.Update(ctx, cc)
		}

		JustAfterEach(func() {
			Expect(rollback(baseCp)).To(Succeed())
			Expect(rollback(instanceCp)).To(Succeed())
			Expect(rollback(tenantCp)).To(Succeed())
			Expect(rollback(clusterccCp)).To(Succeed())
			Expect(rollback(dryrunccCp)).To(Succeed())
		})

		It("Should reconcile all related clusters", func() {
			baseCluster := newCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, baseCluster)).To(Succeed())
			tenantCluster := newCluster(network, tenant.Labels[cloudv1.TenantLabel], "", "test")
			Expect(k8sClient.Create(ctx, tenantCluster)).To(Succeed())
			cluster := newCluster(network, clustercc.Labels[cloudv1.TenantLabel], clustercc.Labels[cloudv1.ClusterLabel], "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			By("Update Base ClusterConfig")
			base.Spec.PD.Tag = pointer.StringPtr("v5.2.1-hotfix")
			Expect(k8sClient.Update(ctx, base)).To(Succeed())
			Eventually(func() error {
				err := checkConfigRevisionPDTag(k8sClient, baseCluster, "v5.2.1-hotfix")
				if err != nil {
					return err
				}
				err = checkConfigRevisionPDTag(k8sClient, tenantCluster, "v5.2.1-hotfix")
				if err != nil {
					return err
				}
				return checkConfigRevisionPDTag(k8sClient, cluster, "v5.2.1-hotfix")
			}, timeout, interval).Should(Succeed())

			By("Update instance-level ClusterConfig")
			Expect(gitopsv2.Merge(instance.Spec.TiKV.Config.MP, map[string]interface{}{
				"overlaytest": "5",
			})).To(Succeed())
			Expect(k8sClient.Update(ctx, instance)).To(Succeed())
			Eventually(func() error {
				baseRev, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(baseCluster))
				if err != nil {
					return err
				}
				if _, ok := baseRev.Spec.TiKV.Config.MP["overlaytest"]; !ok {
					return fmt.Errorf("expected overlaytest to be set for base")
				}
				tenantRev, err := getClusterRevision(k8sClient, client.ObjectKeyFromObject(tenantCluster))
				if err != nil {
					return err
				}
				if _, ok := tenantRev.Spec.TiKV.Config.MP["overlaytest"]; ok {
					return fmt.Errorf("expected overlaytest not be set for tenant")
				}
				return nil
			}, timeout, interval).Should(Succeed())

			By("Update tenant-level ClusterConfig")
			tenant.Spec.PD.Tag = pointer.String("v5.2.1-hotfix-tenant")
			Expect(k8sClient.Update(ctx, tenant)).To(Succeed())
			Eventually(func() error {
				err := checkConfigRevisionPDTag(k8sClient, baseCluster, "v5.2.1-hotfix")
				if err != nil {
					return err
				}
				err = checkConfigRevisionPDTag(k8sClient, tenantCluster, "v5.2.1-hotfix-tenant")
				if err != nil {
					return err
				}
				return checkConfigRevisionPDTag(k8sClient, cluster, "v5.2.1-hotfix-tenant")
			}, timeout, interval).Should(Succeed())

			By("Update cluster-level ClusterConfig")
			clustercc.Spec.PD.Tag = pointer.String("v5.2.1-hotfix-tenant-cluster")
			Expect(k8sClient.Update(ctx, clustercc)).To(Succeed())
			Eventually(func() error {
				err := checkConfigRevisionPDTag(k8sClient, baseCluster, "v5.2.1-hotfix")
				if err != nil {
					return err
				}
				err = checkConfigRevisionPDTag(k8sClient, tenantCluster, "v5.2.1-hotfix-tenant")
				if err != nil {
					return err
				}
				return checkConfigRevisionPDTag(k8sClient, cluster, "v5.2.1-hotfix-tenant-cluster")
			}, timeout, interval).Should(Succeed())
		})

		It("Should not update dryrun clusters", func() {
			dryrunCluster := newCluster(network, dryruncc.Labels[cloudv1.TenantLabel], "1111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, dryrunCluster)).To(Succeed())
			By("Update dryrun ClusterConfig")
			dryruncc.Spec.PD.Tag = pointer.String("v5.2.1-hotfix-dryrun")
			Expect(k8sClient.Update(ctx, dryruncc)).To(Succeed())
			Consistently(func() error {
				c := &cloudv1.Cluster{}
				err := k8sClient.Get(ctx, client.ObjectKeyFromObject(dryrunCluster), c)
				if err != nil {
					return err
				}
				if c.Spec.ConfigRevisionRef.Name != dryrunCluster.Spec.ConfigRevisionRef.Name {
					return fmt.Errorf("expected dryrun cluster to not be updated, got %s", c.Spec.ConfigRevisionRef.Name)
				}
				return nil
			}, timeout, interval).Should(Succeed())
		})
	})

	Context("when clusters are deleted", func() {
		It("Should gc the configRevision", func() {
			baseCluster := newCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, baseCluster)).To(Succeed())
			rev := &cloudv1.ConfigRevision{}
			Expect(k8sClient.Get(ctx, client.ObjectKey{Name: baseCluster.Spec.ConfigRevisionRef.Name}, rev)).To(Succeed())
			By("Delete the cluster")
			Expect(k8sClient.Delete(ctx, baseCluster)).To(Succeed())
			Eventually(func() bool {
				err := k8sClient.Get(ctx, client.ObjectKey{Name: baseCluster.Spec.ConfigRevisionRef.Name}, rev)
				return apierrors.IsNotFound(err)
			}, timeout*2, time.Second*15).Should(BeTrue())
		})
	})
})

// checkConfigRevisionPDTag will return err if the configRevision's pd tag and targetTag don't match
func checkConfigRevisionPDTag(cli client.Client, ref *cloudv1.Cluster, targetTag string) error {
	rev, err := getClusterRevision(cli, client.ObjectKeyFromObject(ref))
	if err != nil {
		return err
	}
	if rev.Spec.PD.Tag == nil || *rev.Spec.PD.Tag != targetTag {
		return fmt.Errorf("expected %s cluster's revision to be %s, got %v", ref.Name, targetTag, rev.Spec.TiDB.Tag)
	}
	return nil
}

func getClusterRevision(cli client.Client, clusterRef types.NamespacedName) (*cloudv1.ConfigRevision, error) {
	c := &cloudv1.Cluster{}
	err := cli.Get(context.TODO(), clusterRef, c)
	if err != nil {
		return nil, err
	}
	rev := &cloudv1.ConfigRevision{}
	err = cli.Get(context.TODO(), types.NamespacedName{Name: c.Spec.ConfigRevisionRef.Name}, rev)
	if err != nil {
		return nil, err
	}
	return rev, nil
}