package cluster

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"reflect"
	"strconv"
	"strings"

	clusteragentv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

const (
	// pdServiceURLTemplate is the template for constructing PD service URLs in Kubernetes
	pdServiceURLTemplate = "https://db-pd-0.db-pd-peer.tidb%d.svc:2379"
)

// buildPdURL constructs the PD service URL for a given cluster
func buildPdURL(clusterID uint64) string {
	return fmt.Sprintf(pdServiceURLTemplate, clusterID)
}

func (s *serviceImpl) SyncPdOnlineConfig(ctx context.Context, req *globaldedicatedv1.SyncPdOnlineConfigRequest) bizerr.BizError {
	ctx, span := tracer.Start(ctx, "Domain.SyncPdOnlineConfig")
	defer span.End()

	// Validate input
	if req.GetSourceClusterId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "source cluster ID is required")
	}
	if req.GetTargetClusterId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "target cluster ID is required")
	}
	if req.GetSourceClusterId() == req.GetTargetClusterId() {
		return bizerr.New(bizerr.ErrCodeParams, "source and target cluster IDs must be different")
	}

	// Validate source cluster
	sourceCluster, err := s.clusterRepo.Get(ctx, req.GetSourceClusterId(), false)
	if err != nil {
		return bizerr.WrapError(err, "failed to get source cluster")
	}
	if sourceCluster.State != globaldedicatedv1.Cluster_AVAILABLE {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"%s",
			fmt.Sprintf("source cluster %d is not available (state: %s)", req.GetSourceClusterId(), sourceCluster.State),
		)
	}

	// Validate target cluster
	targetCluster, err := s.clusterRepo.Get(ctx, req.GetTargetClusterId(), false)
	if err != nil {
		return bizerr.WrapError(err, "failed to get target cluster")
	}
	if targetCluster.State != globaldedicatedv1.Cluster_AVAILABLE {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"%s",
			fmt.Sprintf("target cluster %d is not available (state: %s)", req.GetTargetClusterId(), targetCluster.State),
		)
	}

	tilog.Info(ctx, "starting PD online config sync",
		zap.Uint64("source_cluster_id", req.GetSourceClusterId()),
		zap.Uint64("target_cluster_id", req.GetTargetClusterId()))

	if err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Get source region
		sourceRegion, regionErr := s.regionRepo.SelectRegionById(ctx, sourceCluster.RegionId, sourceCluster.OrganizationId)
		if regionErr != nil {
			return bizerr.WrapError(regionErr, "failed to select source cluster region")
		}
		if sourceRegion.State != globalcommonv1.Region_ACTIVE {
			return bizerr.New(bizerr.ErrCodeParams, "source cluster region is not active")
		}

		// Get target region
		targetRegion, regionErr := s.regionRepo.SelectRegionById(ctx, targetCluster.RegionId, targetCluster.OrganizationId)
		if regionErr != nil {
			return bizerr.WrapError(regionErr, "failed to select target cluster region")
		}
		if targetRegion.State != globalcommonv1.Region_ACTIVE {
			return bizerr.New(bizerr.ErrCodeParams, "target cluster region is not active")
		}

		// Sync regular PD configs
		if !req.GetSkipPdConfig() {
			if err := s.syncPdConfigType(
				ctx,
				sourceRegion,
				sourceCluster,
				targetRegion,
				targetCluster,
				"",
				"pd config",
				"",
				req.GetAllowFallbackToBackup(),
			); err != nil {
				return err
			}
		} else {
			tilog.Info(
				ctx,
				"skip sync pd config",
				zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
				zap.Uint64("target_cluster_id", targetCluster.ClusterId),
			)
		}

		// Sync PD server configs
		if !req.GetSkipPdServerConfig() {
			if err := s.syncPdConfigType(
				ctx,
				sourceRegion,
				sourceCluster,
				targetRegion,
				targetCluster,
				"server",
				"pd server config",
				"pd-server",
				req.GetAllowFallbackToBackup(),
			); err != nil {
				return err
			}
		} else {
			tilog.Info(
				ctx,
				"skip sync pd server config",
				zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
				zap.Uint64("target_cluster_id", targetCluster.ClusterId),
			)
		}

		// Cross-region clone should not blindly copy placement rules, because the rules may
		// contain region/zone labels that are only valid in the source region.
		if !req.GetSkipPlacementRules() {
			if syncRulesErr := s.syncPlacementRules(
				ctx,
				sourceRegion,
				sourceCluster,
				targetRegion,
				targetCluster,
				req.GetAllowFallbackToBackup(),
			); syncRulesErr != nil {
				return syncRulesErr
			}
		} else {
			tilog.Info(
				ctx,
				"skip sync placement rules",
				zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
				zap.Uint64("target_cluster_id", targetCluster.ClusterId),
			)
		}

		// Sync store limits
		if !req.GetSkipStoreLimits() {
			if syncStoreLimitsErr := s.syncStoreLimits(
				ctx,
				sourceRegion,
				sourceCluster,
				targetRegion,
				targetCluster,
				req.GetAllowFallbackToBackup(),
			); syncStoreLimitsErr != nil {
				return syncStoreLimitsErr
			}
		} else {
			tilog.Info(
				ctx,
				"skip sync store limits",
				zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
				zap.Uint64("target_cluster_id", targetCluster.ClusterId),
			)
		}

		return nil
	}); err != nil {
		return bizerr.WrapError(err, "failed to sync pd online config")
	}

	tilog.Info(ctx, "successfully completed PD online config sync",
		zap.Uint64("source_cluster_id", req.GetSourceClusterId()),
		zap.Uint64("target_cluster_id", req.GetTargetClusterId()))
	return nil
}

