package changefeed

import (
	"fmt"
	"reflect"
	"time"

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

type kafkaChangefeedSmokeCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator

	cloudProvider CloudProviderType
}

func NewKafkaChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &kafkaChangefeedSmokeCase{cloudProvider: cloudProvider}
}

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

func (k *kafkaChangefeedSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: k.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
		"kafka": &KafkaRequirement{
			KafkaDimensions: KafkaDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
				AuthType:    DataflowKafkaAuthTypeSASLPLAIN,
				TLSEnabled:  false,
			},
		},
	}
}

func (k *kafkaChangefeedSmokeCase) Run() {
	// declare case related vars
	tidbCluster := k.TiDBDedicated("tidb")
	downstream := k.Kafka("kafka")
	changefeed1 := fmt.Sprintf("%s-1", k.Name())
	changefeed2 := fmt.Sprintf("%s-2", k.Name())
	var changefeed1Id, changefeed2Id string
	instanceProfile := kafkaSmokeInstanceProfileForCloud(k.cloudProvider)
	changefeed1InstaneType_Initial := instanceProfile.changefeed1Initial
	changefeed1InstaneType_ScaleUp := instanceProfile.changefeed1ScaleUp
	changefeed2InstaneType_Initial := instanceProfile.changefeed2Initial
	changefeed2InstaneType_ScaleDown := instanceProfile.changefeed2ScaleDown
	runningCheckRetry := runningCheckRetryForCloud(k.cloudProvider)
	changefeed1Topic := changefeed1
	changefeed2Topic := changefeed2
	dbName := k.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}

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

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

	// TODO: query changefeed instance types and cal_changefeed_fee
	k.
		Step("list changefeed RCUs non-empty", func() error {
			resp, err := cdcApiWrapper.ListChangefeedRCUs(tidbCluster)
			if err != nil {
				return err
			}
			if resp == nil {
				return fmt.Errorf("list changefeed RCUs returned nil response")
			}
			if resp.Items == nil {
				return fmt.Errorf("list changefeed RCUs returned nil items")
			}
			if len(resp.Items) == 0 {
				return fmt.Errorf("list changefeed RCUs returned empty items")
			}
			return nil
		}).
		RetryDo(fmt.Sprintf("test connection to downstream %s", downstream.BootstrapEndpoints), func() (err error) {
			return 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 cfValidator.VerifyTables(tidbCluster, tableFilter, []string{fmt.Sprintf("%s.%s", dbName, sample_data.SAMPLE_TABLE)}, nil)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeed1), func() (err error) {
			changefeed1Id, err = k.cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed1, changefeed1InstaneType_Initial, tidbCluster, downstream, tableFilter, changefeed1Topic)
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeed2), func() (err error) {
			changefeed2Id, err = k.cfValidator.CreateDummyChangefeedWithPublicKafka(changefeed2, changefeed2InstaneType_Initial, tidbCluster, downstream, tableFilter, changefeed2Topic)
			return err
		}, 3, time.Minute)

	if GlobalConfig.ApiMode == ApiModeOpenAPI {
		k.
			Expect("open api: verify changefeed1 details", func() error {
				return k.verifyOpenAPIChangefeedDetails(openApiCli, changefeed1Id, changefeed1)
			}, 3, 10*time.Second).
			Expect("open api: verify changefeed2 details", func() error {
				return k.verifyOpenAPIChangefeedDetails(openApiCli, changefeed2Id, changefeed2)
			}, 3, 10*time.Second).
			Expect("open api: list includes created changefeeds", func() error {
				return k.verifyOpenAPIListContains(openApiCli, tidbCluster.ClusterID, changefeed1Id, changefeed2Id)
			}, 3, 10*time.Second)
	}

	k.
		Expect("changefeeds in list", func() error {
			return cfValidator.AreChangefeedsInList(tidbCluster, changefeed1Id, changefeed2Id)
		}, 3, 10*time.Second).
		Expect("changefeeeds are running", func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id, changefeed2Id)
		}, runningCheckRetry, time.Minute).
		RetryDo(fmt.Sprintf("start write sample data to database %s", dbName),
			sampleDataWriter.StartWriting, 3, time.Minute).
		Step("verify changefeeds' CheckpointTs advance normal", func() error {
			return cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeed1Id, changefeed2Id)
		}).
		RetryDo(fmt.Sprintf("scale up changefeed %s", changefeed1), func() error {
			return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed1Id, changefeed1InstaneType_ScaleUp)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("scale down changefeed %s", changefeed2), func() error {
			return cdcApiWrapper.ScaleChangefeed(tidbCluster, changefeed2Id, changefeed2InstaneType_ScaleDown)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s scale up done", changefeed1), func() error {
			return cfValidator.IsChangefeedScaleDone(tidbCluster, changefeed1Id, changefeed1InstaneType_ScaleUp)
		}, 20, time.Minute).
		Expect(fmt.Sprintf("changefeed %s scale down done", changefeed2), func() error {
			return cfValidator.IsChangefeedScaleDone(tidbCluster, changefeed2Id, changefeed2InstaneType_ScaleDown)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeed1), func() error {
			return cdcApiWrapper.PauseChangefeed(tidbCluster, changefeed1Id)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is paused", changefeed1), func() error {
			return cfValidator.IsChangefeedInStatus(tidbCluster, changefeed1Id, model.DataflowConnectorStatusPAUSED)
		}, 5, time.Minute)

	if GlobalConfig.ApiMode == ApiModeOpenAPI {
		k.RetryDo("open api: set and verify event filters", func() error {
			return k.setAndVerifyOpenAPISimpleEventFilters(openApiCli, changefeed1Id, dbName)
		}, 3, time.Minute)
	}

	k.
		RetryDo(fmt.Sprintf("edit changefeed %s", changefeed1), func() error {
			return cfValidator.ModifyKafkaChangefeedProtocol(tidbCluster, changefeed1Id, model.DataflowProtocolOPENPROTOCOL)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeed1), func() error {
			return cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeed1Id)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s edit done and running", changefeed1), func() error {
			return cfValidator.IsChangefeedUseProtocol(tidbCluster, changefeed1Id, model.DataflowProtocolOPENPROTOCOL)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s changes dispatched to kernel", changefeed1), func() error {
			return cfValidator.IsKernelChangefeedUseProtocol(tidbCluster, changefeed1Id, model.DataflowProtocolOPENPROTOCOL)
		}, 3, time.Minute).
		Expect(fmt.Sprintf("changefeed %s is runining", changefeed1), func() error {
			return cfValidator.AreChangefeedsRunning(tidbCluster, changefeed1Id)
		}, runningCheckRetry, time.Minute).
		Step(fmt.Sprintf("verify changefeed %s CheckpointTs advance normal", changefeed1), func() error {
			return cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeed1Id)
		})
}

