package migrationprecheck

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

	regionaldedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	pkgclusterproxy "github.com/tidbcloud/pkgv2/clusterproxy"

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

const (
	CheckDuplicateCheckConstraints     = "duplicate_check_constraints"
	CheckMySQLDBCollationCompatibility = "mysql_db_collation_compatibility"
	CheckInvalidForeignKey             = "invalid_foreign_key"
	CheckTiDBVersion                   = "tidb_version"
	CheckTiDBGCLifeTime                = "tidb_gc_life_time"
	CheckTablesWithoutValidIndex       = "tables_without_valid_index"
	CheckSequence                      = "sequence"
	CheckNewCollationEnabled           = "new_collation_enabled"
	CheckTTL                           = "ttl"
	CheckTiFlashReplicas               = "tiflash_replicas"
	maximumEvidenceItems               = 50
	maximumEvidenceBytes               = 32 * 1024
	noActionRequired                   = "N/A."
	tidbVersionSQL                     = "SELECT @@version AS tidb_version"
	mysqlDBSchemaSQL                   = "SELECT COUNT(*) AS value FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA='mysql' AND TABLE_NAME='db' AND COLUMN_NAME='DB' AND COLLATION_NAME='utf8mb4_bin'"
	invalidForeignKeyCandidateSQL      = "SELECT DISTINCT TABLE_SCHEMA AS table_schema, TABLE_NAME AS table_name FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS WHERE CONSTRAINT_TYPE = 'FOREIGN KEY' AND TABLE_SCHEMA NOT IN ('mysql','INFORMATION_SCHEMA','PERFORMANCE_SCHEMA','METRICS_SCHEMA') ORDER BY TABLE_SCHEMA, TABLE_NAME"
	ttlJobEnabledSQL                   = "SELECT @@GLOBAL.tidb_ttl_job_enable AS tidb_ttl_job_enable"
	ttlCandidateTablesSQL              = "SELECT DISTINCT t.TABLE_SCHEMA AS table_schema, t.TABLE_NAME AS table_name FROM mysql.tidb_ttl_table_status AS s JOIN information_schema.tables AS t ON s.parent_table_id = t.tidb_table_id ORDER BY t.TABLE_SCHEMA, t.TABLE_NAME"
	tiFlashReplicaTableCountSQL        = "SELECT COUNT(*) AS tables_with_tiflash_replicas FROM INFORMATION_SCHEMA.TIFLASH_REPLICA WHERE REPLICA_COUNT > 0"
	ttlRiskAction                      = "Before cutover:\n" +
		"- Record the fully qualified names of all tables with TTL enabled on the Dedicated cluster.\n" +
		"- Disable cluster-wide TTL job execution on the Dedicated cluster by running:\n" +
		"  set global tidb_ttl_job_enable = 'OFF'\n" +
		"After cutover:\n" +
		"- Verify that @@GLOBAL.tidb_ttl_job_enable is ON on the Premium cluster.\n" +
		"- Manually re-enable table-level TTL for each recorded table on the Premium cluster by running:\n" +
		"  ALTER TABLE <schema_name>.<table_name> TTL_ENABLE = 'ON';"
	tablesWithoutValidIndexBaseSQL = `SELECT table_schema, table_name
FROM information_schema.tables
WHERE table_schema NOT IN ('METRICS_SCHEMA', 'mysql', 'PERFORMANCE_SCHEMA', 'INFORMATION_SCHEMA', 'sys')
  AND table_type = 'BASE TABLE'
  AND (table_schema, table_name) NOT IN (
    SELECT DISTINCT table_schema, table_name
    FROM information_schema.tidb_indexes
    WHERE KEY_NAME = 'PRIMARY'
      AND table_schema NOT IN ('METRICS_SCHEMA', 'mysql', 'PERFORMANCE_SCHEMA', 'INFORMATION_SCHEMA', 'sys')
    UNION ALL
    SELECT DISTINCT a.table_schema, a.table_name
    FROM (
      SELECT table_schema, table_name, column_name
      FROM information_schema.tidb_indexes
      WHERE KEY_NAME <> 'PRIMARY'
        AND NON_UNIQUE = 0
        AND table_schema NOT IN ('METRICS_SCHEMA', 'mysql', 'PERFORMANCE_SCHEMA', 'INFORMATION_SCHEMA', 'sys')
    ) a
    LEFT JOIN (
      SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, IS_NULLABLE, GENERATION_EXPRESSION
      FROM information_schema.columns
      WHERE table_schema NOT IN ('METRICS_SCHEMA', 'mysql', 'PERFORMANCE_SCHEMA', 'INFORMATION_SCHEMA', 'sys')
    ) b
      ON a.table_schema = b.table_schema
      AND a.table_name = b.table_name
      AND a.column_name = b.column_name
    WHERE b.IS_NULLABLE = 'NO' AND b.GENERATION_EXPRESSION = ''
  )`
)

