package e2e

import (
	"context"
	"database/sql"
	"fmt"
	"math/rand"
	"net"
	"time"

	. "github.com/onsi/gomega"
	"github.com/pingcap/errors"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/shared-provider/pkg/types"
	"github.com/tidbcloud/shared-provider/pkg/utils/labels"
)

func newCommonAWSNetwork(region string) *cloudv1.Network {
	projectID := fmt.Sprintf("e2e-%d", rand.Int31n(999999))
	return &cloudv1.Network{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      fmt.Sprintf("e2e-%s", cloudv1.RandomString(6)),
			Labels:    e2eResourceLabels(projectID),
		},
		Spec: cloudv1.NetworkSpec{
			CIDR:     "10.241.0.0/18",
			Provider: types.ProviderTypeAWSSharedTier,
			Region:   region,
		},
	}
}

func newCommonCluster(network *cloudv1.Network) *cloudv1.Cluster {
	instanceType := "aws_sharedtier_6"
	return &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      fmt.Sprintf("e2e-%s-%d", kubernetesLabel, rand.Int31()),
			Labels:    e2eResourceLabels(labels.GetProjectFromLabels(network.Labels)),
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: network.Name,
			SqlEndpoints: cloudv1.SqlEndpoints{
				Public: cloudv1.Service{
					Enabled: pointer.Bool(true),
					// open for all access
					LoadBalancerSourceRanges: []string{
						"0.0.0.0/0",
					},
				},
			},
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     1,
					InstanceType: instanceType,
				},
				PD: cloudv1.PD{
					Replicas:     1,
					InstanceType: instanceType,
				},
				TiKV: cloudv1.TiKV{
					Replicas:     1,
					InstanceType: instanceType,
				},
				TiFlash: cloudv1.TiFlash{
					Replicas:     1,
					InstanceType: instanceType,
				},
			},
		},
	}
}

func newCommonBackup(cluster *cloudv1.Cluster) *cloudv1.Backup {
	return &cloudv1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      fmt.Sprintf("e2e-backup-%s-%d", cluster.Name, rand.Int31()),
			Labels:    e2eResourceLabels(labels.GetProjectFromLabels(cluster.Labels)),
		},
		Spec: cloudv1.BackupSpec{
			ClusterName: cluster.Name,
		},
	}
}

func newCommonRestore(cluster *cloudv1.Cluster, backupName string) *cloudv1.Restore {
	return &cloudv1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      fmt.Sprintf("e2e-restore-%d", rand.Int31()),
			Labels:    e2eResourceLabels(labels.GetProjectFromLabels(cluster.Labels)),
		},
		Spec: cloudv1.RestoreSpec{
			ClusterName: cluster.Name,
			From: cloudv1.From{
				Backup: backupName,
			},
		},
	}
}

func waitNetworkReady(ctx context.Context, network *cloudv1.Network, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, client.ObjectKey{Namespace: network.Namespace, Name: network.Name}, network)
		if err != nil {
			logger.Errorw("error inspecting network status", "network", network.Name, "error", err.Error())
			return err
		}
		if network.Status.Phase != cloudv1.NetworkPhaseNormal {
			logger.Infow("wait network ready", "network", network.Name, "current phase", network.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Network should be ready")
}

func waitClusterReady(ctx context.Context, cluster *cloudv1.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		if err != nil {
			logger.Errorw("error inspecting cluster status", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		if cluster.Status.Phase != cloudv1.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 *cloudv1.Cluster, timeout, internal time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		if err != nil {
			logger.Errorw("error inspecting cluster status", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		if !cluster.Synced() {
			return errWait
		}
		return nil
	}, timeout, internal).Should(
		Succeed(),
		"Cluster should be synced spec: %v, status: %v", cluster.Spec.Cluster, cluster.Status.DB,
	)
}

func endpointConnected(ctx context.Context, endpoint cloudv1.SqlEndpointStatus) error {
	_, err := net.Dial("tcp", fmt.Sprintf("%s:%d", endpoint.Host, endpoint.Port))
	if err != nil {
		return errors.Annotatef(err, "host %q not dialable", endpoint.Host)
	}

	dsn := DSN(endpoint)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return errors.Annotatef(err, "failed to open db %q", dsn)
	}
	defer db.Close()

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

	return nil
}

func waitTiDBNotReady(ctx context.Context, endpoint cloudv1.SqlEndpointStatus, timeout, interval time.Duration) {
	logger.Infof("waiting endpoint %v not connected", endpoint.Host)
	Eventually(func() error {
		err := endpointConnected(ctx, endpoint)
		if err != nil {
			return nil
		}
		return errWait
	}, timeout, interval).Should(Succeed(), "SQL endpoint should be ready")
}

func waitTiDBReady(ctx context.Context, endpoint cloudv1.SqlEndpointStatus, timeout, interval time.Duration) {
	Eventually(func() error {
		err := endpointConnected(ctx, endpoint)
		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 waitBackupDeleted(ctx context.Context, backup *cloudv1.Backup, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, client.ObjectKey{Namespace: backup.Namespace, Name: backup.Name}, backup)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return nil
			}
			logger.Errorw("error inspecting backup status", "backup", backup.Name, "error", err.Error())
			return err
		}
		logger.Infow("wait backup to be deleted", "backup", backup.Name)
		return errWait
	}, timeout, interval).Should(Succeed(), "Backup should be deleted")
}

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

func waitClusterDeleted(ctx context.Context, cluster *cloudv1.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		if err != nil {
			if apierrors.IsNotFound(err) {
				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")
}

// DSN returns a DSN string for a TiDB cluster.
func DSN(endpoint cloudv1.SqlEndpointStatus) string {
	return fmt.Sprintf("root@tcp(%s:%d)/", endpoint.Host, endpoint.Port)
}

// DSNCloudAdmin returns a DNS string a TiDB cluster for cloud admin.
// `password` can be empty.
func DSNCloudAdmin(endpoint cloudv1.SqlEndpointStatus, password string) string {
	return fmt.Sprintf("cloud_admin:%s@tcp(%s:%d)/", password, endpoint.Host, endpoint.Port)
}