package regional_proxy

import (
	"fmt"
	"testing"
	"time"

	"github.com/onsi/gomega"
	grpcmetadata "google.golang.org/grpc/metadata"

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/app/register"
	clustersvc "github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cluster"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/testdata"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/infra/adapterimpl"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/infra/adapterimpl/cli"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

func TestRegionalProxyByListCluster(t *testing.T) {
	g := gomega.NewWithT(t)
	clusterId := allocateClusterId()
	ctx := grpcmetadata.NewIncomingContext(t.Context(), grpcmetadata.New(map[string]string{
		metadata.OrgIDHeaderKey:            "1",
		metadata.DefaultProjectIDHeaderKey: "1",
		metadata.ProjectListHeaderKey:      "1,2",
		metadata.UserIDHeaderKey:           "11",
	}))

	req := &globalserverlessv1.ServerlessServiceCreateRequest{
		Id:            clusterId,
		TenantId:      1,
		ProjectId:     1,
		CloudProvider: provider.Id.String(),
		Region:        regionEn.Id.String(),
		SpendLimit: &globalserverlessv1.SpendLimit{
			Monthly: 1,
		},
	}

	err := register.DiCall(func(svc clustersvc.Service) {
		_, err := svc.Create(ctx, req)
		g.Expect(err).ShouldNot(gomega.HaveOccurred())
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeFalse())
		g.Expect(len(clusters)).To(gomega.Equal(0))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	// in post create action, the cluster will be set in the regional proxy
	time.Sleep(time.Second * 3) // wait for the workflow to finish

	var originalRecord string
	err = register.DiCall(func(svc clustersvc.Service) {
		output, err := svc.List(t.Context(), &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{
				ClusterIds:  []string{clusterId},
				ShowDetails: true,
			},
		})

		g.Expect(err).ShouldNot(gomega.HaveOccurred())
		g.Expect(len(output)).Should(gomega.Equal(1))

		originalRecord = output[0].Cluster.String()
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeTrue())
		g.Expect(len(clusters)).To(gomega.Equal(1))

		t.Logf("cluster string: %s", clusters[0].String())

		g.Expect(clusters[0].String()).To(gomega.Equal(originalRecord))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
}

func TestRegionalProxyByUpdateCluster(t *testing.T) {
	g := gomega.NewWithT(t)
	clusterId := allocateClusterId()
	ctx := grpcmetadata.NewIncomingContext(t.Context(), grpcmetadata.New(map[string]string{
		metadata.OrgIDHeaderKey:            "1",
		metadata.DefaultProjectIDHeaderKey: "1",
		metadata.ProjectListHeaderKey:      "1,2",
		metadata.UserIDHeaderKey:           "11",
	}))

	req := &globalserverlessv1.ServerlessServiceCreateRequest{
		Id:            clusterId,
		TenantId:      1,
		ProjectId:     1,
		CloudProvider: provider.Id.String(),
		Region:        regionEn.Id.String(),
		SpendLimit: &globalserverlessv1.SpendLimit{
			Monthly: 1,
		},
	}

	err := register.DiCall(func(svc clustersvc.Service) {
		_, err := svc.Create(ctx, req)
		g.Expect(err).ShouldNot(gomega.HaveOccurred())
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeFalse())
		g.Expect(len(clusters)).To(gomega.Equal(0))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	// in post create action, the cluster will be set in the regional proxy
	time.Sleep(time.Second * 3) // wait for the workflow to finish

	err = register.DiCall(func(svc clustersvc.Service) {
		output, err := svc.List(t.Context(), &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{
				ClusterIds:  []string{clusterId},
				ShowDetails: true,
			},
		})

		g.Expect(err).ShouldNot(gomega.HaveOccurred())
		g.Expect(len(output)).Should(gomega.Equal(1))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeTrue())
		g.Expect(len(clusters)).To(gomega.Equal(1))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	updateReq := &globalserverlessv1.ServerlessServiceUpdateRequest{
		Cluster: &globalserverlessv1.ServerlessServiceCluster{
			ClusterId: clusterId,
			Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
				ClusterId: clusterId,
				Endpoints: &regionalserverlessv1.Endpoints{
					Public: &regionalserverlessv1.PublicEndpoint{
						Disabled: false,
					},
				},
			},
			TenantId:      1,
			CloudProvider: provider.Id.String(),
			Region:        regionEn.Id.String(),
			HasSetPwd:     false,
			BackupPolicy: &globalserverlessv1.AutomatedBackupPolicy{
				BackupTime: "6:30",
			},
			SpendLimit: &globalserverlessv1.SpendLimit{
				Monthly: 1,
			},
		},
	}
	err = register.DiCall(func(svc clustersvc.Service) {
		_, err := svc.Update(ctx, updateReq)
		g.Expect(err).ShouldNot(gomega.HaveOccurred())
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
	time.Sleep(time.Second * 3) // wait for the workflow to finish

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeFalse())
		g.Expect(len(clusters)).To(gomega.Equal(0))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
}

func TestRegionalProxyByDeleteCluster(t *testing.T) {
	g := gomega.NewWithT(t)
	clusterId := allocateClusterId()
	ctx := grpcmetadata.NewIncomingContext(t.Context(), grpcmetadata.New(map[string]string{
		metadata.OrgIDHeaderKey:            "1",
		metadata.DefaultProjectIDHeaderKey: "1",
		metadata.ProjectListHeaderKey:      "1,2",
		metadata.UserIDHeaderKey:           "11",
	}))

	req := &globalserverlessv1.ServerlessServiceCreateRequest{
		Id:            clusterId,
		TenantId:      1,
		ProjectId:     1,
		CloudProvider: provider.Id.String(),
		Region:        regionEn.Id.String(),
		SpendLimit: &globalserverlessv1.SpendLimit{
			Monthly: 1,
		},
	}

	err := register.DiCall(func(svc clustersvc.Service) {
		_, err := svc.Create(ctx, req)
		g.Expect(err).ShouldNot(gomega.HaveOccurred())
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeFalse())
		g.Expect(len(clusters)).To(gomega.Equal(0))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	// in post create action, the cluster will be set in the regional proxy
	time.Sleep(time.Second * 3) // wait for the workflow to finish

	err = register.DiCall(func(svc clustersvc.Service) {
		output, err := svc.List(t.Context(), &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{
				ClusterIds:  []string{clusterId},
				ShowDetails: true,
			},
		})

		g.Expect(err).ShouldNot(gomega.HaveOccurred())
		g.Expect(len(output)).Should(gomega.Equal(1))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeTrue())
		g.Expect(len(clusters)).To(gomega.Equal(1))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())

	err = register.DiCall(func(svc clustersvc.Service) {
		_, err := svc.Delete(ctx, clusterId, false)
		g.Expect(err).ShouldNot(gomega.HaveOccurred())
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
	time.Sleep(time.Second * 3) // wait for the workflow to finish

	err = register.DiCall(func(proxy *adapterimpl.RegionalServiceProxy) {
		clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{clusterId})
		g.Expect(ok).To(gomega.BeFalse())
		g.Expect(len(clusters)).To(gomega.Equal(0))
	})
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
}

func TestReadyClusterBatchSetCluster(t *testing.T) {
	g := gomega.NewWithT(t)

	proxy := adapterimpl.NewRegionalServiceProxy(cli.NewFakeRegionalProxyClientImpl(NewService()))

	out := make([]*regionalserverlessv1.ServerlessRegionalServiceClusterInfo, 0)
	for i := range 100 {
		cluster := testdata.GetRegionalCluster()
		cluster.ClusterId = fmt.Sprintf("%s-%d", cluster.ClusterId, i)
		if i < 50 {
			cluster.Endpoints.Private = &regionalserverlessv1.PrivateEndpoint{
				Host: "test.tidbcloud.com",
				Port: 4000,
			}
		}

		out = append(out, cluster)
	}

	proxy.ReadyClusterBatchSet(t.Context(), out)

	clusters, ok := proxy.GetReadyClusterByIds(t.Context(), []string{out[0].ClusterId, out[1].ClusterId})
	g.Expect(ok).To(gomega.BeTrue())
	g.Expect(len(clusters)).To(gomega.Equal(2))
	g.Expect(clusters[0].ClusterId).To(gomega.Equal(fmt.Sprintf("%s-%d", testdata.GetRegionalCluster().ClusterId, 0)))

	ids := make([]string, 0)
	for i := range 50 {
		cluster := testdata.GetRegionalCluster()
		cluster.ClusterId = fmt.Sprintf("%s-%d", cluster.ClusterId, i)

		ids = append(ids, cluster.ClusterId)
	}
	clusters, ok = proxy.GetReadyClusterByIds(t.Context(), ids)
	g.Expect(ok).To(gomega.BeTrue())
	g.Expect(len(clusters)).To(gomega.Equal(len(ids)))

	for i := 50; i < 100; i++ {
		cluster := testdata.GetRegionalCluster()
		cluster.ClusterId = fmt.Sprintf("%s-%d", cluster.ClusterId, i)

		clusters, ok = proxy.GetReadyClusterByIds(t.Context(), []string{cluster.ClusterId})
		g.Expect(ok).To(gomega.BeFalse())
		g.Expect(len(clusters)).To(gomega.Equal(0))
	}
}

func TestReset(t *testing.T) {
	g := gomega.NewWithT(t)

	proxy := adapterimpl.NewRegionalServiceProxy(cli.NewFakeRegionalProxyClientImpl(NewService()))
	out := make([]*regionalserverlessv1.ServerlessRegionalServiceClusterInfo, 0)
	ids := make([]string, 0)
	for i := range 100 {
		cluster := testdata.GetRegionalCluster()
		cluster.ClusterId = fmt.Sprintf("%s-%d", cluster.ClusterId, i)
		cluster.Endpoints.Private = &regionalserverlessv1.PrivateEndpoint{
			Host: "test.tidbcloud.com",
			Port: 4000,
		}

		out = append(out, cluster)
		ids = append(ids, cluster.ClusterId)
	}

	proxy.ReadyClusterBatchSet(t.Context(), out)
	g.Expect(proxy.Enabled()).To(gomega.Equal(true))

	infos, ok := proxy.GetReadyClusterByIds(t.Context(), ids)
	g.Expect(ok).To(gomega.Equal(true))
	g.Expect(len(infos)).To(gomega.Equal(100))

	proxy.Reset(t.Context())
	g.Expect(proxy.Enabled()).To(gomega.Equal(false))

	for {
		time.Sleep(time.Millisecond * 100)
		if proxy.Enabled() {
			break
		}
	}

	infos, ok = proxy.GetReadyClusterByIds(t.Context(), ids)
	g.Expect(ok).To(gomega.Equal(false))
	g.Expect(len(infos)).To(gomega.Equal(0))
}