var (
	duplicateCheckConstraintsSQL = fmt.Sprintf(`WITH matches AS (
  SELECT CONSTRAINT_SCHEMA AS constraint_schema, CONSTRAINT_NAME AS constraint_name
  FROM INFORMATION_SCHEMA.CHECK_CONSTRAINTS
  GROUP BY CONSTRAINT_SCHEMA, CONSTRAINT_NAME
  HAVING COUNT(*) > 1
)
SELECT constraint_schema, constraint_name, COUNT(*) OVER () AS total_count
FROM matches
ORDER BY constraint_schema, constraint_name
LIMIT %d`, maximumEvidenceItems)
	tablesWithoutValidIndexSQL = fmt.Sprintf(`WITH matches AS (
%s
)
SELECT table_schema, table_name, COUNT(*) OVER () AS total_count
FROM matches
ORDER BY table_schema, table_name
LIMIT %d`, tablesWithoutValidIndexBaseSQL, maximumEvidenceItems)
	sequencesSQL = fmt.Sprintf(`WITH matches AS (
  SELECT SEQUENCE_SCHEMA AS sequence_schema, SEQUENCE_NAME AS sequence_name
  FROM INFORMATION_SCHEMA.SEQUENCES
)
SELECT sequence_schema, sequence_name, COUNT(*) OVER () AS total_count
FROM matches
ORDER BY sequence_schema, sequence_name
LIMIT %d`, maximumEvidenceItems)
)

var invalidForeignKeyPattern = regexp.MustCompile(
	"(?i)CONSTRAINT\\s+`((?:``|[^`])+)`\\s+FOREIGN\\s+KEY[^\\n]*FOREIGN\\s+KEY\\s+INVALID",
)

var ttlEnabledPattern = regexp.MustCompile(`(?i)TTL_ENABLE\s*=\s*'ON'`)

type checkResult struct {
	status   globaldedicatedv1.MigrationCompatibilityCheckStatus
	evidence string
	action   string
}

type checkDefinition struct {
	run func(context.Context, clusterproxy.ClusterProxy, *entity.Cluster) (checkResult, bizerr.BizError)
}

var checkDefinitions = map[string]checkDefinition{
	CheckDuplicateCheckConstraints: {run: withoutCluster(runDuplicateCheckConstraints)},
	CheckMySQLDBCollationCompatibility: {
		run: withoutCluster(runMySQLDBCollationCompatibility),
	},
	CheckInvalidForeignKey: {run: withoutCluster(runInvalidForeignKeyCheck)},
	CheckTiDBVersion: {
		run: withoutCluster(scalarCheck(tidbVersionSQL, evaluateTiDBVersion)),
	},
	CheckTiDBGCLifeTime: {
		run: withoutCluster(scalarCheck("SELECT @@GLOBAL.tidb_gc_life_time AS tidb_gc_life_time", evaluateTiDBGCLifeTime)),
	},
	CheckTablesWithoutValidIndex: {run: withoutCluster(runTablesWithoutValidIndex)},
	CheckSequence:                {run: withoutCluster(runSequenceCheck)},
	CheckNewCollationEnabled:     {run: withoutCluster(runNewCollationEnabled)},
	CheckTTL:                     {run: withoutCluster(runTTLCheck)},
	CheckTiFlashReplicas:         {run: runTiFlashReplicas},
}

func withoutCluster(
	run func(context.Context, clusterproxy.ClusterProxy) (checkResult, bizerr.BizError),
) func(context.Context, clusterproxy.ClusterProxy, *entity.Cluster) (checkResult, bizerr.BizError) {
	return func(ctx context.Context, proxy clusterproxy.ClusterProxy, _ *entity.Cluster) (checkResult, bizerr.BizError) {
		return run(ctx, proxy)
	}
}

