package v1beta1

import (
	"context"
	"strconv"
	"strings"

	"github.com/tidbcloud/pkgv2/bizerr"
	v1beta1pb "github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

// maxMigrationTaskNameLen defines the maximum allowed length for a migration task name.
const maxMigrationTaskNameLen = 64

func validateCreateTaskReq(
	req *v1beta1pb.CreateMigrationReq,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	// Validate task name
	req.DisplayName = strings.TrimSpace(req.GetDisplayName())
	if req.GetDisplayName() == "" {
		return bizerr.New(bizerr.ErrCodeParams, "migration name should not be empty")
	}
	if len(req.GetDisplayName()) > maxMigrationTaskNameLen {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"migration name is too long, maximum %d characters allowed",
			maxMigrationTaskNameLen,
		)
	}

	// Validate mode
	if req.GetMode() == v1beta1pb.TaskMode_MODE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "invalid task mode")
	}
	if req.ShardMode != nil {
		switch req.GetShardMode() {
		case v1beta1pb.TaskShardMode_PESSIMISTIC, v1beta1pb.TaskShardMode_OPTIMISTIC:
		default:
			return bizerr.New(bizerr.ErrCodeParams, "invalid shard mode")
		}
	}

	// Validate sources
	if len(req.GetSources()) != 1 {
		return bizerr.New(bizerr.ErrCodeParams, "only one source is supported currently")
	}
	for i, source := range req.GetSources() {
		if bizErr := validateSource(source, cluster, i); bizErr != nil {
			return bizErr
		}
	}

	// Validate target database
	if req.GetTarget() == nil {
		return bizerr.New(bizerr.ErrCodeParams, "target_db is required")
	}
	if bizErr := validateTargetDatabase(req.GetTarget()); bizErr != nil {
		return bizErr
	}
	return nil
}

func (s *DMService) validateCluster(
	ctx context.Context,
	clusterID string,
	logger *zap.Logger,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	clusterIDUint, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "Invalid cluster ID: %s", clusterID)
	}
	cluster, bizErr := s.mgmtSvc.GetCluster(ctx, clusterIDUint, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		logger.Error("GetClusterByID failed", zap.String("cluster_id", clusterID), zap.Error(bizErr))
		return nil, bizErr
	}

	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(
			errorcode.ErrCodeClusterNotRunning,
			"Cluster ID %s is not in normal state, please try again later.",
			clusterID,
		)
	}
	// essential plan check
	if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu() == 0 &&
		cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() == 0 {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"Current Data Migration service is only supported for Essential plan clusters.",
		)
	}
	return cluster, nil
}

// validateSource validates a single source configuration
func validateSource(source *v1beta1pb.Source, cluster *tidbmgmtservicev1.Cluster, index int) bizerr.BizError {
	if source == nil {
		return bizerr.New(bizerr.ErrCodeParams, "source[%d] is empty", index)
	}
	if source.SourceType == v1beta1pb.Source_SOURCE_TYPE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "source[%d] source_type is required", index)
	}

	if bizErr := validateConnProfile(source, cluster.CloudProvider, index); bizErr != nil {
		return bizErr
	}

	if source.GetMigrationRules() != nil {
		if bizErr := validateMigrationRules(source); bizErr != nil {
			return bizErr
		}
	}

	if source.GetBinlogPosition() < 0 {
		return bizerr.New(bizerr.ErrCodeParams, "source[%d] binlog position must be non-negative", index)
	}
	return nil
}

func validateMigrationRules(source *v1beta1pb.Source) bizerr.BizError {
	for _, rule := range source.GetMigrationRules() {
		if rule.GetSourceTable() != nil {
			if rule.GetSourceTable().GetSchemaPattern() == "" {
				return bizerr.New(bizerr.ErrCodeParams, "the source schema of migration rule cannot be empty")
			}
			if bizErr := validateWildcardPattern(rule.GetSourceTable().GetSchemaPattern()); bizErr != nil {
				return bizErr
			}
			if bizErr := validateWildcardPattern(rule.GetSourceTable().GetTablePattern()); bizErr != nil {
				return bizErr
			}
		}
		if rule.GetTargetTable() != nil {
			if rule.GetTargetTable().Schema == "" {
				return bizerr.New(bizerr.ErrCodeParams, "the target schema of migration rule cannot be empty")
			}
			if bizErr := validateNoWildcard(
				rule.GetTargetTable().Schema,
				"the target schema of migration rule does not support wildcards",
			); bizErr != nil {
				return bizErr
			}
			if bizErr := validateNoWildcard(
				rule.GetTargetTable().Table,
				"the target table of migration rule does not support wildcards",
			); bizErr != nil {
				return bizErr
			}
		}
	}
	return nil
}

