package changefeed

import (
	"fmt"
	"time"

	"github.com/aws/smithy-go/ptr"
	. "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/tidbcloud/dedicated/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi"
	"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/client/sink_endpoint_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
)

type mysqlPrivateLinkCase struct {
	BaseCase
	cdcApiWrapper          *cdcapi.CDCClientWrapper
	sinkEndpointApiWrapper *consoleapi.SinkEndpointClientWrapper
	cfValidator            *common.ChangfeedValidator
	cloudProvider          CloudProviderType
}

func NewMySQLPrivateLinkCase(cloudProvider CloudProviderType) E2ECase {
	return &mysqlPrivateLinkCase{cloudProvider: cloudProvider}
}

func (m *mysqlPrivateLinkCase) Name() string {
	return fmt.Sprintf("%s%s", m.cloudProvider, "MySQLPrivateLink")
}

func (m *mysqlPrivateLinkCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
		"mysql": &MySQLRequirement{
			MySQLDimensions: MySQLDimensions{
				NetworkType:   DataflowNetworkTypeNetworkTypePrivateLink,
				CloudProvider: m.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
	}
}

func (m *mysqlPrivateLinkCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	downstream := m.MySQL("mysql")
	changefeedName := m.Name()
	runningCheckRetry := runningCheckRetryForCloud(m.cloudProvider)

	m.cdcApiWrapper = cdcapi.NewCDCClientWrapper(m.Logger())
	m.sinkEndpointApiWrapper = consoleapi.NewSinkEndpointClientWrapper(m.Logger())
	m.cfValidator = common.NewChangfeedValidator(m.Logger())

	var (
		availabilityZones []string
		changefeedID      string
		sinkEndpointID    string
	)

	defer func() {
		m.
			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, 10*time.Second).
			RetryDo("delete sink endpoint", func() error {
				return m.sinkEndpointApiWrapper.DeleteSinkEndpoint(tidbCluster, sinkEndpointID)
			}, 3, time.Minute)
	}()

	m.
		RetryDo(fmt.Sprintf("list sink availability zones %s", changefeedName), func() error {
			var err error
			availabilityZones, err = m.getPrivateLinkAvailableZones(tidbCluster)
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("create sink endpoint %s", changefeedName), func() error {
			var err error
			sinkEndpointID, err = m.createPrivateLink(tidbCluster, downstream, availabilityZones)
			return err
		}, 3, time.Minute).
		Expect(fmt.Sprintf("wait sink endpoint available %s", changefeedName), func() error {
			return m.privateLinkAvailable(tidbCluster, sinkEndpointID)
		}, 20, time.Minute).
		Expect(fmt.Sprintf("test connection with private link %s", changefeedName), func() error {
			return m.cdcApiWrapper.TestConnectionWithPrivateLink(tidbCluster, downstream, sinkEndpointID)
		}, 20, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() error {
			spec := m.genCreateMySQLPrivateLinkSpec(changefeedName, "2rcu", downstream, sinkEndpointID)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, spec)
			if err != nil {
				return err
			}
			changefeedID = *resp.Payload.ConnectorID
			return nil
		}, 6, time.Minute).
		Expect("changefeed in list", func() error {
			return m.cfValidator.AreChangefeedsInList(tidbCluster, changefeedID)
		}, 3, 10*time.Second).
		Expect("changefeed is running", func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, runningCheckRetry, time.Minute)
}

func (m *mysqlPrivateLinkCase) getPrivateLinkAvailableZones(tidbCluster *TiDBDedicatedEntity) ([]string, error) {
	zones, err := m.sinkEndpointApiWrapper.ListSinkAvailabilityZones(tidbCluster)
	if err != nil {
		return nil, err
	}
	if len(zones) == 0 {
		return nil, fmt.Errorf("no available zones")
	}
	return zones, nil
}

func (m *mysqlPrivateLinkCase) createPrivateLink(
	tidbCluster *TiDBDedicatedEntity,
	downstream *MySQLEntity,
	availabilityZones []string,
) (string, error) {
	params := &sink_endpoint_service.CreateSinkEndpointParams{
		OrgID:     tidbCluster.OrgID,
		ProjectID: tidbCluster.ProjectID,
		SinkEndpoint: &model.DataflowSinkEndpoint{
			EndpointName:        ptr.String(fmt.Sprintf("%s-mysql-pl-e2e", tidbCluster.ClusterID)),
			EndpointServiceName: ptr.String(downstream.PrivateLinkServiceName),
			Provider:            ptr.String(string(tidbCluster.CloudProvider)),
			Region:              ptr.String(tidbCluster.Region),
			EndpointZones:       availabilityZones,
		},
	}
	return m.sinkEndpointApiWrapper.CreateSinkEndpoint(params)
}

func (m *mysqlPrivateLinkCase) privateLinkAvailable(tidbCluster *TiDBDedicatedEntity, sinkEndpointID string) error {
	params := &sink_endpoint_service.GetSinkEndpointParams{
		OrgID:          tidbCluster.OrgID,
		ProjectID:      tidbCluster.ProjectID,
		SinkEndpointID: sinkEndpointID,
	}
	sinkEndpoint, err := m.sinkEndpointApiWrapper.GetSinkEndpoint(params)
	if err != nil {
		return err
	}
	if *sinkEndpoint.Status != model.DataflowSinkEndpointStatusSINKENDPOINTSTATUSAVAILABLE {
		return fmt.Errorf("private link %s is not available, status %s", sinkEndpointID, *sinkEndpoint.Status)
	}
	return nil
}

func (m *mysqlPrivateLinkCase) genCreateMySQLPrivateLinkSpec(
	name, instanceType string,
	downstream *MySQLEntity,
	sinkEndpointID string,
) connector_service.CreateConnectorBody {
	spec := connector_service.CreateConnectorBody{
		ConnectorType:          model.NewDataflowConnectorType(model.DataflowConnectorTypeMYSQL),
		Name:                   name,
		ChangefeedInstanceType: instanceType,
		NetworkInfo: &model.DataflowNetworkInfo{
			NetworkType:    model.NewDataflowNetworkType(model.DataflowNetworkTypeNetworkTypePrivateLink),
			Ports:          downstream.Ports,
			SinkEndpointID: sinkEndpointID,
		},
		MysqlInfo: &model.DataflowMySQLInfo{
			Connection: &model.DataflowConnection{
				Username: downstream.User,
				Password: downstream.Password,
			},
			TableConfig: &model.DataflowTableConfig{
				FilterRule:      []string{fmt.Sprintf("%s.*", name)},
				Mode:            model.NewDataflowTableMode(model.DataflowTableModeFORCESYNC),
				EventFilterRule: []*model.DataflowEventFilterRule{},
				CaseSensitive:   false,
			},
			StartPosition: &model.DataflowStartPosition{
				Mode: model.NewDataflowStartMode(model.DataflowStartModeFROMNOW),
			},
		},
	}
	return spec
}