package impl

import (
	"context"
	"encoding/json"
	"strconv"
	"strings"

	"github.com/AlekSi/pointer"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

const (
	DateSeparatorNone          = "none"
	DateSeparatorYear          = "year"
	DateSeparatorMonth         = "month"
	DateSeparatorDay           = "day"
	S3ProtocolTypeCSV          = "csv"
	S3ProtocolTypeCanalJson    = "canal-json"
	BinaryEncodingMethodBase64 = "base64"
	BinaryEncodingMethodHex    = "hex"
)

type s3DownstreamAdaptor struct {
	cdcPreCheckCli  regionalapi.TiCDCPreCheckClient
	cdcClusterModel ticdc.ClusterModel
	encryptor       encryptor_wrapper.Encryptor
}

func NewS3DownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient,
	cdcClusterModel ticdc.ClusterModel,
	encryptor encryptor_wrapper.Encryptor,
) biz_changefeed.DownstreamAdaptor {
	return &s3DownstreamAdaptor{
		cdcPreCheckCli:  cdcPreCheckCli,
		cdcClusterModel: cdcClusterModel,
		encryptor:       encryptor,
	}
}

func (m *s3DownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	s3Params, bizErr := m.parseS3Params(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	clusterIDStr := strconv.FormatUint(cluster.Id, 10)
	region := &regionalapi.Region{
		Name:     cluster.RegionName,
		TenantID: cluster.TenantId,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(cluster.InfraProviderType))),
	}
	networkReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: clusterIDStr,
		Type:      string(entity.TiCDCDownStreamTypeS3),
		S3Param: &ticdc_v1.TestNetworkConnectionRequest_S3Parameter{
			S3Uri: s3Params.Endpoint.S3Uri,
		},
	}

	if s3Params.Endpoint.AuthType == dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE {
		externalID, bizErr := m.getExternalID(ctx, cluster)
		if bizErr != nil {
			return bizErr.Chain("get external id")
		}
		networkReq.S3Param.RoleArn = s3Params.Endpoint.IamAssumedRoleArn
		networkReq.S3Param.ExternalId = externalID
	} else {
		networkReq.S3Param.AccessKeyId = s3Params.Endpoint.AccessKey
		networkReq.S3Param.SecretAccessKey = s3Params.Endpoint.SecretKey
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_S3],
			err.ErrMsg())
	}
	return nil
}

//nolint:dupl
func (m *s3DownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	s3Params, bizErr := m.parseS3Params(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	startTSO, bizErr := parseStartTSO(s3Params.StartPosition)
	if bizErr != nil {
		return bizErr
	}
	clusterIDStr := strconv.FormatUint(cluster.Id, 10)
	region := &regionalapi.Region{
		Name:     cluster.RegionName,
		TenantID: cluster.TenantId,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(cluster.InfraProviderType))),
	}
	changefeedParamReq := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: clusterIDStr,
		Type:      string(entity.CDCDownStreamTypeS3),
		StartTso:  startTSO,
	}
	if err := m.cdcPreCheckCli.ValidateChangefeedParameters(ctx, region, changefeedParamReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_S3],
			err.ErrMsg())
	}
	return nil
}

