package changefeed

import (
	"context"
	"fmt"
	"net/url"
	"strings"
	"time"

	"cloud.google.com/go/storage"
	. "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/sample_data"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/cdcapi"
	"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/model"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
)

const gcsRoleName = "roles/storage.admin"

type gcsChangefeedSmokeCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator
	cloudProvider CloudProviderType
}

func NewGcsChangefeedSmokeCase(cloudProvider CloudProviderType) E2ECase {
	return &gcsChangefeedSmokeCase{cloudProvider: cloudProvider}
}

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

func (m *gcsChangefeedSmokeCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: m.cloudProvider,
				Region:        config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
		"gcs": &GcsRequirement{
			GcsDimensions: GcsDimensions{
				Region: config.GlobalConfig.DefaultRegionOf(m.cloudProvider),
			},
		},
	}
}

func (m *gcsChangefeedSmokeCase) Run() {
	tidbCluster := m.TiDBDedicated("tidb")
	downstream := m.GCS("gcs")
	changefeedName := m.Name()
	instanceType := "2rcu"
	dbName := m.Name()
	tableFilter := []string{fmt.Sprintf("%s.*", dbName)}
	runningCheckRetry := runningCheckRetryForCloud(m.cloudProvider)
	ctx := context.Background()

	var (
		changefeedID, gcpServiceAccount string
		createGcsInfo, editGcsInfo      *model.DataflowGcsInfo
		gcsCli                          *storage.Client
		bkt                             *storage.BucketHandle
	)

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

	m.Step("init gcp bucket client", func() error {
		// we should ensure there are gcp permissions in the environment
		bucketUri, err := url.Parse(downstream.BucketURI)
		if err != nil {
			return fmt.Errorf("parse GCS bucket uri %s error: %w", downstream.BucketURI, err)
		}
		cli, err := storage.NewClient(ctx)
		if err != nil {
			return fmt.Errorf("create gcs client error: %w", err)
		}
		bkt = cli.Bucket(bucketUri.Hostname())
		gcsCli = cli
		return nil
	})

	defer gcsCli.Close()

	m.RetryDo("get gcp service account", func() error {
		adminProfileParams := &connector_service.ConnectorServiceDescribeAdminProfileParams{
			ClusterID: tidbCluster.ClusterID,
			ProjectID: tidbCluster.ProjectID,
			OrgID:     tidbCluster.OrgID,
		}
		resp, err := m.cdcApiWrapper.DescribeAdminProfile(adminProfileParams)
		if err != nil {
			return err
		}
		gcpServiceAccount = *resp.Payload.GcpAccountID
		return nil
	}, 3, time.Second*10)

	m.RetryDo("grant bucket permission", func() error {
		return m.grantBucketPermission(ctx, bkt, gcpServiceAccount)
	}, 3, time.Second*5)

	defer func() {
		m.RetryDo("revoke bucket permission", func() error {
			return m.revokeBucketPermission(ctx, bkt, gcpServiceAccount)
		}, 3, time.Second*5)
	}()

	// test create request validation
	invalidCreateReqOpts := []gcsChangefeedOpt{
		gcsChangefeedWithInvalidDataFormat,
		gcsChangefeedWithInvalidFlushConfig,
		gcsChangefeedWithInvalidTableFilter,
		gcsChangefeedWithInvalidEventFilter,
	}

	for _, opt := range invalidCreateReqOpts {
		m.Expect("create gcs changefeed with invalid params", func() error {
			req := genCreateGcsChangefeedReqWithOpt(changefeedName, instanceType, dbName, tidbCluster.ClusterID, downstream, opt)
			_, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *req)
			return ExpectError(err)
		}, 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)
	}()

	m.
		RetryDo(fmt.Sprintf("test connection to downstream %s", downstream.BucketURI), func() (err error) {
			return m.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 m.cfValidator.VerifyTables(tidbCluster, tableFilter, []string{fmt.Sprintf("%s.%s", dbName, sample_data.SAMPLE_TABLE)}, nil)
		}, 3, time.Second*10).
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			createReq := genCreateGcsChangefeedReq(changefeedName, instanceType, dbName, tidbCluster.ClusterID, downstream)
			resp, err := m.cdcApiWrapper.CreateChangefeed(tidbCluster, *createReq)
			if err == nil {
				changefeedID = *resp.Payload.ConnectorID
				createGcsInfo = createReq.GcsInfo
			}
			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)
		}, runningCheckRetry, time.Minute).
		Expect("compare changefeed kernel config after create", func() error {
			cfg, err := m.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedID)
			if err != nil {
				return err
			}
			return compareGcsChangefeedKernelConfig(createGcsInfo, 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)
		}).
		Step("verifyGetCheckpointWithoutCache", func() error {
			return m.cfValidator.VerifyGetCheckpointWithoutCache(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 := []gcsChangefeedOpt{
		gcsChangefeedWithInvalidFlushConfig,
		gcsChangefeedWithInvalidDataFormat,
		gcsChangefeedWithInvalidTableFilter,
		gcsChangefeedWithInvalidEventFilter,
	}
	for _, opt := range invalidEditReqOpts {
		m.Expect("edit mysql changefeed with invalid params", func() error {
			changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
			gcsInfo := changefeedInfo.Payload.Request.GcsInfo
			opt(gcsInfo)
			editReq := connector_service.EditConnectorBody{
				ConnectorID:   changefeedInfo.Payload.ConnectorID,
				ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
				Name:          changefeedInfo.Payload.Name,
				GcsInfo:       gcsInfo,
			}
			err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
			return ExpectError(err)
		}, 3, time.Second*10)
	}

	m.
		RetryDo(fmt.Sprintf("edit changefeed %s", changefeedName), func() error {
			gcsInfo, err := m.editChangefeed(tidbCluster, changefeedID, downstream)
			if err == nil {
				editGcsInfo = gcsInfo
			}
			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 compareGcsChangefeedKernelConfig(editGcsInfo, 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)
		}).
		Step("verifyGetCheckpointWithoutCache", func() error {
			return m.cfValidator.VerifyGetCheckpointWithoutCache(tidbCluster, changefeedID)
		})
}

func (m *gcsChangefeedSmokeCase) editChangefeed(
	tidbCluster *TiDBDedicatedEntity, changefeedID string, downstream *GcsEntity,
) (*model.DataflowGcsInfo, error) {
	changefeedInfo, err := m.cdcApiWrapper.GetChangefeed(tidbCluster, changefeedID)
	if err != nil {
		return nil, err
	}
	gcsInfo := changefeedInfo.Payload.Request.GcsInfo
	editGcsInfoNormally(gcsInfo)
	editReq := connector_service.EditConnectorBody{
		ConnectorID:   changefeedInfo.Payload.ConnectorID,
		ConnectorType: changefeedInfo.Payload.Request.ConnectorType,
		Name:          changefeedInfo.Payload.Name,
		GcsInfo:       gcsInfo,
	}
	err = m.cdcApiWrapper.EditChangefeed(tidbCluster, changefeedID, editReq)
	if err != nil {
		return nil, err
	}
	return gcsInfo, nil
}

func (m *gcsChangefeedSmokeCase) grantBucketPermission(ctx context.Context, bkt *storage.BucketHandle, sa string) error {
	member := genGcpServiceAccountMember(sa)
	policy, err := bkt.IAM().Policy(ctx)
	if err != nil {
		return fmt.Errorf("get bucket policy error: %w", err)
	}
	if policy.HasRole(member, gcsRoleName) {
		m.Logger().Info("bucket policy already exists")
		return nil
	}

	policy.Add(member, gcsRoleName)
	if err := bkt.IAM().SetPolicy(ctx, policy); err != nil {
		return fmt.Errorf("set bucket policy error: %w", err)
	}
	return nil
}

func (m *gcsChangefeedSmokeCase) revokeBucketPermission(ctx context.Context, bkt *storage.BucketHandle, sa string) error {
	member := genGcpServiceAccountMember(sa)
	policy, err := bkt.IAM().Policy(ctx)
	if err != nil {
		return fmt.Errorf("get bucket policy error: %w", err)
	}
	if !policy.HasRole(member, gcsRoleName) {
		m.Logger().Info("bucket policy does not exist")
		return nil
	}

	policy.Remove(member, gcsRoleName)
	if err := bkt.IAM().SetPolicy(ctx, policy); err != nil {
		return fmt.Errorf("set bucket policy error: %w", err)
	}
	return nil
}

type gcsChangefeedOpt func(gcsInfo *model.DataflowGcsInfo)

func gcsChangefeedWithInvalidDataFormat(gcsInfo *model.DataflowGcsInfo) {
	gcsInfo.DataFormat.Delimiter = ""
}

func gcsChangefeedWithInvalidFlushConfig(gcsInfo *model.DataflowGcsInfo) {
	gcsInfo.StorageFlushConfig.FlushInterval = "20m"
}

func gcsChangefeedWithInvalidTableFilter(gcsInfo *model.DataflowGcsInfo) {
	gcsInfo.TableConfig.FilterRule = []string{"abc", "def"}
}

func gcsChangefeedWithInvalidEventFilter(gcsInfo *model.DataflowGcsInfo) {
	gcsInfo.TableConfig.EventFilterRule = []*model.DataflowEventFilterRule{
		{
			IgnoreEvent: []string{"all dml", "all ddl", "update"},
			Matcher:     []string{"a.b.d", "c.d.d"},
		},
	}
}

func genCreateGcsChangefeedReqWithOpt(
	name, instanceType, dbName, clusterID string, downstream *GcsEntity, opt gcsChangefeedOpt,
) *connector_service.CreateConnectorBody {
	req := genCreateGcsChangefeedReq(name, instanceType, dbName, clusterID, downstream)
	opt(req.GcsInfo)
	return req
}

func genCreateGcsChangefeedReq(
	name, instanceType, dbName, clusterID string, downstream *GcsEntity,
) *connector_service.CreateConnectorBody {
	changefeedSpec := &connector_service.CreateConnectorBody{}
	changefeedSpec.Name = name
	connectorType := model.DataflowConnectorTypeGCS
	encodingMethod := model.DataflowBinaryEncodingMethodBASE64
	s3Protocol := model.DataflowS3ProtocolS3PROTOCOLCSV
	changefeedSpec.ConnectorType = &connectorType
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.GcsInfo = &model.DataflowGcsInfo{}
	bucketURI := downstream.BucketURI
	if strings.HasSuffix(bucketURI, "/") {
		bucketURI = bucketURI[:len(bucketURI)-1]
	}
	changefeedSpec.GcsInfo.Endpoint = &model.DataflowGcsEndpoint{
		GcsURI: fmt.Sprintf("%s/%s/%s", bucketURI, clusterID, name),
	}
	changefeedSpec.GcsInfo.DataFormat = &model.DataflowS3Dataformat{
		BinaryEncodingMethod: &encodingMethod,
		DateSeparator:        "day",
		Terminator:           "\\n",
		Delimiter:            ",",
		Quote:                "\"",
		Null:                 "\\N",
		Protocol:             &s3Protocol,
	}
	changefeedSpec.GcsInfo.StorageFlushConfig = &model.DataflowStorageFlushConfig{
		FlushInterval: "60s",
		FileSize:      "67108864",
	}
	changefeedSpec.GcsInfo.TableConfig = &model.DataflowTableConfig{
		FilterRule: []string{fmt.Sprintf("%s.*", dbName)},
		EventFilterRule: []*model.DataflowEventFilterRule{
			{
				Matcher:     []string{fmt.Sprintf("%s.*", dbName)},
				IgnoreEvent: []string{"delete"},
			},
		},
	}
	startMode := model.DataflowStartModeFROMNOW
	changefeedSpec.GcsInfo.StartPosition = &model.DataflowStartPosition{
		Mode: &startMode,
	}
	return changefeedSpec
}

func compareGcsChangefeedKernelConfig(gcsInfo *model.DataflowGcsInfo, kernelConfig *ticdc_v1.Changefeed) error {
	sinkUri, err := url.Parse(kernelConfig.SinkUri)
	if err != nil {
		return fmt.Errorf("parse sink uri error: %w", err)
	}
	gcsURI := fmt.Sprintf("%s://%s%s", sinkUri.Scheme, sinkUri.Host, sinkUri.Path)
	if gcsURI != gcsInfo.Endpoint.GcsURI {
		return fmt.Errorf("gcs uri mismatch, expected %s, got %s", gcsInfo.Endpoint.GcsURI, gcsURI)
	}

	return compareCloudStorageCommonKernelConfig(
		gcsInfo.TableConfig, gcsInfo.DataFormat, gcsInfo.StorageFlushConfig, sinkUri, kernelConfig)
}

func editGcsInfoNormally(gcsInfo *model.DataflowGcsInfo) {
	gcsInfo.TableConfig.FilterRule = append(gcsInfo.TableConfig.FilterRule, "test.*")
	gcsInfo.TableConfig.EventFilterRule = append(gcsInfo.TableConfig.EventFilterRule, &model.DataflowEventFilterRule{
		Matcher:     []string{"test.*"},
		IgnoreEvent: []string{"update", "delete"},
	})
}

func genGcpServiceAccountMember(sa string) string {
	return fmt.Sprintf("serviceAccount:%s", sa)
}