package migrationdedicated

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/avast/retry-go"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	openapiruntimeclient "github.com/go-openapi/runtime/client"
	"github.com/samber/lo"

	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/coreportalapi/model"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/client/backup_service"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/client/changefeed_instance_type_service"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/client/connector_service"
	dataflowmodel "github.com/tidbcloud/tidb-management-service/functional-test/pkg/dataflow/model"
	_ "github.com/tidbcloud/tidb-management-service/functional-test/pkg/openapiv1beta/client/backup"
	"github.com/tidbcloud/tidb-management-service/functional-test/pkg/openapiv1beta/client/cluster"
)

// Test_201_XXX to Test_300_XXX are run before migration

type TimeStampRecord struct {
	ID uint64    `gorm:"primarykey"`
	TS time.Time `gorm:"column:ts"`
}

func (su *TestDedicatedMigrationSuite) Test_201_EnsureConnectionsNotInterrupted() {
	if len(su.privateLinkEndpointWorkloadPID)+
		len(su.vpcPeeringEndpointWorkloadPID)+
		len(su.publicEndpointWorkloadPID) == 0 {
		return
	}
	db := su.portalCli.ConnectDB(su.orgID, su.projectID, su.clusterID, "root", su.password, "test", "skip-verify")
	// for privatelink records
	if su.privateLinkEndpointWorkloadPID != "" {
		var records []TimeStampRecord
		err := db.Raw("select * from test.privatelink order by id desc limit 1").Find(&records).Error
		su.Require().NoError(err, "query test.privatelink failed")
		su.Require().NotEmptyf(records, "table test.privatelink is empty")
		currUnixSec := time.Now().Unix()
		lastUnixSec := records[0].TS.Unix()
		delta := currUnixSec - lastUnixSec
		su.Require().
			Lessf(delta, int64(20), "last ts(%v) in test.privatelink does not catch up current ts(%v)", lastUnixSec, currUnixSec)
	}
	// for vpcpeering records
	if su.vpcPeeringEndpointWorkloadPID != "" {
		var records []TimeStampRecord
		err := db.Raw("select * from test.vpcpeering order by id desc limit 1").Find(&records).Error
		su.Require().NoError(err, "query test.vpcpeering failed")
		su.Require().NotEmptyf(records, "table test.vpcpeering is empty")
		currUnixSec := time.Now().Unix()
		lastUnixSec := records[0].TS.Unix()
		delta := currUnixSec - lastUnixSec
		su.Require().
			Lessf(delta, int64(20), "last ts(%v) in test.vpcpeering does not catch up current ts(%v)", lastUnixSec, currUnixSec)
	}
	// for publicendpoint records
	if su.publicEndpointWorkloadPID != "" {
		var records []TimeStampRecord
		err := db.Raw("select * from test.publicendpoint order by id desc limit 1").Find(&records).Error
		su.Require().NoError(err, "query test.publicendpoint failed")
		su.Require().NotEmptyf(records, "table test.publicendpoint is empty")
		currUnixSec := time.Now().Unix()
		lastUnixSec := records[0].TS.Unix()
		delta := currUnixSec - lastUnixSec
		su.Require().
			Lessf(delta, int64(20), "last ts(%v) in test.publicendpoint does not catch up current ts(%v)", lastUnixSec, currUnixSec)
	}
}

func (su *TestDedicatedMigrationSuite) Test_202_EnsurePodNotRebuild() {
	if len(su.podStartTimestampBeforeMig) == 0 {
		return
	}
	out, err := su.getProcessStartTimeMillSeconds()
	su.Require().NoError(err)
	for k, afterMigTs := range out {
		beforeMigTs := su.podStartTimestampBeforeMig[k]
		su.Require().
			Equalf(beforeMigTs, afterMigTs, "pod %s for cluster %s has been rebuild; uptime before mig is %v, uptime after mig is %v", k, su.clusterID, beforeMigTs, afterMigTs)
	}
}

func (su *TestDedicatedMigrationSuite) Test_203_EnsureAccesspointNotInterrupted() {
	if len(su.accesspointPrivateLinkEndpointWorkloadPID)+
		len(su.accesspointVpcPeeringEndpointWorkloadPID)+
		len(su.accesspointPublicEndpointWorkloadPID) == 0 {
		return
	}
	db := su.portalCli.ConnectDatabase(
		su.accesspointPublicEndpointHost,
		"4000",
		"root",
		su.accesspointClusterPassword,
		"test",
		"skip-verify",
	)
	if su.accesspointPrivateLinkEndpointWorkloadPID != "" {
		var records []TimeStampRecord
		err := db.Raw("select * from test.ap_privatelink order by id desc limit 1").Find(&records).Error
		su.Require().NoError(err, "query test.ap_privatelink failed")
		su.Require().NotEmptyf(records, "table test.ap_privatelink is empty")
		currUnixSec := time.Now().Unix()
		lastUnixSec := records[0].TS.Unix()
		delta := currUnixSec - lastUnixSec
		su.Require().
			Lessf(delta, int64(20), "last ts(%v) in test.ap_privatelink does not catch up current ts(%v)", lastUnixSec, currUnixSec)
	}
	if su.accesspointVpcPeeringEndpointWorkloadPID != "" {
		var records []TimeStampRecord
		err := db.Raw("select * from test.ap_vpcpeering order by id desc limit 1").Find(&records).Error
		su.Require().NoError(err, "query test.ap_vpcpeering failed")
		su.Require().NotEmptyf(records, "table test.vpcpeering is empty")
		currUnixSec := time.Now().Unix()
		lastUnixSec := records[0].TS.Unix()
		delta := currUnixSec - lastUnixSec
		su.Require().
			Lessf(delta, int64(20), "last ts(%v) in test.vpcpeering does not catch up current ts(%v)", lastUnixSec, currUnixSec)
	}
	if su.accesspointPublicEndpointWorkloadPID != "" {
		var records []TimeStampRecord
		err := db.Raw("select * from test.ap_publicendpoint order by id desc limit 1").Find(&records).Error
		su.Require().NoError(err, "query test.ap_publicendpoint failed")
		su.Require().NotEmpty(records, "table test.ap_publicendpoint is empty")
		currUnixSec := time.Now().Unix()
		lastUnixSec := records[0].TS.Unix()
		delta := currUnixSec - lastUnixSec
		su.Require().
			Lessf(delta, int64(20), "last ts(%v) in test.ap_publicendpoint does not catch up current ts(%v)", lastUnixSec, currUnixSec)
	}
}

