package common

import (
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
	"github.com/tidbcloud/dataflow-service-e2e/internal/misc"
	. "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/cdc/model"
)

type ChangfeedValidator struct {
	lgr  log.SimpleLogger
	cdcw *cdcapi.CDCClientWrapper
}

func NewChangfeedValidator(lgr log.SimpleLogger) *ChangfeedValidator {
	return &ChangfeedValidator{lgr: lgr, cdcw: cdcapi.NewCDCClientWrapper(lgr)}
}

func (c *ChangfeedValidator) VerifyTables(tidbCluster *TiDBPremiumEntity, tableFilter []string, allowTables, denyTables []string) error {
	resp, err := c.cdcw.DescribeSchemaTable(tidbCluster, tableFilter)
	if err != nil {
		return err
	}
	sort.Strings(allowTables)
	sort.Strings(denyTables)
	sort.Strings(resp.Payload.ReplicateAllowSchemaTable)
	sort.Strings(resp.Payload.ReplicateDenySchemaTable)
	if !misc.AreStringArrayEqual(allowTables, resp.Payload.ReplicateAllowSchemaTable) {
		return fmt.Errorf("allow tables are not equal, expected: %v, actual: %v", allowTables, resp.Payload.ReplicateAllowSchemaTable)
	}
	if !misc.AreStringArrayEqual(denyTables, resp.Payload.ReplicateDenySchemaTable) {
		return fmt.Errorf("deny tables are not equal, expected: %v, actual: %v", allowTables, resp.Payload.ReplicateAllowSchemaTable)
	}
	return nil
}

func (c *ChangfeedValidator) CreateDummyChangefeedWithPublicKafka(
	name, instanceType string,
	tidbCluster *TiDBPremiumEntity,
	downstream *KafkaEntity,
	tableFilter []string,
	topic string,
) (changefeedId string, err error) {
	changefeedSpec := GenDefaultKafkaChangefeedSpec()
	changefeedSpec.Name = name
	changefeedSpec.ChangefeedSpec = instanceType
	changefeedSpec.NetworkInfo.NetworkType = model.NewConsoleapiNetworkType(model.ConsoleapiNetworkTypeNetworkTypePublic)
	changefeedSpec.KafkaInfo.Broker.BrokerEndpoints = downstream.BootstrapEndpoints
	changefeedSpec.KafkaInfo.Authentication = kafkaAuthentication(downstream)
	changefeedSpec.KafkaInfo.TableConfig.FilterRule = tableFilter
	changefeedSpec.KafkaInfo.TopicPartitionConfig.DefaultTopic = topic
	resp, err := c.cdcw.CreateChangefeed(tidbCluster, changefeedSpec)
	if err != nil {
		return "", err
	}
	return *resp.Payload.ChangefeedID, nil
}

func kafkaAuthentication(downstream *KafkaEntity) *model.ConsoleapiAuthentication {
	return &model.ConsoleapiAuthentication{
		AuthType: model.ConsoleapiKafkaAuthType(downstream.AuthType).Pointer(),
		UserName: downstream.User,
		Password: downstream.Password,
	}
}

