package changefeed

import (
	"fmt"
	"sort"
	"strings"
	"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 kafkaPrivateLinkCase struct {
	BaseCase
	cdcApiWrapper          *cdcapi.CDCClientWrapper
	sinkEndpointApiWrapper *consoleapi.SinkEndpointClientWrapper
	cfValidator            *common.ChangfeedValidator
	cloudProvider          CloudProviderType
}

func NewKafkaPrivateLinkCase(cloudProvider CloudProviderType) E2ECase {
	return &kafkaPrivateLinkCase{
		cloudProvider: cloudProvider,
	}
}

func (k *kafkaPrivateLinkCase) Name() string {
	return fmt.Sprintf("%s%s", k.cloudProvider, "KafkaPrivateLink")
}

func (k *kafkaPrivateLinkCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	tlsEnabled := k.cloudProvider == AZURE
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: k.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
		"kafka": &KafkaRequirement{
			KafkaDimensions: KafkaDimensions{
				NetworkType:    DataflowNetworkTypeNetworkTypePrivateLink,
				AuthType:       DataflowKafkaAuthTypeDISABLE,
				TLSEnabled:     tlsEnabled,
				CloudProvider:  k.cloudProvider,
				Region:         GlobalConfig.DefaultRegionOf(k.cloudProvider),
				IsCustomDomain: true,
			},
		},
	}
}

func (k *kafkaPrivateLinkCase) Run() {
	k.Logger().Info("run %s", k.Name())

	// declare case related vars
	tidbCluster := k.TiDBDedicated("tidb")
	downstream := k.Kafka("kafka")
	changefeedName := k.Name()
	changefeedId := ""
	// init reusable api client and validator
	cdcApiWrapper := cdcapi.NewCDCClientWrapper(k.Logger())
	k.cdcApiWrapper = cdcApiWrapper
	sinkEndpointApiWrapper := consoleapi.NewSinkEndpointClientWrapper(k.Logger())
	k.sinkEndpointApiWrapper = sinkEndpointApiWrapper
	k.cfValidator = common.NewChangfeedValidator(k.Logger())

	var (
		azs                []string
		advertisedListener *model.DataflowSinkAdvertisedListener
		privateLinkID      string
	)

	// cleanup changefeeds if exists
	defer func() {
		k.
			RetryDo("delete changefeeds", func() error {
				return k.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedName)
			}, 3, time.Minute).
			Expect("changefeeds not in list", func() error {
				return k.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedName)
			}, 3, 10*time.Second).
			RetryDo("delete private link", func() error {
				return k.sinkEndpointApiWrapper.DeleteSinkEndpoint(tidbCluster, privateLinkID)
			}, 3, time.Minute)
	}()

	// create changefeed
	k.RetryDo(fmt.Sprintf("gen azs  %s", changefeedName), func() (err error) {
		azs, err = k.getPrivateLinkAvailableZones(tidbCluster)
		return err
	}, 3, time.Minute).
		RetryDo(fmt.Sprintf("get auth info without kafka id  %s", changefeedName), func() (err error) {
			return k.getPrivateLinkAuthInfo(tidbCluster)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("generateAdvertisedPatternGroup  %s", changefeedName), func() (err error) {
			advertisedListener, err = k.generateAdvertisedPatternGroup(tidbCluster, downstream, azs)
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("create private link  %s", changefeedName), func() (err error) {
			privateLinkID, err = k.createPrivateLink(tidbCluster, downstream, azs, advertisedListener)
			return err
		}, 3, time.Minute).
		Expect(fmt.Sprintf("get private link  %s", changefeedName), func() (err error) {
			return k.privateLinkAvailable(tidbCluster, privateLinkID)
		}, 20, 1*time.Minute).
		Expect(fmt.Sprintf("test connection  %s", changefeedName), func() (err error) {
			return k.cdcApiWrapper.TestConnectionWithPrivateLink(tidbCluster, downstream, privateLinkID)
		}, 20, 1*time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			changefeedSpec := k.genAddChangefeedSpec(changefeedName, "2rcu", downstream, privateLinkID)
			resp, err := k.cdcApiWrapper.CreateChangefeed(tidbCluster, changefeedSpec)
			if err != nil {
				return err
			}
			changefeedId = *resp.Payload.ConnectorID
			return nil
		}, 6, time.Minute).
		Expect("changefeeds in list", func() error {
			return k.cfValidator.AreChangefeedsInList(tidbCluster, changefeedId)
		}, 3, 10*time.Second).
		Expect("changefeeeds are running", func() error {
			return k.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedId)
		}, 20, time.Minute)
}

