package impl

import (
	"context"
	"encoding/json"
	"fmt"
	"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/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	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"
)

type pulsarDownstreamAdaptor struct {
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient
	encryptor      encryptor_wrapper.Encryptor
}

func NewPulsarDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient, encryptor encryptor_wrapper.Encryptor,
) biz_changefeed.DownstreamAdaptor {
	return &pulsarDownstreamAdaptor{
		cdcPreCheckCli: cdcPreCheckCli,
		encryptor:      encryptor,
	}
}

func (m *pulsarDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	pulsarParams, bizErr := m.parsePulsarParams(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.TiCDCDownStreamTypePulsar),
		PulsarParam: &ticdc_v1.TestNetworkConnectionRequest_PulsarParameter{
			ServiceUrl: fmt.Sprintf("%s://%s",
				pulsarProtocolDataflow2TiCDC(pulsarParams.DestinationProtocol), pulsarParams.Endpoint),
			AuthenticationType: pulsarAuthTypeDataflow2ClusterProxy(pulsarParams.Authentication.AuthType),
			Username:           pulsarParams.Authentication.UserName,
			Password:           pulsarParams.Authentication.Password,
			Token:              pulsarParams.Authentication.Token,
			ConnectionTimeout:  pulsarParams.AdvancedSetting.ConnectionTimeout,
		},
	}

	if pulsarParams.TopicDistribution != nil {
		networkReq.PulsarParam.Tenant = pulsarParams.TopicDistribution.PulsarTenant
		networkReq.PulsarParam.Namespace = pulsarParams.TopicDistribution.PulsarNamespace
		networkReq.PulsarParam.Topic = pulsarParams.TopicDistribution.DefaultTopic
	}

	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_PULSAR],
			err.ErrMsg())
	}
	return nil
}

func (m *pulsarDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	pulsarParams, bizErr := m.parsePulsarParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	startTSO, bizErr := parseStartTSO(pulsarParams.GetStartPosition())
	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.TiCDCDownStreamTypePulsar),
		StartTso:  startTSO,
	}
	if err := m.cdcPreCheckCli.ValidateChangefeedParameters(ctx, region, changefeedParamReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_PULSAR],
			err.ErrMsg())
	}
	return nil
}

func (m *pulsarDownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	pulsarParams, bizErr := m.parsePulsarParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}

	if pulsarParams.Authentication.AuthType == dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC {
		if bizErr = m.encryptor.EncryptBase64Inplace(&pulsarParams.Authentication.Password); bizErr != nil {
			return bizErr.Chain("encrypt password of authentication")
		}
	} else if pulsarParams.Authentication.AuthType == dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN {
		if bizErr = m.encryptor.EncryptBase64Inplace(&pulsarParams.Authentication.Token); bizErr != nil {
			return bizErr.Chain("encrypt token of authentication")
		}
	}
	return nil
}

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

	uriConfig := m.assembleURIConfigData(pulsarParams)
	changefeedConfig, bizErr := m.assembleChangefeedConfigData(pulsarParams)
	if bizErr != nil {
		return nil, bizErr.Chain("assemble changefeed config")
	}
	return &domain.TiCDCChangefeedConfig{
		URIConfig: uriConfig,
		Config:    changefeedConfig,
	}, nil
}

func (m *pulsarDownstreamAdaptor) assembleURIConfigData(
	pulsarParams *dataflow.PulsarInfo,
) domain.TiCDCChangefeedURIConfig {
	uriConfig := &domain.TiCDCChangefeedPulsarURIConfigData{
		PulsarProtocol:  pulsarProtocolDataflow2TiCDC(pulsarParams.GetDestinationProtocol()),
		Endpoint:        pulsarParams.GetEndpoint(),
		PulsarTenant:    pulsarParams.GetTopicDistribution().GetPulsarTenant(),
		PulsarNamespace: pulsarParams.GetTopicDistribution().GetPulsarNamespace(),
		DefaultTopic:    pulsarParams.GetTopicDistribution().GetDefaultTopic(),
		DataProtocol:    utils.TransferProtocol(pulsarParams.GetDataFormat().GetProtocol()),
	}
	if pulsarParams.GetDataFormat().GetProtocol() != dataflow.Protocol_OPEN_PROTOCOL {
		uriConfig.EnableTiDBExtension = pointer.ToBool(pulsarParams.GetDataFormat().GetEnableTidbExtension())
	}
	return uriConfig
}

