package changefeed

import (
	"fmt"
	"net/url"
	"strings"
	"time"

	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/dedicated/common"
	"github.com/tidbcloud/dataflow-service-e2e/internal/config"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/sample_data"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/client/connector_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
)

const azureBlobStorageHostname = "blob.core.windows.net"

type azureBlobStorageChangefeedSmokeCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator
	cloudProvider CloudProviderType
}

func NewAzureBlobStorageChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &azureBlobStorageChangefeedSmokeCase{cloudProvider: cloudProvider}
}

func (m *azureBlobStorageChangefeedSmokeCase) Name() string {
	return "azureBlobStorageChangefeedSmoke"
}

func (m *azureBlobStorageChangefeedSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
		"azblob": &AzureBlobStorageRequirement{
			AzureBlobStorageDimensions: AzureBlobStorageDimensions{
				Region: config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
	}
}

func (m *azureBlobStorageChangefeedSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	downstream := m.AzureBlobStorage("azblob")
	changefeedName := m.Name()
	instanceType := "2rcu"
	dbName := m.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}

	var (
		changefeedID                     string
		createAzBlobInfo, editAzBlobInfo *model.DataflowAzureBlobStorageInfo
	)

	// init reusable api client and validator
	m.cdcApiWrapper = cdcapi.NewCDCClientWrapper(m.Logger())
	m.cfValidator = common.NewChangfeedValidator(m.Logger())
	sampleDataWriter := sample_data.NewSampleDataWriter(m.Logger(), tidbCluster.MysqlParameter(), dbName, 3*time.Second)

	// test create request validation
	invalidCreateReqOpts := []azureBlobStorageChangefeedOpt{
		azureBlobStorageChangefeedWithInvalidDataFormat,
		azureBlobStorageChangefeedWithInvalidFlushConfig,
		azureBlobStorageChangefeedWithInvalidTableFilter,
		azureBlobStorageChangefeedWithInvalidEventFilter,
	}

	for _, opt := range invalidCreateReqOpts {
		m.Expect("create azblob changefeed with invalid params", func() error {
			req := genCreateAzureBlobStorageChangefeedReqWithOpt(changefeedName, instanceType, dbName, tidbCluster.ClusterID, downstream, opt)
			_, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *req)
			return ExpectError(err)
		}, 3, time.Second*10)
	}

	// cleanup changefeeds if exists
	defer func() {
		m.
			Step("stop writing data", func() error {
				sampleDataWriter.StopWriting()
				return nil
			}).
			RetryDo("delete changefeeds", func() error {
				return m.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedName)
			}, 3, time.Minute).
			Expect("changefeeds not in list", func() error {
				return m.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedName)
			}, 3, time.Second*10)
	}()

	m.
		RetryDo(fmt.Sprintf("test connection to downstream %s", downstream.ContainerURI), func() (err error) {
			return m.cdcApiWrapper.TestConnection(tidbCluster, downstream)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("init database and table in %s", tidbCluster.DisplayName),
			sampleDataWriter.InitDBTable, 3, time.Minute).
		RetryDo(fmt.Sprintf("verify tables with tableFilter %v", tableFilter), func() (err error) {
			return m.cfValidator.VerifyTables(tidbCluster, tableFilter, []string{fmt.Sprintf("%s.%s", dbName, sample_data.SAMPLE_TABLE)}, nil)
		}, 3, time.Second*10).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			createReq := genCreateAzureBlobStorageChangefeedReq(changefeedName, instanceType, dbName, tidbCluster.ClusterID, downstream)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *createReq)
			if err == nil {
				changefeedID = *resp.Payload.ConnectorID
				createAzBlobInfo = createReq.AzureBlobStorageInfo
			}
			return err
		}, 3, time.Minute).
		Expect("changefeed in list", func() error {
			return m.cfValidator.AreChangefeedsInList(tidbCluster, changefeedID)
		}, 3, time.Second*10).
		Expect("changefeeed is running", func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, 20, time.Minute).
		Expect("compare changefeed kernel config after create", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return compareAzureBlobStorageChangefeedKernelConfig(createAzBlobInfo, cfg)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("start write sample data to database %s", dbName),
			sampleDataWriter.StartWriting, 3, time.Minute).
		Step("verify changefeed's CheckpointTs advance normal", func() error {
			return m.cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeedID)
		}).
		Step("verifyGetCheckpointWithoutCache", func() error {
			return m.cfValidator.VerifyGetCheckpointWithoutCache(tidbCluster, changefeedID)
		}).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeedName), func() error {
			return m.cdcApiWrapper.PauseChangefeed(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is paused", changefeedName), func() error {
			return m.cfValidator.IsChangefeedInStatus(tidbCluster, changefeedID, model.DataflowConnectorStatusPAUSED)
		}, 3, time.Minute)

	// test edit request validation
	invalidEditReqOpts := []azureBlobStorageChangefeedOpt{
		azureBlobStorageChangefeedWithInvalidFlushConfig,
		azureBlobStorageChangefeedWithInvalidDataFormat,
		azureBlobStorageChangefeedWithInvalidTableFilter,
		azureBlobStorageChangefeedWithInvalidEventFilter,
	}
	for _, opt := range invalidEditReqOpts {
		m.Expect("edit azblob changefeed with invalid params", func() error {
			changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
			azBlobInfo := changefeedInfo.Payload.Request.AzureBlobStorageInfo
			opt(azBlobInfo)
			editReq := connector_service.EditConnectorBody{
				ConnectorID:          changefeedInfo.Payload.ConnectorID,
				ConnectorType:        changefeedInfo.Payload.Request.ConnectorType,
				Name:                 changefeedInfo.Payload.Name,
				AzureBlobStorageInfo: azBlobInfo,
			}
			err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
			return ExpectError(err)
		}, 3, time.Second*10)
	}

	m.
		RetryDo(fmt.Sprintf("edit changefeed %s without password", changefeedName), func() error {
			azBlobInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, false)
			if err == nil {
				editAzBlobInfo = azBlobInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s with password", changefeedName), func() error {
			azBlobInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, true)
			if err == nil {
				editAzBlobInfo = azBlobInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeedName), func() error {
			return m.cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is runining", changefeedName), func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Expect("compare changefeed kernel config after edit", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return compareAzureBlobStorageChangefeedKernelConfig(editAzBlobInfo, cfg)
		}, 3, time.Minute).
		Step(fmt.Sprintf("verify changefeed %s CheckpointTs advance normal", changefeedName), func() error {
			return m.cfValidator.VerifyCheckpointTsAdvanceNormal(2*time.Minute, tidbCluster, changefeedID)
		}).
		Step("verifyGetCheckpointWithoutCache", func() error {
			return m.cfValidator.VerifyGetCheckpointWithoutCache(tidbCluster, changefeedID)
		})
}

