package changefeed

import (
	"fmt"
	"net/http"
	"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/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"
	"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 serverlessChangefeedSmokeCase struct {
	BaseCase
	cdcApiWrapper          *cdcapi.CDCClientWrapper
	sinkEndpointApiWrapper *consoleapi.SinkEndpointClientWrapper
	cfValidator            *common.ChangfeedValidator
	cloudProvider          CloudProviderType
	region                 string
}

func NewServerlessChangefeedSmokeCase(cloudProvider CloudProviderType, region string) E2ECase {
	return &serverlessChangefeedSmokeCase{cloudProvider: cloudProvider, region: region}
}

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

func (m *serverlessChangefeedSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        m.region,
			},
		},
		"serverless": &TiDBServerlessRequirement{
			Dedicated: true,
			TiDBServerlessDimensions: TiDBServerlessDimensions{
				CloudProvider: m.cloudProvider,
				Region:        m.region,
				ServicePlan:   "Starter",
			},
		},
	}
}

func (m *serverlessChangefeedSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	downstream := m.TiDBServerless("serverless")
	changefeedName := m.Name()
	instanceType := "2rcu"
	dbName := m.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}
	mysqlEntity := &MySQLEntity{
		User:     downstream.User,
		Password: downstream.Password,
	}

	var (
		changefeedID                             string
		createServerlessInfo, editServerlessInfo *model.DataflowServerlessTidbInfo
	)

	// init reusable api client and validator
	m.cdcApiWrapper = cdcapi.NewCDCClientWrapper(m.Logger())
	m.sinkEndpointApiWrapper = consoleapi.NewSinkEndpointClientWrapper(m.Logger())
	m.cfValidator = common.NewChangfeedValidator(m.Logger())
	sampleDataWriter := sample_data.NewSampleDataWriter(m.Logger(), tidbCluster.MysqlParameter(), dbName, 3*time.Second)
	mysqlParams := &sample_data.MySQLParameter{
		Endpoint:   downstream.Host,
		User:       downstream.User,
		Password:   downstream.Password,
		TLSEnabled: true,
	}
	downstreamMgr := sample_data.NewMySQLDownstreamManager(m.Logger(), mysqlParams, dbName)

	m.
		RetryDo("create serverless private link endpoint", func() error {
			params := &sink_endpoint_service.CreateSinkEndpointParams{
				OrgID:     tidbCluster.OrgID,
				ProjectID: tidbCluster.ProjectID,
				SinkEndpoint: &model.DataflowSinkEndpoint{
					EndpointServiceName: &downstream.PLServiceName,
					EndpointName:        ptr.String(fmt.Sprintf("%s-serverless-pl-e2e", tidbCluster.ClusterID)),
					Provider:            ptr.String(string(tidbCluster.CloudProvider)),
					Region:              ptr.String(tidbCluster.Region),
				},
			}
			endpointID, err := m.sinkEndpointApiWrapper.CreateSinkEndpoint(params)
			if err != nil {
				return err
			}
			downstream.PLEndpointID = endpointID
			return nil
		}, 3, time.Minute).
		Expect("serverless private link endpoint is available", func() error {
			params := &sink_endpoint_service.GetSinkEndpointParams{
				OrgID:          tidbCluster.OrgID,
				ProjectID:      tidbCluster.ProjectID,
				SinkEndpointID: downstream.PLEndpointID,
			}
			endpoint, err := m.sinkEndpointApiWrapper.GetSinkEndpoint(params)
			if err != nil {
				return err
			}
			if *endpoint.Status == model.DataflowSinkEndpointStatusSINKENDPOINTSTATUSAVAILABLE {
				return nil
			}
			return fmt.Errorf("serverless private link endpoint is not available, phase is %s", *endpoint.Status)
		}, 40, time.Second*30)

	// test create request validation
	invalidCreateReqOpts := []mysqlChangefeedOpt{
		mysqlChangefeedWithInvalidTableFilter,
		mysqlChangefeedWithInvalidEventFilter,
	}

	for _, opt := range invalidCreateReqOpts {
		m.Expect("create serverless changefeed with invalid params", func() error {
			req := genCreateServerlessChangefeedReqWithOpt(changefeedName, instanceType, dbName, downstream, mysqlEntity, opt)
			_, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *req)
			return ExpectHTTPStatus(err, http.StatusBadRequest)
		}, 3, time.Second*10)
	}

	// cleanup changefeeds if exists
	defer func() {
		m.
			Step("stop writing data", func() error {
				sampleDataWriter.StopWriting()
				return nil
			}).
			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, time.Second*10).
			RetryDo("delete private link", func() error {
				return m.sinkEndpointApiWrapper.DeleteSinkEndpoint(tidbCluster, downstream.PLEndpointID)
			}, 3, time.Minute).
			RetryDo("delete database and tables in downstream", downstreamMgr.DropTables, 3, time.Second*10)
	}()

	m.
		RetryDo(fmt.Sprintf("test connection to serverless cluster %s", downstream.ClusterID), func() (err error) {
			return m.cdcApiWrapper.TestConnection(tidbCluster, downstream)
		}, 5, 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 m.cfValidator.VerifyTables(tidbCluster, tableFilter, []string{fmt.Sprintf("%s.%s", dbName, sample_data.SAMPLE_TABLE)}, nil)
		}, 3, time.Second*10).
		RetryDo("init database and table in downstream", downstreamMgr.InitTables, 3, time.Minute).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			createReq := genCreateServerlessChangefeedReq(changefeedName, instanceType, dbName, downstream, mysqlEntity)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *createReq)
			if err == nil {
				changefeedID = *resp.Payload.ConnectorID
				createServerlessInfo = createReq.ServerlessTidbInfo
			}
			return err
		}, 3, time.Minute).
		Expect("changefeed in list", func() error {
			return m.cfValidator.AreChangefeedsInList(tidbCluster, changefeedID)
		}, 3, time.Second*10).
		Expect("changefeeed is running", func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, 20, time.Minute).
		Expect("compare changefeed kernel config after create", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return compareMySQLChangefeedKernelConfig(createServerlessInfo.MysqlInfo, cfg)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("start write sample data to database %s", dbName),
			sampleDataWriter.StartWriting, 3, time.Minute).
		Step("verify changefeed's CheckpointTs advance normal", func() error {
			return m.cfValidator.VerifyCheckpointTsAdvanceNormal(5*time.Minute, tidbCluster, changefeedID)
		}).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeedName), func() error {
			return m.cdcApiWrapper.PauseChangefeed(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is paused", changefeedName), func() error {
			return m.cfValidator.IsChangefeedInStatus(tidbCluster, changefeedID, model.DataflowConnectorStatusPAUSED)
		}, 3, time.Minute)

	// test edit request validation
	invalidEditReqOpts := []mysqlChangefeedOpt{
		mysqlChangefeedWithInvalidTableFilter,
		mysqlChangefeedWithInvalidEventFilter,
	}
	for _, opt := range invalidEditReqOpts {
		m.Expect("edit serverless changefeed with invalid params", func() error {
			changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			serverlessInfo, err := buildServerlessInfoForEdit(changefeedInfo, downstream)
			if err != nil {
				return err
			}
			if serverlessInfo.MysqlInfo == nil || serverlessInfo.MysqlInfo.Connection == nil {
				return fmt.Errorf("changefeed mysql connection is empty")
			}
			serverlessInfo.MysqlInfo.Connection.Password = downstream.Password
			opt(serverlessInfo.MysqlInfo)
			editReq, err := buildServerlessEditChangefeedReq(changefeedInfo, serverlessInfo)
			if err != nil {
				return err
			}
			err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
			return ExpectHTTPStatus(err, http.StatusBadRequest)
		}, 3, time.Second*10)
	}

	m.
		RetryDo(fmt.Sprintf("edit changefeed %s without password", changefeedName), func() error {
			serverlessInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, false)
			if err == nil {
				editServerlessInfo = serverlessInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s with password", changefeedName), func() error {
			serverlessInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream, true)
			if err == nil {
				editServerlessInfo = serverlessInfo
			}
			return err
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeedName), func() error {
			return m.cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Still(fmt.Sprintf("changefeed %s is runining", changefeedName), func() error {
			return m.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedID)
		}, 3, time.Minute).
		Expect("compare changefeed kernel config after edit", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return compareMySQLChangefeedKernelConfig(editServerlessInfo.MysqlInfo, cfg)
		}, 3, time.Minute).
		Step(fmt.Sprintf("verify changefeed %s CheckpointTs advance normal", changefeedName), func() error {
			return m.cfValidator.VerifyCheckpointTsAdvanceNormal(2*time.Minute, tidbCluster, changefeedID)
		})
}

