package migrationdedicated

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/avast/retry-go"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	"github.com/google/go-github/v63/github"
	"github.com/spf13/viper"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/pkgv2/gormutil"
	"gorm.io/gorm"

	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/account"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi/model"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/openapiv1beta"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/ssh"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/tidbcloudlogin"
	"github.com/tidbcloud/tidb-management-service/utilities/randstr"
)

var (
	cloudRegionCIDR           = "172.16.0.0/19"
	gcpVpcPeeringName         = "gcp-vpc-peering-test"
	gcpPrivateLinkNetworkName = "dedicated-migration-private-link"
	gcpVpcPeeringNetworkName  = "dedicated-migration-vpc-peering"
	gcpPrivateLinkSubnetName  = "dedicated-migration-private-link"
	awsCumstomerAccountID     = "385595570414"
	gcpCustomerProjectID      = "pingcap-gardener"
	awsCustomerVpcID          = "vpc-0f9fbff83daa0895c"
	awsCustomerVpcSubnetIds   = "subnet-059a36ed42fc29507 subnet-0568f949ecd9d238b"
	awsCumstomerVpcCidr       = "10.0.0.0/16"
	gcpCustomerVpcCidr        = "10.0.0.0/24"
)

func TestDedicatedMigration(t *testing.T) {
	testSuite := &TestDedicatedMigrationSuite{
		WebApiLoginContext: &tidbcloudlogin.WebApiLoginContext{
			Host:              viper.GetString("host"),
			Auth0Domain:       viper.GetString("auth0_domain"),
			Auth0ClientID:     viper.GetString("auth0_client_id"),
			Auth0ClientSecret: viper.GetString("auth0_client_secret"),
			UserEmail:         viper.GetString("user_email"),
			BearerToken:       viper.GetString("bearer_token"),
		},

		skipTearDown:      viper.GetBool("skip_tear_down"),
		orgID:             viper.GetString("org_id"),
		projectID:         viper.GetString("project_id"),
		clusterNamePrefix: viper.GetString("cluster_name_prefix"),
		clusterID:         viper.GetString("cluster_id"),

		provider:             viper.GetString("cloud_provider"),
		cmekEnabled:          viper.GetBool("cmek_enabled"),
		addTiflash:           viper.GetBool("add_tiflash"),
		skipMigration:        viper.GetBool("skip_migration"),
		skipAccesspoint:      viper.GetBool("skip_accesspoint"),
		skipPauseClusterCase: viper.GetBool("skip_pause_cluster"),
		customerKMSKey:       viper.GetString("customer_kms_key"),
		password:             randstr.GenWithCharset(16, randstr.AlphabetCharset),

		AwsEc2Host:                 viper.GetString("ec2_host"),
		AwsEc2User:                 viper.GetString("ec2_user"),
		AwsEc2Key:                  viper.GetString("ec2_key"),
		GcpVpcPeeringVmHost:        viper.GetString("gcp_vpc_peering_vm_host"),
		GcpPrivateLinkVmHost:       viper.GetString("gcp_private_link_vm_host"),
		GcpVmUser:                  viper.GetString("gcp_vm_user"),
		GcpVmKey:                   viper.GetString("gcp_vm_key"),
		gcpPrivateLinkEndpointName: fmt.Sprintf("endpoint%s", strconv.FormatInt(time.Now().Unix(), 10)),
		s3AK:                       viper.GetString("s3_ak"),
		s3SK:                       viper.GetString("s3_sk"),
		githubToken:                viper.GetString("gh_token"),
		changefeedtKafkaEndpoint:   viper.GetString("changefeed_kafka_endpoint"),
		changefeedMysqlEndpoint:    viper.GetString("changefeed_mysql_endpoint"),
		changefeedMysqlUsername:    viper.GetString("changefeed_mysql_username"),
		changefeedMysqlPassword:    viper.GetString("changefeed_mysql_password"),
	}

	testSuite.clusterName = testSuite.clusterNamePrefix + "default-keep-48h"
	testSuite.tiflashClusterName = testSuite.clusterNamePrefix + "tiflash-keep-48h"
	testSuite.restoreClusterNameBeforeMig = testSuite.clusterNamePrefix + "restore-before-mig-keep-48h"
	testSuite.restoreClusterNameAfterMig = testSuite.clusterNamePrefix + "restore-after-mig-keep-48h"
	testSuite.pausedClusterNameBeforeMig = testSuite.clusterNamePrefix + "paused-before-mig-keep-48h"
	testSuite.accesspointClusterName = testSuite.clusterNamePrefix + "accesspoint-keep-48h"
	testSuite.accesspointClusterPassword = randstr.GenWithCharset(16, randstr.AlphabetCharset)

	switch {
	case strings.Contains(testSuite.Host, "dev") || strings.Contains(testSuite.Host, "nightly"):
		testSuite.env = "dev"
	case strings.Contains(testSuite.Host, "staging"):
		testSuite.env = "staging"
	case strings.Contains(testSuite.Host, "prod"):
		testSuite.env = "prod"
	}

	switch testSuite.provider {
	case "aws":
		testSuite.region = "us-west-2"
	case "gcp":
		testSuite.region = "us-west1"
	}
	// testSuite.o11yDSHost = fmt.Sprintf("https://www.ds.%s.%s.observability.tidbcloud.com", testSuite.region, testSuite.provider)

	suite.Run(t, testSuite)
}