func (m *s3DownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	s3Params, bizErr := m.parseS3Params(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	if s3Params.GetEndpoint().GetSecretKey() == "" {
		return nil
	}

	return m.encryptor.EncryptBase64Inplace(&s3Params.Endpoint.SecretKey)
}

func (m *s3DownstreamAdaptor) AssembleChangefeedConfig(
	ctx context.Context, cluster *core.Cluster, bizCf *entity.BizChangefeed,
) (*domain.TiCDCChangefeedConfig, bizerr.BizError) {
	s3Params, bizErr := m.parseS3ParamsStr(bizCf.DownstreamParams)
	if bizErr != nil {
		return nil, bizErr
	}

	uriConfig, bizErr := m.assembleURIConfig(ctx, s3Params, cluster)
	if bizErr != nil {
		return nil, bizErr.Chain("assemble uri config")
	}
	changefeedConfig, bizErr := assembleCloudStorageChangefeedConfigData(
		s3Params.GetTableConfig(), s3Params.GetStartPosition(), s3Params.GetDataFormat())
	if bizErr != nil {
		return nil, bizErr.Chain("assemble changefeed config data")
	}
	return &domain.TiCDCChangefeedConfig{
		URIConfig: uriConfig,
		Config:    changefeedConfig,
	}, nil
}

func (m *s3DownstreamAdaptor) assembleURIConfig(
	ctx context.Context, s3Params *dataflow.S3Info, cluster *core.Cluster,
) (*domain.TiCDCChangefeedS3URIConfigData, bizerr.BizError) {
	uriConfig := &domain.TiCDCChangefeedS3URIConfigData{
		S3URI:                s3Params.GetEndpoint().GetS3Uri(),
		ForcePathStyle:       pointer.ToBool(true),
		FlushInterval:        s3Params.GetStorageFlushConfig().GetFlushInterval(),
		FileSize:             s3Params.GetStorageFlushConfig().GetFileSize(),
		TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
	}
	if s3Params.GetEndpoint().GetAuthType() == dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE {
		externalID, bizErr := m.getExternalID(ctx, cluster)
		if bizErr != nil {
			return nil, bizErr.Chain("get external id")
		}
		uriConfig.RoleARN = pointer.ToString(s3Params.GetEndpoint().GetIamAssumedRoleArn())
		uriConfig.ExternalID = pointer.ToString(externalID)
	} else {
		uriConfig.AccessKey = pointer.ToString(s3Params.GetEndpoint().GetAccessKey())
		secretKey, err := m.encryptor.DecryptBase64(s3Params.GetEndpoint().GetSecretKey())
		if err != nil {
			return nil, err.Chain("decrypt secret key")
		}
		uriConfig.SecretAccessKey = pointer.ToString(secretKey)
	}
	if s3Params.GetDataFormat().GetProtocol() == dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON &&
		s3Params.GetDataFormat().GetEnableTidbExtension() {
		uriConfig.EnableTiDBExtension = pointer.ToBool(true)
	}
	return uriConfig, nil
}

func (m *s3DownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editS3Params, err := m.parseS3Params(editDownstreamParams)
	if err != nil {
		return err
	}
	currentS3Params, err := m.parseS3ParamsStr(currentDownstreamParams)
	if err != nil {
		return err
	}

	editS3Params.Endpoint.S3Uri = currentS3Params.Endpoint.S3Uri
	if editS3Params.Endpoint.AuthType != dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK {
		return nil
	}

	if editS3Params.Endpoint.SecretKey == "" {
		editS3Params.Endpoint.SecretKey = currentS3Params.Endpoint.SecretKey
	} else {
		if err = m.encryptor.EncryptBase64Inplace(&editS3Params.Endpoint.SecretKey); err != nil {
			return err.Chain("encrypt password")
		}
	}
	return nil
}

func (m *s3DownstreamAdaptor) GetPrivateLinkEndpointID(_ interface{}) entity.ID {
	// s3 downstream doesn't support private link
	return entity.ID(0)
}

func (m *s3DownstreamAdaptor) BackfillConfig(
	ctx context.Context, currentDownstreamParams string, kernelConfig *ticdc_v1.ReplicaConfig,
) (downstreamParams string, changed bool, bizErr bizerr.BizError) {
	s3Params, bizErr := m.parseS3ParamsStr(currentDownstreamParams)
	if bizErr != nil {
		return "", false, bizErr
	}
	if kernelConfig.GetSink().GetCloudStorageConfig().GetOutputRawChangeEvent() &&
		!s3Params.GetDataFormat().GetOutputRawChangeEvent() {
		s3Params.DataFormat.OutputRawChangeEvent = true
		changed = true
	}
	if kernelConfig.GetCaseSensitive() && !s3Params.GetTableConfig().GetCaseSensitive() {
		s3Params.TableConfig.CaseSensitive = true
		changed = true
	}

	downstreamParamsBytes, err := json.Marshal(s3Params)
	if err != nil {
		return "", false, bizerr.WrapError(err, "marshal downstream params")
	}
	downstreamParams = string(downstreamParamsBytes)
	return downstreamParams, changed, nil
}

func (m *s3DownstreamAdaptor) parseS3Params(downstreamParams interface{}) (*dataflow.S3Info, bizerr.BizError) {
	s3Params, ok := downstreamParams.(*dataflow.S3Info)
	if !ok || s3Params == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstreamParams is not *dataflow.S3Info type")
	}
	if _, bizErr := normalizeCloudStorageDateSeparator(s3Params.GetDataFormat()); bizErr != nil {
		return nil, bizErr
	}
	return s3Params, nil
}

func (m *s3DownstreamAdaptor) parseS3ParamsStr(downstreamParams string) (*dataflow.S3Info, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	s3Params := new(dataflow.S3Info)
	if err := decoder.Decode(s3Params); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal s3 params error")
	}
	return s3Params, nil
}