// syncPdConfigType synchronizes a specific type of PD config from source to target cluster
// configType: the config type suffix (e.g., "" for regular config, "server" for server config)
// configName: the name of the config for logging (e.g., "pd config", "pd server config")
// configPathPrefix: the prefix to add to the config path (e.g., "pd-server" for server config)
func (s *serviceImpl) syncPdConfigType(
	ctx context.Context,
	sourceRegion *entity.Region,
	sourceCluster *entity.Cluster,
	targetRegion *entity.Region,
	targetCluster *entity.Cluster,
	configType string,
	configName string,
	configPathPrefix string,
	allowFallbackToBackup bool,
) error {
	// Get config from source cluster or backup
	sourceConfig, found, err := s.getSourcePdConfig(ctx, sourceRegion, sourceCluster, configType, allowFallbackToBackup)
	if err != nil {
		return err
	}
	if !found {
		tilog.Warn(ctx, "skip syncing pd config because neither live config nor backup is available",
			zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
			zap.String("config_type", configType))
		return nil
	}

	// Get config from target cluster
	targetConfig, err := s.getPdConfig(ctx, targetRegion, targetCluster, configType)
	if err != nil {
		return err
	}

	// Compare configs and find differences
	diffConfigs := comparePdConfigs(sourceConfig, targetConfig)

	// Filter out cluster-specific configs that shouldn't be synced
	filteredDiffConfigs := make(map[string]any)
	for configPath, sourceValue := range diffConfigs {
		// Skip cluster-specific configs that shouldn't be synced
		if strings.HasSuffix(configPath, "dashboard-address") {
			tilog.Debug(ctx, "skipping cluster-specific config", zap.String("config_path", configPath))
			continue
		}
		filteredDiffConfigs[configPath] = sourceValue
	}

	// Apply differences to target cluster
	if len(filteredDiffConfigs) > 0 {
		tilog.Info(ctx, fmt.Sprintf("found %s differences", configName), zap.Int("count", len(filteredDiffConfigs)))
		for configPath, sourceValue := range filteredDiffConfigs {
			// Prepend path prefix if specified
			if configPathPrefix != "" {
				configPath = configPathPrefix + "." + configPath
			}
			if setErr := s.setPdOnlineConfig(ctx, targetRegion, targetCluster, configPath, sourceValue); setErr != nil {
				tilog.Error(ctx, fmt.Sprintf("failed to set %s", configName),
					zap.String("path", configPath),
					zap.Any("value", sourceValue),
					zap.Error(setErr))
				return setErr
			}
			tilog.Info(ctx, fmt.Sprintf("successfully set %s", configName),
				zap.String("path", configPath),
				zap.Any("value", sourceValue))
		}
	}

	return nil
}

// getPdConfig retrieves the PD configuration from a cluster and returns it as a flattened map
// configType specifies the config type suffix (e.g., "" for regular config, "server" for server config)
func (s *serviceImpl) getPdConfig(ctx context.Context, region *entity.Region, cluster *entity.Cluster, configType string) (map[string]any, error) {
	pdURL := buildPdURL(cluster.ClusterId)
	var command string
	var configName string
	if configType == "" {
		command = fmt.Sprintf("pd-ctl -u %s config show", pdURL)
		configName = "pd config"
	} else {
		command = fmt.Sprintf("pd-ctl -u %s config show %s", pdURL, configType)
		configName = fmt.Sprintf("pd %s config", configType)
	}

	resp, err := s.regionServerCli.ExecuteClusterAgentCommand(
		ctx,
		region,
		strconv.FormatUint(cluster.ClusterId, 10),
		&clusteragentv1.ExecuteCommandReq{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
			Command:   command,
		},
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to execute %s", command)
	}

	// Validate response is not empty
	if strings.TrimSpace(resp.StdoutMsg) == "" {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s", fmt.Sprintf("empty response from %s", command))
	}

	if resp.ExitCode != nil && *resp.ExitCode != 0 {
		tilog.Error(ctx, fmt.Sprintf("%s failed", command),
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("stderr", resp.StderrMsg),
			zap.Int("exit_code", int(*resp.ExitCode)))
		return nil, bizerr.New(
			bizerr.ErrCodeUnknown,
			"%s",
			fmt.Sprintf("%s failed with exit code %d: %s", command, *resp.ExitCode, resp.StderrMsg),
		)
	}

	// Parse JSON output
	var configMap map[string]any
	if err := json.Unmarshal([]byte(resp.StdoutMsg), &configMap); err != nil {
		tilog.Error(ctx, fmt.Sprintf("failed to parse %s JSON", configName),
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("response", resp.StdoutMsg),
			zap.Error(err))
		return nil, bizerr.WrapError(err, "failed to parse %s JSON", configName)
	}

	tilog.Debug(ctx, fmt.Sprintf("retrieved %s", configName),
		zap.Uint64("cluster_id", cluster.ClusterId),
		zap.Int("config_keys_count", len(configMap)))

	// Flatten nested config to dot notation
	flatConfig := make(map[string]any)
	flattenConfig(configMap, "", flatConfig)

	return flatConfig, nil
}