func (m *azureBlobStorageChangefeedSmokeCase) editChangefeed(
	tidbCluster *TiDBDedicatedEntity, changefeedID string, downstream *AzureBlobStorageEntity, updatePassword bool,
) (*model.DataflowAzureBlobStorageInfo, error) {
	changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
	if err != nil {
		return nil, err
	}
	azBlobInfo := changefeedInfo.Payload.Request.AzureBlobStorageInfo
	if updatePassword {
		azBlobInfo.Endpoint.SasToken = downstream.SASToken
	}
	editAzureBlobStorageInfoNormally(azBlobInfo)
	editReq := connector_service.EditConnectorBody{
		ConnectorID:          changefeedInfo.Payload.ConnectorID,
		ConnectorType:        changefeedInfo.Payload.Request.ConnectorType,
		Name:                 changefeedInfo.Payload.Name,
		AzureBlobStorageInfo: azBlobInfo,
	}
	err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
	if err != nil {
		return nil, err
	}
	return azBlobInfo, nil
}

type azureBlobStorageChangefeedOpt func(azBlobInfo *model.DataflowAzureBlobStorageInfo)

func azureBlobStorageChangefeedWithInvalidDataFormat(azBlobInfo *model.DataflowAzureBlobStorageInfo) {
	azBlobInfo.DataFormat.Delimiter = ""
}

func azureBlobStorageChangefeedWithInvalidFlushConfig(azBlobInfo *model.DataflowAzureBlobStorageInfo) {
	azBlobInfo.StorageFlushConfig.FlushInterval = "20m"
}

func azureBlobStorageChangefeedWithInvalidTableFilter(azBlobInfo *model.DataflowAzureBlobStorageInfo) {
	azBlobInfo.TableConfig.FilterRule = []string{"abc", "def"}
}

func azureBlobStorageChangefeedWithInvalidEventFilter(azBlobInfo *model.DataflowAzureBlobStorageInfo) {
	azBlobInfo.TableConfig.EventFilterRule = []*model.DataflowEventFilterRule{
		{
			IgnoreEvent: []string{"all dml", "all ddl", "update"},
			Matcher:     []string{"a.b.d", "c.d.d"},
		},
	}
}