func (su *TestDedicatedMigrationSuite) Test_204_EnsureTiCDCCheckpointLat() {
	if su.connectIDBeforeMig == "" {
		return
	}
	params := connector_service.NewGetConnectorParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	params.SetConnectorID(su.connectIDBeforeMig)
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	resp, err := su.dataflowCli.ConnectorService.GetConnector(params, authInfoWriter)
	su.Require().NoError(err)
	su.Require().Equal(dataflowmodel.DataflowConnectorStatusRUNNING, *(resp.Payload.Status))
	currentUnixSec := time.Now().Unix()
	delta := currentUnixSec - time.Time(resp.Payload.CheckpointUtc).Unix()
	su.Require().Lessf(delta, int64(180), "changefeed %s checkpoint fall behind", su.connectIDBeforeMig)
}

func (su *TestDedicatedMigrationSuite) Test_205_EnsureTiCDCCanScale() {
	getConnectorParams := connector_service.NewGetConnectorParams()
	getConnectorParams.SetOrgID(su.orgID)
	getConnectorParams.SetProjectID(su.projectID)
	getConnectorParams.SetClusterID(su.clusterID)
	getConnectorParams.SetConnectorID(su.connectIDBeforeMig)
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	getConnectorResp, err := su.dataflowCli.ConnectorService.GetConnector(getConnectorParams, authInfoWriter)
	su.Require().NoError(err)
	currentInstanceType := getConnectorResp.GetPayload().ChangefeedInstanceType

	listInstanceTypesParams := changefeed_instance_type_service.NewListChangefeedInstanceTypesParams()
	listInstanceTypesResp, err := su.dataflowCli.ChangefeedInstanceTypeService.ListChangefeedInstanceTypes(
		listInstanceTypesParams, authInfoWriter)
	su.Require().NoError(err)
	var scaledInstanceType string
	for _, instanceType := range listInstanceTypesResp.Payload.Items {
		if instanceType.Name != currentInstanceType {
			scaledInstanceType = instanceType.Name
			su.T().Log("get scaled instance type:", instanceType.Name, "rcu:", instanceType.RcuNum)
			break
		}
	}
	su.Require().NotEmpty(scaledInstanceType)

	scaleConnectorParams := connector_service.NewScaleChangefeedParams()
	scaleConnectorParams.SetOrgID(su.orgID)
	scaleConnectorParams.SetProjectID(su.projectID)
	scaleConnectorParams.SetClusterID(su.clusterID)
	scaleConnectorParams.SetChangefeedID(su.connectIDBeforeMig)
	scaleConnectorParams.SetBody(connector_service.ScaleChangefeedBody{
		ChangefeedInstanceType: &scaledInstanceType,
	})

	err = retry.Do(
		func() error {
			_, err := su.dataflowCli.ConnectorService.ScaleChangefeed(scaleConnectorParams, authInfoWriter)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(err)

	for {
		time.Sleep(time.Second * 10)
		getConnectorResp, err := su.dataflowCli.ConnectorService.GetConnector(getConnectorParams, authInfoWriter)
		su.Require().NoError(err)
		su.Require().NotNil(getConnectorResp.GetPayload().Status)
		if *getConnectorResp.GetPayload().Status == dataflowmodel.DataflowConnectorStatusRUNNING {
			su.T().Log("changefeed scales successfully")
			break
		}
		su.T().Log("changefeed status after scale:", *getConnectorResp.GetPayload().Status)
	}
}

func (su *TestDedicatedMigrationSuite) Test_206_DeleteTiCDC() {
	if su.connectIDBeforeMig == "" {
		return
	}
	params := connector_service.NewDeleteConnectorParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	params.SetConnectorID(su.connectIDBeforeMig)
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	_, err := su.dataflowCli.ConnectorService.DeleteConnector(params, authInfoWriter)
	su.Require().NoError(err)
}

func (su *TestDedicatedMigrationSuite) Test_207_EnsureTiFlashClusterCanWork() {
	if su.tiflashClusterID == "" {
		su.T().Logf("Skip, tiflash cluster is not created")
		return
	}

	status := su.portalCli.PollClusterStatus(su.orgID, su.projectID, su.tiflashClusterID)
	su.Require().Equal(model.CentralClusterStatusNORMAL, *status)
	su.T().Logf("cluster %s is normal", su.tiflashClusterID)

	resp := su.portalCli.GetTiFlashNodes(su.orgID, su.projectID, su.tiflashClusterID)
	su.Require().Len(resp.Items, 1)
	for _, node := range resp.Items {
		su.Require().Equal(model.CentralTiFlashNodeStatusAVAILABLE, *node.Status)
	}

	su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.tiflashClusterName)
	su.T().Logf("cluster %s is deleted", su.tiflashClusterID)
}

func (su *TestDedicatedMigrationSuite) Test_208_EnsureBackupCanWork() {
	currUTC := time.Now().UTC()
	backupName := fmt.Sprintf("bak-after-mig-%s", time.Now().Format("15-04-05"))
	backupTime := currUTC.Format(time.RFC3339)

	retryErr := retry.Do(
		func() error {
			response, err := su.dataflowCli.CreateBackupSchedule(
				su.orgID,
				su.projectID,
				su.clusterID,
				backupName,
				backupTime,
			)
			if err != nil {
				return err
			}
			if response.Payload == nil {
				return errors.New("response payload is nil")
			}
			su.bakIDAfterMig = response.GetPayload().BackupID
			return nil
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	su.Require().NoError(retryErr)
	su.Require().NotEmpty(su.bakIDAfterMig)
	su.T().Logf("bakIDAfterMig is %s", su.bakIDAfterMig)

	{
		params := backup_service.NewListBackupsParams()
		params.SetClusterID(su.clusterID)
		params.SetProjectID(su.projectID)
		params.SetOrgID(su.orgID)
		params.SetBackupID(&su.bakIDAfterMig)
		params.SetPage(1)
		params.SetPerPage(10)
		authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
		for {
			time.Sleep(30 * time.Second)
			resp, err := su.dataflowCli.BackupService.ListBackups(params, authInfoWriter)
			su.Require().NoError(err, "ListBackups failed")
			su.Require().NotEmpty(resp.Payload.Items)
			bak := resp.Payload.Items[0]
			su.Require().NotEqual(dataflowmodel.BackupExternalStatusFAILED, *bak.Status)
			su.T().Logf("backup %s status: %s", su.bakIDAfterMig, *bak.Status)
			if *bak.Status == dataflowmodel.BackupExternalStatusSUCCESS {
				break
			}
		}
	}
}

func (su *TestDedicatedMigrationSuite) Test_209_RestoreCluster() {
	respBeforeMig := su.dataflowCli.RestoreBackup(
		su.orgID,
		su.projectID,
		su.clusterID,
		su.bakIDAfterMig,
		su.restoreClusterNameBeforeMig,
		su.region,
	)
	respAfterMig := su.dataflowCli.RestoreBackup(
		su.orgID,
		su.projectID,
		su.clusterID,
		su.bakIDAfterMig,
		su.restoreClusterNameAfterMig,
		su.region,
	)
	su.Require().NotEmpty(respBeforeMig)
	su.Require().NotEmpty(respAfterMig)

	su.restoreClusterIDBeforeMig = lo.FromPtr(respBeforeMig.ClusterID)
	su.restoreClusterIDAfterMig = lo.FromPtr(respAfterMig.ClusterID)
	su.Require().NotEmpty(su.restoreClusterIDBeforeMig)
	su.Require().NotEmpty(su.restoreClusterIDAfterMig)
}

func (su *TestDedicatedMigrationSuite) Test_210_ResumeCluster() {
	if su.skipPauseClusterCase {
		su.T().Logf("Skip resuming cluster, clusterID: %s", su.pausedClusterIDBeforeMig)
		return
	}

	su.portalCli.ResumeCluster(su.orgID, su.projectID, su.pausedClusterIDBeforeMig)
	su.T().Logf("cluster %s is resumed", su.pausedClusterIDBeforeMig)
}

func (su *TestDedicatedMigrationSuite) Test_211_EnsureScaleOutCanWork() {
	resp, err := su.openApiCli.ScaleCluster(
		su.projectID,
		su.clusterID,
		&cluster.UpdateClusterParamsBodyConfigComponents{
			Tidb: &cluster.UpdateClusterParamsBodyConfigComponentsTidb{
				NodeQuantity: 2,
			},
			Tikv: &cluster.UpdateClusterParamsBodyConfigComponentsTikv{
				NodeQuantity: 6,
			},
		},
	)
	su.Require().True(resp.IsSuccess())
	su.Require().NoError(err)
	su.T().Logf("cluster %s is scaling out tidb and tikv", su.clusterID)

	for {
		status := su.portalCli.PollClusterStatus(su.orgID, su.projectID, su.clusterID)
		if *status == model.CentralClusterStatusNORMAL {
			break
		}
		su.T().Logf("cluster %s status: %s", su.clusterID, *status)
		time.Sleep(10 * time.Second)
	}
}

func (su *TestDedicatedMigrationSuite) Test_212_EnsureCMEKCanWork() {
	if !su.cmekEnabled || su.provider != "aws" {
		return
	}
	// {
	//	// original volumes
	//	input := new(ec2.DescribeVolumesInput)
	//	input.Filters = append(input.Filters,
	//		ec2types.Filter{
	//			Name:   lo.ToPtr("tag:component"),
	//			Values: []string{"tikv"},
	//		},
	//		ec2types.Filter{
	//			Name:   lo.ToPtr("tag:usedby"),
	//			Values: []string{"customer-tidb"},
	//		},
	//		ec2types.Filter{
	//			Name:   lo.ToPtr("tag:project"),
	//			Values: []string{su.projectID},
	//		},
	//		ec2types.Filter{
	//			Name:   lo.ToPtr("tag:cluster"),
	//			Values: []string{su.clusterID},
	//		},
	//		ec2types.Filter{
	//			Name:   lo.ToPtr("encrypted"),
	//			Values: []string{"true"},
	//		},
	//	)
	//	out, err := su.ec2Cli.DescribeVolumes(context.Background(), input)
	//	su.Require().NoError(err, "DescribeVolumes failed")
	//	su.Require().NotEmpty(out.Volumes, "filter volumes get empty result")
	//	for i := range out.Volumes {
	//		v := out.Volumes[i]
	//		if su.customerKMSKey != lo.FromPtr(v.KmsKeyId) {
	//			su.Require().FailNowf(fmt.Sprintf("volume %s kms key(%s) not match ", *v.VolumeId, lo.FromPtr(v.KmsKeyId)), "customer kms key is %s", su.customerKMSKey)
	//			break
	//		}
	//	}
	// }
	{
		// scale out volumes
		input := new(ec2.DescribeVolumesInput)
		input.Filters = append(input.Filters,
			ec2types.Filter{
				Name:   lo.ToPtr("tag:kubernetes.io/created-for/pvc/namespace"),
				Values: []string{"tidb" + su.clusterID},
			},
			ec2types.Filter{
				Name:   lo.ToPtr("encrypted"),
				Values: []string{"true"},
			},
		)
		out, err := su.ec2Cli.DescribeVolumes(context.Background(), input)
		su.Require().NoError(err, "DescribeVolumes failed")
		su.Require().NotEmpty(out.Volumes, "filter volumes get empty result")
		// filter tikv volumes by tag
		// kubernetes.io/created-for/pvc/name: tikv-db-tikv-0
		for i := range out.Volumes {
			v := out.Volumes[i]
			pvcName := ""
			for _, tag := range v.Tags {
				if lo.FromPtr(tag.Key) == "kubernetes.io/created-for/pvc/name" {
					pvcName = lo.FromPtr(tag.Value)
					break
				}
			}
			if !strings.HasPrefix(pvcName, "tikv-db") {
				continue
			}
			if su.customerKMSKey != lo.FromPtr(v.KmsKeyId) {
				su.Require().
					FailNowf(fmt.Sprintf("volume %s kms key(%s) not match ", *v.VolumeId, lo.FromPtr(v.KmsKeyId)), "customer kms key is %s", su.customerKMSKey)
				break
			}
		}
	}
}

func (su *TestDedicatedMigrationSuite) Test_213_EnsureClusterScaleInCanWork() {
	resp, err := su.openApiCli.ScaleCluster(
		su.projectID,
		su.clusterID,
		&cluster.UpdateClusterParamsBodyConfigComponents{
			Tidb: &cluster.UpdateClusterParamsBodyConfigComponentsTidb{
				NodeQuantity: 1,
			},
			Tikv: &cluster.UpdateClusterParamsBodyConfigComponentsTikv{
				NodeQuantity: 3,
			},
		},
	)
	su.Require().True(resp.IsSuccess())
	su.Require().NoError(err)
	su.T().Logf("cluster %s is scaling in tidb and tikv", su.clusterID)

	for {
		status := su.portalCli.PollClusterStatus(su.orgID, su.projectID, su.clusterID)
		if *status == model.CentralClusterStatusNORMAL {
			break
		}
		su.T().Logf("cluster %s status: %s", su.clusterID, *status)
		time.Sleep(10 * time.Second)
	}
}

func (su *TestDedicatedMigrationSuite) Test_214_EnsureRestoreCanWork() {
	su.waitForClusterStatusNormal(
		model.CentralClusterStatusNORMAL,
		su.restoreClusterIDBeforeMig,
		su.restoreClusterNameBeforeMig,
	)
	su.waitForClusterStatusNormal(
		model.CentralClusterStatusNORMAL,
		su.restoreClusterIDAfterMig,
		su.restoreClusterNameAfterMig,
	)
}

func (su *TestDedicatedMigrationSuite) Test_215_DeleteRestoreClusterNameBeforeMig() {
	su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.restoreClusterNameBeforeMig)
}

func (su *TestDedicatedMigrationSuite) Test_216_CheckRestoreClusterO11yDataAfterMig() {
	if su.skipO11yCase {
		su.T().Logf("Skip checking o11y case")
		return
	}

	ctx := context.Background()
	now := int(time.Now().Unix())
	su.checkO11yData(ctx, su.restoreClusterIDAfterMig, now-10*60, now)
}

func (su *TestDedicatedMigrationSuite) Test_217_DeleteRestoreClusterNameAfterMig() {
	su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.restoreClusterNameAfterMig)
}

func (su *TestDedicatedMigrationSuite) Test_218_EnsureResumeClusterCanWork() {
	if su.skipPauseClusterCase {
		su.T().Logf("Skip resuming cluster, clusterID: %s", su.pausedClusterIDBeforeMig)
		return
	}

	su.waitForClusterStatusNormal(
		model.CentralClusterStatusNORMAL,
		su.pausedClusterIDBeforeMig,
		su.pausedClusterNameBeforeMig,
	)
	su.T().Logf("cluster %s is normal", su.pausedClusterIDBeforeMig)

	resp := su.portalCli.GetTiDBNodes(su.orgID, su.projectID, su.pausedClusterIDBeforeMig)
	su.Require().Len(resp.Items, 1)
	for _, node := range resp.Items {
		su.Require().Equal(model.CentralTiDBNodeStatusAVAILABLE, *node.Status)
	}

	su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.pausedClusterNameBeforeMig)
	su.T().Logf("cluster %s is deleted", su.pausedClusterIDBeforeMig)
}