// setPdOnlineConfig sets a specific PD configuration value on the target cluster
func (s *serviceImpl) setPdOnlineConfig(ctx context.Context, region *entity.Region, cluster *entity.Cluster, configPath string, value any) error {
	valueStr := formatPdConfigValue(value)
	pdURL := buildPdURL(cluster.ClusterId)
	command := fmt.Sprintf("pd-ctl -u %s config set %s %s", pdURL, configPath, valueStr)

	resp, err := s.regionServerCli.ExecuteClusterAgentCommand(
		ctx,
		region,
		strconv.FormatUint(cluster.ClusterId, 10),
		&clusteragentv1.ExecuteCommandReq{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
			Command:   command,
		},
	)
	if err != nil {
		return bizerr.WrapError(err, "failed to execute pd-ctl config set")
	}

	// Check exit code and stderr
	if resp.ExitCode != nil && *resp.ExitCode != 0 {
		tilog.Error(ctx, "pd-ctl config set failed",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("config_path", configPath),
			zap.Any("value", value),
			zap.String("stderr", resp.StderrMsg),
			zap.Int("exit_code", int(*resp.ExitCode)))
		return bizerr.New(bizerr.ErrCodeUnknown, "%s", fmt.Sprintf("pd-ctl config set failed with exit code %d: %s", *resp.ExitCode, resp.StderrMsg))
	}

	if resp.StderrMsg != "" {
		tilog.Warn(ctx, "pd-ctl config set stderr",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("config_path", configPath),
			zap.String("stderr", resp.StderrMsg))
	}

	return nil
}

// formatPdConfigValue formats a configuration value for use in pd-ctl commands
func formatPdConfigValue(value any) string {
	switch v := value.(type) {
	case string:
		// Remove quotes if already present to avoid double quoting
		return strings.Trim(v, "\"")
	case float64:
		// Check if it's an integer to avoid unnecessary decimal notation
		if v == float64(int64(v)) {
			return fmt.Sprintf("%d", int64(v))
		}
		return fmt.Sprintf("%v", v)
	case bool:
		return fmt.Sprintf("%t", v)
	default:
		return fmt.Sprintf("%v", v)
	}
}

// flattenConfig converts nested map to flat map with dot notation keys
func flattenConfig(data map[string]any, prefix string, result map[string]any) {
	for key, value := range data {
		fullKey := key
		if prefix != "" {
			fullKey = prefix + "." + key
		}

		switch v := value.(type) {
		case map[string]any:
			// Recursively flatten nested maps
			flattenConfig(v, fullKey, result)
		default:
			// Store leaf values
			result[fullKey] = value
		}
	}
}

// comparePdConfigs compares two PD configs and returns a map of differences (source values that differ from target)
func comparePdConfigs(source, target map[string]any) map[string]any {
	diff := make(map[string]any)

	for key, sourceValue := range source {
		targetValue, exists := target[key]
		if !exists {
			// Config exists in source but not in target
			diff[key] = sourceValue
			continue
		}

		// Compare values
		if !reflect.DeepEqual(sourceValue, targetValue) {
			diff[key] = sourceValue
		}
	}

	return diff
}