func scalarCheck(
	query string,
	evaluate func(string) (checkResult, error),
) func(context.Context, clusterproxy.ClusterProxy) (checkResult, bizerr.BizError) {
	return func(ctx context.Context, proxy clusterproxy.ClusterProxy) (checkResult, bizerr.BizError) {
		raw, err := proxy.QuerySQL(ctx, query)
		if err != nil {
			return checkResult{}, err
		}
		value, parseErr := firstScalar(raw)
		if parseErr != nil {
			return checkResult{}, bizerr.WrapError(parseErr, "parse scalar query result")
		}
		result, evaluateErr := evaluate(strings.TrimSpace(value))
		if evaluateErr != nil {
			return checkResult{}, bizerr.WrapError(evaluateErr, "evaluate scalar query result")
		}
		return result, nil
	}
}

func (s *service) RunDedicatedToPremiumCompatibilityCheck(
	ctx context.Context,
	req *globaldedicatedv1.RunDedicatedToPremiumCompatibilityCheckRequest,
) (*globaldedicatedv1.RunDedicatedToPremiumCompatibilityCheckResponse, bizerr.BizError) {
	if req.GetClusterId() == 0 || req.GetCheckKey() == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id and check_key are required")
	}

	definition, ok := checkDefinitions[req.GetCheckKey()]
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported compatibility check key: %s", req.GetCheckKey())
	}
	cluster, proxy, err := s.getClusterAndProxy(ctx, req.GetClusterId())
	if err != nil {
		return nil, err
	}
	result, err := definition.run(ctx, proxy, cluster)
	if err != nil {
		return nil, err
	}
	return &globaldedicatedv1.RunDedicatedToPremiumCompatibilityCheckResponse{
		CheckKey: req.GetCheckKey(),
		Status:   result.status,
		Evidence: result.evidence,
		Action:   result.action,
	}, nil
}

func (s *service) getClusterAndProxy(
	ctx context.Context,
	clusterID uint64,
) (*entity.Cluster, clusterproxy.ClusterProxy, bizerr.BizError) {
	cluster, err := s.clusterRepo.GetAndLoadComponents(ctx, clusterID)
	if err != nil {
		return nil, nil, err
	}
	regionEntity, err := s.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, nil, err
	}
	regionalCluster, err := s.regionServerCli.GetCluster(ctx, regionEntity, &regionaldedicatedv1.GetClusterRequest{
		ClusterId: strconv.FormatUint(clusterID, 10),
	})
	if err != nil {
		return nil, nil, err
	}
	password, err := s.encryptor.DecryptBase64(ctx, cluster.EncryptedCloudAdminPassword)
	if err != nil {
		return nil, nil, err
	}
	operator := &pkgclusterproxy.ClusterOperator{
		Username: clusterproxy.CloudAdminUser, AuthMethod: pkgclusterproxy.AuthMethodPassword, Credential: password,
	}
	proxy, err := s.clusterProxyProvider.GetClusterProxy(ctx, clusterID, regionalCluster.GetProxyBaseUrl(), cluster.Version, operator)
	if err != nil {
		return nil, nil, err
	}
	return cluster, proxy, nil
}

func runDuplicateCheckConstraints(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, tidbVersionSQL)
	if err != nil {
		return checkResult{}, err
	}
	version, parseErr := firstScalar(raw)
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse TiDB version query result")
	}
	atLeast74, versionErr := tidbVersionAtLeast(strings.TrimSpace(version), 7, 4)
	if versionErr != nil {
		return checkResult{}, bizerr.WrapError(versionErr, "evaluate TiDB version for duplicate check constraints")
	}
	if !atLeast74 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: "The information_schema.CHECK_CONSTRAINTS table is added starting with v7.4.0.",
			action:   noActionRequired,
		}, nil
	}

	raw, err = proxy.QuerySQL(ctx, duplicateCheckConstraintsSQL)
	if err != nil {
		return checkResult{}, err
	}
	rows, total, parseErr := stringColumnsWithTotal(raw, "constraint_schema", "constraint_name")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse duplicate check constraints query result")
	}
	if len(rows) == 0 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: "Every check-constraint name is unique within its database.",
			action:   noActionRequired,
		}, nil
	}
	duplicates := make([]string, 0, len(rows))
	for _, row := range rows {
		duplicates = append(duplicates, fmt.Sprintf("database %s, constraint name %s", row[0], row[1]))
	}
	return checkResult{
		status: globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
		evidence: summarizedEvidence(
			"Duplicate check constraints detected: ", duplicates, total,
		),
		action: "Resolve the affected constraints so that every check-constraint name is unique within its database, then rerun the check before backup.",
	}, nil
}