func (k *kafkaChangefeedSmokeCase) verifyOpenAPIChangefeedDetails(
	cli *openapi.ChangefeedOpenApiClientWrapper,
	changefeedID,
	changefeedName string,
) error {
	resp, err := cli.GetChangefeed(changefeedID)
	if err != nil {
		return err
	}
	if resp == nil || resp.Payload == nil {
		return fmt.Errorf("get changefeed returned empty response for %s", changefeedID)
	}
	payload := resp.Payload
	if payload.ID != changefeedID {
		return fmt.Errorf("changefeed id mismatch, expected %s, got %s", changefeedID, payload.ID)
	}
	if payload.Name == nil || *payload.Name != changefeedName {
		return fmt.Errorf("changefeed name mismatch, expected %s, got %v", changefeedName, payload.Name)
	}
	return nil
}

func (k *kafkaChangefeedSmokeCase) verifyOpenAPIListContains(
	cli *openapi.ChangefeedOpenApiClientWrapper,
	clusterID string,
	changefeedIDs ...string,
) error {
	resp, err := cli.ListChangefeed(clusterID)
	if err != nil {
		return err
	}
	if resp == nil || resp.Payload == nil {
		return fmt.Errorf("list changefeeds returned empty response")
	}
	ids := pie.Map(resp.Payload.Changefeeds, func(it *openapi_model.V1beta1Changefeed) string {
		return it.ID
	})
	for _, id := range changefeedIDs {
		if !pie.Contains(ids, id) {
			return fmt.Errorf("changefeed id %s not found in openapi list: %v", id, ids)
		}
	}
	return nil
}

func (k *kafkaChangefeedSmokeCase) setAndVerifyOpenAPISimpleEventFilters(
	cli *openapi.ChangefeedOpenApiClientWrapper,
	changefeedID,
	dbName string,
) error {
	expected := []*openapi_model.V1beta1EventFilter{
		{
			TableMatchers: []string{fmt.Sprintf("%s.*", dbName)},
			IgnoredEvents: []string{"update"},
		},
	}
	if err := cli.SetEventFilters(changefeedID, openapi.SetEventFiltersBody{EventFilters: expected}); err != nil {
		return err
	}
	resp, err := cli.GetEventFilters(changefeedID)
	if err != nil {
		return err
	}
	if resp == nil || resp.Payload == nil {
		return fmt.Errorf("get event filters returned empty response")
	}
	if !isEventFiltersEquals(expected, resp.Payload.EventFilters) {
		return fmt.Errorf("event filters mismatch, expected %s, got %s", misc.JSON(expected), misc.JSON(resp.Payload.EventFilters))
	}
	return nil
}

func isEventFiltersEquals(expected, actual []*openapi_model.V1beta1EventFilter) bool {
	if len(expected) != len(actual) {
		return false
	}
	for idx, item := range actual {
		expectedItem := expected[idx]
		if expectedItem.IgnoredDeleteValueExpression != item.IgnoredDeleteValueExpression {
			return false
		}
		if expectedItem.IgnoredInsertValueExpression != item.IgnoredInsertValueExpression {
			return false
		}
		if expectedItem.IgnoredUpdateNewValueExpression != item.IgnoredUpdateNewValueExpression {
			return false
		}
		if expectedItem.IgnoredUpdateOldValueExpression != item.IgnoredUpdateOldValueExpression {
			return false
		}
		if !equalStringSliceAllowNilEmpty(expectedItem.IgnoredEvents, item.IgnoredEvents) {
			return false
		}
		if !equalStringSliceAllowNilEmpty(expectedItem.IgnoredSQLStatements, item.IgnoredSQLStatements) {
			return false
		}
		if !equalStringSliceAllowNilEmpty(expectedItem.TableMatchers, item.TableMatchers) {
			return false
		}
	}
	return true
}

func equalStringSliceAllowNilEmpty(expected, actual []string) bool {
	if len(expected) == 0 && len(actual) == 0 {
		return true
	}
	return reflect.DeepEqual(expected, actual)
}