// validateNoWildcard ensures the provided value does not include unsupported wildcard characters.
func validateNoWildcard(value, errMsg string) bizerr.BizError {
	if value == "" {
		return nil
	}
	if strings.ContainsAny(value, "*?[]") {
		return bizerr.New(bizerr.ErrCodeParams, "%s", errMsg)
	}
	return nil
}

// validateConnProfile validates connection profile
func validateConnProfile(source *v1beta1pb.Source, provider string, sourceIndex int) bizerr.BizError {
	connProfile := source.GetConnProfile()
	if connProfile == nil {
		return bizerr.New(bizerr.ErrCodeParams, "source[%d] conn_profile is required", sourceIndex)
	}
	if connProfile.GetConnType() == v1beta1pb.ConnType_CONN_TYPE_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "conn_profile.conn_type is required")
	}
	host := strings.TrimSpace(connProfile.GetHost())
	id := strings.TrimSpace(connProfile.GetEndpointId())
	if connProfile.GetConnType() == v1beta1pb.ConnType_PUBLIC && host == "" {
		return bizerr.New(bizerr.ErrCodeParams, "source[%d] host is required for public connection", sourceIndex)
	}
	if connProfile.GetConnType() == v1beta1pb.ConnType_PRIVATE_LINK && id == "" {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"source[%d] endpoint_id is required for private connection",
			sourceIndex,
		)
	}
	connProfile.User = strings.TrimSpace(connProfile.GetUser())
	if connProfile.User == "" {
		return bizerr.New(bizerr.ErrCodeParams, "source[%d] user is empty", sourceIndex)
	}

	if source.ConnProfile.ConnType == v1beta1pb.ConnType_PRIVATE_LINK {
		if source.SourceType == v1beta1pb.Source_AWS_RDS_MYSQL && provider != config.ProviderAWS {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"AWS endpoint service private link connection can only be used for AWS cluster",
			)
		}
		if source.SourceType == v1beta1pb.Source_ALICLOUD_RDS_MYSQL &&
			provider != config.ProviderAliCloud {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"Alicloud endpoint service private link connection can only be used for Alicloud cluster",
			)
		}
	}
	return nil
}

// validateWildcardPattern enforces supported wildcards: '*', '?', and character classes '[...]'.
// Constraints: only one '*', and it must appear at the end of the pattern.
func validateWildcardPattern(pattern string) bizerr.BizError {
	if pattern == "" || pattern == "*" {
		return nil
	}

	asteriskCount := strings.Count(pattern, "*")
	if asteriskCount > 1 {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"invalid pattern '%s': multiple '*' characters detected, only one is allowed",
			pattern,
		)
	}
	if asteriskCount == 1 {
		asteriskIndex := strings.Index(pattern, "*")
		if asteriskIndex != len(pattern)-1 {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"invalid pattern '%s': '*' can only be placed at the end",
				pattern,
			)
		}
	}

	// Validate character classes '[]' and reject misplaced '*'.
	inClass := false
	classHasChar := false
	for i := 0; i < len(pattern); i++ {
		switch pattern[i] {
		case '[':
			if inClass {
				return bizerr.New(bizerr.ErrCodeParams, "invalid pattern '%s': nested '[' not allowed", pattern)
			}
			inClass = true
			classHasChar = false
		case ']':
			if !inClass {
				return bizerr.New(bizerr.ErrCodeParams, "invalid pattern '%s': unmatched ']'", pattern)
			}
			if !classHasChar {
				return bizerr.New(bizerr.ErrCodeParams, "invalid pattern '%s': empty character class []", pattern)
			}
			inClass = false
		case '*':
			if inClass {
				return bizerr.New(bizerr.ErrCodeParams, "invalid pattern '%s': '*' cannot appear inside []", pattern)
			}
		case '?':
			// allowed anywhere
		default:
			if inClass {
				classHasChar = true
			}
		}
	}

	if inClass {
		return bizerr.New(bizerr.ErrCodeParams, "invalid pattern '%s': unclosed '['", pattern)
	}

	return nil
}

// validateTargetDatabase validates target database configuration
func validateTargetDatabase(targetDb *v1beta1pb.Target) bizerr.BizError {
	if targetDb == nil {
		return bizerr.New(bizerr.ErrCodeParams, "target_db is required")
	}
	targetDb.User = strings.TrimSpace(targetDb.GetUser())
	if targetDb.User == "" {
		return bizerr.New(bizerr.ErrCodeParams, "target_db user is empty")
	}
	if targetDb.Password == "" {
		return bizerr.New(bizerr.ErrCodeParams, "target_db password is required")
	}
	return nil
}