type privilegeTableCount struct {
	table   string
	total   int64
	grouped int64
}

func runMySQLDBCollationCompatibility(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, mysqlDBSchemaSQL)
	if err != nil {
		return checkResult{}, err
	}
	value, parseErr := firstScalar(raw)
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse mysql.db schema query result")
	}
	matches, parseErr := strconv.ParseInt(strings.TrimSpace(value), 10, 64)
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse mysql.db schema match count")
	}
	if matches != 1 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: "The mysql.db schema does not match the known privilege-table compatibility-risk pattern.",
			action:   noActionRequired,
		}, nil
	}

	checks := []struct {
		table string
		query string
	}{
		{
			table: "mysql.db",
			query: "SELECT (SELECT COUNT(1) FROM mysql.db) AS total_count, " +
				"(SELECT COUNT(1) FROM (SELECT Host, DB COLLATE utf8mb4_general_ci, User FROM mysql.db " +
				"GROUP BY Host, DB COLLATE utf8mb4_general_ci, User) AS grouped_rows) AS grouped_count",
		},
		{
			table: "mysql.tables_priv",
			query: "SELECT (SELECT COUNT(1) FROM mysql.tables_priv) AS total_count, " +
				"(SELECT COUNT(1) FROM (SELECT Host, DB COLLATE utf8mb4_general_ci, User, " +
				"Table_name COLLATE utf8mb4_general_ci FROM mysql.tables_priv GROUP BY Host, " +
				"DB COLLATE utf8mb4_general_ci, User, Table_name COLLATE utf8mb4_general_ci) AS grouped_rows) AS grouped_count",
		},
		{
			table: "mysql.columns_priv",
			query: "SELECT (SELECT COUNT(1) FROM mysql.columns_priv) AS total_count, " +
				"(SELECT COUNT(1) FROM (SELECT Host, DB COLLATE utf8mb4_general_ci, User, " +
				"Table_name COLLATE utf8mb4_general_ci, Column_name COLLATE utf8mb4_general_ci " +
				"FROM mysql.columns_priv GROUP BY Host, DB COLLATE utf8mb4_general_ci, User, " +
				"Table_name COLLATE utf8mb4_general_ci, Column_name COLLATE utf8mb4_general_ci) AS grouped_rows) AS grouped_count",
		},
	}
	counts := make([]privilegeTableCount, 0, len(checks))
	for _, check := range checks {
		count, queryErr := queryPrivilegeTableCount(ctx, proxy, check.table, check.query)
		if queryErr != nil {
			return checkResult{}, queryErr
		}
		counts = append(counts, count)
	}

	evidence := make([]string, 0, len(counts))
	collisions := make([]string, 0, len(counts))
	for _, count := range counts {
		value := fmt.Sprintf("%s: %d=%d", count.table, count.total, count.grouped)
		evidence = append(evidence, value)
		if count.total != count.grouped {
			collisions = append(collisions, fmt.Sprintf(
				"%s: total_count=%d, grouped_count=%d", count.table, count.total, count.grouped,
			))
		}
	}
	if len(collisions) > 0 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
			evidence: "Collation-equivalent privilege keys were detected in " + strings.Join(collisions, "; ") + ".",
			action:   "Stop migration planning and engage the DBA or migration engineering team to define a supported remediation or migration path.",
		}, nil
	}
	return checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
		evidence: "No collation collision was detected. " + strings.Join(evidence, "; ") + ".",
		action:   "Restore the system tables with --sys-check-collation.",
	}, nil
}

