package e2e

import (
	"context"
	"database/sql"
	"encoding/json"
	stderr "errors"
	"fmt"
	"math/rand"
	"time"

	"github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const throttledMode = "throttled"

type statusDumpTokens struct {
	Tokens []tokenInfo `json:"tokens"`
}

type tokenInfo struct {
	Token     string   `json:"token"`
	Endpoints []string `json:"endpoints"`
	Replicas  int      `json:"replicas"`
	Comment   string   `json:"comment"`
}

func NewE2EAnno(ls map[string]string, base string) map[string]string {
	if ls == nil {
		ls = map[string]string{}
	}

	ls["testing.tidbcloud.com/for"] = "e2e"
	ls["testing.tidbcloud.com/e2e-id"] = base

	return ls
}

func newCommonCluster() *infra.Cluster {
	return &infra.Cluster{
		ObjectMeta: meta.ObjectMeta{
			Namespace:      namespace,
			Name:           fmt.Sprintf("e2e-%s-%d", kubernetesLabel, rand.Int31()),
			TagsTenant:     "e2e-tenant-id",
			TagsKubernetes: kubernetesLabel,
			Region:         awsRegion,
			Annotations:    NewE2EAnno(nil, kubernetesLabel),
		},
		Spec: infra.ClusterSpec{
			SqlEndpoints: infra.SqlEndpointsSpec{
				Public: &infra.PublicSpec{Enabled: true},
			},
			TiDB: infra.TiDBSpec{
				Replicas: 1,
			},
		},
	}
}

func newCommonVipCluster() *infra.Cluster {
	cluster := newCommonCluster()
	cluster.Spec.TiDB.PoolSelector = fmt.Sprintf("e2etesting-system-%s-vip-tidb-pool", namespacePrefix)
	return cluster
}

//nolint:unparam // xx
func waitClusterReady(ctx context.Context, cluster *infra.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraClusterCli.Get(ctx, cluster)
		if err != nil {
			logger.Errorw("error inspecting cluster status", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		if cluster.Status.Phase != infra.ClusterPhaseRunning {
			logger.Infow("wait cluster ready",
				"cluster", cluster.Name, "current phase", cluster.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Cluster should be ready")
}

// waitClusterSynced wait status synced with spec when you update replicas.
func waitClusterSynced(ctx context.Context, cluster *infra.Cluster, timeout, internal time.Duration) { //nolint:unparam // by design
	time.Sleep(3 * time.Second)
	Eventually(func() error {
		err := infraClusterCli.Get(ctx, cluster)
		if err != nil {
			logger.Errorw("error inspecting cluster", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		if cluster.Status.Phase != infra.ClusterPhaseRunning {
			return errWait
		}
		if cluster.Status.TiDB.UpdatedReplicas != cluster.Spec.TiDB.Replicas {
			return errWait
		}
		if cluster.Spec.TiDB.PoolSelector != "" &&
			cluster.Spec.TiDB.PoolSelector != cluster.Status.TidbPool &&
			len(cluster.Spec.TiDB.ResourceGroups) != len(cluster.Status.ResourceGroups) &&
			cluster.Spec.EnableZoneRedundant != cluster.Status.ZoneRedundantEnabled {
			return errWait
		}
		return nil
	}, timeout, internal).Should(
		Succeed(),
		"Cluster should be synced spec: %v, status: %v", cluster.Spec, cluster.Status,
	)
}

func waitClusterPoolChanged(ctx context.Context, cluster *infra.Cluster, timeout, internal time.Duration) {
	time.Sleep(3 * time.Second)
	Eventually(func() error {
		err := infraClusterCli.Get(ctx, cluster)
		if err != nil {
			logger.Errorw("error inspecting cluster", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		if cluster.Annotations[types.AnnoKeyTidbPoolChanging] != "" {
			return errWait
		}
		if cluster.Spec.TiDB.PoolSelector != "" &&
			cluster.Spec.TiDB.PoolSelector != cluster.Status.TidbPool &&
			len(cluster.Spec.TiDB.ResourceGroups) != len(cluster.Status.ResourceGroups) {
			return errWait
		}
		return nil
	}, timeout, internal).Should(
		Succeed(), "Cluster AnnoKeyTidbPoolChanging should be removed: %v, status: %v", cluster.Annotations, cluster.Status,
	)
}

func endpointConnected(ctx context.Context, keyspaceToken string, endpoint infra.PublicStatus) error {
	dsn := DSN(keyspaceToken, endpoint)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return errors.Annotatef(err, "failed to open db %q", dsn)
	}
	defer db.Close()

	ctx2, cancel := context.WithCancel(ctx)
	defer cancel()
	row := db.QueryRowContext(ctx2, "select 1")
	if row.Err() != nil {
		return errors.Annotatef(row.Err(), "failed to run query")
	}

	return nil
}

func endpointConnectedDB(ctx context.Context, keyspaceToken string, endpoint infra.PublicStatus) (*sql.DB, error) {
	dsn := DSN(keyspaceToken, endpoint)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to open db %q", dsn)
	}

	row := db.QueryRowContext(ctx, "select 1")
	if row.Err() != nil {
		return nil, errors.Annotatef(row.Err(), "failed to run query")
	}

	return db, nil
}

// func endpointConnectedDBWithCloudAdmin(ctx context.Context, keyspaceToken string, endpoint infra.PublicStatus) (*sql.DB, error) {
// 	dsn := CloudAdminDSN(keyspaceToken, endpoint)
// 	db, err := sql.Open("mysql", dsn)
// 	if err != nil {
// 		return nil, errors.Annotatef(err, "failed to open db %q", dsn)
// 	}

// 	row := db.QueryRowContext(ctx, "select 1")
// 	if row.Err() != nil {
// 		return nil, errors.Annotatef(row.Err(), "failed to run query")
// 	}

// 	return db, nil
// }

func waitTiDBReady(ctx context.Context, cluster *infra.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)
		if err != nil {
			logger.Infow(err.Error())
		}
		return err
	}, timeout, interval).Should(Succeed(), "SQL endpoint should be ready")
}

func waitImportDeleted(ctx context.Context, cloudImport *cloudv1.Import, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, client.ObjectKey{Namespace: cloudImport.Namespace, Name: cloudImport.Name}, cloudImport)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return nil
			}
			logger.Errorw("error inspecting import status", "import", cloudImport.Name, "error", err.Error())
			return err
		}
		logger.Infow("wait import to be deleted", "import", cloudImport.Name)
		return errWait
	}, timeout, interval).Should(Succeed(), "Import should be deleted")
}

