package main

import (
	"context"
	"fmt"
	"testing"

	pb "github.com/tidbcloud/config-center/proto/config_center"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/emptypb"
)

type fakeClusterMetaDeleteClient struct {
	searchResponses          []*pb.SearchClusterComponentsResponse
	searchRequests           []*pb.SearchClusterComponentsRequest
	searchBatchRequestCounts []int
	batchRequests            []*pb.BatchDeleteClusterMetaRequest
}

func (f *fakeClusterMetaDeleteClient) SearchClusterComponents(ctx context.Context, req *pb.SearchClusterComponentsRequest, opts ...grpc.CallOption) (*pb.SearchClusterComponentsResponse, error) {
	f.searchRequests = append(f.searchRequests, req)
	f.searchBatchRequestCounts = append(f.searchBatchRequestCounts, len(f.batchRequests))
	if len(f.searchResponses) == 0 {
		return &pb.SearchClusterComponentsResponse{}, nil
	}
	resp := f.searchResponses[0]
	f.searchResponses = f.searchResponses[1:]
	return resp, nil
}

func (f *fakeClusterMetaDeleteClient) BatchDeleteClusterMeta(ctx context.Context, req *pb.BatchDeleteClusterMetaRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
	f.batchRequests = append(f.batchRequests, req)
	return &emptypb.Empty{}, nil
}

func TestDeleteStaleClusterMetaFromConfigCenterDeletesAfterScanCompletes(t *testing.T) {
	logicalRegion := "us-west-2"
	firstPageComponents := make([]*pb.ClusterComponent, 0, metaDeletePageSize)
	for i := 0; i < metaDeletePageSize-1; i++ {
		firstPageComponents = append(firstPageComponents, &pb.ClusterComponent{
			ClusterId:     fmt.Sprintf("live-cluster-%03d", i),
			ComponentName: "tidb",
		})
	}
	firstPageComponents = append(firstPageComponents, &pb.ClusterComponent{
		ClusterId:     "stale-cluster-page-1",
		ComponentName: "tidb",
	})

	client := &fakeClusterMetaDeleteClient{
		searchResponses: []*pb.SearchClusterComponentsResponse{
			{
				Components: firstPageComponents,
				Total:      int64(metaDeletePageSize + 2),
			},
			{
				Components: []*pb.ClusterComponent{
					{ClusterId: "stale-cluster-page-2", ComponentName: "tidb"},
					{ClusterId: "stale-cluster-page-1", ComponentName: "tidb"},
				},
				Total: int64(metaDeletePageSize + 2),
			},
		},
	}
	liveClusterIDs := map[string]struct{}{
		"stale-cluster-page-2": {},
	}
	for i := 0; i < metaDeletePageSize-1; i++ {
		liveClusterIDs[fmt.Sprintf("live-cluster-%03d", i)] = struct{}{}
	}

	err := deleteStaleClusterMetaFromConfigCenter(context.Background(), client, pb.ClusterType_DEDICATED_TIDB, "tidb", logicalRegion, false, liveClusterIDs)
	if err != nil {
		t.Fatalf("deleteStaleClusterMetaFromConfigCenter returned error: %v", err)
	}

	if len(client.searchRequests) != 2 {
		t.Fatalf("expected 2 search requests, got %d", len(client.searchRequests))
	}
	for i, searchReq := range client.searchRequests {
		if searchReq.GetComponentName() != "tidb" {
			t.Fatalf("expected component_name tidb, got %q", searchReq.GetComponentName())
		}
		if searchReq.GetShowDeleted() {
			t.Fatal("expected show_deleted to be false")
		}
		if searchReq.GetLogicalRegion() != logicalRegion {
			t.Fatalf("expected logical_region %q, got %q", logicalRegion, searchReq.GetLogicalRegion())
		}
		if client.searchBatchRequestCounts[i] != 0 {
			t.Fatalf("expected no batch delete before scan completes, got %d batch requests before search %d", client.searchBatchRequestCounts[i], i+1)
		}
	}

	if len(client.batchRequests) != 1 {
		t.Fatalf("expected 1 batch delete request, got %d", len(client.batchRequests))
	}
	batchReq := client.batchRequests[0]
	if len(batchReq.GetClusterIds()) != 1 || batchReq.GetClusterIds()[0] != "stale-cluster-page-1" {
		t.Fatalf("expected only stale cluster missing from management service to be deleted, got %v", batchReq.GetClusterIds())
	}
	if batchReq.GetLogicalRegion() != logicalRegion {
		t.Fatalf("expected batch logical_region %q, got %q", logicalRegion, batchReq.GetLogicalRegion())
	}
}