package e2e

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

	"github.com/pingcap/errors"

	. "github.com/onsi/gomega"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

const (
	GB = 1024 * 1024 * 1024
)

func newProviderType(privoderName string) *cloudv1.ProviderType {
	return &cloudv1.ProviderType{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eEnv.Namespace,
			Name:      privoderName,
			Labels:    e2eResourceLabels(),
		},
	}
}

func newProvider(privoderName string) *cloudv1.Provider {
	return &cloudv1.Provider{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eEnv.Namespace,
			Name:      privoderName,
			Labels:    e2eResourceLabels(),
		},
		Spec: cloudv1.ProviderSpec{
			Type: privoderName,
		},
	}
}

func newCommonNetwork(region, project, provider string) *cloudv1.Network {
	commonLabels := e2eResourceLabels()
	commonLabels[cloudv1.ProjectLabel] = project

	return &cloudv1.Network{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eEnv.Namespace,
			Name:      fmt.Sprintf("e2e-%d", time.Now().UnixNano()),
			Labels:    commonLabels,
		},
		Spec: cloudv1.NetworkSpec{
			CIDR:     "10.241.0.0/18",
			Provider: provider,
			Region:   region,
		},
	}
}

func newCommonCluster(networkName string, instanceType string) *cloudv1.Cluster {
	return &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eEnv.Namespace,
			Name:      fmt.Sprintf("%d", time.Now().UnixNano()),
			Labels:    e2eResourceLabels(),
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: networkName,
			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:     2,
					InstanceType: instanceType,
				},
				PD: cloudv1.PD{
					Replicas:     3,
					InstanceType: instanceType,
					StorageSize:  resource.MustParse("10Gi"),
				},
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: instanceType,
					StorageSize:  resource.MustParse("50Gi"),
				},
				TiFlash: cloudv1.TiFlash{
					Replicas:     1,
					InstanceType: instanceType,
					StorageSize:  resource.MustParse("50Gi"),
				},
			},
		},
	}
}

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

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

func newImportFromS3(clusterName string, format cloudv1.DataFormat, url string, region string) *cloudv1.Import {
	return &cloudv1.Import{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eEnv.Namespace,
			Name:      fmt.Sprintf("e2e-import-%s-%d", clusterName, rand.Int31()),
			Labels:    e2eResourceLabels(),
		},
		Spec: cloudv1.ImportSpec{
			ClusterName: clusterName,
			Source: cloudv1.Source{
				Format:   format,
				UserName: "cloud_admin",
				Storage: cloudv1.Storage{
					S3: &cloudv1.S3{
						URL:    url,
						Region: region,
					},
				},
			},
			StorageRequest: "10Gi",
		},
	}
}

func newTiCDC(networkName, clusterName string) *cloudv1.TiCDC {
	return &cloudv1.TiCDC{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: e2eEnv.Namespace,
			Name:      fmt.Sprintf("e2e-cdc-%d", rand.Int31()),
			Labels:    e2eResourceLabels(),
		},
		Spec: cloudv1.TiCDCSpec{
			NetworkName:  networkName,
			ClusterName:  clusterName,
			Replicas:     1,
			InstanceType: "r5.2xlarge",
			StorageSize:  *resource.NewQuantity(10*GB, resource.BinarySI),
		},
	}
}

