package e2e

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

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/labels"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var errWait = fmt.Errorf("wait for condition met")

// 1. create old cluster & new cluster
// 2. import data to old cluster
var _ = Describe("full link test", func() {
	interval := 10 * time.Second
	ctx := context.Background()

	It("Should take import for aws shared provider successfully", func() {
		clusters := make([]*infra.Cluster, 0, 2)
		By("create cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		clusters = append(clusters, cluster)

		By("create vip cluster")
		vipCluster := newCommonVipCluster()
		Expect(infraClusterCli.Create(ctx, vipCluster)).Should(Succeed())
		clusters = append(clusters, vipCluster)

		for _, cluster := range clusters {
			By("wait cluster ready")
			waitClusterReady(ctx, cluster, 5*time.Minute, interval)
			Expect(cluster.Status.SqlEndpoints.Public).NotTo(BeNil())

			By("create import")
			cloudImport := &cloudv1.Import{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: namespace,
					Name:      fmt.Sprintf("import-%v", rand.Int63()),
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel:        types.ProviderTypeAWSFreeTier,
						cloudv1.RegionLabel:              awsRegion,
						labels.LabelKeyKubernetesCluster: cluster.TagsKubernetes,
					},
					Annotations: map[string]string{
						cloudv1.ImportAnnotationKeyNetworkName:     "",
						"tags.tidbcloud.com/migrated-from-central": "true",
					},
				},
				Spec: cloudv1.ImportSpec{
					ClusterName:    cluster.Name,
					StorageRequest: "200Mi",
					Source: cloudv1.Source{
						Format:   cloudv1.DataFormatSQL,
						UserName: "cloud_admin",
						Storage: cloudv1.Storage{
							S3: &cloudv1.S3{
								Region: "us-west-2",
								URL:    "s3://dbaas-import-e2e-test/files/e2e/import/simple_dump/",
							},
						},
					},

					//nolint: lll // no need
					LightningConfig: `
					[checkpoint]
					driver = "file"
					dsn = "/etc/endpoint/cp.pb"
					enable = true
					
					[cron]
					log-progress = "5m"
					switch-mode = "5m"
					
					[lightning]
					check-requirements = false
					index-concurrency = 1
					level = "info"
					pprof-port = 8289
					table-concurrency = 1
					
					[mydumper]
					data-source-dir = "s3://dbaas-import-e2e-test/files/e2e/import/simple_dump/?region=us-west-2&role-arn=arn:aws:iam::385595570414:role/dbaas-import-e2e-test&external-id=696e6672612d6170696cd2d628c999b4b6a51db7e6b53a7cc1"
					no-schema = false
					strict-format = false

					[security]
					ca-path = "/var/lib/lightning-tls/ca.crt"
					cert-path = "/var/lib/lightning-tls/tls.crt"
					key-path = "/var/lib/lightning-tls/tls.key"
					
					[tikv-importer]
					backend = "tidb"
					sorted-kv-dir = "/etc/endpoint"
					`,
				},
			}
			Expect(infraCli.Create(ctx, cloudImport)).Should(Succeed())

			By("wait import complete")
			Eventually(func() error {
				err := infraCli.Get(ctx, client.ObjectKey{Namespace: cloudImport.Namespace, Name: cloudImport.Name}, cloudImport)
				if err != nil {
					logger.Errorw("error inspecting import status", "import", cloudImport.Name, "error", err.Error())
					return err
				}
				logger.Infow("dump import status", "status", cloudImport.Status)
				if cloudImport.Status.Phase != cloudv1.ImportCompleted {
					logger.Infow("wait import complete", "import", cloudImport.Name, "current phase", cloudImport.Status.Phase)
					return errWait
				}
				return nil
			}, 15*time.Minute, interval).Should(Succeed(), "Import should complete")
			Expect(cloudImport.Status.CompletedTables).Should(Equal(uint32(2)))
			// TODO inspect why sometimes failed:
			// Expected
			//     <uint32>: 2
			// to equal
			//     <uint32>: 0
			//
			// Expect(cloudImport.Status.PendingTables).Should(Equal(uint32(0)))
			Expect(cloudImport.Status.TotalSize > uint64(0)).To(BeTrue())
			Expect(cloudImport.Status.CurrentTables).Should(BeEmpty())

			By("wait TiDB ready") // should wait TiDB to be ready in order to execute SQL statements
			waitTiDBReady(ctx, cluster, 10*time.Minute, interval)

			By("Verify data in cluster for import")
			verifyImportSampleData(ctx, cluster)

			By("delete import")
			Expect(infraCli.Delete(ctx, cloudImport)).Should(Succeed())

			By("wait import to be deleted")
			waitImportDeleted(ctx, cloudImport, 10*time.Minute, interval)

			By("delete clusters")
			Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed())

			By("wait cluster to be deleted")
			waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
		}
	})
})

// verifyImportSampleData verify import sample data for a TidbCluster.
func verifyImportSampleData(ctx context.Context, cluster *infra.Cluster) {
	dsn := DSN(cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)
	db, err := sql.Open("mysql", dsn)
	Expect(err).ShouldNot(HaveOccurred())
	row := db.QueryRowContext(ctx, "SELECT count(*) FROM `test2`.`test1`")
	var got int
	Expect(row.Scan(&got)).Should(Succeed())
	Expect(got).Should(Equal(3))

	row = db.QueryRowContext(ctx, "SELECT count(*) FROM `test2`.`test2`")
	Expect(row.Scan(&got)).Should(Succeed())
	Expect(got).Should(Equal(3))
	Expect(db.Close()).Should(Succeed())
}