func queryPrivilegeTableCount(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
	table string,
	query string,
) (privilegeTableCount, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, query)
	if err != nil {
		return privilegeTableCount{}, err
	}
	totalValue, totalErr := scalarByKey(raw, "total_count")
	groupedValue, groupedErr := scalarByKey(raw, "grouped_count")
	if totalErr != nil || groupedErr != nil {
		return privilegeTableCount{}, bizerr.WrapError(
			errors.Join(totalErr, groupedErr), "parse privilege table count query result",
		)
	}
	total, totalErr := strconv.ParseInt(strings.TrimSpace(totalValue), 10, 64)
	grouped, groupedErr := strconv.ParseInt(strings.TrimSpace(groupedValue), 10, 64)
	if totalErr != nil || groupedErr != nil {
		return privilegeTableCount{}, bizerr.WrapError(
			errors.Join(totalErr, groupedErr), "parse privilege table count values",
		)
	}
	return privilegeTableCount{table: table, total: total, grouped: grouped}, nil
}

func runInvalidForeignKeyCheck(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, invalidForeignKeyCandidateSQL)
	if err != nil {
		return checkResult{}, err
	}
	rows, parseErr := stringColumns(raw, "table_schema", "table_name")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse invalid foreign key candidate query result")
	}
	invalidForeignKeys := make([]string, 0, maximumEvidenceItems)
	totalInvalidForeignKeys := 0
	for _, row := range rows {
		if !validIdentifier(row[0]) || !validIdentifier(row[1]) {
			return checkResult{}, bizerr.New(
				bizerr.ErrCodeUnknown, "invalid foreign key query returned an unsupported schema or table identifier",
			)
		}
		showSQL := fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", row[0], row[1])
		createRaw, queryErr := proxy.QuerySQL(ctx, showSQL)
		if queryErr != nil {
			return checkResult{}, queryErr
		}
		createSQL, valueErr := scalarByKeyContains(createRaw, "create table")
		if valueErr != nil {
			return checkResult{}, bizerr.WrapError(valueErr, "parse SHOW CREATE TABLE result")
		}
		if !strings.Contains(strings.ToUpper(createSQL), "FOREIGN KEY INVALID") {
			continue
		}
		constraintNames := invalidForeignKeyNames(createSQL)
		if len(constraintNames) == 0 {
			return checkResult{}, bizerr.New(
				bizerr.ErrCodeUnknown, "FOREIGN KEY INVALID marker found but no constraint name could be parsed",
			)
		}
		for _, constraintName := range constraintNames {
			totalInvalidForeignKeys++
			if len(invalidForeignKeys) < maximumEvidenceItems {
				invalidForeignKeys = append(invalidForeignKeys, row[0]+"."+row[1]+"."+constraintName)
			}
		}
	}
	if totalInvalidForeignKeys == 0 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: "No invalid foreign-key definition was detected.",
			action:   noActionRequired,
		}, nil
	}
	return checkResult{
		status: globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
		evidence: summarizedEvidence(
			"Invalid foreign keys detected: ", invalidForeignKeys, totalInvalidForeignKeys,
		),
		action: "Clean up (ALTER TABLE {table_name} DROP FOREIGN KEY {foreign_key_name}) or recreate the affected constraints, and rerun the check before backup.",
	}, nil
}

func evaluateTiDBVersion(value string) (checkResult, error) {
	result := checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
		evidence: fmt.Sprintf("Dedicated cluster TiDB version is %s.", value),
		action:   noActionRequired,
	}
	atLeast75, err := tidbVersionAtLeast(value, 7, 5)
	if err != nil {
		return checkResult{}, err
	}
	if !atLeast75 {
		result.action = "For TiDB versions below v7.5.x, use the TiCDC old architecture and configure the blocklist via dataflow-service."
	}
	return result, nil
}

func evaluateTiDBGCLifeTime(value string) (checkResult, error) {
	duration, err := time.ParseDuration(value)
	if err != nil {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NEEDS_INPUT,
			evidence: fmt.Sprintf("tidb_gc_life_time=%s could not be interpreted.", value),
			action:   "Confirm tidb_gc_life_time manually and rerun the check.",
		}, nil
	}
	if duration < 72*time.Hour {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
			evidence: fmt.Sprintf("tidb_gc_life_time=%s, which is shorter than the required migration window.", value),
			action:   "Set tidb_gc_life_time system variable to at least 72h before starting the migration backup.",
		}, nil
	}
	return checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
		evidence: fmt.Sprintf("tidb_gc_life_time=%s, which meets the minimum migration window.", value),
		action:   noActionRequired,
	}, nil
}

