package impl

import (
	"context"
	"encoding/json"
	"errors"
	"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/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	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 (
	defaultTransactionAtomicity = "none"
)

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

func NewMySQLDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient, encryptor encryptor_wrapper.Encryptor,
) biz_changefeed.DownstreamAdaptor {
	return newMySQLDownstreamAdaptor(cdcPreCheckCli, encryptor)
}

func newMySQLDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient, encryptor encryptor_wrapper.Encryptor,
) *mysqlDownstreamAdaptor {
	return &mysqlDownstreamAdaptor{
		cdcPreCheckCli: cdcPreCheckCli,
		encryptor:      encryptor,
	}
}

func (m *mysqlDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	mysqlParams, bizErr := m.parseMySQLParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}

	region := &regionalapi.Region{
		Name:     cluster.RegionName,
		TenantID: cluster.TenantId,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(cluster.InfraProviderType))),
	}
	networkReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: strconv.FormatUint(cluster.Id, 10),
		Type:      string(entity.TiCDCDownStreamTypeMySQL),
		MysqlParam: &ticdc_v1.TestNetworkConnectionRequest_MySQLParameter{
			User:     mysqlParams.GetConnection().GetUsername(),
			Password: mysqlParams.GetConnection().GetPassword(),
			Url:      mysqlParams.GetConnection().GetEndpoint(),
		},
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_MYSQL],
			err.ErrMsg())
	}
	return nil
}

func (m *mysqlDownstreamAdaptor) parseTestMySQLConnectionError(connErr error) bizerr.BizError {
	if connErr == nil {
		return nil
	}

	errMsg := connErr.Error()
	var bzErr bizerr.BizError
	if ok := errors.As(connErr, &bzErr); ok {
		errMsg = bzErr.ErrMsg()
	}
	if strings.Contains(errMsg, "Access denied for user") {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr, config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_MYSQL], config.ErrMsgConnectToMySQL_AccessDenied)
	}
	return bizerr.New(config.ErrCodeCDCFailedToConnectErr, config.ErrMsgCDCFailedToConnectErr,
		DownstreamDisplayNameMapping[dataflow.ConnectorType_MYSQL], errMsg)
}

func (m *mysqlDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	mysqlParams, bizErr := m.parseMySQLParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	startTSO, bizErr := parseStartTSO(mysqlParams.GetStartPosition())
	if bizErr != nil {
		return bizErr
	}
	if startTSO == 0 {
		// start tso 0 means replication from now, no need check
		return nil
	}

	region := &regionalapi.Region{
		Name:     cluster.RegionName,
		TenantID: cluster.TenantId,
		CloudProvider: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(cluster.InfraProviderType))),
	}
	changefeedParamReq := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: strconv.FormatUint(cluster.Id, 10),
		Type:      string(entity.CDCDownStreamTypeMysql),
		StartTso:  startTSO,
	}
	if err := m.cdcPreCheckCli.ValidateChangefeedParameters(ctx, region, changefeedParamReq); err != nil {
		return err
	}
	return nil
}

func (m *mysqlDownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	mysqlParams, bizErr := m.parseMySQLParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	if mysqlParams.GetConnection().GetPassword() == "" {
		return nil
	}

	return m.encryptor.EncryptBase64Inplace(&mysqlParams.Connection.Password)
}

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

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

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

func (m *mysqlDownstreamAdaptor) assembleURIConfigData(
	ctx context.Context, mysqlParams *dataflow.MySQLInfo,
) (*domain.TiCDCChangefeedMySQLURIConfigData, bizerr.BizError) {
	transactionAtomicity := defaultTransactionAtomicity
	uriConfig := &domain.TiCDCChangefeedMySQLURIConfigData{
		TransactionAtomicity: &transactionAtomicity,
	}
	if mysqlParams.GetConnection().GetUsername() != "" {
		uriConfig.User = pointer.ToString(mysqlParams.GetConnection().GetUsername())
	}
	if mysqlParams.GetConnection().GetPassword() != "" {
		password, bizErr := m.encryptor.DecryptBase64(mysqlParams.GetConnection().GetPassword())
		if bizErr != nil {
			return nil, bizErr
		}
		uriConfig.Password = &password
	}
	if mysqlParams.GetConnection().GetEndpoint() != "" {
		uriConfig.Endpoint = pointer.ToString(mysqlParams.GetConnection().GetEndpoint())
	}
	return uriConfig, nil
}

func (m *mysqlDownstreamAdaptor) assembleChangefeedConfigData(
	ctx context.Context, cluster *core.Cluster, mysqlParams *dataflow.MySQLInfo,
) (*domain.TiCDCChangefeedConfigData, bizerr.BizError) {
	startTSO, bizErr := parseStartTSO(mysqlParams.GetStartPosition())
	if bizErr != nil {
		return nil, bizErr
	}

	sqlMode := internal_config.Global().ChangefeedConfig.GetMatchedSQLMode(
		cluster.TenantId, cluster.ProjectId, cluster.Id)
	changefeedConfig := &domain.TiCDCChangefeedConfigData{
		StartTs: &startTSO,
		ReplicaConfig: &clusterproxy.ReplicaConfig{
			CaseSensitive:  pointer.ToBool(mysqlParams.GetTableConfig().GetCaseSensitive()),
			EnableOldValue: pointer.ToBool(true),
			SQLMode:        sqlMode,
		},
	}

	assembleChangefeedSinkConfigTableConfig(mysqlParams.GetTableConfig(), changefeedConfig.ReplicaConfig)
	return changefeedConfig, nil
}

func (m *mysqlDownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editMysqlParams, err := m.parseMySQLParams(editDownstreamParams)
	if err != nil {
		return err
	}
	currentMysqlParams, err := m.parseMySQLParamsStr(currentDownstreamParams)
	if err != nil {
		return err
	}

	editMysqlParams.GetConnection().Endpoint = currentMysqlParams.GetConnection().GetEndpoint()

	if editMysqlParams.Connection.Password == "" {
		editMysqlParams.Connection.Password = currentMysqlParams.GetConnection().GetPassword()
	} else {
		if err = m.encryptor.EncryptBase64Inplace(&editMysqlParams.Connection.Password); err != nil {
			return err.Chain("encrypt password")
		}
	}
	return nil
}

func (m *mysqlDownstreamAdaptor) BackfillConfig(
	ctx context.Context, currentDownstreamParams string, kernelConfig *ticdc_v1.ReplicaConfig,
) (downstreamParams string, changed bool, bizErr bizerr.BizError) {
	mysqlParams, bizErr := m.parseMySQLParamsStr(currentDownstreamParams)
	if bizErr != nil {
		return "", false, bizErr
	}
	if kernelConfig.GetCaseSensitive() && !mysqlParams.GetTableConfig().GetCaseSensitive() {
		mysqlParams.TableConfig.CaseSensitive = true
		changed = true
	}

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

func (m *mysqlDownstreamAdaptor) parseMySQLParams(downstreamParams interface{}) (*dataflow.MySQLInfo, bizerr.BizError) {
	mysqlParams, ok := downstreamParams.(*dataflow.MySQLInfo)
	if !ok || mysqlParams == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstreamParams is not *dataflow.MySQLInfo type")
	}
	return mysqlParams, nil
}

func (m *mysqlDownstreamAdaptor) parseMySQLParamsStr(downstreamParams string) (*dataflow.MySQLInfo, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	mysqlParams := new(dataflow.MySQLInfo)
	if err := decoder.Decode(mysqlParams); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal mysql params error")
	}
	return mysqlParams, nil
}