package main

import (
	"context"
	"fmt"
	"sort"

	"github.com/spf13/cobra"
	pb "github.com/tidbcloud/config-center/proto/config_center"
	nextgenv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/emptypb"
)

const (
	metaDeletePageSize    = 100
	defaultMgmtSvcAddress = "tidb-mgmt-service.tidb-management-service:10001"
)

type clusterMetaDeleteClient interface {
	SearchClusterComponents(context.Context, *pb.SearchClusterComponentsRequest, ...grpc.CallOption) (*pb.SearchClusterComponentsResponse, error)
	BatchDeleteClusterMeta(context.Context, *pb.BatchDeleteClusterMetaRequest, ...grpc.CallOption) (*emptypb.Empty, error)
}

func newMetaDeleteStaleCmd() *cobra.Command {
	var clusterTypeStr, mgmtSvcAddr, logicalRegion string
	var dryRun bool

	cmd := &cobra.Command{
		Use:   "delete-stale",
		Short: "delete stale cluster meta missing from management service",
		Long:  `delete stale cluster meta missing from management service`,
		RunE: func(cmd *cobra.Command, args []string) error {
			clusterType, err := parseRequiredClusterType(clusterTypeStr)
			if err != nil {
				return err
			}
			if mgmtSvcAddr == "" {
				return fmt.Errorf("mgmt-svc-addr is required")
			}
			componentName, err := componentNameForMetaDelete(clusterType)
			if err != nil {
				return err
			}

			fmt.Printf("start stale cluster meta cleanup: cluster_type=%s component_name=%s config_center=%s mgmt_svc_addr=%s dry_run=%v logical_region=%q page_size=%d\n",
				clusterType.String(), componentName, configCenterEndpoint, mgmtSvcAddr, dryRun, logicalRegion, metaDeletePageSize)

			ctx := context.Background()
			var opts []grpc.DialOption
			opts = append(opts, grpc.WithInsecure())

			mgmtConn, err := grpc.Dial(mgmtSvcAddr, opts...)
			if err != nil {
				return fmt.Errorf("dial management service %q: %w", mgmtSvcAddr, err)
			}
			defer mgmtConn.Close()

			liveClusterIDs, err := listLiveClusterIDsFromManagement(ctx, mgmtConn, clusterType)
			if err != nil {
				return err
			}
			fmt.Printf("management service live cluster ids loaded: count=%d\n", len(liveClusterIDs))

			ccConn, err := grpc.Dial(configCenterEndpoint, opts...)
			if err != nil {
				return fmt.Errorf("dial config-center %q: %w", configCenterEndpoint, err)
			}
			defer ccConn.Close()

			ccClient := pb.NewConfigCenterServiceClient(ccConn)
			return deleteStaleClusterMetaFromConfigCenter(ctx, ccClient, clusterType, componentName, logicalRegion, dryRun, liveClusterIDs)
		},
	}

	cmd.Flags().StringVar(&clusterTypeStr, "cluster-type", "", "cluster type")
	cmd.Flags().StringVar(&mgmtSvcAddr, "mgmt-svc-addr", defaultMgmtSvcAddress, "management service gRPC endpoint")
	cmd.Flags().StringVar(&logicalRegion, "logical-region", "", "logical region when using a global config-center endpoint")
	cmd.Flags().BoolVar(&dryRun, "dry-run", false, "print stale cluster ids without deleting data")

	return cmd
}

func parseRequiredClusterType(clusterTypeStr string) (pb.ClusterType, error) {
	if clusterTypeStr == "" {
		return pb.ClusterType_CLUSTER_TYPE_UNSPECIFIED, fmt.Errorf("cluster-type is required")
	}
	value, ok := pb.ClusterType_value[clusterTypeStr]
	if !ok {
		return pb.ClusterType_CLUSTER_TYPE_UNSPECIFIED, fmt.Errorf("unknown cluster-type %q", clusterTypeStr)
	}
	clusterType := pb.ClusterType(value)
	switch clusterType {
	case pb.ClusterType_DEDICATED_TIDB, pb.ClusterType_LOGICAL_CLUSTER, pb.ClusterType_SHARED_POOL:
		return clusterType, nil
	default:
		return pb.ClusterType_CLUSTER_TYPE_UNSPECIFIED, fmt.Errorf("unsupported cluster-type %s", clusterType.String())
	}
}