func (c *ChangfeedValidator) VerifyCheckpointTsAdvanceNormal(period time.Duration, tidbCluster *TiDBPremiumEntity, changfeedIds ...string) error {
	interval := time.Second * 30
	times := period/interval + 1

	// collect data
	changefeedsTSOs := map[string][]string{}
	for _, id := range changfeedIds {
		changefeedsTSOs[id] = []string{}
	}
	var errs []error
	for i := 0; i < int(times); i++ {
		resp, err := c.cdcw.ListChangefeed(tidbCluster)
		errs = append(errs, err)
		if err != nil {
			continue
		}
		for _, it := range resp.Payload.Items {
			tsos, ok := changefeedsTSOs[*it.ChangefeedID]
			if ok {
				changefeedsTSOs[*it.ChangefeedID] = append(tsos, it.CheckpointTso)
			}
		}
		time.Sleep(interval)
	}

	// handling data
	// 1. if all error, return error
	if pie.All(errs, func(value error) bool { return value != nil }) {
		c.lgr.Warn("collect checkpointTS failed, all http calls failed, cluster(id=%s, name=%s), errors: %+v",
			tidbCluster.ClusterID, tidbCluster.DisplayName, errs)
		return fmt.Errorf("collect checkpointTS failed, all http calls failed")
	}

	// 2. not all error, log warning, verify tso increasing
	c.lgr.Info("strings checkpointTS of changefeed: %+v", changefeedsTSOs)
	changefeedsTSOsUint64 := map[string][]uint64{}
	for id, tsos := range changefeedsTSOs {
		if _, ok := changefeedsTSOsUint64[id]; !ok {
			changefeedsTSOsUint64[id] = []uint64{}
		}
		for _, v := range tsos {
			num, err := strconv.ParseUint(v, 10, 64)
			if err != nil {
				panic(fmt.Sprintf("parse checkpointTS failed, changefeedId: %s, checkpointTS: %s", id, v))
			}
			// workaround: sometime kernel return zero value
			if num == 0 {
				continue
			}
			changefeedsTSOsUint64[id] = append(changefeedsTSOsUint64[id], num)
		}
	}
	c.lgr.Info("uint64 checkpointTS of changefeed: %+v", changefeedsTSOsUint64)
	// verify increasing
	var checkpointTSNotMonotonicallyIncreased []string
	for id, tsos := range changefeedsTSOsUint64 {
		if len(tsos) < 2 || tsos[0] >= tsos[len(tsos)-1] {
			checkpointTSNotMonotonicallyIncreased = append(checkpointTSNotMonotonicallyIncreased, id)
			continue
		}
		pre := tsos[0]
		for i := 1; i < len(tsos); i++ {
			if tsos[i] < pre {
				checkpointTSNotMonotonicallyIncreased = append(checkpointTSNotMonotonicallyIncreased, id)
				break
			}
			pre = tsos[i]
		}
	}
	if len(checkpointTSNotMonotonicallyIncreased) > 0 {
		for _, id := range checkpointTSNotMonotonicallyIncreased {
			c.lgr.Error("checkpointTS of changefeed(id=%s) are not monotonically increased, strings of checkpointTSs: %v, uinte64 of checkpointTSs: %v",
				id, changefeedsTSOs[id], changefeedsTSOsUint64[id])
		}
		return fmt.Errorf("checkpointTS of changefeeds(ids=%s) are not monotonically increased", strings.Join(checkpointTSNotMonotonicallyIncreased, ","))
	}

	return nil
}

func (c *ChangfeedValidator) VerifyRealtimeCheckpointNotBehind(
	tidbCluster *TiDBPremiumEntity,
	changefeedID string,
) error {
	cached, err := c.cdcw.GetChangefeedWithInvalidateCache(tidbCluster, changefeedID, false)
	if err != nil {
		return err
	}
	realtime, err := c.cdcw.GetChangefeedWithInvalidateCache(tidbCluster, changefeedID, true)
	if err != nil {
		return err
	}
	cachedTSO, err := strconv.ParseUint(cached.Payload.CheckpointTso, 10, 64)
	if err != nil {
		return fmt.Errorf("parse cached checkpoint TSO %q: %w", cached.Payload.CheckpointTso, err)
	}
	realtimeTSO, err := strconv.ParseUint(realtime.Payload.CheckpointTso, 10, 64)
	if err != nil {
		return fmt.Errorf("parse realtime checkpoint TSO %q: %w", realtime.Payload.CheckpointTso, err)
	}
	if realtimeTSO < cachedTSO {
		return fmt.Errorf("realtime checkpoint TSO %d is behind cached checkpoint TSO %d", realtimeTSO, cachedTSO)
	}
	return nil
}

func (c *ChangfeedValidator) VerifyCCUsAvailable(tidbCluster *TiDBPremiumEntity) error {
	ccus, err := c.cdcw.ListChangefeedCapacityUnits(tidbCluster)
	if err != nil {
		return err
	}
	if len(ccus) == 0 {
		return fmt.Errorf("premium OpenAPI returned no changefeed CCU options")
	}
	return nil
}

func (c *ChangfeedValidator) AreChangefeedsInList(tidbCluster *TiDBPremiumEntity, changefeedIds ...string) error {
	resp, err := c.cdcw.ListChangefeed(tidbCluster)
	if err != nil {
		return err
	}
	changefeeds := map[string]*model.ConsoleapiChangefeed{}
	for _, it := range resp.Payload.Items {
		changefeeds[*it.ChangefeedID] = it
	}

	var notFound []string
	for _, id := range changefeedIds {
		if _, ok := changefeeds[id]; !ok {
			notFound = append(notFound, id)
		}
	}

	if len(notFound) > 0 {
		return fmt.Errorf("changefeeds(ids=%s) not found in cluster %s(id=%s)", strings.Join(notFound, ","), tidbCluster.DisplayName, tidbCluster.ClusterID)
	}

	c.lgr.Info("changefeeds(ids=%s) are found in cluster %s(id=%s)", strings.Join(changefeedIds, ","), tidbCluster.DisplayName, tidbCluster.ClusterID)

	return nil
}

