package changefeed

import (
	"fmt"
	"strings"
	"time"

	"github.com/aws/smithy-go/ptr"
	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/premium/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/premium/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/consoleapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/premium/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 {
	return map[string]DependencyRequirement{
		"premium": &TiDBPremiumRequirement{
			Dedicated: false,
			TiDBPremiumDimensions: TiDBPremiumDimensions{
				CloudProvider: k.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
		"kafka": &KafkaRequirement{
			KafkaDimensions: KafkaDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePrivateLink,
				AuthType:    DataflowKafkaAuthTypeDISABLE,
				// TLSEnabled:     true,
				IsCustomDomain: true,
				CloudProvider:  k.cloudProvider,
				Region:         GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
	}
}

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

	// declare case related vars
	tidbCluster := k.TiDBPremium("premium")
	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.ConsoleapiSinkAdvertisedListener
		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)
		}, 6, 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.ChangefeedID
			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 *TiDBPremiumEntity,
	downstream *KafkaEntity,
	allAzs []string,
	advertisedListener *model.ConsoleapiSinkAdvertisedListener,
) (string, error) {
	var azs []string
	if downstream.IsCustomDomain {
		for az, _ := range downstream.DnsRecords {
			azs = append(azs, az)
		}
	} else {
		azs = allAzs
	}
	sinkEndpoint := &model.ConsoleapiSinkEndpoint{
		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(tidbCluster, sinkEndpoint)
}

func (k *kafkaPrivateLinkCase) privateLinkAvailable(tidbCluster *TiDBPremiumEntity, privateLinkID string) error {
	sinkEndpoint, err := k.sinkEndpointApiWrapper.GetSinkEndpoint(tidbCluster, privateLinkID)
	if err != nil {
		return err
	}
	if *sinkEndpoint.Status != model.ConsoleapiSinkEndpointStatusSINKENDPOINTSTATUSAVAILABLE {
		return fmt.Errorf("private link %s is not available, status %s", privateLinkID, *sinkEndpoint.Status)
	}
	return nil
}

func (k *kafkaPrivateLinkCase) generateAdvertisedPatternGroup(
	tidbCluster *TiDBPremiumEntity, downstream *KafkaEntity, azs []string,
) (*model.ConsoleapiSinkAdvertisedListener, error) {
	if downstream.IsCustomDomain {
		var records []*model.SinkAdvertisedListenerRecord
		for az, record := range downstream.DnsRecords {
			records = append(records, &model.SinkAdvertisedListenerRecord{
				SubdomainName: record,
				Zone:          az,
			})
		}
		return &model.ConsoleapiSinkAdvertisedListener{
			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), downstream.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 *TiDBPremiumEntity) error {
	resp, err := k.sinkEndpointApiWrapper.GetSinkEndpointAuthInfo(tidbCluster)
	if err != nil {
		return err
	}
	if len(resp.CloudAccountID) == 0 {
		return fmt.Errorf("no CloudAccountID")
	}
	if tidbCluster.CloudProvider == AWS && len(resp.AwsPrincipalID) == 0 {
		return fmt.Errorf("no AwsPrincipalID")
	}
	return nil
}

func (k *kafkaPrivateLinkCase) getPrivateLinkAvailableZones(tidbCluster *TiDBPremiumEntity) ([]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,
) *model.ChangefeedServiceCreateChangefeedBody {
	changefeedSpec := &model.ChangefeedServiceCreateChangefeedBody{
		ChangefeedType: model.ConsoleapiChangefeedTypeKAFKA.Pointer(),
	}
	changefeedSpec.Name = name
	changefeedSpec.ChangefeedSpec = instanceType
	changefeedSpec.NetworkInfo = &model.ConsoleapiNetworkInfo{
		NetworkType:    model.ConsoleapiNetworkTypeNetworkTypePrivateLink.Pointer(),
		Ports:          downstream.Ports,
		SinkEndpointID: privateLinkID,
	}
	changefeedSpec.KafkaInfo = &model.ConsoleapiKafkaInfo{
		Authentication: &model.ConsoleapiAuthentication{
			AuthType: model.ConsoleapiKafkaAuthTypeDISABLE.Pointer(),
		},
		Broker: &model.ConsoleapiBroker{
			Compression:  model.ConsoleapiCompressionTypeGZIP.Pointer(),
			KafkaVersion: model.ConsoleapiKafkaVersionKAFKAVERSION3XX.Pointer(),
			// TLSEnable:    downstream.TLSEnabled,
		},
		ColumnSelectors: []*model.ConsoleapiColumnSelector{
			{
				Columns: []string{"abc"},
				Matcher: []string{
					"*.*"},
			},
		},
		DataFormat: &model.ConsoleapiDataFormat{
			Protocol: model.ConsoleapiProtocolCANALJSON.Pointer(),
		},
		StartPosition: &model.ConsoleapiStartPosition{
			Mode: model.ConsoleapiStartModeFROMNOW.Pointer(),
		},
		TableConfig: &model.ConsoleapiTableConfig{
			EventFilterRule: []*model.ConsoleapiEventFilterRule{
				{
					IgnoreEvent: []string{"insert", "update"},
					Matcher:     []string{"e.f", "h.j"},
				},
			},
			FilterRule: []string{"test.a", "test.b"},
			Mode:       model.ConsoleapiTableModeFORCESYNC.Pointer(),
		},
		TopicPartitionConfig: &model.ConsoleapiTopicPartitionConfig{
			DefaultTopic: "test2",
			DispatchType: model.ConsoleapiDispatchTypeDISPATCHTYPEONETOPIC.Pointer(),
			PartitionDispatchers: []*model.ConsoleapiPartitionDispatcherConfig{
				{
					Columns: []string{
						"abc",
					},
					Matcher:       []string{"a.b"},
					PartitionType: model.ConsoleapiPartitionDispatcherCOLUMNS.Pointer(),
				},
			},
			PartitionNum:      1,
			ReplicationFactor: 1,
			Separator:         "abc",
			TopicPrefix:       "efg",
			TopicSuffix:       "hij",
		},
	}
	return changefeedSpec
}