type TestDedicatedMigrationSuite struct {
	suite.Suite

	*tidbcloudlogin.WebApiLoginContext

	skipTearDown bool

	orgID             string
	projectID         string
	clusterNamePrefix string

	portalCli   *coreportalapi.Cli
	openApiCli  *openapiv1beta.Cli
	accountCli  *account.Cli
	ec2Cli      *ec2.Client
	dataflowCli *dataflow.Cli
	ghCli       *github.Client

	provider    string
	env         string
	region      string
	password    string
	clusterName string
	clusterID   string

	cmekEnabled          bool
	addTiflash           bool
	skipMigration        bool
	skipAccesspoint      bool
	skipPauseClusterCase bool
	customerKMSKey       string
	dedicatedGlobalDB    *gorm.DB

	publicEndpointHost                 string
	vpcPeeringEndpointHost             string
	privateLinkEndpointHost            string
	accesspointPublicEndpointHost      string
	accesspointVpcPeeringEndpointHost  string
	accesspointPrivateLinkEndpointHost string
	accesspointClusterPassword         string

	publicEndpointWorkloadPID                 string
	vpcPeeringEndpointWorkloadPID             string
	privateLinkEndpointWorkloadPID            string
	accesspointPublicEndpointWorkloadPID      string
	accesspointVpcPeeringEndpointWorkloadPID  string
	accesspointPrivateLinkEndpointWorkloadPID string

	AwsEc2Host           string
	AwsEc2User           string
	AwsEc2Key            string
	GcpVpcPeeringVmHost  string
	GcpPrivateLinkVmHost string
	GcpVmUser            string
	GcpVmKey             string

	s3AK string
	s3SK string

	privateLinkEndpointID            string
	accesspointPrivateLinkEndpointID string
	awsVpcEndpointName               string
	accesspointAwsVpcEndpointName    string
	gcpVpcEndpointName               string
	gcpPrivateLinkEndpointName       string
	dbaasAwsVpcPeeringID             string
	dbaasGcpVpcPeeringID             string

	bakIDBeforeMig             string
	bakIDAfterMig              string
	connectIDBeforeMig         string
	podStartTimestampBeforeMig map[string]uint64
	githubToken                string

	tiflashClusterName          string
	tiflashClusterID            string
	restoreClusterNameBeforeMig string
	restoreClusterIDBeforeMig   string
	restoreClusterNameAfterMig  string
	restoreClusterIDAfterMig    string
	pausedClusterNameBeforeMig  string
	pausedClusterIDBeforeMig    string
	accesspointID               string
	// newAccesspointID            string
	accesspointClusterName string
	accesspointClusterID   string

	migratedClusterMetering schema.ClusterMetering

	changefeedtKafkaEndpoint string
	changefeedMysqlEndpoint  string
	changefeedMysqlUsername  string
	changefeedMysqlPassword  string

	clusterBeforeMigPaused bool

	// manual test only
	regionCidrSet bool
	skipO11yCase  bool
}