func runTablesWithoutValidIndex(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, tablesWithoutValidIndexSQL)
	if err != nil {
		return checkResult{}, err
	}
	rows, total, parseErr := stringColumnsWithTotal(raw, "table_schema", "table_name")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse tables without valid index query result")
	}
	tables := qualifiedNames(rows)
	if len(tables) == 0 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: "Every base table in the TiCDC replication scope has a valid primary key or qualifying unique index.",
			action:   noActionRequired,
		}, nil
	}
	return checkResult{
		status: globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
		evidence: summarizedEvidence(
			"The following tables in the replication scope do not have a valid primary key or qualifying unique index: ",
			tables,
			total,
		),
		action: "Add a valid primary key or qualifying unique index. If that is not possible, agree on an explicit alternative handling or exclusion plan with the customer and DBA.",
	}, nil
}

func runSequenceCheck(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, sequencesSQL)
	if err != nil {
		return checkResult{}, err
	}
	rows, total, parseErr := stringColumnsWithTotal(raw, "sequence_schema", "sequence_name")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse sequence query result")
	}
	sequences := qualifiedNames(rows)
	if len(sequences) == 0 {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: "None of the tables to be replicated are of type SEQUENCE.",
			action:   noActionRequired,
		}, nil
	}
	return checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
		evidence: summarizedEvidence("The following tables are of type SEQUENCE: ", sequences, total),
		action:   "Define how to handle tables of type SEQUENCE during migration.",
	}, nil
}

func runNewCollationEnabled(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(
		ctx,
		"SELECT VARIABLE_VALUE AS new_collation_enabled FROM mysql.tidb WHERE VARIABLE_NAME = 'new_collation_enabled'",
	)
	if err != nil {
		return checkResult{}, err
	}
	value, parseErr := firstScalar(raw)
	if parseErr != nil {
		value = "unknown"
	}
	if parseBool(value) {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: fmt.Sprintf("new_collation_enabled=%s.", value),
			action:   noActionRequired,
		}, nil
	}
	return checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
		evidence: fmt.Sprintf("new_collation_enabled=%s; the dedicated cluster does not meet the migration requirement.", value),
		action:   "Stop migration planning and engage the DBA or migration engineering team to define a supported remediation or migration path.",
	}, nil
}

func runTTLCheck(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (checkResult, bizerr.BizError) {
	raw, err := proxy.QuerySQL(ctx, ttlJobEnabledSQL)
	if err != nil {
		return checkResult{}, err
	}
	value, parseErr := scalarByKey(raw, "tidb_ttl_job_enable")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse tidb_ttl_job_enable query result")
	}
	jobEnabled, parseErr := parseTTLJobEnabled(value)
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse tidb_ttl_job_enable value")
	}

	raw, err = proxy.QuerySQL(ctx, ttlCandidateTablesSQL)
	if err != nil {
		return checkResult{}, err
	}
	rows, parseErr := stringColumns(raw, "table_schema", "table_name")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse TTL table query result")
	}

	enabledTables := make([]string, 0, len(rows))
	for _, row := range rows {
		if !validIdentifier(row[0]) || !validIdentifier(row[1]) {
			return checkResult{}, bizerr.New(
				bizerr.ErrCodeUnknown, "TTL table query returned an unsupported schema or table identifier",
			)
		}
		showSQL := fmt.Sprintf("SHOW CREATE TABLE `%s`.`%s`", row[0], row[1])
		createRaw, queryErr := proxy.QuerySQL(ctx, showSQL)
		if queryErr != nil {
			return checkResult{}, queryErr
		}
		createSQL, valueErr := scalarByKeyContains(createRaw, "create table")
		if valueErr != nil {
			return checkResult{}, bizerr.WrapError(valueErr, "parse TTL SHOW CREATE TABLE result")
		}
		if ttlEnabledPattern.MatchString(createSQL) {
			enabledTables = append(enabledTables, row[0]+"."+row[1])
		}
	}

	state := "OFF"
	if jobEnabled {
		state = "ON"
	}
	if len(enabledTables) == 0 {
		evidence := fmt.Sprintf(
			"The global variable tidb_ttl_job_enable is %s, but table-level TTL is not enabled on any table.",
			state,
		)
		if !jobEnabled {
			evidence = "The global variable tidb_ttl_job_enable is OFF,  and table-level TTL is not enabled on any table."
		}
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: evidence,
			action:   noActionRequired,
		}, nil
	}

	evidence := summarizedEvidence(
		fmt.Sprintf(
			"The global variable tidb_ttl_job_enable is %s, and table-level TTL is enabled on the following tables: ",
			state,
		),
		enabledTables,
		len(enabledTables),
	)
	if !jobEnabled {
		return checkResult{
			status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
			evidence: evidence,
			action:   noActionRequired,
		}, nil
	}
	return checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_RISK_FOUND,
		evidence: evidence,
		action:   ttlRiskAction,
	}, nil
}