func (c *ChangfeedValidator) AllChangefeedsNotInListByNames(tidbCluster *TiDBPremiumEntity, changefeedNames ...string) error {
	resp, err := c.cdcw.ListChangefeed(tidbCluster)
	if err != nil {
		return err
	}
	changefeedSet := map[string]bool{}
	for _, name := range changefeedNames {
		changefeedSet[name] = true
	}

	var inList []string
	for _, it := range resp.Payload.Items {
		if changefeedSet[it.Name] {
			inList = append(inList, it.Name)
		}
	}

	if len(inList) > len(changefeedNames) {
		c.lgr.Error("there msut be changefeeds have duplicated name,  changefeeds(names=%s), cluster %s(id=%s)", strings.Join(changefeedNames, ","), tidbCluster.DisplayName, tidbCluster.ClusterID)
	}

	if len(inList) > 0 {
		return fmt.Errorf("changefeeds(names=%s) still can be found in cluster %s(id=%s)", strings.Join(inList, ","), tidbCluster.DisplayName, tidbCluster.ClusterID)
	}

	c.lgr.Info("changefeeds(names=%s) can not be found in cluster %s(id=%s)", strings.Join(changefeedNames, ","), tidbCluster.DisplayName, tidbCluster.ClusterID)

	return nil
}

func (c *ChangfeedValidator) AreChangefeedsRunning(tidbCluster *TiDBPremiumEntity, changefeedIds ...string) error {
	resp, err := c.cdcw.ListChangefeed(tidbCluster)
	if err != nil {
		return err
	}
	changefeeds := map[string]*model.ConsoleapiChangefeed{}
	for _, it := range resp.Payload.Items {
		changefeeds[*it.ChangefeedID] = it
	}

	notRunning := map[string]string{}
	for _, id := range changefeedIds {
		cf, ok := changefeeds[id]
		if !ok {
			notRunning[id] = "changefeed not found"
		}
		if *cf.Status != model.ConsoleapiChangefeedStatusRUNNING {
			notRunning[id] = string(*cf.Status)
		}
	}

	if len(notRunning) > 0 {
		return fmt.Errorf("not all changefeeds status are running in cluster %s(id=%s), not running changefeeds' status: %s",
			tidbCluster.DisplayName, tidbCluster.ClusterID, misc.JSON(notRunning))
	}

	c.lgr.Info("all changefeeds(ids=%s) status are running in cluster %s(id=%s)",
		strings.Join(changefeedIds, ","), tidbCluster.DisplayName, tidbCluster.ClusterID)
	return nil
}

func (c *ChangfeedValidator) IsChangefeedScaleDone(tidbCluster *TiDBPremiumEntity, changefeedId string, expectedInstanceType string) error {
	resp, err := c.cdcw.GetChangefeed(tidbCluster, changefeedId)
	if err != nil {
		return err
	}
	if resp.Payload.ChangefeedSpec != expectedInstanceType {
		return fmt.Errorf("instance type of changefeed(%s) hasn't changed to %s", changefeedId, expectedInstanceType)
	}
	if *resp.Payload.Status != model.ConsoleapiChangefeedStatusRUNNING {
		return fmt.Errorf("instance type of changefeed(%s) has alreay changed to %s, but status not running", changefeedId, expectedInstanceType)
	}
	return nil
}

func (c *ChangfeedValidator) DeleteChangefeedByName(tidbCluster *TiDBPremiumEntity, name string) error {
	resp, err := c.cdcw.ListChangefeed(tidbCluster)
	if err != nil {
		return err
	}
	var errs []error
	for _, it := range resp.Payload.Items {
		if it.Name != name {
			continue
		}
		err = c.cdcw.DeleteChangefeedForCleanup(tidbCluster, *it.ChangefeedID)
		if err != nil {
			c.lgr.Warn("delete changefeed failed, id: %s, name: %s, err: %s", *it.ChangefeedID, name, err.Error())
			errs = append(errs, err)
		}
	}
	if len(errs) != 0 {
		return fmt.Errorf("not all changefeeds delete successfully")
	}
	return nil
}

func (c *ChangfeedValidator) IsChangefeedInStatus(tidbCluster *TiDBPremiumEntity, changefeedId string, expectedStatus model.ConsoleapiChangefeedStatus) error {
	resp, err := c.cdcw.GetChangefeed(tidbCluster, changefeedId)
	if err != nil {
		return err
	}
	if *resp.Payload.Status != expectedStatus {
		return fmt.Errorf("expected status: %s, acutal status: %s", expectedStatus, *resp.Payload.Status)
	}

	return nil
}