func (m *pulsarDownstreamAdaptor) assembleChangefeedConfigData(
	pulsarParams *dataflow.PulsarInfo,
) (*domain.TiCDCChangefeedConfigData, bizerr.BizError) {
	startTSO, bizErr := parseStartTSO(pulsarParams.GetStartPosition())
	if bizErr != nil {
		return nil, bizErr
	}
	pulsarConfig, bizErr := m.assemblePulsarConfig(pulsarParams)
	if bizErr != nil {
		return nil, bizErr
	}

	replicaConfig := &clusterproxy.ReplicaConfig{
		CaseSensitive: pointer.ToBool(pulsarParams.GetTableConfig().GetCaseSensitive()),
		Sink: &clusterproxy.SinkConfigV2{
			DispatchRules: m.assemblePulsarDispatchers(
				pulsarParams.GetTopicDistribution(), pulsarParams.GetPartitionDispatchers()),
			PulsarConfig: pulsarConfig,
		},
	}
	assembleChangefeedSinkConfigTableConfig(pulsarParams.GetTableConfig(), replicaConfig)
	return &domain.TiCDCChangefeedConfigData{
		StartTs:       &startTSO,
		ReplicaConfig: replicaConfig,
	}, nil
}

func (m *pulsarDownstreamAdaptor) assemblePulsarConfig(
	pulsarParams *dataflow.PulsarInfo,
) (*clusterproxy.PulsarConfig, bizerr.BizError) {
	pulsarConfig := &clusterproxy.PulsarConfig{
		BasicUserName:           pulsarParams.GetAuthentication().GetUserName(),
		CompressionType:         pulsarCompressionTypeDataflow2TiCDC(pulsarParams.GetCompressionType()),
		ConnectionTimeout:       int32(pulsarParams.GetAdvancedSetting().GetConnectionTimeout()),
		OperationTimeout:        int32(pulsarParams.GetAdvancedSetting().GetOperationTimeout()),
		SendTimeout:             int32(pulsarParams.GetAdvancedSetting().GetSendTimeout()),
		BatchingMaxMessages:     pulsarParams.GetAdvancedSetting().GetBatchingMaxMessages(),
		BatchingMaxPublishDelay: int32(pulsarParams.GetAdvancedSetting().GetBatchingMaxPublishDelay()),
		OutputRawChangeEvent:    pointer.ToBool(pulsarParams.GetDataFormat().GetOutputRawChangeEvent()),
	}
	if pulsarParams.GetAuthentication().GetToken() != "" {
		token, err := m.encryptor.DecryptBase64(pulsarParams.GetAuthentication().GetToken())
		if err != nil {
			return nil, err.Chain("decrypt pulsar token")
		}
		pulsarConfig.AuthenticationToken = token
	} else if pulsarParams.GetAuthentication().GetPassword() != "" {
		password, err := m.encryptor.DecryptBase64(pulsarParams.GetAuthentication().GetPassword())
		if err != nil {
			return nil, err.Chain("decrypt pulsar password")
		}
		pulsarConfig.BasicPassword = password
	}
	return pulsarConfig, nil
}

func (m *pulsarDownstreamAdaptor) assemblePulsarDispatchers(
	td *dataflow.PulsarTopicDistribution, pds []*dataflow.PartitionDispatcherConfig,
) []*clusterproxy.DispatchRuleV2 {
	topicExpr := m.generatePulsarTopicExpression(td)
	dispatchRules := make([]*clusterproxy.DispatchRuleV2, 0, len(pds))
	for _, pd := range pds {
		rule := &clusterproxy.DispatchRuleV2{
			Topic:     topicExpr,
			Partition: utils.TransferPartitionDispatcher(pd.GetPartitionType()),
			Matcher:   pd.Matcher,
			IndexName: pd.IndexName,
			Columns:   pd.Columns,
		}
		dispatchRules = append(dispatchRules, rule)
	}
	return dispatchRules
}