// syncPlacementRules synchronizes placement rules from source to target cluster, excluding TiFlash group
func (s *serviceImpl) syncPlacementRules(
	ctx context.Context,
	sourceRegion *entity.Region,
	sourceCluster *entity.Cluster,
	targetRegion *entity.Region,
	targetCluster *entity.Cluster,
	allowFallbackToBackup bool,
) error {
	// Get placement rules from source cluster or backup
	sourceBundles, found, err := s.getSourcePlacementRules(ctx, sourceRegion, sourceCluster, allowFallbackToBackup)
	if err != nil {
		return err
	}
	if !found {
		tilog.Warn(ctx, "skip syncing placement rules because neither live config nor backup is available",
			zap.Uint64("source_cluster_id", sourceCluster.ClusterId))
		return nil
	}

	tilog.Info(ctx, "retrieved placement rules from source cluster",
		zap.Int("total_bundles", len(sourceBundles)),
		zap.Int("total_rules", countRulesInBundles(sourceBundles)))

	// Filter out rules with specific key ranges
	filteredBundles := filterTableSpecificRules(sourceBundles)

	tilog.Info(ctx, "filtered placement rules",
		zap.Int("bundles_after_filter", len(filteredBundles)),
		zap.Int("rules_after_filter", countRulesInBundles(filteredBundles)))

	if len(filteredBundles) == 0 {
		tilog.Info(ctx, "no placement rules to sync (after filtering table-specific rules)")
		return nil
	}

	// Get placement rules from target cluster for comparison
	targetBundles, err := s.getNonTiflashPlacementRules(ctx, targetRegion, targetCluster)
	if err != nil {
		return err
	}

	// Compare and identify differences
	bundlesToSync := comparePlacementRuleBundles(filteredBundles, filterTableSpecificRules(targetBundles))

	if len(bundlesToSync) == 0 {
		tilog.Info(ctx, "placement rules are already in sync")
		return nil
	}

	// Apply placement rules to target cluster
	tilog.Info(ctx, "syncing placement rule bundles", zap.Int("count", len(bundlesToSync)))
	for _, bundle := range bundlesToSync {
		if err := s.setPlacementRuleBundle(ctx, targetRegion, targetCluster, bundle); err != nil {
			tilog.Error(ctx, "failed to set placement rule bundle",
				zap.String("group_id", bundle.GroupID),
				zap.Error(err))
			return err
		}
		tilog.Info(ctx, "successfully set placement rule bundle",
			zap.String("group_id", bundle.GroupID),
			zap.Int("rules_count", len(bundle.Rules)))
	}

	return nil
}

// getNonTiflashPlacementRules retrieves placement rule bundles from a cluster excluding the TiFlash group
func (s *serviceImpl) getNonTiflashPlacementRules(
	ctx context.Context,
	region *entity.Region,
	cluster *entity.Cluster,
) ([]entity.PlacementRuleBundle, error) {
	pdURL := buildPdURL(cluster.ClusterId)
	command := fmt.Sprintf("pd-ctl -u %s config placement-rules rule-group show", pdURL)

	resp, err := s.regionServerCli.ExecuteClusterAgentCommand(
		ctx,
		region,
		strconv.FormatUint(cluster.ClusterId, 10),
		&clusteragentv1.ExecuteCommandReq{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
			Command:   command,
		},
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to execute pd-ctl config placement-rules rule-group show")
	}

	// If response is empty, return empty array
	if strings.TrimSpace(resp.StdoutMsg) == "" {
		tilog.Warn(ctx, "empty response from pd-ctl config placement-rules rule-group show",
			zap.Uint64("cluster_id", cluster.ClusterId))
		return nil, nil
	}

	// Check exit code
	if resp.ExitCode != nil && *resp.ExitCode != 0 {
		tilog.Error(ctx, "pd-ctl config placement-rules rule-group show failed",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("stderr", resp.StderrMsg),
			zap.Int("exit_code", int(*resp.ExitCode)))
		return nil, bizerr.New(
			bizerr.ErrCodeUnknown,
			"%s",
			fmt.Sprintf("pd-ctl placement-rules rule-group show failed with exit code %d: %s", *resp.ExitCode, resp.StderrMsg),
		)
	}

	type ruleGroup struct {
		ID    string `json:"id"`
		Index int    `json:"index"`
	}

	// Parse JSON output
	var ruleGroups []ruleGroup
	if err := json.Unmarshal([]byte(resp.StdoutMsg), &ruleGroups); err != nil {
		tilog.Error(ctx, "failed to parse placement rules rule-group JSON",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("response", resp.StdoutMsg),
			zap.Error(err))
		return nil, bizerr.WrapError(err, "failed to parse placement rules rule-group JSON")
	}
	var bundles []entity.PlacementRuleBundle
	for _, ruleGroup := range ruleGroups {
		// Skip TiFlash group (case-insensitive comparison)
		if strings.EqualFold(ruleGroup.ID, "tiflash") {
			continue
		}

		getBundleCommand := fmt.Sprintf("pd-ctl -u %s config placement-rules rule-bundle get %s", pdURL, ruleGroup.ID)
		getBundleResp, err := s.regionServerCli.ExecuteClusterAgentCommand(
			ctx,
			region,
			strconv.FormatUint(cluster.ClusterId, 10),
			&clusteragentv1.ExecuteCommandReq{
				ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
				Command:   getBundleCommand,
			},
		)
		if err != nil {
			return nil, bizerr.WrapError(err, "failed to execute pd-ctl config placement-rules rule-bundle get")
		}
		if getBundleResp.ExitCode != nil && *getBundleResp.ExitCode != 0 {
			tilog.Error(ctx, "pd-ctl config placement-rules rule-bundle get failed",
				zap.Uint64("cluster_id", cluster.ClusterId),
				zap.String("group_id", ruleGroup.ID),
				zap.String("stderr", getBundleResp.StderrMsg),
				zap.Int("exit_code", int(*getBundleResp.ExitCode)))
			return nil, bizerr.New(
				bizerr.ErrCodeUnknown,
				"%s",
				fmt.Sprintf(
					"pd-ctl config placement-rules rule-bundle get failed with exit code %d: %s",
					*getBundleResp.ExitCode,
					getBundleResp.StderrMsg,
				),
			)
		}
		if getBundleResp.StdoutMsg == "" {
			tilog.Warn(ctx, "empty response from pd-ctl config placement-rules rule-bundle get",
				zap.Uint64("cluster_id", cluster.ClusterId),
				zap.String("group_id", ruleGroup.ID))
			continue // Skip empty bundle response
		}
		var bundle entity.PlacementRuleBundle
		if err := json.Unmarshal([]byte(getBundleResp.StdoutMsg), &bundle); err != nil {
			tilog.Error(ctx, "failed to parse placement rule bundle JSON",
				zap.Uint64("cluster_id", cluster.ClusterId),
				zap.String("group_id", ruleGroup.ID),
				zap.String("response", getBundleResp.StdoutMsg),
				zap.Error(err))
			return nil, bizerr.WrapError(err, "failed to parse placement rule bundle JSON")
		}
		bundles = append(bundles, bundle) // Add valid bundle
	}
	return bundles, nil
}