func (su *TestDedicatedMigrationSuite) Test_219_EnsureClusterScaleUpCanWork() {
	resp, err := su.openApiCli.ScaleCluster(
		su.projectID,
		su.clusterID,
		&cluster.UpdateClusterParamsBodyConfigComponents{
			Tidb: &cluster.UpdateClusterParamsBodyConfigComponentsTidb{
				NodeSize: "8C16G",
			},
			Tikv: &cluster.UpdateClusterParamsBodyConfigComponentsTikv{
				NodeSize: "8C32G",
			},
		},
	)
	su.Require().True(resp.IsSuccess())
	su.Require().NoError(err)
	su.T().Logf("cluster %s is scaling up tikv", su.clusterID)

	for {
		status := su.portalCli.PollClusterStatus(su.orgID, su.projectID, su.clusterID)
		if *status == model.CentralClusterStatusNORMAL {
			break
		}
		su.T().Logf("cluster %s status: %s", su.clusterID, *status)
		time.Sleep(10 * time.Second)
	}
}

func (su *TestDedicatedMigrationSuite) Test_220_EnsureClusterProxyCanWork() {
	type ConfigValue struct {
		Id string `json:"id"`
	}
	type Items struct {
		PdConfig []ConfigValue `json:"pd_config"`
	}

	type Config struct {
		Items Items `json:"items"`
	}

	req, err := http.NewRequest(
		http.MethodGet,
		fmt.Sprintf("https://%s/proxy/orgs/%s/clusters/%s/pd/dashboard/api/configuration/all",
			su.Host,
			su.orgID,
			su.clusterID),
		http.NoBody,
	)
	su.Require().NoError(err)
	req.Header.Add("authorization",
		"Bearer "+su.BearerToken)

	resp, err := http.DefaultClient.Do(req)
	su.Require().NoError(err)
	su.Require().Equal(http.StatusOK, resp.StatusCode)
	defer resp.Body.Close()
	data, err := io.ReadAll(resp.Body)
	su.Require().NoError(err)

	var cfg Config
	err = json.Unmarshal(data, &cfg)
	su.Require().NoError(err)
	su.Require().NotEmpty(cfg.Items.PdConfig)
}