func (m *serverlessChangefeedSmokeCase) editChangefeed(
	tidbCluster *TiDBDedicatedEntity, changefeedID string, downstream *TiDBServerlessEntity, updatePassword bool,
) (*model.DataflowServerlessTidbInfo, error) {
	changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
	if err != nil {
		return nil, err
	}
	serverlessInfo, err := buildServerlessInfoForEdit(changefeedInfo, downstream)
	if err != nil {
		return nil, err
	}
	if updatePassword {
		if serverlessInfo.MysqlInfo == nil || serverlessInfo.MysqlInfo.Connection == nil {
			return nil, fmt.Errorf("changefeed mysql connection is empty")
		}
		serverlessInfo.MysqlInfo.Connection.Password = downstream.Password
	}
	editMySQLInfoNormally(serverlessInfo.MysqlInfo)
	editReq, err := buildServerlessEditChangefeedReq(changefeedInfo, serverlessInfo)
	if err != nil {
		return nil, err
	}
	err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
	if err != nil {
		return nil, err
	}
	return serverlessInfo, nil
}

func buildServerlessEditChangefeedReq(
	changefeedInfo *connector_service.GetConnectorOK,
	serverlessInfo *model.DataflowServerlessTidbInfo,
) (connector_service.EditConnectorBody, error) {
	if changefeedInfo == nil || changefeedInfo.Payload == nil || changefeedInfo.Payload.Request == nil {
		return connector_service.EditConnectorBody{}, fmt.Errorf("changefeed response is empty")
	}

	editReq := connector_service.EditConnectorBody{
		ConnectorID:   changefeedInfo.Payload.ConnectorID,
		ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
		Name:          changefeedInfo.Payload.Name,
	}

	// OpenAPI get/list returns MYSQL connector type for serverless sinks.
	if changefeedInfo.Payload.Request.ConnectorType != nil &&
		*changefeedInfo.Payload.Request.ConnectorType == model.DataflowConnectorTypeMYSQL {
		if serverlessInfo == nil || serverlessInfo.MysqlInfo == nil {
			return connector_service.EditConnectorBody{}, fmt.Errorf("changefeed mysql info is empty")
		}
		editReq.MysqlInfo = serverlessInfo.MysqlInfo
		return editReq, nil
	}

	editReq.ServerlessTidbInfo = serverlessInfo
	return editReq, nil
}

