package impl

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

	"github.com/AlekSi/pointer"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/config"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"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"
	// S3ProviderAlibaba is the `provider` value understood by TiCDC/BR for
	// Alibaba Cloud OSS. It makes BR skip AWS-specific bucket-region discovery
	// and use path-style addressing suited to OSS.
	S3ProviderAlibaba = "alibaba"
)

// s3SinkProvider maps the API S3Provider enum to the `provider` query-parameter
// value understood by TiCDC/BR. An empty string keeps the default AWS S3
// behavior (no `provider` parameter on the sink URI).
func s3SinkProvider(provider dataflow.S3Provider) string {
	if provider == dataflow.S3Provider_S3_PROVIDER_ALIBABA_OSS {
		return S3ProviderAlibaba
	}
	return ""
}

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

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

func (m *s3DownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *utils.Cluster, downstreamParams interface{},
) bizerr.BizError {
	s3Params, bizErr := m.parseS3Params(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	clusterIDStr := strconv.FormatUint(cluster.GetClusterID(), 10)
	region := &regionalapi.Region{SharedPoolID: cluster.GetSharedPoolId()}
	networkReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: clusterIDStr,
		Type:      string(entity.TiCDCDownStreamTypeS3),
		S3Param: &ticdc_v1.TestNetworkConnectionRequest_S3Parameter{
			S3Uri:    s3Params.Endpoint.S3Uri,
			Endpoint: s3Params.Endpoint.Endpoint,
		},
	}
	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.ChangefeedType_S3],
			err.ErrMsg())
	}
	return nil
}

func (m *s3DownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *utils.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.GetClusterID(), 10)
	region := &regionalapi.Region{SharedPoolID: cluster.GetSharedPoolId()}
	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.ChangefeedType_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 *utils.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 *utils.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 endpoint := s3Params.GetEndpoint().GetEndpoint(); endpoint != "" {
		uriConfig.Endpoint = pointer.ToString(endpoint)
	}
	if provider := s3SinkProvider(s3Params.GetEndpoint().GetProvider()); provider != "" {
		uriConfig.Provider = pointer.ToString(provider)
	}
	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
	editS3Params.Endpoint.Endpoint = currentS3Params.Endpoint.Endpoint
	editS3Params.Endpoint.Provider = currentS3Params.Endpoint.Provider
	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) 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 *utils.Cluster) (string, bizerr.BizError) {
	req := &domain.GetTiCDCClusterRoleReq{
		SharedPoolID:  cluster.GetSharedPoolId(),
		TiDBClusterID: cluster.GetClusterID(),
		Region:        cluster.GetRegionName(),
		CloudProvider: cluster.GetCloudProvider(),
	}
	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
}