func parseTTLJobEnabled(value string) (bool, error) {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "1", "on", "true":
		return true, nil
	case "0", "off", "false":
		return false, nil
	default:
		return false, fmt.Errorf("unsupported value %q", value)
	}
}

func runTiFlashReplicas(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
	cluster *entity.Cluster,
) (checkResult, bizerr.BizError) {
	var nodeCount uint64
	for _, component := range cluster.GetClusterComponents() {
		if component.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH {
			nodeCount += uint64(component.GetCount())
		}
	}

	raw, err := proxy.QuerySQL(ctx, tiFlashReplicaTableCountSQL)
	if err != nil {
		return checkResult{}, err
	}
	tableCountValue, parseErr := scalarByKey(raw, "tables_with_tiflash_replicas")
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse TiFlash replica table count query result")
	}
	tableCount, parseErr := strconv.ParseInt(strings.TrimSpace(tableCountValue), 10, 64)
	if parseErr != nil {
		return checkResult{}, bizerr.WrapError(parseErr, "parse TiFlash replica table count")
	}

	return checkResult{
		status:   globaldedicatedv1.MigrationCompatibilityCheckStatus_NO_RISK_FOUND,
		evidence: fmt.Sprintf("TiFlash nodes: %d; tables with TiFlash replicas: %d.", nodeCount, tableCount),
		action:   noActionRequired,
	}, nil
}

func tidbVersionAtLeast(value string, wantMajor, wantMinor int) (bool, error) {
	lower := strings.ToLower(strings.TrimSpace(value))
	const marker = "tidb-v"
	_, after, ok := strings.Cut(lower, marker)
	if !ok {
		return false, fmt.Errorf("TiDB version marker not found")
	}
	parts := strings.Split(after, ".")
	if len(parts) < 2 {
		return false, fmt.Errorf("TiDB version has fewer than two components")
	}
	major, err := strconv.Atoi(parts[0])
	if err != nil {
		return false, fmt.Errorf("parse TiDB major version: %w", err)
	}
	minor, err := strconv.Atoi(parts[1])
	if err != nil {
		return false, fmt.Errorf("parse TiDB minor version: %w", err)
	}
	return major > wantMajor || major == wantMajor && minor >= wantMinor, nil
}

func qualifiedNames(rows [][]string) []string {
	values := make([]string, 0, len(rows))
	for _, row := range rows {
		values = append(values, row[0]+"."+row[1])
	}
	return values
}

func invalidForeignKeyNames(createSQL string) []string {
	normalized := strings.ReplaceAll(createSQL, `\n`, "\n")
	matches := invalidForeignKeyPattern.FindAllStringSubmatch(normalized, -1)
	names := make([]string, 0, len(matches))
	for _, match := range matches {
		names = append(names, strings.ReplaceAll(match[1], "``", "`"))
	}
	return names
}

func summarizedEvidence(prefix string, values []string, total int) string {
	if total < len(values) {
		total = len(values)
	}
	shown := min(len(values), maximumEvidenceItems)
	for shown >= 0 {
		omitted := total - shown
		var evidence string
		if shown == 0 {
			evidence = prefix + "No matching item details are shown."
		} else {
			evidence = prefix + strings.Join(values[:shown], ", ") + "."
		}
		if omitted > 0 {
			if omitted == 1 {
				evidence += fmt.Sprintf(
					" Showing the first %d of %d matches; 1 additional match is omitted.", shown, total,
				)
			} else {
				evidence += fmt.Sprintf(
					" Showing the first %d of %d matches; %d additional matches are omitted.",
					shown,
					total,
					omitted,
				)
			}
		}
		if len(evidence) <= maximumEvidenceBytes {
			return evidence
		}
		shown--
	}
	return "Matching item details are omitted because the Evidence limit is exceeded."
}