func (su *TestDedicatedMigrationSuite) Test_221_OpsAccesspoint() {
	if su.accesspointClusterID == "" {
		return
	}

	// delete accesspoint
	su.openApiCli.DeleteAccessPoint(su.projectID, su.accesspointClusterID, su.accesspointID)

	su.waitForClusterStatusNormal(model.CentralClusterStatusNORMAL, su.accesspointClusterID, su.accesspointClusterName)
	su.T().Logf("cluster %s is normal", su.accesspointClusterID)

	if su.accesspointClusterID != "" {
		su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.accesspointClusterName)
	}

	// create accesspoint
	// accesspointResp, err := su.openApiCli.CreateAccessPoint(su.projectID, su.accesspointClusterID)
	// su.Require().NoError(err)
	// su.Require().NotNil(accesspointResp)
	// su.newAccesspointID = accesspointResp.ID

	// scale accesspoint
	// resp, err := su.openApiCli.ScaleClusterByAccessPoint(su.projectID, su.accesspointClusterID, su.newAccesspointID, &idl.ScaleTiDBConfig{
	//	NodeQuantity: 1,
	// })
	// su.Require().True(resp.IsSuccess())
	// su.Require().NoError(err)
}

// func (su *TestDedicatedMigrationSuite) Test_222_EnsureAccesspointCanWork() {
//	if su.accesspointClusterID == "" {
//		return
//	}
//
//	su.waitForClusterStatusNormal(model.CentralClusterStatusNORMAL, su.accesspointClusterID, su.accesspointClusterName)
//	su.T().Logf("cluster %s is normal", su.accesspointClusterID)
//
//	// list accesspoint
//	listAccessPointResp := su.openApiCli.ListAccessPoint(su.projectID, su.accesspointClusterID)
//	_, ok := lo.Find(listAccessPointResp.Items, func(item *cluster.ListAccesspointsOKBodyItemsItems0) bool {
//		return lo.FromPtr(item.ID) == su.newAccesspointID
//	})
//	su.Require().True(ok)
//
//	// delete cluster
//	su.portalCli.EnsureDeleteCluster(su.orgID, su.projectID, su.accesspointClusterName)
//	su.T().Logf("cluster %s is deleted", su.accesspointClusterID)
// }