func waitNetworkReady(ctx context.Context, network *cloudv1.Network, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{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 waitAWSNetworkPeerReady(ctx context.Context, networkPeer *cloudv1.NetworkPeer, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: networkPeer.Namespace, Name: networkPeer.Name}, networkPeer)
		if err != nil {
			logger.Errorw("error get network peer status", "error", err.Error())
			return err
		}
		if networkPeer.Status.PeeringID == "" {
			logger.Infow("wait network peer ready", "current peeringId", networkPeer.Status.PeeringID)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Network peering be ready")
}

func waitGCPNetworkPeerReady(ctx context.Context, networkPeer *cloudv1.NetworkPeer, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: networkPeer.Namespace, Name: networkPeer.Name}, networkPeer)
		if err != nil {
			logger.Errorw("error get network peer status", "error", err.Error())
			return err
		}

		if networkPeer.Status.Phase == "" {
			logger.Errorw("create network peer error", "current network peer phase", networkPeer.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Network peering be ready")
}

// waitClusterReady wait cluster.Status.Phase become running.
func waitClusterReady(ctx context.Context, cluster *cloudv1.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{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")
}

func waitClusterSynced(ctx context.Context, cluster *cloudv1.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{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() {
			logger.Infow("wait cluster synced", "cluster", cluster.Name, "current phase", cluster.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Cluster should be ready")
}

func waitTiDBReady(ctx context.Context, cluster *cloudv1.Cluster, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		if err != nil {
			logger.Errorw("error inspecting cluster status", "cluster", cluster.Name, "error", err.Error())
			return err
		}
		_, err = net.Dial("tcp", fmt.Sprintf("%s:%d", cluster.Status.SqlEndpoints.PublicEndpoint.Host, cluster.Status.SqlEndpoints.PublicEndpoint.Port))
		if err != nil {
			logger.Infow("wait host available", "host", cluster.Status.SqlEndpoints.PublicEndpoint.Host, "error", err.Error())
			return err
		}

		dsn := DSN(*cluster.Status.SqlEndpoints.PublicEndpoint)
		db, err := sql.Open("mysql", dsn)
		if err != nil {
			logger.Infow("wait sql interface active", "dsn", dsn, "error", err.Error())
			return err
		}
		defer db.Close()
		_, err = db.Query("SHOW DATABASES")
		if err != nil {
			logger.Errorw("error executing sql", "error", err.Error())
		}
		return err
	}, timeout, interval).Should(Succeed(), "SQL endpoint should be ready")
}

func waitBackupReady(ctx context.Context, backup *cloudv1.Backup, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, backup)
		if err != nil {
			logger.Errorw("error inspecting backup status", "error", err.Error())
			return err
		}
		if backup.Status.Phase != cloudv1.BackupComplete {
			logger.Infow("wait backup ready", "current phase", backup.Status.Phase)
			return errWait
		}
		if backup.Status.BackupSize == 0 {
			logger.Errorw("backup size is wrong", "error")
			return err
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Backup should be ready")
}

func waitRestoreReady(ctx context.Context, restore *cloudv1.Restore, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: restore.Namespace, Name: restore.Name}, restore)
		if err != nil {
			logger.Errorw("error inspecting restore status", "error", err.Error())
			return err
		}
		if restore.Status.Phase != cloudv1.RestoreComplete {
			logger.Infow("wait restore ready", "current phase", restore.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Restore should be ready")
}

func waitImportReady(ctx context.Context, imp *cloudv1.Import, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: imp.Namespace, Name: imp.Name}, imp)
		if err != nil {
			logger.Errorw("error inspecting import status", "error", err.Error())
			return err
		}

		if imp.Status.Phase == cloudv1.ImportFailed {
			logger.Errorw("import fail!")
			return errors.New("import fail!")
		}
		if imp.Status.Phase != cloudv1.ImportCompleted {
			logger.Infow("wait import ready", "current phase", imp.Status.Phase)
			return errWait
		}
		return nil
	}, timeout, interval).Should(Succeed(), "Import should be ready")
}

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

func waitImportDeleted(ctx context.Context, cloudImport *cloudv1.Import, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{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, types.NamespacedName{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, types.NamespacedName{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, types.NamespacedName{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")
}

func waitNetworkDeleted(ctx context.Context, network *cloudv1.Network, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: network.Namespace, Name: network.Name}, network)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return nil
			}
			logger.Errorw("error inspecting network status", "network", network.Name, "error", err.Error())
			return err
		}
		logger.Infow("wait network to be cleaned", "network", network.Name)
		return errWait
	}, timeout, interval).Should(Succeed(), "Network should be cleaned up")
}

func waitNetworkPeerDeleted(ctx context.Context, networkPeer *cloudv1.NetworkPeer, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: networkPeer.Namespace, Name: networkPeer.Name}, networkPeer)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return nil
			}
			logger.Errorw("error inspecting delete network peer status", "error", err.Error())
			return err
		}
		logger.Infow("wait network peer to be cleaned")
		return errWait
	}, timeout, interval).Should(Succeed(), "Network peering should be cleaned up")
}

func waitTiCDCDeleted(ctx context.Context, TiCDC *cloudv1.TiCDC, timeout, interval time.Duration) {
	Eventually(func() error {
		err := infraCli.Get(ctx, types.NamespacedName{Namespace: TiCDC.Namespace, Name: TiCDC.Name}, TiCDC)
		if err != nil {
			if apierrors.IsNotFound(err) {
				return nil
			}
			logger.Errorw("error inspecting delete TiCDC status", "error", err.Error())
			return err
		}
		logger.Infow("wait TiCDC to be cleaned", "current phase", TiCDC.Status.Phase)
		return errWait
	}, timeout, interval).Should(Succeed(), "TiCDC 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)
}