package cluster

import (
	"context"
	"fmt"
	"strconv"
	"testing"

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

	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	openapiserverlessv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/serverless/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/infra/repository/db"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/infra/repository/schema"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

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

	service, err := initService()
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
	dbProvider := db.NewProvider(nil)

	type testCase struct {
		name      string
		inputsFn  func() (context.Context, *openapiserverlessv1beta1.DeleteClusterRequest)
		outputsFn func(*openapiserverlessv1beta1.Cluster, error)
	}

	cases := []testCase{
		{
			name: "Delete should success",
			inputsFn: func() (context.Context, *openapiserverlessv1beta1.DeleteClusterRequest) {
				// Create a cluster first
				ctx := grpcmetadata.NewIncomingContext(t.Context(), grpcmetadata.New(map[string]string{
					metadata.UserIDHeaderKey:           "1",
					metadata.OrgIDHeaderKey:            "1",
					metadata.DefaultProjectIDHeaderKey: "1",
					metadata.ProjectListHeaderKey:      "1,2,3",
					metadata.ClusterIDHeaderKey:        "1",
					metadata.IsCheckPermissionKey:      "true",
				}))

				req := &openapiserverlessv1beta1.CreateClusterRequest{
					Cluster: &openapiserverlessv1beta1.Cluster{
						DisplayName: "delete-test-1",
						Region:      &openapicommonv1beta1.Region{Name: "regions/aws-us-east-1"},
						Labels: map[string]string{
							common.ProjectIDLabelKey: "1",
						},
					},
				}
				c, err := service.CreateCluster(ctx, req)
				g.Expect(err).ShouldNot(gomega.HaveOccurred())

				// Build delete request
				return ctx, &openapiserverlessv1beta1.DeleteClusterRequest{ClusterId: c.ClusterId}
			},
			outputsFn: func(cluster *openapiserverlessv1beta1.Cluster, err error) {
				g.Expect(err).ShouldNot(gomega.HaveOccurred())
				g.Expect(cluster).ShouldNot(gomega.BeNil())
				g.Expect(cluster.ClusterId).ShouldNot(gomega.BeEmpty())
				g.Expect(cluster.Name).Should(gomega.Equal(fmt.Sprintf("clusters/%s", cluster.ClusterId)))
				g.Expect(cluster.DisplayName).Should(gomega.Equal("delete-test-1"))
				g.Expect(cluster.Version).ShouldNot(gomega.BeEmpty())
				g.Expect(cluster.UserPrefix).ShouldNot(gomega.BeEmpty())
				g.Expect(cluster.GetRegion().GetDisplayName()).
					Should(gomega.Equal(fakeRegionDisplayName(cluster.GetRegion().GetRegionId())))

				// Check the cluster in database
				var record schema.Cluster

				// Find the soft deleted record
				err = dbProvider.ReadDB(t.Context()).
					Model(&schema.Cluster{}).
					Unscoped().
					Where("cluster_id = ?", cluster.ClusterId).
					Take(&record).
					Error

				g.Expect(err).ShouldNot(gomega.HaveOccurred())
				g.Expect(record).ShouldNot(gomega.BeNil())
				g.Expect(record.DeletedAt.Valid).Should(gomega.BeTrue())
			},
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			ctx, req := tc.inputsFn()
			id, err := strconv.ParseUint(req.ClusterId, 10, 64)
			g.Expect(err).ShouldNot(gomega.HaveOccurred())
			ctx = metadata.AppendToIncomingMeta(ctx, &metadata.RequestMeta{
				ClusterID: id,
			})
			cluster, err := service.DeleteCluster(ctx, req)
			tc.outputsFn(cluster, err)
		})
	}
}

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

	service, err := initService()
	g.Expect(err).ShouldNot(gomega.HaveOccurred())
	dbProvider := db.NewProvider(nil)

	buildCtx := func() context.Context {
		return grpcmetadata.NewIncomingContext(t.Context(), grpcmetadata.New(map[string]string{
			metadata.UserIDHeaderKey:           "1",
			metadata.OrgIDHeaderKey:            "1",
			metadata.DefaultProjectIDHeaderKey: "1",
			metadata.ProjectListHeaderKey:      "1,2,3",
			metadata.IsCheckPermissionKey:      "true",
		}))
	}

	createCluster := func(displayName string) string {
		cluster, createErr := service.CreateCluster(buildCtx(), &openapiserverlessv1beta1.CreateClusterRequest{
			Cluster: &openapiserverlessv1beta1.Cluster{
				DisplayName: displayName,
				Region:      &openapicommonv1beta1.Region{Name: "regions/aws-us-east-1"},
				Labels: map[string]string{
					common.ProjectIDLabelKey: "1",
				},
			},
		})
		g.Expect(createErr).ShouldNot(gomega.HaveOccurred())
		return cluster.ClusterId
	}

	t.Run("Batch delete should partially succeed and deduplicate ids", func(t *testing.T) {
		clusterID1 := createCluster("batch-delete-test-1")
		clusterID2 := createCluster("batch-delete-test-2")

		resp, batchErr := service.BatchDeleteClusters(buildCtx(), &openapiserverlessv1beta1.BatchDeleteClustersRequest{
			ClusterIds: []string{clusterID1, clusterID1, "999999999999999999", clusterID2},
		})
		g.Expect(batchErr).ShouldNot(gomega.HaveOccurred())
		g.Expect(resp.Total).Should(gomega.Equal(int32(3)))
		g.Expect(resp.SuccessClusterIds).Should(gomega.ContainElements(clusterID1, clusterID2))
		g.Expect(resp.FailedClusterIds).Should(gomega.ContainElement("999999999999999999"))

		for _, deletedID := range []string{clusterID1, clusterID2} {
			var record schema.Cluster
			err = dbProvider.ReadDB(t.Context()).
				Model(&schema.Cluster{}).
				Unscoped().
				Where("cluster_id = ?", deletedID).
				Take(&record).
				Error
			g.Expect(err).ShouldNot(gomega.HaveOccurred())
			g.Expect(record.DeletedAt.Valid).Should(gomega.BeTrue())
		}
	})

	t.Run("Batch delete should validate empty ids", func(t *testing.T) {
		resp, batchErr := service.BatchDeleteClusters(buildCtx(), &openapiserverlessv1beta1.BatchDeleteClustersRequest{})
		g.Expect(resp).Should(gomega.BeNil())
		g.Expect(batchErr).Should(gomega.HaveOccurred())
		g.Expect(batchErr.Error()).Should(gomega.ContainSubstring("cluster ids must not be empty"))
	})

	t.Run("Batch delete should validate invalid id format", func(t *testing.T) {
		resp, batchErr := service.BatchDeleteClusters(buildCtx(), &openapiserverlessv1beta1.BatchDeleteClustersRequest{
			ClusterIds: []string{"invalid-cluster-id"},
		})
		g.Expect(resp).Should(gomega.BeNil())
		g.Expect(batchErr).Should(gomega.HaveOccurred())
		g.Expect(batchErr.Error()).Should(gomega.ContainSubstring("invalid cluster id"))
	})

	t.Run("Batch delete should validate max batch size", func(t *testing.T) {
		ids := make([]string, maxBatchSize+1)
		for i := range ids {
			ids[i] = strconv.Itoa(i + 1)
		}

		resp, batchErr := service.BatchDeleteClusters(buildCtx(), &openapiserverlessv1beta1.BatchDeleteClustersRequest{
			ClusterIds: ids,
		})
		g.Expect(resp).Should(gomega.BeNil())
		g.Expect(batchErr).Should(gomega.HaveOccurred())
		g.Expect(batchErr.Error()).Should(gomega.ContainSubstring("too many cluster ids"))
	})
}