func (su *TestDedicatedMigrationSuite) Test_223_CheckClusterMetering() {
	if su.skipMigration {
		su.T().Log("Skip check cluster metering case")
		return
	}

	if su.env == "prod" {
		su.T().Log("Skip test for prod environment")
		return
	}

	// check if new cluster metering records are created after unlock project
	var count int64
	err := su.dedicatedGlobalDB.Model(&schema.ClusterMetering{}).Unscoped().
		Where("`cluster_id` = ?", su.clusterID).
		Count(&count).Error
	su.Require().NoError(err)
	su.Require().Greaterf(count, int64(1), "cluster metering record count is not greater than 1 after unlock project")

	// query the latest cluster metering record after unlock project
	var latestClusterMetering schema.ClusterMetering
	err = su.dedicatedGlobalDB.Model(&schema.ClusterMetering{}).Unscoped().
		Where("`cluster_id` = ?", su.clusterID).
		Order("start_ts DESC").
		First(&latestClusterMetering).Error
	su.Require().NoError(err)
	su.Require().Equal(strconv.FormatUint(latestClusterMetering.TenantID, 10), su.orgID)
	su.Require().Equal(strconv.FormatUint(latestClusterMetering.ProjectID, 10), su.projectID)
	// check if cluster metering record interval is 1 minute
	su.Require().Equal(time.Minute, latestClusterMetering.EndTS.Sub(latestClusterMetering.StartTS))

	// query all cluster metering records between the migrated cluster metering and the latest cluster metering
	var clusterMeterings []*schema.ClusterMetering
	err = su.dedicatedGlobalDB.Model(&schema.ClusterMetering{}).Unscoped().
		Where("`cluster_id` = ?", su.clusterID).
		Where("`start_ts` >= ?", su.migratedClusterMetering.StartTS).
		Where("`end_ts` <= ?", latestClusterMetering.EndTS).
		Find(&clusterMeterings).Error
	su.Require().NoError(err)

	// check if the sum of cluster metering duration is equal to the diff duration
	diffDuration := latestClusterMetering.EndTS.Sub(su.migratedClusterMetering.StartTS)
	var meteringDuration time.Duration
	for _, v := range clusterMeterings {
		meteringDuration += v.EndTS.Sub(v.StartTS)
	}
	su.T().Logf("diff duration: %s", diffDuration)
	su.T().Logf("metering duration: %s", meteringDuration)
	su.Require().Equal(diffDuration, meteringDuration, "cluster metering duration is not equal to the diff duration")
}