func (m *s3DownstreamAdaptor) getExternalID(ctx context.Context, cluster *core.Cluster) (string, bizerr.BizError) {
	req := &domain.GetTiCDCClusterRoleReq{
		TenantID:      cluster.TenantId,
		ProjectID:     cluster.ProjectId,
		TiDBClusterID: cluster.Id,
		Region:        cluster.RegionName,
		CloudProvider: domain.CloudProviderType(cluster.CloudProviderName),
	}
	roleInfo, err := m.cdcClusterModel.GetRole(ctx, req)
	if err != nil {
		return "", err.Chain("get ticdc cluster role")
	}
	return roleInfo.AWSExternalID, nil
}

func assembleCloudStorageChangefeedConfigData(
	tableConfig *dataflow.TableConfig, startPosition *dataflow.StartPosition, dataFormat *dataflow.S3Dataformat,
) (*domain.TiCDCChangefeedConfigData, bizerr.BizError) {
	startTSO, bizErr := parseStartTSO(startPosition)
	if bizErr != nil {
		return nil, bizErr
	}
	dateSeparator, bizErr := normalizeCloudStorageDateSeparator(dataFormat)
	if bizErr != nil {
		return nil, bizErr
	}

	sinkConfig := &clusterproxy.SinkConfigV2{
		Terminator:    "\n",
		DateSeparator: dateSeparator,
	}
	if dataFormat.GetTerminator() != "" {
		sinkConfig.Terminator = dataFormat.GetTerminator()
	}

	if dataFormat.GetProtocol() == dataflow.S3Protocol_S3_PROTOCOL_CSV {
		sinkConfig.Protocol = S3ProtocolTypeCSV
		sinkConfig.CSVConfig = assembleCSVConfig(dataFormat)
	} else if dataFormat.GetProtocol() == dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON {
		sinkConfig.Protocol = S3ProtocolTypeCanalJson
	}

	if sinkConfig.CloudStorageConfig == nil {
		sinkConfig.CloudStorageConfig = &clusterproxy.CloudStorageConfig{}
	}
	sinkConfig.CloudStorageConfig.OutputRawChangeEvent = pointer.ToBool(dataFormat.GetOutputRawChangeEvent())

	changefeedConfig := &domain.TiCDCChangefeedConfigData{
		StartTs: &startTSO,
		ReplicaConfig: &clusterproxy.ReplicaConfig{
			CaseSensitive: pointer.ToBool(tableConfig.GetCaseSensitive()),
			Sink:          sinkConfig,
		},
	}

	assembleChangefeedSinkConfigTableConfig(tableConfig, changefeedConfig.ReplicaConfig)
	return changefeedConfig, nil
}

func normalizeCloudStorageDateSeparator(dataFormat *dataflow.S3Dataformat) (string, bizerr.BizError) {
	if dataFormat == nil {
		return DateSeparatorNone, nil
	}

	dateSeparator := strings.ToLower(strings.TrimSpace(dataFormat.GetDateSeparator()))
	if dateSeparator == "" {
		dateSeparator = DateSeparatorNone
	}
	switch dateSeparator {
	case DateSeparatorNone, DateSeparatorYear, DateSeparatorMonth, DateSeparatorDay:
		dataFormat.DateSeparator = dateSeparator
		return dateSeparator, nil
	default:
		return "", bizerr.New(bizerr.ErrCodeParams, "invalid date separator %q", dataFormat.GetDateSeparator())
	}
}

func assembleCSVConfig(dataFormat *dataflow.S3Dataformat) *clusterproxy.CSVConfig {
	csvConfig := &clusterproxy.CSVConfig{
		Delimiter:            ",",
		Quote:                "\"",
		NullString:           "\\N",
		IncludeCommitTs:      pointer.ToBool(dataFormat.GetIncludeCommitTs()),
		BinaryEncodingMethod: BinaryEncodingMethodBase64,
	}
	if dataFormat.GetBinaryEncodingMethod() == dataflow.BinaryEncodingMethod_HEX {
		csvConfig.BinaryEncodingMethod = BinaryEncodingMethodHex
	}
	if dataFormat.GetDelimiter() != "" {
		csvConfig.Delimiter = dataFormat.GetDelimiter()
	}
	if dataFormat.GetNull() != "" {
		csvConfig.NullString = dataFormat.GetNull()
	}
	if dataFormat.GetQuote() != "" {
		csvConfig.Quote = dataFormat.GetQuote()
	}
	csvConfig.OutputFieldHeader = pointer.ToBool(dataFormat.GetOutputFieldHeader())
	return csvConfig
}