package storage

import (
	"encoding/json"
	"fmt"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/testcontainers/testcontainers-go"
	mysqlcontainer "github.com/testcontainers/testcontainers-go/modules/mysql"
	inframysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/mysql"
	gormmysql "gorm.io/driver/mysql"
	"gorm.io/gorm"
)

func TestLatestClusterEventPageQueryMySQLPlan(t *testing.T) {
	testcontainers.SkipIfProviderIsNotHealthy(t)

	ctx := t.Context()
	mysqlContainer, err := mysqlcontainer.Run(
		ctx,
		"mysql:8.0.36",
		mysqlcontainer.WithDatabase("metaserver"),
		mysqlcontainer.WithUsername("root"),
		mysqlcontainer.WithPassword("password"),
	)
	require.NoError(t, err)
	t.Cleanup(func() {
		require.NoError(t, testcontainers.TerminateContainer(mysqlContainer))
	})

	dsn, err := mysqlContainer.ConnectionString(ctx, "parseTime=true")
	require.NoError(t, err)
	db, err := gorm.Open(gormmysql.Open(dsn), &gorm.Config{})
	require.NoError(t, err)
	require.NoError(t, db.AutoMigrate(&inframysql.ClusterEvent{}))

	dbEvents := make([]inframysql.ClusterEvent, 0, 1020)
	for range 1000 {
		dbEvents = append(dbEvents, inframysql.ClusterEvent{
			Name:       "cluster-hot",
			Kubernetes: "kube-a",
		})
	}
	for i := range 20 {
		dbEvents = append(dbEvents, inframysql.ClusterEvent{
			Name:       fmt.Sprintf("cluster-%03d", i),
			Kubernetes: "kube-a",
		})
	}
	require.NoError(t, db.CreateInBatches(&dbEvents, 500).Error)
	require.NoError(t, db.Exec("ANALYZE TABLE cluster_events").Error)

	var events []ClusterEvent
	query := latestClusterEventPageQuery(
		db.Session(&gorm.Session{DryRun: true}),
		"kube-a",
		"cluster-005",
	).Find(&events)
	require.NoError(t, query.Error)
	require.Contains(t, query.Statement.SQL.String(), latestClusterEventTableWithIndexHint)

	var plan string
	require.NoError(t, db.Raw(
		"EXPLAIN FORMAT=JSON "+query.Statement.SQL.String(),
		query.Statement.Vars...,
	).Scan(&plan).Error)
	requireMySQLPlanAccessPath(
		t,
		plan,
		inframysql.ClusterEventACKIndex,
		"range",
	)
}

func requireMySQLPlanAccessPath(t *testing.T, plan, index, accessType string) {
	t.Helper()

	var root any
	require.NoError(t, json.Unmarshal([]byte(plan), &root))
	require.Truef(
		t,
		mysqlPlanHasAccessPath(root, index, accessType),
		"MySQL plan does not use index %q with access type %q:\n%s",
		index,
		accessType,
		plan,
	)
}

func mysqlPlanHasAccessPath(node any, index, accessType string) bool {
	switch value := node.(type) {
	case map[string]any:
		if value["key"] == index && value["access_type"] == accessType {
			return true
		}
		for _, child := range value {
			if mysqlPlanHasAccessPath(child, index, accessType) {
				return true
			}
		}
	case []any:
		for _, child := range value {
			if mysqlPlanHasAccessPath(child, index, accessType) {
				return true
			}
		}
	}
	return false
}