func firstScalar(raw []byte) (string, error) {
	var result map[string][]json.RawMessage
	if err := json.Unmarshal(raw, &result); err != nil {
		return "", err
	}
	for _, values := range result {
		if len(values) > 0 {
			return decodeScalar(values[0])
		}
	}
	return "", fmt.Errorf("empty result")
}

func scalarByKey(raw []byte, name string) (string, error) {
	var result map[string][]json.RawMessage
	if err := json.Unmarshal(raw, &result); err != nil {
		return "", err
	}
	for key, values := range result {
		if strings.EqualFold(key, name) && len(values) > 0 {
			return decodeScalar(values[0])
		}
	}
	return "", fmt.Errorf("scalar not found")
}

func scalarByKeyContains(raw []byte, keyPart string) (string, error) {
	var result map[string][]json.RawMessage
	if err := json.Unmarshal(raw, &result); err != nil {
		return "", err
	}
	for key, values := range result {
		if strings.Contains(strings.ToLower(key), strings.ToLower(keyPart)) && len(values) > 0 {
			return decodeScalar(values[0])
		}
	}
	return "", fmt.Errorf("matching scalar not found")
}

func stringColumns(raw []byte, names ...string) ([][]string, error) {
	var result map[string][]json.RawMessage
	if err := json.Unmarshal(raw, &result); err != nil {
		return nil, err
	}
	if len(result) == 0 {
		return [][]string{}, nil
	}
	columns := make([][]json.RawMessage, 0, len(names))
	for _, name := range names {
		var found []json.RawMessage
		for key, values := range result {
			if strings.EqualFold(key, name) {
				found = values
				break
			}
		}
		if found == nil {
			return nil, fmt.Errorf("missing column")
		}
		columns = append(columns, found)
	}
	if len(columns) == 0 {
		return nil, nil
	}
	rows := make([][]string, 0, len(columns[0]))
	for i := range columns[0] {
		row := make([]string, 0, len(columns))
		for _, column := range columns {
			if i >= len(column) {
				return nil, fmt.Errorf("column length mismatch")
			}
			value, err := decodeScalar(column[i])
			if err != nil {
				return nil, err
			}
			row = append(row, value)
		}
		rows = append(rows, row)
	}
	return rows, nil
}

func stringColumnsWithTotal(raw []byte, names ...string) ([][]string, int, error) {
	rows, err := stringColumns(raw, names...)
	if err != nil || len(rows) == 0 {
		return rows, 0, err
	}
	totalValue, err := scalarByKey(raw, "total_count")
	if err != nil {
		return nil, 0, err
	}
	total, err := strconv.Atoi(totalValue)
	if err != nil {
		return nil, 0, fmt.Errorf("parse total_count: %w", err)
	}
	if total < len(rows) {
		return nil, 0, fmt.Errorf("total_count %d is smaller than returned row count %d", total, len(rows))
	}
	return rows, total, nil
}

func decodeScalar(raw json.RawMessage) (string, error) {
	var text string
	if err := json.Unmarshal(raw, &text); err == nil {
		decoded, decodeErr := base64.StdEncoding.DecodeString(text)
		if decodeErr == nil && utf8.Valid(decoded) && isPrintable(decoded) {
			return string(decoded), nil
		}
		return text, nil
	}
	var number json.Number
	if err := json.Unmarshal(raw, &number); err == nil {
		return number.String(), nil
	}
	return "", fmt.Errorf("unsupported scalar")
}

func isPrintable(value []byte) bool {
	for _, r := range string(value) {
		if r < 0x20 && r != '\t' && r != '\n' && r != '\r' {
			return false
		}
	}
	return true
}

func parseBool(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "1", "on", "true", "yes":
		return true
	default:
		return false
	}
}

func validIdentifier(value string) bool {
	if value == "" {
		return false
	}
	for _, r := range value {
		if r != '_' && r != '$' && (r < '0' || r > '9') && (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') {
			return false
		}
	}
	return true
}