func buildServerlessInfoForEdit(
	changefeedInfo *connector_service.GetConnectorOK,
	downstream *TiDBServerlessEntity,
) (*model.DataflowServerlessTidbInfo, error) {
	if changefeedInfo == nil || changefeedInfo.Payload == nil || changefeedInfo.Payload.Request == nil {
		return nil, fmt.Errorf("changefeed response is empty")
	}

	req := changefeedInfo.Payload.Request
	if req.ServerlessTidbInfo != nil && req.ServerlessTidbInfo.MysqlInfo != nil {
		return req.ServerlessTidbInfo, nil
	}

	if req.MysqlInfo == nil {
		return nil, fmt.Errorf("changefeed mysql info is empty")
	}
	if downstream == nil {
		return nil, fmt.Errorf("downstream serverless info is empty")
	}

	return &model.DataflowServerlessTidbInfo{
		ClusterID: downstream.ClusterID,
		MysqlInfo: req.MysqlInfo,
	}, nil
}

func genCreateServerlessChangefeedReqWithOpt(
	name, instanceType, dbName string, serverlessEnt *TiDBServerlessEntity, mysqlEnt *MySQLEntity, opt mysqlChangefeedOpt,
) *connector_service.CreateConnectorBody {
	req := genCreateServerlessChangefeedReq(name, instanceType, dbName, serverlessEnt, mysqlEnt)
	opt(req.ServerlessTidbInfo.MysqlInfo)
	return req
}

func genCreateServerlessChangefeedReq(
	name, instanceType, dbName string, serverlessEnt *TiDBServerlessEntity, mysqlEnt *MySQLEntity,
) *connector_service.CreateConnectorBody {
	req := genCreateMySQLChangefeedReq(name, instanceType, dbName, mysqlEnt)
	connectorType := model.DataflowConnectorTypeSERVERLESSTIDB
	req.ConnectorType = &connectorType
	req.ServerlessTidbInfo = &model.DataflowServerlessTidbInfo{
		ClusterID: serverlessEnt.ClusterID,
		MysqlInfo: req.MysqlInfo,
	}
	req.MysqlInfo = nil
	req.ServerlessTidbInfo.MysqlInfo.PrivatelinkInfo = &model.DataflowPrivatelinkInfo{
		DfPrivatelinkEndpointID: serverlessEnt.PLEndpointID,
	}
	return req
}