func waitClusterUpdated(ctx context.Context, mutate func() *infra.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		cluster := mutate()
		if err := infraClusterCli.Update(ctx, cluster); err != nil {
			if stderr.Is(err, infraclient.ErrConflict) {
				_ = infraClusterCli.Get(ctx, cluster)
				return fmt.Errorf("cluster %s: %w", cluster.Name, errWait)
			}
			ginkgo.Fail(fmt.Sprintf("cluster %s non-conflict update error: %v", cluster.Name, err))
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Cluster should be updated")
}

//nolint:unparam // xx
func waitClusterDeleted(ctx context.Context, cluster *infra.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraClusterCli.Get(ctx, cluster)
		if err != nil {
			if stderr.Is(err, infraclient.ErrNotFound) {
				return nil
			}
			logger.Errorw("error inspecting cluster status", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		logger.Infow("wait cluster to be cleaned", "cluster", cluster.Name)
		return errWait
	}, timeout, interval).Should(Succeed(), "Cluster should be cleaned up")
}

func SqlEndpointStatus2PublicStatus(endpoint cloudv1.SqlEndpointStatus) infra.PublicStatus {
	return infra.PublicStatus{
		Host: endpoint.Host,
		Port: uint(endpoint.Port),
	}
}

// DSN returns a DSN string for a TiDB cluster.
func DSN(keyspaceToken string, endpoint infra.PublicStatus) string {
	return fmt.Sprintf("%s.root@tcp(%s:%d)/?tls=preferred", keyspaceToken, endpoint.Host, endpoint.Port)
}

func CloudAdminDSN(keyspaceToken string, endpoint infra.PublicStatus) string {
	return fmt.Sprintf("%s.cloud_admin@tcp(%s:%d)/?tls=preferred", keyspaceToken, endpoint.Host, endpoint.Port)
}

// func requestTidbExit(keyspaceToken, ip string) error {
// 	// request /tidb-pool/exit in specified tidb
// 	kubectlBuilder := NewKubectlCommand(
// 		eksKubeconfig,
// 		eksKubecontext,
// 		managerNs,
// 		"run",
// 		fmt.Sprintf("curl-tidb-exit-%d", rand.Int31()),
// 		"--image=gcr.io/pingcap-public/dbaas/curl:7.74.0", "-i", "--restart=Never", "-q", "--rm", "--",
// 		"curl", "-v", fmt.Sprintf("http://%s:10080/tidb-pool/exit?keyspace=%s", ip, keyspaceToken),
// 	).WithStdinData("")
// 	_, err := kubectlBuilder.Exec()
// 	return err
// }

func checkTokenInfoFromManager(token, pool string, f func(t *tokenInfo) error) error {
	// request /api/tidb/status in specified tidb pool
	kubectlBuilder := NewKubectlCommand(
		eksKubeconfig,
		eksKubecontext,
		managerNs,
		"run",
		fmt.Sprintf("curl-%d", rand.Int31()),
		"--image=gcr.io/pingcap-public/dbaas/curl:7.74.0", "-i", "--restart=Never", "-q", "--rm", "--",
		"curl", "-s", "-k", "https://manager-server:8000/api/tidb/status?ns="+pool,
	).WithStdinData("")
	stdout, _, err := kubectlBuilder.ExecWithFullOutput()
	if err != nil {
		return err
	}
	var managerStatusDumpTokens statusDumpTokens
	if err := json.Unmarshal([]byte(stdout), &managerStatusDumpTokens); err != nil {
		return err
	}
	for _, t := range managerStatusDumpTokens.Tokens {
		if t.Token == token {
			logger.Infof("get tokenInfo: %+v", t)
			return f(&t)
		}
	}

	return errors.Errorf("not found token %v from managerStatusDumpTokens", token)
}

func setResourceGroupByToken(token, user, resourceGroup, pool string) error {
	// request /api/tidb/resource_group
	kubectlBuilder := NewKubectlCommand(
		eksKubeconfig,
		eksKubecontext,
		managerNs,
		"run",
		fmt.Sprintf("curl-%d", rand.Int31()),
		"--image=gcr.io/pingcap-public/dbaas/curl:7.74.0", "-i", "--restart=Never", "-q", "--rm", "--",
		"curl", "-s", "-k",
		fmt.Sprintf("https://manager-server:8000/api/tidb/resource_group?token=%s&user=%s&resource_group=%s&pool=%s",
			token, user, resourceGroup, pool),
	).WithStdinData("")
	stdout, _, err := kubectlBuilder.ExecWithFullOutput()
	if err != nil {
		return err
	}
	if stdout != "" {
		return errors.Errorf("unexpected stdout: %s", stdout)
	}
	return nil
}