func componentNameForMetaDelete(clusterType pb.ClusterType) (string, error) {
	switch clusterType {
	case pb.ClusterType_DEDICATED_TIDB, pb.ClusterType_LOGICAL_CLUSTER:
		return "tidb", nil
	case pb.ClusterType_SHARED_POOL:
		return "tikv", nil
	default:
		return "", fmt.Errorf("unsupported cluster-type %s", clusterType.String())
	}
}

func listLiveClusterIDsFromManagement(ctx context.Context, conn grpc.ClientConnInterface, clusterType pb.ClusterType) (map[string]struct{}, error) {
	switch clusterType {
	case pb.ClusterType_DEDICATED_TIDB, pb.ClusterType_LOGICAL_CLUSTER:
		return listLiveClusters(ctx, mgmtv1.NewClusterServiceClient(conn), clusterType)
	case pb.ClusterType_SHARED_POOL:
		return listLiveResourcePools(ctx, mgmtv1.NewResourcePoolServiceClient(conn))
	default:
		return nil, fmt.Errorf("unsupported cluster-type %s", clusterType.String())
	}
}

func listLiveClusters(ctx context.Context, client mgmtv1.ClusterServiceClient, clusterType pb.ClusterType) (map[string]struct{}, error) {
	deleted := false
	filter := &mgmtv1.ClusterFilter{
		Deleted: &deleted,
		View:    mgmtv1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	}
	switch clusterType {
	case pb.ClusterType_DEDICATED_TIDB:
		deployType := mgmtv1.DeployType_DEPLOY_TYPE_DEDICATED
		filter.ClusterType = &deployType
	case pb.ClusterType_LOGICAL_CLUSTER:
		deployType := mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN
		filter.ClusterType = &deployType
	default:
		return nil, fmt.Errorf("unsupported ListClusters cluster-type %s", clusterType.String())
	}

	liveClusterIDs := make(map[string]struct{})
	for page := uint32(1); ; page++ {
		resp, err := client.ListClusters(ctx, &mgmtv1.ListClustersRequest{
			Filter:   filter,
			PageSize: metaDeletePageSize,
			Page:     page,
		})
		if err != nil {
			return nil, fmt.Errorf("list management clusters page=%d: %w", page, err)
		}
		clusters := resp.GetClusters()
		for _, cluster := range clusters {
			liveClusterIDs[fmt.Sprintf("%d", cluster.GetClusterId())] = struct{}{}
		}

		fmt.Printf("management ListClusters page=%d returned=%d total=%d accumulated_live=%d\n",
			page, len(clusters), resp.GetTotal(), len(liveClusterIDs))
		if len(clusters) < metaDeletePageSize || (resp.GetTotal() > 0 && page*metaDeletePageSize >= resp.GetTotal()) {
			break
		}
	}
	return liveClusterIDs, nil
}

func listLiveResourcePools(ctx context.Context, client mgmtv1.ResourcePoolServiceClient) (map[string]struct{}, error) {
	liveClusterIDs := make(map[string]struct{})
	for page := uint32(1); ; page++ {
		resp, err := client.ListResourcePools(ctx, &nextgenv1.ListResourcePoolsRequest{
			PageSize:    metaDeletePageSize,
			Page:        page,
			ShowDeleted: false,
		})
		if err != nil {
			return nil, fmt.Errorf("list resource pools page=%d: %w", page, err)
		}

		resourcePools := resp.GetResourcePools()
		for _, profile := range resourcePools {
			resourcePoolID := profile.GetResourcePool().GetId()
			if resourcePoolID == 0 {
				continue
			}
			liveClusterIDs[fmt.Sprintf("%d", resourcePoolID)] = struct{}{}
		}

		fmt.Printf("management ListResourcePools page=%d returned=%d total_size=%d next_page_token=%q accumulated_live=%d\n",
			page, len(resourcePools), resp.GetTotalSize(), resp.GetNextPageToken(), len(liveClusterIDs))
		if len(resourcePools) < metaDeletePageSize || (resp.GetTotalSize() > 0 && int32(page*metaDeletePageSize) >= resp.GetTotalSize()) {
			break
		}
	}
	return liveClusterIDs, nil
}