func (m *pulsarDownstreamAdaptor) generatePulsarTopicExpression(
	distribution *dataflow.PulsarTopicDistribution,
) string {
	if distribution.DispatchType == dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE {
		return fmt.Sprintf("persistent://%s/%s/%s{schema}%s",
			distribution.GetPulsarTenant(), distribution.GetPulsarNamespace(),
			distribution.GetTopicPrefix(), distribution.GetTopicSuffix())
	} else if distribution.DispatchType == dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE {
		return fmt.Sprintf("persistent://%s/%s/%s{schema}%s{table}%s",
			distribution.GetPulsarTenant(), distribution.GetPulsarNamespace(),
			distribution.GetTopicPrefix(), distribution.GetSeparator(), distribution.GetTopicSuffix())
	} else {
		return ""
	}
}

//nolint:nestif
func (m *pulsarDownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editPulsarParams, bizErr := m.parsePulsarParams(editDownstreamParams)
	if bizErr != nil {
		return bizErr
	}
	currentPulsarParams, bizErr := m.parsePulsarParamsStr(currentDownstreamParams)
	if bizErr != nil {
		return bizErr
	}

	editPulsarParams.Endpoint = currentPulsarParams.Endpoint
	editAuth := editPulsarParams.Authentication
	if editAuth.AuthType == dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC {
		if editAuth.Password == "" {
			if currentPulsarParams.GetAuthentication().GetPassword() == "" {
				// if the auth type is changed, the password of current params may be empty
				return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "password")
			}
			editAuth.Password = currentPulsarParams.GetAuthentication().GetPassword()
		} else {
			if err := m.encryptor.EncryptBase64Inplace(&editAuth.Password); err != nil {
				return err.Chain("encrypt password of authentication")
			}
		}
	} else if editAuth.AuthType == dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN {
		if editAuth.Token == "" {
			if currentPulsarParams.GetAuthentication().GetToken() == "" {
				// if the auth type is changed, the token of current params may be empty
				return bizerr_wrapper.NewInvalidInputError(config.ErrEmptyParam, "token")
			}
			editAuth.Token = currentPulsarParams.GetAuthentication().GetToken()
		} else {
			if err := m.encryptor.EncryptBase64Inplace(&editAuth.Token); err != nil {
				return err.Chain("encrypt token of authentication")
			}
		}
	}

	return nil
}

func (m *pulsarDownstreamAdaptor) GetPrivateLinkEndpointID(downstreamParams interface{}) entity.ID {
	return entity.ID(0)
}

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

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

func (m *pulsarDownstreamAdaptor) parsePulsarParams(
	downstreamParams interface{},
) (*dataflow.PulsarInfo, bizerr.BizError) {
	pulsarParams, ok := downstreamParams.(*dataflow.PulsarInfo)
	if !ok || pulsarParams == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstreamParams is not *dataflow.PulsarInfo type")
	}
	return pulsarParams, nil
}

func (m *pulsarDownstreamAdaptor) parsePulsarParamsStr(
	downstreamParams string,
) (*dataflow.PulsarInfo, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	pulsarParams := new(dataflow.PulsarInfo)
	if err := decoder.Decode(pulsarParams); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal pulsar params error")
	}
	return pulsarParams, nil
}

func pulsarProtocolDataflow2TiCDC(protocol dataflow.PulsarDestinationProtocol) string {
	switch protocol {
	case dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR:
		return "pulsar"
	case dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL:
		return "pulsar+ssl"
	case dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_HTTP:
		return "pulsar+http"
	case dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_HTTPS:
		return "pulsar+https"
	default:
		return ""
	}
}

func pulsarCompressionTypeDataflow2TiCDC(compression dataflow.PulsarCompressionType) string {
	switch compression {
	case dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4:
		return "lz4"
	case dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZLIB:
		return "zlib"
	case dataflow.PulsarCompressionType_PULSAR_COMPRESSION_ZSTD:
		return "zstd"
	default:
		return ""
	}
}

func pulsarAuthTypeDataflow2ClusterProxy(authType dataflow.PulsarAuthType) string {
	switch authType {
	case dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC:
		return "basic"
	case dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN:
		return "token"
	default:
		return ""
	}
}