func (k *kafkaPrivateLinkCase) createPrivateLink(
	tidbCluster *TiDBDedicatedEntity,
	downstream *KafkaEntity,
	allAzs []string,
	advertisedListener *model.DataflowSinkAdvertisedListener,
) (string, error) {
	azs := allAzs
	if downstream.IsCustomDomain {
		azs = make([]string, 0, len(downstream.DnsRecords))
		for az := range downstream.DnsRecords {
			azs = append(azs, az)
		}
		sort.Strings(azs)
	}
	params := &sink_endpoint_service.CreateSinkEndpointParams{
		OrgID:     tidbCluster.OrgID,
		ProjectID: tidbCluster.ProjectID,
		SinkEndpoint: &model.DataflowSinkEndpoint{
			EndpointName:        ptr.String(fmt.Sprintf("%s-kafka-pl-e2e", tidbCluster.ClusterID)),
			EndpointServiceName: ptr.String(downstream.PrivateLinkServiceName),
			Provider:            ptr.String(string(tidbCluster.CloudProvider)),
			Region:              ptr.String(tidbCluster.Region),
			EndpointZones:       azs,
			AdvertisedListener:  advertisedListener,
		},
	}
	return k.sinkEndpointApiWrapper.CreateSinkEndpoint(params)
}

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

func (k *kafkaPrivateLinkCase) generateAdvertisedPatternGroup(
	tidbCluster *TiDBDedicatedEntity, downstream *KafkaEntity, azs []string,
) (*model.DataflowSinkAdvertisedListener, error) {
	if downstream.IsCustomDomain {
		zones := make([]string, 0, len(downstream.DnsRecords))
		for zone := range downstream.DnsRecords {
			zones = append(zones, zone)
		}
		sort.Strings(zones)

		records := make([]*model.SinkAdvertisedListenerRecord, 0, len(zones))
		for _, zone := range zones {
			records = append(records, &model.SinkAdvertisedListenerRecord{
				SubdomainName: downstream.DnsRecords[zone],
				Zone:          zone,
			})
		}
		return &model.DataflowSinkAdvertisedListener{
			DomainType:     model.SinkAdvertisedListenerDomainTypeCUSTOM.Pointer(),
			Records:        records,
			RootDomainName: &downstream.RootDomainName,
		}, nil
	}

	advertisedListener, err := k.sinkEndpointApiWrapper.GenerateSinkAdvertisedListener(
		tidbCluster, downstream.KafkaID, downstream.PrivateLinkServiceName, azs)
	if err != nil {
		return nil, err
	}

	if *advertisedListener.DomainType != model.SinkAdvertisedListenerDomainTypeGENERATED {
		return nil, fmt.Errorf("invalid domain type: %s", *advertisedListener.DomainType)
	}
	expectedRootDomainName := fmt.Sprintf("%s.%s.%s.%s.tidbcloud.com",
		downstream.KafkaID, tidbCluster.Region, string(tidbCluster.CloudProvider), tidbCluster.ProjectID)
	if !strings.HasPrefix(*advertisedListener.RootDomainName, expectedRootDomainName) {
		return nil, fmt.Errorf("invalid root domain name: %s", *advertisedListener.RootDomainName)
	}

	isValidZone := func(zone string) bool {
		for _, az := range azs {
			if az == zone {
				return true
			}
		}
		return false
	}
	if tidbCluster.CloudProvider == AWS {
		if len(advertisedListener.Records) != len(azs) {
			return nil, fmt.Errorf("invalid records length for aws: %d", len(advertisedListener.Records))
		}
		for _, record := range advertisedListener.Records {
			if !isValidZone(record.Zone) {
				return nil, fmt.Errorf("invalid zone in records for aws: %s", record.Zone)
			}
			expectedRecord := fmt.Sprintf("%s.%s", record.Zone, *advertisedListener.RootDomainName)
			if record.SubdomainName != expectedRecord {
				return nil, fmt.Errorf("invalid sub domain name in records for aws: %s", record.SubdomainName)
			}
		}
	} else {
		if len(advertisedListener.Records) != 1 {
			return nil, fmt.Errorf("invalid records length for non-aws: %d", len(advertisedListener.Records))
		}
		record := advertisedListener.Records[0]
		if record.Zone != "all zones" {
			return nil, fmt.Errorf("invalid zone in records for non-aws: %s", record.Zone)
		}
		if record.SubdomainName != *advertisedListener.RootDomainName {
			return nil, fmt.Errorf("invalid sub domain name in records for non-aws: %s", record.SubdomainName)
		}
	}
	return advertisedListener, nil
}