// setPlacementRuleBundle sets a placement rule bundle on the target cluster
func (s *serviceImpl) setPlacementRuleBundle(
	ctx context.Context,
	region *entity.Region,
	cluster *entity.Cluster,
	bundle entity.PlacementRuleBundle,
) error {
	// Marshal bundle to JSON
	bundleJSON, err := json.Marshal(bundle)
	if err != nil {
		return bizerr.WrapError(err, "failed to marshal placement rule bundle")
	}

	// Generate a cryptographically random unique identifier
	randomBytes := make([]byte, 8)
	if _, err := rand.Read(randomBytes); err != nil {
		return bizerr.WrapError(err, "failed to generate random identifier")
	}
	randomID := hex.EncodeToString(randomBytes)

	// Sanitize groupID for use in filename (replace invalid characters)
	safeGroupID := strings.Map(func(r rune) rune {
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '-' || r == '_' {
			return r
		}
		return '_'
	}, bundle.GroupID)

	// Create a unique temporary file name to avoid collisions
	// Format: /tmp/pd-rule-bundle-{clusterId}-{safeGroupId}-{randomID}.json
	tmpFile := fmt.Sprintf("/tmp/pd-rule-bundle-%d-%s-%s.json",
		cluster.ClusterId,
		safeGroupID,
		randomID)

	// Escape JSON for shell - replace single quotes with shell-safe equivalent
	escapedJSON := strings.ReplaceAll(string(bundleJSON), "'", "'\"'\"'")

	pdURL := buildPdURL(cluster.ClusterId)
	// Create a compound command: write JSON to file, then execute pd-ctl, then cleanup
	command := fmt.Sprintf("echo '%s' > %s && pd-ctl -u %s config placement-rules rule-bundle set %s --in=%s",
		escapedJSON,
		tmpFile,
		pdURL,
		bundle.GroupID,
		tmpFile)

	tilog.Debug(ctx, "setting placement rule bundle",
		zap.Uint64("cluster_id", cluster.ClusterId),
		zap.String("group_id", bundle.GroupID),
		zap.Int("rules_count", len(bundle.Rules)))

	resp, err := s.regionServerCli.ExecuteClusterAgentCommand(
		ctx,
		region,
		strconv.FormatUint(cluster.ClusterId, 10),
		&clusteragentv1.ExecuteCommandReq{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
			Command:   command,
		},
	)
	if err != nil {
		return bizerr.WrapError(err, "failed to execute pd-ctl config placement-rules rule-bundle set")
	}

	// Check exit code and stderr
	if resp.ExitCode != nil && *resp.ExitCode != 0 {
		tilog.Error(ctx, "pd-ctl placement-rules set failed",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("group_id", bundle.GroupID),
			zap.String("stderr", resp.StderrMsg),
			zap.Int("exit_code", int(*resp.ExitCode)))
		return bizerr.New(
			bizerr.ErrCodeUnknown,
			"%s",
			fmt.Sprintf("pd-ctl placement-rules set failed with exit code %d: %s", *resp.ExitCode, resp.StderrMsg),
		)
	}

	if resp.StderrMsg != "" {
		tilog.Warn(ctx, "pd-ctl placement-rules set stderr",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("stderr", resp.StderrMsg),
			zap.String("group_id", bundle.GroupID))
	}

	return nil
}