func genCreateAzureBlobStorageChangefeedReqWithOpt(
	name, instanceType, dbName, clusterID string, downstream *AzureBlobStorageEntity, opt azureBlobStorageChangefeedOpt,
) *connector_service.CreateConnectorBody {
	req := genCreateAzureBlobStorageChangefeedReq(name, instanceType, dbName, clusterID, downstream)
	opt(req.AzureBlobStorageInfo)
	return req
}

func genCreateAzureBlobStorageChangefeedReq(
	name, instanceType, dbName, clusterID string, downstream *AzureBlobStorageEntity,
) *connector_service.CreateConnectorBody {
	changefeedSpec := &connector_service.CreateConnectorBody{}
	changefeedSpec.Name = name
	connectorType := model.DataflowConnectorTypeAZUREBLOBSTORAGE
	encodingMethod := model.DataflowBinaryEncodingMethodBASE64
	s3Protocol := model.DataflowS3ProtocolS3PROTOCOLCSV
	changefeedSpec.ConnectorType = &connectorType
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.AzureBlobStorageInfo = &model.DataflowAzureBlobStorageInfo{}
	containerUri := downstream.ContainerURI
	if strings.HasSuffix(containerUri, "/") {
		containerUri = containerUri[:len(containerUri)-1]
	}
	changefeedSpec.AzureBlobStorageInfo.Endpoint = &model.DataflowAzureBlobStorageEndpoint{
		BlobURI:  fmt.Sprintf("%s/%s", containerUri, clusterID),
		SasToken: downstream.SASToken,
	}
	changefeedSpec.AzureBlobStorageInfo.DataFormat = &model.DataflowS3Dataformat{
		BinaryEncodingMethod: &encodingMethod,
		DateSeparator:        "day",
		Terminator:           "\\n",
		Delimiter:            ",",
		Quote:                "\"",
		Null:                 "\\N",
		Protocol:             &s3Protocol,
	}
	changefeedSpec.AzureBlobStorageInfo.StorageFlushConfig = &model.DataflowStorageFlushConfig{
		FlushInterval: "60s",
		FileSize:      "67108864",
	}
	changefeedSpec.AzureBlobStorageInfo.TableConfig = &model.DataflowTableConfig{
		FilterRule: []string{fmt.Sprintf("%s.*", dbName)},
		EventFilterRule: []*model.DataflowEventFilterRule{
			{
				Matcher:     []string{fmt.Sprintf("%s.*", dbName)},
				IgnoreEvent: []string{"delete"},
			},
		},
	}
	startMode := model.DataflowStartModeFROMNOW
	changefeedSpec.AzureBlobStorageInfo.StartPosition = &model.DataflowStartPosition{
		Mode: &startMode,
	}
	return changefeedSpec
}

func compareAzureBlobStorageChangefeedKernelConfig(azBlobInfo *model.DataflowAzureBlobStorageInfo, kernelConfig *ticdc_v1.Changefeed) error {
	sinkUri, err := url.Parse(kernelConfig.SinkUri)
	if err != nil {
		return fmt.Errorf("parse sink uri error: %w", err)
	}
	storageAccount, sasToken := sinkUri.Query().Get("account-name"), sinkUri.Query().Get("sas-token")

	blobUri := fmt.Sprintf("https://%s.blob.core.windows.net/%s%s", storageAccount, sinkUri.Host, sinkUri.Path)
	if strings.HasSuffix(blobUri, "/") {
		blobUri = blobUri[:len(blobUri)-1]
	}
	if blobUri != azBlobInfo.Endpoint.BlobURI {
		return fmt.Errorf("blob uri mismatch, kernel: %s, dataflow: %s", blobUri, azBlobInfo.Endpoint.BlobURI)
	}
	if sasToken != azBlobInfo.Endpoint.SasToken {
		return fmt.Errorf("sas token mismatch")
	}

	return compareCloudStorageCommonKernelConfig(
		azBlobInfo.TableConfig, azBlobInfo.DataFormat, azBlobInfo.StorageFlushConfig, sinkUri, kernelConfig)
}

func editAzureBlobStorageInfoNormally(azBlobInfo *model.DataflowAzureBlobStorageInfo) {
	azBlobInfo.TableConfig.FilterRule = append(azBlobInfo.TableConfig.FilterRule, "test.*")
	azBlobInfo.TableConfig.EventFilterRule = append(azBlobInfo.TableConfig.EventFilterRule, &model.DataflowEventFilterRule{
		Matcher:     []string{"test.*"},
		IgnoreEvent: []string{"update", "delete"},
	})
}