func (k *kafkaPrivateLinkCase) getPrivateLinkAuthInfo(tidbCluster *TiDBDedicatedEntity) error {
	resp, err := k.sinkEndpointApiWrapper.GetSinkEndpointAuthInfo(tidbCluster)
	if err != nil {
		return err
	}
	if len(resp.CloudAccountID) == 0 {
		return fmt.Errorf("no CloudAccountID")
	}
	if len(resp.Network) == 0 {
		return fmt.Errorf("no Network field")
	}
	if tidbCluster.CloudProvider == AWS && len(resp.AwsPrincipalID) == 0 {
		return fmt.Errorf("no AwsPrincipalID")
	}
	return nil
}

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

func (k *kafkaPrivateLinkCase) genAddChangefeedSpec(
	name, instanceType string,
	downstream *KafkaEntity,
	privateLinkID string,
) connector_service.CreateConnectorBody {
	changefeedSpec := connector_service.CreateConnectorBody{
		ConnectorType: model.NewDataflowConnectorType(model.DataflowConnectorTypeKAFKA),
	}
	changefeedSpec.Name = name
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.NetworkInfo = &model.DataflowNetworkInfo{
		NetworkType:    model.NewDataflowNetworkType(model.DataflowNetworkTypeNetworkTypePrivateLink),
		Ports:          downstream.Ports,
		SinkEndpointID: privateLinkID,
	}
	changefeedSpec.KafkaInfo = &model.DataflowKafkaInfo{
		Authentication: &model.DataflowAuthentication{
			AuthType: model.NewDataflowKafkaAuthType(model.DataflowKafkaAuthTypeDISABLE),
		},
		Broker: &model.DataflowBroker{
			Compression:  model.NewDataflowCompressionType(model.DataflowCompressionTypeGZIP),
			KafkaVersion: model.NewDataflowKafkaVersion(model.DataflowKafkaVersionKAFKAVERSION3XX),
			TLSEnable:    downstream.TLSEnabled,
		},
		ColumnSelectors: []*model.DataflowColumnSelector{
			{
				Columns: []string{"abc"},
				Matcher: []string{
					"*.*"},
			},
		},
		DataFormat: &model.DataflowDataFormat{
			Protocol: model.NewDataflowProtocol(model.DataflowProtocolCANALJSON),
		},
		StartPosition: &model.DataflowStartPosition{
			Mode: model.NewDataflowStartMode(model.DataflowStartModeFROMNOW),
		},
		TableConfig: &model.DataflowTableConfig{
			EventFilterRule: []*model.DataflowEventFilterRule{
				{
					IgnoreEvent: []string{"insert", "update"},
					Matcher:     []string{"e.f", "h.j"},
				},
			},
			FilterRule: []string{"test.a", "test.b"},
			Mode:       model.NewDataflowTableMode(model.DataflowTableModeFORCESYNC),
		},
		TopicPartitionConfig: &model.DataflowTopicPartitionConfig{
			DefaultTopic: "test2",
			DispatchType: model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEONETOPIC),
			PartitionDispatchers: []*model.DataflowPartitionDispatcherConfig{
				{
					Columns: []string{
						"abc",
					},
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherCOLUMNS),
				},
			},
			PartitionNum:      1,
			ReplicationFactor: 1,
			Separator:         "abc",
			TopicPrefix:       "efg",
			TopicSuffix:       "hij",
		},
	}
	return changefeedSpec
}