func (c *ChangfeedValidator) ModifyKafkaChangefeedProtocol(tidbCluster *TiDBPremiumEntity,
	changefeedId string, newProtocol model.ConsoleapiProtocol) error {
	resp, err := c.cdcw.GetChangefeed(tidbCluster, changefeedId)
	if err != nil {
		return err
	}
	modify := &model.ChangefeedServiceEditChangefeedBody{
		ChangefeedType: resp.Payload.ChangefeedType,
		Name:           resp.Payload.Name,
		KafkaInfo:      resp.Payload.Request.KafkaInfo,
	}
	modify.KafkaInfo.DataFormat.Protocol = &newProtocol
	return c.cdcw.EditChangefeed(tidbCluster, changefeedId, modify)
}

func (c *ChangfeedValidator) IsChangefeedUseProtocol(tidbCluster *TiDBPremiumEntity, changefeedId string, expectProtocol model.ConsoleapiProtocol) error {
	resp, err := c.cdcw.GetChangefeed(tidbCluster, changefeedId)
	if err != nil {
		return err
	}
	actualProtocol := *resp.Payload.Request.KafkaInfo.DataFormat.Protocol
	if actualProtocol != expectProtocol {
		return fmt.Errorf("expected protocol: %s, actual protocol: %s", expectProtocol, actualProtocol)
	}
	return nil
}

// TODO
func (c *ChangfeedValidator) IsKernelChangefeedUseProtocol(tidbCluster *TiDBPremiumEntity, changefeedId string, protocol model.ConsoleapiProtocol) error {
	return nil
}

type cdcTopology struct {
	Replica     int
	MachineType string
}

func (c *ChangfeedValidator) getExpectedTiCDCTopology(cp CloudProviderType, totalRCUs string) cdcTopology {
	expectedTopologyMap := map[string]struct {
		Replica int
		CPUCnt  int
	}{

		"2rcu":   {Replica: 1, CPUCnt: 2},
		"4rcu":   {Replica: 1, CPUCnt: 4},
		"6rcu":   {Replica: 1, CPUCnt: 8},
		"8rcu":   {Replica: 1, CPUCnt: 8},
		"16rcu":  {Replica: 2, CPUCnt: 8},
		"24rcu":  {Replica: 3, CPUCnt: 8},
		"32rcu":  {Replica: 4, CPUCnt: 8},
		"40rcu":  {Replica: 5, CPUCnt: 8},
		"64rcu":  {Replica: 4, CPUCnt: 16},
		"128rcu": {Replica: 4, CPUCnt: 32},
		"256rcu": {Replica: 4, CPUCnt: 64},
	}

	topo, ok := expectedTopologyMap[totalRCUs]
	if !ok {
		panic(fmt.Sprintf("can not found abstract topo for %s", totalRCUs))
	}
	c.lgr.Info("the abstract cdc topology for %s is %s", totalRCUs, misc.JSON(topo))
	return cdcTopology{
		Replica: topo.Replica,
	}
}

func (c *ChangfeedValidator) IsTiCDCTopologyAsExpected(cluster *TiDBPremiumEntity, changfeedId string, totalRCUs string) error {
	actual, err := c.cdcw.GetTiCDCClusterByChangefeed(cluster, changfeedId)
	if err != nil {
		return err
	}
	expect := c.getExpectedTiCDCTopology(cluster.CloudProvider, totalRCUs)
	if expect.Replica != int(actual.NodeSpec.Count) {
		c.lgr.Warn("topology is not as expected, expect: %d x %s, actual: %d x ",
			expect.Replica, expect.MachineType, actual.NodeSpec.Count)
		return fmt.Errorf("topology is not as expected, expect: %d x %s, actual: %d x",
			expect.Replica, expect.MachineType, actual.NodeSpec.Count)
	}
	c.lgr.Info("topology is expected, expect: %d x %s, actual: %d x %s",
		expect.Replica, expect.MachineType, actual.NodeSpec.Count)
	return nil
}

func (c *ChangfeedValidator) NoTiCDCClusterFound(cluster *TiDBPremiumEntity) error {
	tiCDCClusters, err := c.cdcw.GetTiCDCClusters(cluster)
	if err != nil {
		return err
	}
	if len(tiCDCClusters) != 0 {
		c.lgr.Warn("there are still %d TiCDC clusters in cluster %s(id=%s), TiCDC clusters: %s",
			len(tiCDCClusters), cluster.DisplayName, cluster.ClusterID, misc.JSON(tiCDCClusters))
		return fmt.Errorf("there are still %d TiCDC clusters in cluster %s(id=%s)", len(tiCDCClusters), cluster.DisplayName, cluster.ClusterID)
	}
	c.lgr.Info("there are no TiCDC clusters in cluster %s(id=%s)", cluster.DisplayName, cluster.ClusterID)
	return nil
}