func (su *TestDedicatedMigrationSuite) TearDownSuite() {
	if su.T().Failed() {
		return
	}

	if su.skipTearDown {
		return
	}

	if su.bakIDBeforeMig != "" {
		su.dataflowCli.DeleteBackup(su.orgID, su.projectID, su.clusterID, su.bakIDBeforeMig)
	}
	if su.bakIDAfterMig != "" {
		su.dataflowCli.DeleteBackup(su.orgID, su.projectID, su.clusterID, su.bakIDAfterMig)
	}
	if su.awsVpcEndpointName != "" {
		su.deleteAwsVpcEndpoint(su.awsVpcEndpointName)
	}
	if su.accesspointAwsVpcEndpointName != "" {
		su.deleteAwsVpcEndpoint(su.accesspointAwsVpcEndpointName)
	}

	if su.gcpVpcEndpointName != "" {
		deleteVpcEndpointCmd := fmt.Sprintf(
			"gcloud compute forwarding-rules delete %s --region=%s --project=%s --quiet\n\n"+
				"gcloud compute addresses delete %s --region=%s --project=%s --quiet\n",
			su.gcpPrivateLinkEndpointName,
			su.region,
			gcpCustomerProjectID,
			su.gcpPrivateLinkEndpointName,
			su.region,
			gcpCustomerProjectID,
		)
		su.T().Logf("deleteVpcEndpointCmd: %s", deleteVpcEndpointCmd)
		var (
			deleteVpcEndpointCmdOutput string
			err                        error
		)
		retryErr := retry.Do(
			func() error {
				deleteVpcEndpointCmdOutput, err = su.runCommandOnGcpPrivateLinkVm(deleteVpcEndpointCmd)
				su.T().Logf("deleteVpcEndpointCmdOutput: %s, err: %v", deleteVpcEndpointCmdOutput, err)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
		su.Require().NoError(retryErr)
	}

	if su.dbaasAwsVpcPeeringID != "" {
		su.portalCli.DeleteAwsVpcPeering(su.orgID, su.projectID, su.dbaasAwsVpcPeeringID)

		su.deleteAwsVpcPeeringRoute()
	}
	if su.dbaasGcpVpcPeeringID != "" {
		su.T().Logf("Delete GCP VPC peering: %s", su.dbaasGcpVpcPeeringID)
		su.portalCli.DeleteGcpVpcPeering(su.orgID, su.projectID, su.dbaasGcpVpcPeeringID)

		deleteVpcPeeringCmd := fmt.Sprintf(
			"gcloud compute networks peerings delete %s --network=dedicated-migration-vpc-peering",
			gcpVpcPeeringName,
		)
		su.T().Logf("deleteVpcPeeringCmd: %s", deleteVpcPeeringCmd)
		var deleteVpcPeeringCmdOutput string
		_ = retry.Do(
			func() error {
				var err error
				deleteVpcPeeringCmdOutput, err = su.runCommandOnGcpVpcPeeringVm(deleteVpcPeeringCmd)
				su.T().Logf("deleteVpcPeeringCmdOutput: %s, err: %v", deleteVpcPeeringCmdOutput, err)
				return err
			},
			retry.Attempts(3),
			retry.DelayType(retry.FixedDelay),
			retry.Delay(5*time.Second),
		)
	}

	if su.clusterID != "" {
		su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.clusterName)
	}

	// if su.projectID != "" {
	//	su.accountCli.DeleteProject(su.orgID, su.projectID)
	// }
}

func (su *TestDedicatedMigrationSuite) SetupSuite() {
	if su.provider == "gcp" {
		su.Require().Falsef(su.cmekEnabled, "CMEK is not supported on GCP")
	}

	token, err := su.LoginAndGetToken(su.orgID)
	su.Require().NoError(err, "LoginAndGetToken failed")

	su.portalCli, err = coreportalapi.NewClient(&su.Suite, su.Host, token)
	su.Require().NoError(err)
	su.Require().NotNil(su.portalCli)

	su.accountCli, err = account.NewClient(&su.Suite, su.Host, token)
	su.Require().NoError(err)
	su.Require().NotNil(su.accountCli)

	su.dataflowCli, err = dataflow.NewClient(&su.Suite, su.Host, token)
	su.Require().NoError(err)
	su.Require().NotNil(su.dataflowCli)

	apikey := viper.GetString("open_api_key")
	keyPair := strings.Split(apikey, ":")
	su.openApiCli, err = openapiv1beta.NewClient(&su.Suite, viper.GetString("open_api_host"), keyPair[0], keyPair[1])
	su.Require().NoError(err)
	su.Require().NotNil(su.openApiCli)

	if su.provider == "aws" {
		awsCfg, err := config.LoadDefaultConfig(context.Background())
		su.Require().NoError(err, "load default aws config failed")
		su.ec2Cli = ec2.NewFromConfig(awsCfg)
	}

	if pwd := viper.GetString("db_password"); pwd != "" {
		su.dedicatedGlobalDB, err = gormutil.Connect(gormutil.Config{
			User:     viper.GetString("db_user"),
			Host:     viper.GetString("db_host"),
			Database: viper.GetString("db_database"),
			Password: viper.GetString("db_password"),
			Port:     3306,
		})
		su.Require().NoError(err)
		su.Require().NotNil(su.dedicatedGlobalDB)
	}

	if su.githubToken != "" {
		su.ghCli = github.NewClient(nil).WithAuthToken(su.githubToken)
	}

	// manual test only
	// su.projectID = "1372813089454556169"
	// su.regionCidrSet = true
	// su.clusterID = "1379661944646415686"
	// su.pausedClusterIDBeforeMig = "1379661944646415687"
	// su.accesspointClusterID = "1379661944646415688"
	// // su.clusterPaused = true
	// su.accesspointID = "1826234864769175552"
	// su.privateLinkEndpointID = "362146"
	// su.awsVpcEndpointName = "vpce-044f73b224b9dee25"
	// // su.gcpVpcEndpointName = "projects/tidbcloud-prod-000/regions/us-west1/serviceAttachments/tidb1379661944646415651"
	// su.dbaasAwsVpcPeeringID = "1200267"
	// // su.dbaasGcpVpcPeeringID = "1200104"
	// su.bakIDBeforeMig = "2862695"
	// su.accesspointPrivateLinkEndpointID = "362147"
	// su.accesspointAwsVpcEndpointName = "vpce-0b499c21266c59d28"
	// su.bakIDAfterMig = "2857801"
	// su.restoreClusterIDBeforeMig = "10224926992459212628"
	// su.restoreClusterNameBeforeMig = "migration-e2e-staging-bDEC"
	// su.restoreClusterIDAfterMig = "10245384390101209715"
	// su.restoreClusterNameAfterMig = "migration-e2e-staging-sqAa"
	// su.newAccesspointID = "1821388157011099648"
	// su.skipO11yCase = true
}

func (su *TestDedicatedMigrationSuite) runCommandOnAwsEc2(cmd string) (string, error) {
	su.T().Logf("Running command on AWS EC2: %s", cmd)
	return ssh.RunCommandViaSSH(su.AwsEc2Host, su.AwsEc2User, su.AwsEc2Key, cmd)
}

func (su *TestDedicatedMigrationSuite) runCommandOnGcpVpcPeeringVm(cmd string) (string, error) {
	su.T().Logf("Running command on GCP Vpc Peering VM: %s", cmd)
	return ssh.RunCommandViaSSH(su.GcpVpcPeeringVmHost, su.GcpVmUser, su.GcpVmKey, cmd)
}

func (su *TestDedicatedMigrationSuite) runCommandOnGcpPrivateLinkVm(cmd string) (string, error) {
	su.T().Logf("Running command on GCP Private Link VM: %s", cmd)
	return ssh.RunCommandViaSSH(su.GcpPrivateLinkVmHost, su.GcpVmUser, su.GcpVmKey, cmd)
}

func (su *TestDedicatedMigrationSuite) waitForClusterStatusNormal(
	expectedStatus model.CentralClusterStatus,
	clusterID, clusterName string,
) {
	for {
		status := su.portalCli.PollClusterStatus(su.orgID, su.projectID, clusterID)
		if *status == expectedStatus {
			break
		}
		su.T().Logf("cluster status: %s, clusterName: %s, clusterID: %s", *status, clusterName, clusterID)
		time.Sleep(10 * time.Second)
	}
}

func (su *TestDedicatedMigrationSuite) deleteAwsVpcEndpoint(awsVpcEndpointName string) {
	deleteVpcEndpointCmd := fmt.Sprintf(
		"aws ec2 delete-vpc-endpoints --vpc-endpoint-ids %s --region %s",
		awsVpcEndpointName,
		su.region,
	)
	su.T().Logf("deleteVpcEndpointCmd: %s", deleteVpcEndpointCmd)
	var (
		deleteVpcEndpointCmdOutput string
		err                        error
	)
	retryErr := retry.Do(
		func() error {
			deleteVpcEndpointCmdOutput, err = su.runCommandOnAwsEc2(deleteVpcEndpointCmd)
			su.T().Logf("deleteVpcEndpointCmdOutput: %s, err: %v", deleteVpcEndpointCmdOutput, err)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(deleteVpcEndpointCmdOutput)
}

func (su *TestDedicatedMigrationSuite) checkO11yData(ctx context.Context, clusterID string, start, end int) {
	owner := "tidbcloud"
	repo := "observability"
	workflowFileName := "datacheck.yaml"
	event := github.CreateWorkflowDispatchEventRequest{
		Ref: "main",
		Inputs: map[string]any{
			"git-ref":  "main",
			"provider": su.provider,
			"env":      su.env,
			"region":   su.region,
			"tenant":   su.orgID,
			"project":  su.projectID,
			"cluster":  clusterID,
			"start":    strconv.Itoa(start),
			"end":      strconv.Itoa(end),
		},
	}
	resp, err := su.ghCli.Actions.CreateWorkflowDispatchEventByFileName(ctx, owner, repo, workflowFileName, event)
	su.Require().NoError(err)
	su.Require().NotNil(resp)

	time.Sleep(2 * time.Minute)

	err = su.waitLatestRunCompletedByFileName(ctx, repo, workflowFileName)
	su.Require().NoError(err)
}