package integration

import (
	"context"
	"fmt"
	"math/rand"
	"net/http"
	"net/url"
	"reflect"
	"sync"
	"time"

	gh "github.com/google/go-github/v56/github"
	gitopsv2 "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	"github.com/tidbcloud/infra-provider/util/config"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"

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

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

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 = genNamespace()
		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(ctx context.Context) {
			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.TenantTagKey]
			_, existp := rev.Labels[cloudv1.TenantTagKey]
			_, existc := rev.Labels[cloudv1.ClusterTagKey]
			Expect(existt && existp && existc).
				To(BeTrue(), "%s should have both tenant, project and cluster label", rev.Labels)

			Expect(rev.Labels[cloudv1.VersionLabel] == base.Spec.Version).
				To(BeTrue(), "expected %s cluster's revision inherited from base only", cluster.Name)
			Expect(rev.Spec.PD.Config.MP["schedule"] == nil).
				To(BeTrue(), "%s should use base config", rev.Name)
		}, NodeTimeout(time.Second*10))

		It("Should ref different configRevision but with the same config when creating two cluster", func(ctx context.Context) {
			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.Spec, rev2.Spec)).To(BeTrue(), "expected the same, but got cluster1: %+v, cluster2: %+v", rev1, rev2)
			Expect(rev1.Name != rev2.Name).To(BeTrue(), "expected the same, but got cluster1: %+v, cluster2: %+v", rev1, rev2)
		}, NodeTimeout(time.Second*10))

		It("Should use tenant clusterconfig", func(ctx context.Context) {
			By("Create a cluster")
			cluster := newCluster(network, tenant.Labels[cloudv1.TenantTagKey], "", "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())
			Expect(rev.Labels[cloudv1.TenantTagKey] == tenant.Labels[cloudv1.TenantTagKey]).
				To(BeTrue(), "expected %s cluster's revision inherited from tenant", cluster.Name)

			conf := config.New(rev.Spec.PD.Config.MP).Get("schedule.enable-cross-table-merge")
			Expect(conf.MustBool() == true).
				To(BeTrue(), "%s should use tenant config", rev.Name)
		}, NodeTimeout(time.Second*10))

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

			conf := config.New(rev.Spec.PD.Config.MP).Get("schedule.enable-cross-table-merge")
			Expect(conf.MustBool() == false).
				To(BeTrue(), "%s should use cluster config", rev.Name)
		}, NodeTimeout(time.Second*10))
	})

	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(ctx context.Context) {
			baseCluster := newCluster(network, "1", "11111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, baseCluster)).To(Succeed())
			tenantCluster := newCluster(network, tenant.Labels[cloudv1.TenantTagKey], "", "test")
			Expect(k8sClient.Create(ctx, tenantCluster)).To(Succeed())
			cluster := newCluster(network, clustercc.Labels[cloudv1.TenantTagKey], clustercc.Labels[cloudv1.ClusterTagKey], "i3.2xlarge")
			Expect(k8sClient.Create(ctx, cluster)).To(Succeed())
			commentsAppender.resetUpdateCnt()
			By("Update Base ClusterConfig")
			base.Spec.PD.Tag = ptr.To("v5.2.1-hotfix")
			Expect(k8sClient.Update(ctx, base)).To(Succeed())
			Eventually(func() error {
				if cnt := commentsAppender.getUpdateCnt(baseCluster.Name); cnt == 0 {
					return fmt.Errorf("expected base cc uptdate cnt larger than 0, got %v", cnt)
				}
				if cnt := commentsAppender.getUpdateCnt(tenantCluster.Name); cnt == 0 {
					return fmt.Errorf("expected tenant cc uptdate cnt larger than 0, got %v", cnt)
				}
				if cnt := commentsAppender.getUpdateCnt(cluster.Name); cnt == 0 {
					return fmt.Errorf("expected cluster cc uptdate cnt larger than 0, got %v", cnt)
				}
				return nil
			}, timeout, interval).Should(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, baseCluster, commentsAppender.getNextRevision(baseCluster.Name))).To(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, tenantCluster, commentsAppender.getNextRevision(tenantCluster.Name))).To(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, cluster, commentsAppender.getNextRevision(cluster.Name))).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())
			commentsAppender.resetUpdateCnt()
			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 {
				if cnt := commentsAppender.getUpdateCnt(baseCluster.Name); cnt != 1 {
					return fmt.Errorf("expected base cc uptdate cnt is 1, got %v", cnt)
				}
				if cnt := commentsAppender.getUpdateCnt(tenantCluster.Name); cnt != 0 {
					return fmt.Errorf("expected tenant cc uptdate cnt is 0, got %v", cnt)
				}
				if cnt := commentsAppender.getUpdateCnt(cluster.Name); cnt != 1 {
					return fmt.Errorf("expected cluster cc uptdate cnt is 1, got %v", cnt)
				}
				return nil
			}, timeout, interval).Should(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, baseCluster, commentsAppender.getNextRevision(baseCluster.Name))).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())
			commentsAppender.resetUpdateCnt()
			By("Update tenant-level ClusterConfig")
			tenant.Spec.PD.Tag = ptr.To("v5.2.1-hotfix-tenant")
			Expect(k8sClient.Update(ctx, tenant)).To(Succeed())
			Eventually(func() error {
				if cnt := commentsAppender.getUpdateCnt(tenantCluster.Name); cnt == 0 {
					return fmt.Errorf("expected tenant cc uptdate cnt larger than 0, got %v", cnt)
				}
				if cnt := commentsAppender.getUpdateCnt(cluster.Name); cnt == 0 {
					return fmt.Errorf("expected cluster cc uptdate cnt larger than 0, got %v", cnt)
				}
				return nil
			}, timeout, interval).Should(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, tenantCluster, commentsAppender.getNextRevision(tenantCluster.Name))).To(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, cluster, commentsAppender.getNextRevision(cluster.Name))).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())
			commentsAppender.resetUpdateCnt()
			By("Update cluster-level ClusterConfig")
			clustercc.Spec.PD.Tag = ptr.To("v5.2.1-hotfix-tenant-cluster")
			Expect(k8sClient.Update(ctx, clustercc)).To(Succeed())
			Eventually(func() error {
				if cnt := commentsAppender.getUpdateCnt(cluster.Name); cnt == 0 {
					return fmt.Errorf("expected cluster cc uptdate cnt larger than 0, got %v", cnt)
				}
				return nil
			}, timeout, interval).Should(Succeed())
			Expect(patchConfigRevisionRef(k8sClient, cluster, commentsAppender.getNextRevision(cluster.Name))).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())
			commentsAppender.resetUpdateCnt()
		}, NodeTimeout(time.Second*10))

		It("Should not update dryrun clusters", func() {
			dryrunCluster := newCluster(network, dryruncc.Labels[cloudv1.TenantTagKey], "1111", "i3.2xlarge")
			Expect(k8sClient.Create(ctx, dryrunCluster)).To(Succeed())
			By("Update dryrun ClusterConfig")
			dryruncc.Spec.PD.Tag = ptr.To("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
			}, interval*10, interval).Should(Succeed())
		})
	})

	Context("when clusters are deleted", func() {
		It("Should gc the configRevision", func(ctx context.Context) {
			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.PD.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
}

func newClusterRevision(cluster *cloudv1.Cluster) *cloudv1.ConfigRevision {
	rev := &cloudv1.ConfigRevision{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("cr-%d", rand.Int()),
			Namespace: "default",
			Labels: map[string]string{
				cloudv1.ClusterTagKey: cluster.Name,
				cloudv1.ProjectTagKey: cluster.GetProjectID(),
			},
		},
		Spec: cloudv1.ConfigRevisionSpec{
			Version: "v7.1.1",
		},
	}
	return rev
}

func patchConfigRevisionRef(cli client.Client, cluster *cloudv1.Cluster, revisionRef string) error {
	patch := []byte(fmt.Sprintf(`{"spec":{"configRevisionRef":{"name": "%s"}}}`, revisionRef))
	return cli.Patch(context.Background(), cluster, client.RawPatch(types.MergePatchType, patch))
}

type fakeCommentsAppender struct {
	mu            sync.Mutex
	client        client.Client
	nextRevisions map[string]string
	updateCnt     map[string]int
}

func (fca *fakeCommentsAppender) ListPullRequestsWithCommit(_ context.Context, _, _, _ string, _ *gh.ListOptions) ([]*gh.PullRequest, *gh.Response, error) {
	prNumber := 1
	now := gh.Timestamp{Time: time.Now()}
	return []*gh.PullRequest{{Number: &prNumber, State: ptr.To("closed"), Merged: ptr.To(false), MergedAt: &now}},
		&gh.Response{Response: &http.Response{StatusCode: http.StatusOK}},
		nil
}

func (fca *fakeCommentsAppender) CreateComment(_ context.Context, _ string, _ string, _ int, comment *gh.IssueComment) (*gh.IssueComment, *gh.Response, error) {
	body := *comment.Body
	u, err := url.Parse(body)
	if err != nil {
		return nil, nil, err
	}
	clusterID := u.Query().Get("clusterid")
	nextRevisionID := u.Query().Get("configrevisionid")
	//if currRevisionID, ok := fca.nextRevisions[clusterID]; ok {
	//	nextRev := &cloudv1.ConfigRevision{}
	//	_ = fca.client.Get(context.TODO(), types.NamespacedName{Name: nextRevisionID}, nextRev)
	//	currRev := &cloudv1.ConfigRevision{}
	//	_ = fca.client.Get(context.TODO(), types.NamespacedName{Name: currRevisionID}, currRev)
	//}
	fca.mu.Lock()
	fca.nextRevisions[clusterID] = nextRevisionID
	fca.updateCnt[clusterID]++
	fca.mu.Unlock()
	return nil,
		&gh.Response{Response: &http.Response{StatusCode: http.StatusOK}},
		nil
}

func (fca *fakeCommentsAppender) getNextRevision(clusterID string) string {
	fca.mu.Lock()
	defer fca.mu.Unlock()
	return fca.nextRevisions[clusterID]
}

func (fca *fakeCommentsAppender) getUpdateCnt(clusterID string) int {
	fca.mu.Lock()
	defer fca.mu.Unlock()
	return fca.updateCnt[clusterID]
}

func (fca *fakeCommentsAppender) resetUpdateCnt() {
	fca.mu.Lock()
	defer fca.mu.Unlock()
	fca.updateCnt = make(map[string]int)
}