package impl

import (
	"context"
	"encoding/json"
	"errors"
	"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/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 (
	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 &mysqlDownstreamAdaptor{
		cdcPreCheckCli: cdcPreCheckCli,
		encryptor:      encryptor,
	}
}

func (m *mysqlDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *utils.Cluster, downstreamParams interface{},
) bizerr.BizError {
	mysqlParams, bizErr := m.parseMySQLParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	region := &regionalapi.Region{SharedPoolID: cluster.GetSharedPoolId()}
	networkReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: strconv.FormatUint(cluster.GetClusterID(), 10),
		Type:      string(entity.TiCDCDownStreamTypeMySQL),
		MysqlParam: &ticdc_v1.TestNetworkConnectionRequest_MySQLParameter{
			Url:      mysqlParams.Connection.Endpoint,
			User:     mysqlParams.Connection.Username,
			Password: mysqlParams.Connection.Password,
		},
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return m.parseTestMySQLConnectionError(err)
	}
	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.ChangefeedType_MYSQL], config.ErrMsgConnectToMySQL_AccessDenied)
	}
	return bizerr.New(config.ErrCodeCDCFailedToConnectErr, config.ErrMsgCDCFailedToConnectErr,
		DownstreamDisplayNameMapping[dataflow.ChangefeedType_MYSQL], errMsg)
}

func (m *mysqlDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *utils.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
	}

	return checkStartTSO(ctx, m.cdcPreCheckCli, cluster, startTSO)
}

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 *utils.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) 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 *utils.Cluster, mysqlParams *dataflow.MySQLInfo,
) (*domain.TiCDCChangefeedConfigData, bizerr.BizError) {
	startTSO, bizErr := parseStartTSO(mysqlParams.GetStartPosition())
	if bizErr != nil {
		return nil, bizErr
	}

	changefeedConfig := &domain.TiCDCChangefeedConfigData{
		StartTs: &startTSO,
		ReplicaConfig: &clusterproxy.ReplicaConfig{
			CaseSensitive:  pointer.ToBool(mysqlParams.GetTableConfig().GetCaseSensitive()),
			EnableOldValue: pointer.ToBool(true),
		},
	}

	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) 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
}