func (su *TestDedicatedMigrationSuite) getCreateChangefeedInstanceType() string {
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	listInstanceTypesParams := changefeed_instance_type_service.NewListChangefeedInstanceTypesParams()
	listInstanceTypesResp, err := su.dataflowCli.ChangefeedInstanceTypeService.ListChangefeedInstanceTypes(
		listInstanceTypesParams, authInfoWriter)
	su.Require().NoError(err)
	su.Require().NotEmpty(listInstanceTypesResp.Payload.Items)
	return listInstanceTypesResp.Payload.Items[0].Name
}

func (su *TestDedicatedMigrationSuite) createChangefeedWithRetry(
	params *connector_service.CreateConnectorParams,
) (resp *connector_service.CreateConnectorOK, err error) {
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	retryErr := retry.Do(
		func() error {
			resp, err = su.dataflowCli.ConnectorService.CreateConnector(params, authInfoWriter)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
	if retryErr == nil {
		return resp, nil
	}
	return nil, retryErr
}

func (su *TestDedicatedMigrationSuite) Test_224_CreateChangefeed_Kafka() {
	instanceType := su.getCreateChangefeedInstanceType()
	name := "after-migration-kafka"
	params := connector_service.NewCreateConnectorParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	body := connector_service.CreateConnectorBody{}
	body.ConnectorType = dataflowmodel.DataflowConnectorTypeKAFKA.Pointer()
	body.ChangefeedInstanceType = instanceType
	body.Name = name
	body.KafkaInfo = new(dataflowmodel.DataflowKafkaInfo)
	body.KafkaInfo.Authentication = &dataflowmodel.DataflowAuthentication{
		AuthType: dataflowmodel.DataflowKafkaAuthTypeDISABLE.Pointer(),
	}
	body.KafkaInfo.Broker = &dataflowmodel.DataflowBroker{
		BrokerEndpoints: su.changefeedtKafkaEndpoint,
		Compression:     dataflowmodel.DataflowCompressionTypeNONE.Pointer(),
		KafkaVersion:    dataflowmodel.DataflowKafkaVersionKAFKAVERSION2XX.Pointer(),
	}
	body.KafkaInfo.DataFormat = &dataflowmodel.DataflowDataFormat{}
	body.KafkaInfo.TableConfig = &dataflowmodel.DataflowTableConfig{
		Mode:       dataflowmodel.DataflowTableModeIGNORENOTSUPPORTTABLE.Pointer(),
		FilterRule: []string{"test.*"},
	}
	body.KafkaInfo.TopicPartitionConfig = &dataflowmodel.DataflowTopicPartitionConfig{
		DefaultTopic:        "test",
		DispatchType:        dataflowmodel.DataflowDispatchTypeDISPATCHTYPEONETOPIC.Pointer(),
		ReplicationFactor:   1,
		PartitionNum:        1,
		PartitionDispatcher: dataflowmodel.DataflowPartitionDispatcherINDEXVALUE.Pointer(),
	}
	body.KafkaInfo.StartPosition = &dataflowmodel.DataflowStartPosition{
		Mode: dataflowmodel.DataflowStartModeFROMNOW.Pointer(),
	}
	params.SetBody(body)
	resp, err := su.createChangefeedWithRetry(params)
	su.Require().NoError(err)
	su.Require().NotNil(resp.GetPayload().ConnectorID)
	su.T().Log("create changefeed with kafka successfully, id:", *resp.GetPayload().ConnectorID)
}

func (su *TestDedicatedMigrationSuite) Test_225_CreateChangefeed_Mysql() {
	instanceType := su.getCreateChangefeedInstanceType()
	name := "after-migration-mysql"
	params := connector_service.NewCreateConnectorParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	body := connector_service.CreateConnectorBody{}
	body.ConnectorType = dataflowmodel.DataflowConnectorTypeMYSQL.Pointer()
	body.ChangefeedInstanceType = instanceType
	body.Name = name
	body.MysqlInfo = new(dataflowmodel.DataflowMySQLInfo)
	body.MysqlInfo.Connection = &dataflowmodel.DataflowConnection{
		Endpoint: su.changefeedMysqlEndpoint,
		Username: su.changefeedMysqlUsername,
		Password: su.changefeedMysqlPassword,
	}
	body.MysqlInfo.TableConfig = &dataflowmodel.DataflowTableConfig{
		Mode:       dataflowmodel.DataflowTableModeIGNORENOTSUPPORTTABLE.Pointer(),
		FilterRule: []string{"test.*"},
	}
	body.MysqlInfo.StartPosition = &dataflowmodel.DataflowStartPosition{
		Mode: dataflowmodel.DataflowStartModeFROMNOW.Pointer(),
	}
	params.SetBody(body)
	resp, err := su.createChangefeedWithRetry(params)
	su.Require().NoError(err)
	su.Require().NotNil(resp.GetPayload().ConnectorID)
	su.T().Log("create changefeed with mysql successfully, id:", *resp.GetPayload().ConnectorID)
}

func (su *TestDedicatedMigrationSuite) Test_226_CreateChangefeed_ObjectStorage() {
	instanceType := su.getCreateChangefeedInstanceType()
	name := "after-migration-oss"
	params := connector_service.NewCreateConnectorParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	body := connector_service.CreateConnectorBody{}
	switch su.provider {
	case "aws":
		s3URI := fmt.Sprintf("s3://dedicated-migration-cdc-us-west-2/%s/", su.clusterID)
		body.ConnectorType = dataflowmodel.DataflowConnectorTypeS3.Pointer()
		body.ChangefeedInstanceType = instanceType
		body.URL = lo.ToPtr(s3URI)
		body.Name = name
		body.S3Info = new(dataflowmodel.DataflowS3Info)
		body.S3Info.Endpoint = &dataflowmodel.DataflowS3Endpoint{
			S3URI:     s3URI,
			AccessKey: su.s3AK,
			SecretKey: su.s3SK,
		}
		body.S3Info.DataFormat = &dataflowmodel.DataflowS3Dataformat{
			Protocol:             dataflowmodel.DataflowS3ProtocolS3PROTOCOLCSV.Pointer(),
			Delimiter:            ",",
			BinaryEncodingMethod: dataflowmodel.DataflowBinaryEncodingMethodBASE64.Pointer(),
			Quote:                "\"",
			Null:                 "\\N",
		}
		body.S3Info.StartPosition = &dataflowmodel.DataflowStartPosition{
			Mode: dataflowmodel.DataflowStartModeFROMNOW.Pointer(),
		}
		body.S3Info.StorageFlushConfig = &dataflowmodel.DataflowStorageFlushConfig{
			FlushInterval: "60s",
			FileSize:      "67108864",
		}
		body.S3Info.TableConfig = &dataflowmodel.DataflowTableConfig{
			Mode:       dataflowmodel.DataflowTableModeFORCESYNC.Pointer(),
			FilterRule: []string{"*.*"},
		}
	case "gcp":
		// get admin profile first
		// the gcp account id is service account
		serviceAccount, err := su.getGCPAccountID(su.orgID, su.projectID, su.clusterID)
		su.Require().NoError(err)
		err = su.allowSeviceAccountAccessGCSBucket(serviceAccount)
		su.Require().NoError(err)

		gcsURI := fmt.Sprintf("gs://dedicated-migration-cdc-us-west1/%s/", su.clusterID)
		body.ConnectorType = dataflowmodel.DataflowConnectorTypeGCS.Pointer()
		body.ChangefeedInstanceType = "2rcu"
		body.URL = lo.ToPtr(gcsURI)
		body.Name = name
		body.GcsInfo = new(dataflowmodel.DataflowGcsInfo)
		body.GcsInfo.Endpoint = &dataflowmodel.DataflowGcsEndpoint{
			GcsURI: gcsURI,
		}
		body.GcsInfo.DataFormat = &dataflowmodel.DataflowS3Dataformat{
			Protocol:             dataflowmodel.DataflowS3ProtocolS3PROTOCOLCSV.Pointer(),
			Delimiter:            ",",
			BinaryEncodingMethod: dataflowmodel.DataflowBinaryEncodingMethodBASE64.Pointer(),
			Quote:                "\"",
			Null:                 "\\N",
		}
		body.GcsInfo.StartPosition = &dataflowmodel.DataflowStartPosition{
			Mode: dataflowmodel.DataflowStartModeFROMNOW.Pointer(),
		}
		body.GcsInfo.StorageFlushConfig = &dataflowmodel.DataflowStorageFlushConfig{
			FlushInterval: "60s",
			FileSize:      "67108864",
		}
		body.GcsInfo.TableConfig = &dataflowmodel.DataflowTableConfig{
			Mode:       dataflowmodel.DataflowTableModeFORCESYNC.Pointer(),
			FilterRule: []string{"*.*"},
		}
	}
	params.SetBody(body)
	resp, err := su.createChangefeedWithRetry(params)
	su.Require().NoError(err)
	su.Require().NotNil(resp.GetPayload().ConnectorID)
	su.T().Log("create changefeed with object storage successfully, id:", *resp.GetPayload().ConnectorID)
}

func (su *TestDedicatedMigrationSuite) Test_227_EnsureAllChangefeedsCreated() {
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	params := connector_service.NewListConnectorsParams()
	params.SetClusterID(su.clusterID)
	params.SetProjectID(su.projectID)
	params.SetOrgID(su.orgID)
	params.SetPage(1)
	params.SetPerPage(10)
	for {
		time.Sleep(time.Second * 10)
		resp, err := su.dataflowCli.ConnectorService.ListConnectors(params, authInfoWriter)
		su.Require().NoError(err)

		changefeedNotRunning := false
		for _, item := range resp.Payload.Items {
			su.Require().NotNil(item.Status)
			su.Require().NotNil(item.ConnectorID)
			if *item.Status != dataflowmodel.DataflowConnectorStatusRUNNING {
				changefeedNotRunning = true
				su.T().Log("changefeed is not running, id:", *item.ConnectorID, "status:", *item.Status)
			}
		}
		if !changefeedNotRunning {
			su.T().Log("all the changefeeds are running")
			break
		}
	}
}

func (su *TestDedicatedMigrationSuite) deleteChangefeedWithRetry(
	params *connector_service.DeleteConnectorParams,
) error {
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	return retry.Do(
		func() error {
			_, err := su.dataflowCli.ConnectorService.DeleteConnector(params, authInfoWriter)
			return err
		},
		retry.Attempts(3),
		retry.DelayType(retry.FixedDelay),
		retry.Delay(5*time.Second),
	)
}

func (su *TestDedicatedMigrationSuite) Test_228_DeleteAllChangefeeds() {
	authInfoWriter := openapiruntimeclient.BearerToken(su.BearerToken)
	listParams := connector_service.NewListConnectorsParams()
	listParams.SetClusterID(su.clusterID)
	listParams.SetProjectID(su.projectID)
	listParams.SetOrgID(su.orgID)
	listParams.SetPage(1)
	listParams.SetPerPage(10)
	resp, err := su.dataflowCli.ConnectorService.ListConnectors(listParams, authInfoWriter)
	su.Require().NoError(err)

	for _, item := range resp.Payload.Items {
		su.Require().NotNil(item.ConnectorID)
		deleteParams := connector_service.NewDeleteConnectorParams()
		deleteParams.SetOrgID(su.orgID)
		deleteParams.SetProjectID(su.projectID)
		deleteParams.SetClusterID(su.clusterID)
		deleteParams.SetConnectorID(*item.ConnectorID)
		err = su.deleteChangefeedWithRetry(deleteParams)
		su.Require().NoError(err)
	}
}

func (su *TestDedicatedMigrationSuite) Test_229_CheckO11yData() {
	if su.skipO11yCase {
		su.T().Logf("Skip checking o11y case")
		return
	}

	ctx := context.Background()
	now := int(time.Now().Unix())
	su.checkO11yData(ctx, su.clusterID, now-10*60, now)
}

// Optimization: Ensure changefeeds are deleted before pausing the cluster.
// {"base_resp":{"err_code":"49900003","err_msg":"Forbidden. The cluster is in the CDC state and cannot execute this operation."}}}
func (su *TestDedicatedMigrationSuite) Test_230_PauseCluster() {
	su.portalCli.PauseCluster(su.orgID, su.projectID, su.clusterID)
	su.T().Logf("cluster %s is paused", su.clusterID)
}

func (su *TestDedicatedMigrationSuite) Test_231_EnsurePauseClusterCanWork() {
	su.waitForClusterStatusNormal(model.CentralClusterStatusPAUSED, su.clusterID, su.clusterName)
	su.T().Logf("cluster %s is paused", su.clusterID)
}