// filterTableSpecificRules filters out rules with specific key ranges from placement rule bundles
func filterTableSpecificRules(bundles []entity.PlacementRuleBundle) []entity.PlacementRuleBundle {
	filtered := make([]entity.PlacementRuleBundle, 0)
	for _, bundle := range bundles {
		// Filter rules within the bundle - skip rules with specific start_key or end_key
		filteredRules := make([]entity.PlacementRule, 0)
		for i := range bundle.Rules {
			// Skip rules with non-empty start_key or end_key (table-specific rules)
			if bundle.Rules[i].StartKey != "" || bundle.Rules[i].EndKey != "" {
				continue
			}
			filteredRules = append(filteredRules, bundle.Rules[i])
		}

		// Only include bundle if it has at least one rule after filtering
		if len(filteredRules) > 0 {
			bundle.Rules = filteredRules
			filtered = append(filtered, bundle)
		}
	}
	return filtered
}

// comparePlacementRuleBundles compares source and target rule bundles and returns bundles that differ
func comparePlacementRuleBundles(source, target []entity.PlacementRuleBundle) []entity.PlacementRuleBundle {
	// Create a map of target bundles for quick lookup
	targetMap := make(map[string]entity.PlacementRuleBundle)
	for _, bundle := range target {
		targetMap[bundle.GroupID] = bundle
	}

	bundlesToSync := make([]entity.PlacementRuleBundle, 0)
	for _, sourceBundle := range source {
		targetBundle, exists := targetMap[sourceBundle.GroupID]
		if !exists {
			// Bundle doesn't exist in target, need to sync
			bundlesToSync = append(bundlesToSync, sourceBundle)
			continue
		}

		// Compare bundle content
		if !reflect.DeepEqual(sourceBundle, targetBundle) {
			bundlesToSync = append(bundlesToSync, sourceBundle)
		}
	}

	return bundlesToSync
}

// countRulesInBundles counts the total number of rules across all bundles
func countRulesInBundles(bundles []entity.PlacementRuleBundle) int {
	count := 0
	for _, bundle := range bundles {
		count += len(bundle.Rules)
	}
	return count
}

// syncStoreLimits synchronizes store limits from source to target cluster
// It finds the minimum add-peer and remove-peer limits from source cluster
// and applies them to target cluster if target's limits are lower
func (s *serviceImpl) syncStoreLimits(
	ctx context.Context,
	sourceRegion *entity.Region,
	sourceCluster *entity.Cluster,
	targetRegion *entity.Region,
	targetCluster *entity.Cluster,
	allowFallbackToBackup bool,
) error {
	// Get store limits from source cluster or backup
	sourceLimits, found, err := s.getSourceStoreLimits(ctx, sourceRegion, sourceCluster, allowFallbackToBackup)
	if err != nil {
		return err
	}
	if !found {
		tilog.Warn(ctx, "skip syncing store limits because neither live config nor backup is available",
			zap.Uint64("source_cluster_id", sourceCluster.ClusterId))
		return nil
	}

	if len(sourceLimits) == 0 {
		tilog.Info(ctx, "no store limits found in source cluster")
		return nil
	}

	// Find minimum add-peer and remove-peer limits from source
	minAddPeer := -1
	minRemovePeer := -1
	for storeID, limits := range sourceLimits {
		if minAddPeer < 0 || limits.AddPeer < minAddPeer {
			minAddPeer = limits.AddPeer
		}
		if minRemovePeer < 0 || limits.RemovePeer < minRemovePeer {
			minRemovePeer = limits.RemovePeer
		}
		tilog.Debug(ctx, "source store limit",
			zap.String("store_id", storeID),
			zap.Int("add_peer", limits.AddPeer),
			zap.Int("remove_peer", limits.RemovePeer))
	}

	tilog.Info(ctx, "minimum store limits from source cluster",
		zap.Int("min_add_peer", minAddPeer),
		zap.Int("min_remove_peer", minRemovePeer))

	// Get store limits from target cluster
	targetLimits, err := s.getStoreLimits(ctx, targetRegion, targetCluster)
	if err != nil {
		return err
	}

	// Check if target has any store limits configured
	if len(targetLimits) == 0 {
		tilog.Info(ctx, "target cluster has no store limits configured, setting minimum from source")
		if err := s.setStoreLimitAll(ctx, targetRegion, targetCluster, minAddPeer, "add-peer"); err != nil {
			return err
		}
		if err := s.setStoreLimitAll(ctx, targetRegion, targetCluster, minRemovePeer, "remove-peer"); err != nil {
			return err
		}
		tilog.Info(ctx, "successfully synced store limits")
		return nil
	}

	// Check if all target store limits are the same
	targetAddPeerLimitIsSame := true
	targetRemovePeerLimitIsSame := true
	var targetAddPeerLimit, targetRemovePeerLimit int

	// Get the first store's limits as reference
	firstIteration := true
	for storeID, limits := range targetLimits {
		if firstIteration {
			targetAddPeerLimit = limits.AddPeer
			targetRemovePeerLimit = limits.RemovePeer
			firstIteration = false
		} else {
			if limits.AddPeer != targetAddPeerLimit {
				targetAddPeerLimitIsSame = false
			}
			if limits.RemovePeer != targetRemovePeerLimit {
				targetRemovePeerLimitIsSame = false
			}
		}
		tilog.Debug(ctx, "target store limit",
			zap.String("store_id", storeID),
			zap.Int("add_peer", limits.AddPeer),
			zap.Int("remove_peer", limits.RemovePeer))
	}

	tilog.Info(ctx, "target cluster store limits status",
		zap.Bool("add_peer_all_same", targetAddPeerLimitIsSame),
		zap.Int("add_peer_value", targetAddPeerLimit),
		zap.Bool("remove_peer_all_same", targetRemovePeerLimitIsSame),
		zap.Int("remove_peer_value", targetRemovePeerLimit))

	// Compare and apply if target limits are uniform and lower than source minimum
	needUpdateAddPeer := targetAddPeerLimitIsSame && targetAddPeerLimit < minAddPeer
	needUpdateRemovePeer := targetRemovePeerLimitIsSame && targetRemovePeerLimit < minRemovePeer

	if !needUpdateAddPeer && !needUpdateRemovePeer {
		tilog.Info(ctx, "target cluster store limits are already equal or higher than source minimum, no update needed")
		return nil
	}

	// Apply store limits to all stores in target cluster
	if needUpdateAddPeer {
		tilog.Info(ctx, "setting add-peer limit for all stores in target cluster",
			zap.Int("limit", minAddPeer))
		if err := s.setStoreLimitAll(ctx, targetRegion, targetCluster, minAddPeer, "add-peer"); err != nil {
			return err
		}
	}

	if needUpdateRemovePeer {
		tilog.Info(ctx, "setting remove-peer limit for all stores in target cluster",
			zap.Int("limit", minRemovePeer))
		if err := s.setStoreLimitAll(ctx, targetRegion, targetCluster, minRemovePeer, "remove-peer"); err != nil {
			return err
		}
	}

	tilog.Info(ctx, "successfully synced store limits")
	return nil
}