func deleteStaleClusterMetaFromConfigCenter(ctx context.Context, client clusterMetaDeleteClient, clusterType pb.ClusterType, componentName, logicalRegion string, dryRun bool, liveClusterIDs map[string]struct{}) error {
	processedClusterIDs := make(map[string]struct{})
	staleClusterIDs := make([]string, 0)
	var totalComponents, totalUniqueClusters, totalLiveClusters, totalDeletedClusters int

	for page := int32(1); ; page++ {
		req := &pb.SearchClusterComponentsRequest{
			ClusterType:   clusterType,
			ComponentName: componentName,
			PageSize:      metaDeletePageSize,
			Page:          page,
			ShowDeleted:   false,
		}
		if logicalRegion != "" {
			req.LogicalRegion = &logicalRegion
		}

		resp, err := client.SearchClusterComponents(ctx, req)
		if err != nil {
			return fmt.Errorf("search config-center cluster components page=%d: %w", page, err)
		}

		components := resp.GetComponents()
		totalComponents += len(components)
		pageStaleClusterIDs := make([]string, 0)
		pageUniqueClusters := 0
		pageLiveClusters := 0
		pageAlreadyProcessed := 0

		for _, component := range components {
			clusterID := component.GetClusterId()
			if clusterID == "" {
				continue
			}
			if _, ok := processedClusterIDs[clusterID]; ok {
				pageAlreadyProcessed++
				continue
			}
			processedClusterIDs[clusterID] = struct{}{}
			pageUniqueClusters++
			totalUniqueClusters++

			if _, ok := liveClusterIDs[clusterID]; ok {
				pageLiveClusters++
				totalLiveClusters++
				continue
			}
			pageStaleClusterIDs = append(pageStaleClusterIDs, clusterID)
			staleClusterIDs = append(staleClusterIDs, clusterID)
		}

		sort.Strings(pageStaleClusterIDs)
		fmt.Printf("config-center SearchClusterComponents page=%d returned_components=%d total_components=%d page_unique_clusters=%d page_live_clusters=%d page_stale_clusters=%d page_duplicate_components=%d scanned_unique_clusters=%d collected_stale_clusters=%d\n",
			page, len(components), resp.GetTotal(), pageUniqueClusters, pageLiveClusters, len(pageStaleClusterIDs), pageAlreadyProcessed, totalUniqueClusters, len(staleClusterIDs))

		if len(pageStaleClusterIDs) > 0 {
			fmt.Printf("collected stale cluster ids on page=%d: %v\n", page, pageStaleClusterIDs)
		}

		if len(components) < metaDeletePageSize || (resp.GetTotal() > 0 && int64(page*metaDeletePageSize) >= resp.GetTotal()) {
			break
		}
	}

	sort.Strings(staleClusterIDs)
	if dryRun {
		fmt.Printf("dry-run completed: scanned_components=%d scanned_unique_clusters=%d live_clusters=%d stale_clusters=%d deleted_clusters=0\n",
			totalComponents, totalUniqueClusters, totalLiveClusters, len(staleClusterIDs))
		return nil
	}

	for start := 0; start < len(staleClusterIDs); start += metaDeletePageSize {
		end := start + metaDeletePageSize
		if end > len(staleClusterIDs) {
			end = len(staleClusterIDs)
		}
		deleteReq := &pb.BatchDeleteClusterMetaRequest{
			ClusterType: clusterType,
			ClusterIds:  staleClusterIDs[start:end],
		}
		if logicalRegion != "" {
			deleteReq.LogicalRegion = &logicalRegion
		}
		_, err := client.BatchDeleteClusterMeta(ctx, deleteReq)
		if err != nil {
			return fmt.Errorf("batch delete cluster meta batch=%d cluster_ids=%v: %w", start/metaDeletePageSize+1, deleteReq.GetClusterIds(), err)
		}
		totalDeletedClusters += len(deleteReq.GetClusterIds())
		fmt.Printf("BatchDeleteClusterMeta completed batch=%d cluster_count=%d\n", start/metaDeletePageSize+1, len(deleteReq.GetClusterIds()))
	}

	fmt.Printf("cleanup completed: scanned_components=%d scanned_unique_clusters=%d live_clusters=%d stale_clusters=%d deleted_clusters=%d\n",
		totalComponents, totalUniqueClusters, totalLiveClusters, len(staleClusterIDs), totalDeletedClusters)
	return nil
}