// getStoreLimits retrieves store limits from a cluster
func (s *serviceImpl) getStoreLimits(ctx context.Context, region *entity.Region, cluster *entity.Cluster) (map[string]entity.StoreLimits, error) {
	pdURL := buildPdURL(cluster.ClusterId)
	command := fmt.Sprintf("pd-ctl -u %s store limit", pdURL)

	resp, err := s.regionServerCli.ExecuteClusterAgentCommand(
		ctx,
		region,
		strconv.FormatUint(cluster.ClusterId, 10),
		&clusteragentv1.ExecuteCommandReq{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
			Command:   command,
		},
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to execute pd-ctl store limit")
	}

	// Validate response is not empty
	if strings.TrimSpace(resp.StdoutMsg) == "" {
		tilog.Info(ctx, "empty response from pd-ctl store limit, no limits configured",
			zap.Uint64("cluster_id", cluster.ClusterId))
		return make(map[string]entity.StoreLimits), nil
	}

	// Check exit code
	if resp.ExitCode != nil && *resp.ExitCode != 0 {
		tilog.Error(ctx, "pd-ctl store limit failed",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("stderr", resp.StderrMsg),
			zap.Int("exit_code", int(*resp.ExitCode)))
		return nil, bizerr.New(
			bizerr.ErrCodeUnknown,
			"%s",
			fmt.Sprintf("pd-ctl store limit failed with exit code %d: %s", *resp.ExitCode, resp.StderrMsg),
		)
	}

	// Parse JSON output - map of store ID to limits
	var limits map[string]entity.StoreLimits
	if err := json.Unmarshal([]byte(resp.StdoutMsg), &limits); err != nil {
		tilog.Error(ctx, "failed to parse store limits JSON",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("response", resp.StdoutMsg),
			zap.Error(err))
		return nil, bizerr.WrapError(err, "failed to parse store limits JSON")
	}

	return limits, nil
}

// setStoreLimitAll sets store limit for all stores in the cluster
func (s *serviceImpl) setStoreLimitAll(ctx context.Context, region *entity.Region, cluster *entity.Cluster, rate int, limitType string) error {
	if rate < 0 {
		return bizerr.New(bizerr.ErrCodeParams, "%s", fmt.Sprintf("invalid rate value: %d", rate))
	}
	if limitType != "add-peer" && limitType != "remove-peer" {
		return bizerr.New(bizerr.ErrCodeParams, "%s", fmt.Sprintf("invalid limit type: %s", limitType))
	}

	pdURL := buildPdURL(cluster.ClusterId)
	// Construct command: pd-ctl store limit all <rate> <type>
	command := fmt.Sprintf("pd-ctl -u %s store limit all %d %s", pdURL, rate, limitType)

	resp, err := s.regionServerCli.ExecuteClusterAgentCommand(
		ctx,
		region,
		strconv.FormatUint(cluster.ClusterId, 10),
		&clusteragentv1.ExecuteCommandReq{
			ClusterId: strconv.FormatUint(cluster.ClusterId, 10),
			Command:   command,
		},
	)
	if err != nil {
		return bizerr.WrapError(err, "failed to execute pd-ctl store limit all")
	}

	// Check exit code and stderr
	if resp.ExitCode != nil && *resp.ExitCode != 0 {
		tilog.Error(ctx, "pd-ctl store limit all failed",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("limit_type", limitType),
			zap.Int("rate", rate),
			zap.String("stderr", resp.StderrMsg),
			zap.Int("exit_code", int(*resp.ExitCode)))
		return bizerr.New(
			bizerr.ErrCodeUnknown,
			"%s",
			fmt.Sprintf("pd-ctl store limit all failed with exit code %d: %s", *resp.ExitCode, resp.StderrMsg),
		)
	}

	if resp.StderrMsg != "" {
		tilog.Warn(ctx, "pd-ctl store limit all stderr",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.String("stderr", resp.StderrMsg),
			zap.String("limit_type", limitType),
			zap.Int("rate", rate))
	}

	return nil
}

func (s *serviceImpl) getSourcePdConfig(
	ctx context.Context,
	sourceRegion *entity.Region,
	sourceCluster *entity.Cluster,
	configType string,
	allowFallbackToBackup bool,
) (map[string]any, bool, error) {
	cfg, liveErr := s.getPdConfig(ctx, sourceRegion, sourceCluster, configType)
	if liveErr == nil {
		return cfg, true, nil
	}

	if !allowFallbackToBackup {
		tilog.Info(ctx, "failed to get source pd config from live cluster and backup fallback is disabled",
			zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
			zap.String("config_type", configType),
			zap.Error(liveErr),
		)
		return nil, false, liveErr
	}

	tilog.Warn(ctx, "failed to get source pd config from live cluster, fallback to backup",
		zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
		zap.String("config_type", configType),
		zap.Error(liveErr),
	)

	backup, backupErr := s.configBackupRepo.GetByClusterID(ctx, sourceCluster.ClusterId)
	if backupErr != nil {
		if backupErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, false, nil
		}
		return nil, false, backupErr
	}

	switch configType {
	case "":
		if len(backup.PdConfig) == 0 {
			return nil, false, nil
		}
		return backup.PdConfig, true, nil
	case "server":
		if len(backup.PdServerConfig) == 0 {
			return nil, false, nil
		}
		return backup.PdServerConfig, true, nil
	default:
		return nil, false, nil
	}
}

func (s *serviceImpl) getSourcePlacementRules(
	ctx context.Context,
	sourceRegion *entity.Region,
	sourceCluster *entity.Cluster,
	allowFallbackToBackup bool,
) ([]entity.PlacementRuleBundle, bool, error) {
	bundles, liveErr := s.getNonTiflashPlacementRules(ctx, sourceRegion, sourceCluster)
	if liveErr == nil {
		return bundles, true, nil
	}

	if !allowFallbackToBackup {
		tilog.Info(ctx, "failed to get source placement rules from live cluster and backup fallback is disabled",
			zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
			zap.Error(liveErr),
		)
		return nil, false, liveErr
	}

	tilog.Warn(ctx, "failed to get source placement rules from live cluster, fallback to backup",
		zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
		zap.Error(liveErr),
	)
	backup, backupErr := s.configBackupRepo.GetByClusterID(ctx, sourceCluster.ClusterId)
	if backupErr != nil {
		if backupErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, false, nil
		}
		return nil, false, backupErr
	}
	if len(backup.PlacementRules) == 0 {
		return nil, false, nil
	}
	return backup.PlacementRules, true, nil
}

func (s *serviceImpl) getSourceStoreLimits(
	ctx context.Context,
	sourceRegion *entity.Region,
	sourceCluster *entity.Cluster,
	allowFallbackToBackup bool,
) (map[string]entity.StoreLimits, bool, error) {
	limits, liveErr := s.getStoreLimits(ctx, sourceRegion, sourceCluster)
	if liveErr == nil {
		return limits, true, nil
	}

	if !allowFallbackToBackup {
		tilog.Info(ctx, "failed to get source store limits from live cluster and backup fallback is disabled",
			zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
			zap.Error(liveErr),
		)
		return nil, false, liveErr
	}

	tilog.Warn(ctx, "failed to get source store limits from live cluster, fallback to backup",
		zap.Uint64("source_cluster_id", sourceCluster.ClusterId),
		zap.Error(liveErr),
	)

	backup, backupErr := s.configBackupRepo.GetByClusterID(ctx, sourceCluster.ClusterId)
	if backupErr != nil {
		if backupErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, false, nil
		}
		return nil, false, backupErr
	}
	if len(backup.StoreLimits) == 0 {
		return nil, false, nil
	}
	return backup.StoreLimits, true, nil
}