package impl

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

	"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"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"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"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	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"
)

type serverlessDownstreamAdaptor struct {
	encryptor               encryptor_wrapper.Encryptor
	privatelinkEndpointRepo repo.BizPrivateLinkEndpointRepo
	coreGrpcClient          grpcli.CoreGrpcClient
	mysqlAdaptor            *mysqlDownstreamAdaptor
}

func NewServerlessDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient,
	encryptor encryptor_wrapper.Encryptor,
	privatelinkEndpointRepo repo.BizPrivateLinkEndpointRepo,
	coreGrpcClient grpcli.CoreGrpcClient,
) biz_changefeed.DownstreamAdaptor {
	return &serverlessDownstreamAdaptor{
		encryptor:               encryptor,
		privatelinkEndpointRepo: privatelinkEndpointRepo,
		coreGrpcClient:          coreGrpcClient,
		mysqlAdaptor:            newMySQLDownstreamAdaptor(cdcPreCheckCli, encryptor),
	}
}

func (m *serverlessDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	serverlessParams, bizErr := m.parseServerlessParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	if err := m.fillServerlessEndpoint(ctx, serverlessParams); err != nil {
		return err.Chain("fill serverless endpoint")
	}
	return m.mysqlAdaptor.TestConnection(ctx, cluster, serverlessParams.GetMysqlInfo())
}

func (m *serverlessDownstreamAdaptor) fillServerlessEndpoint(
	ctx context.Context, serverlessParams *dataflow.ServerlessTidbInfo,
) bizerr.BizError {
	endpointAddr, err := m.queryRegionalDNSByPlEndpointID(
		ctx, serverlessParams.GetMysqlInfo().GetPrivatelinkInfo().GetDfPrivatelinkEndpointId())
	if err != nil {
		return err
	}
	serverlessTiDBPort, err := m.queryServerlessTiDBPort(ctx, serverlessParams.GetClusterId())
	if err != nil {
		return err
	}
	serverlessParams.MysqlInfo.Connection.Endpoint = fmt.Sprintf("%s:%d", endpointAddr, serverlessTiDBPort)
	return nil
}

func (m *serverlessDownstreamAdaptor) queryRegionalDNSByPlEndpointID(
	ctx context.Context, id uint64,
) (string, bizerr.BizError) {
	dfPrivateLinkEndpoint, err := m.privatelinkEndpointRepo.Load(ctx, id)
	if err != nil {
		return "", err.Chain("query private link endpoint")
	}
	if dfPrivateLinkEndpoint == nil || dfPrivateLinkEndpoint.Status != entity.BizPLEndpointStatusAvailable {
		return "", bizerr.New(config.ErrCodeCDCPrivatelinkIsExpirationErr, config.ErrMsgCDCPrivatelinkIsExpirationErr)
	}
	return dfPrivateLinkEndpoint.EndpointAddress, err
}

func (m *serverlessDownstreamAdaptor) queryServerlessTiDBPort(
	ctx context.Context, clusterId uint64,
) (int32, bizerr.BizError) {
	cluster, err := m.coreGrpcClient.GetAvailableClusterByID(ctx, clusterId)
	if err != nil {
		return 0, err.Chain("get serverless TiDB cluster")
	}

	for _, c := range cluster.Connections {
		if c.CreateSuccess && c.Port != 0 {
			return c.Port, nil
		}
	}
	return 0, bizerr.New(bizerr.ErrCodeUnknown, "port of serverless TiDB cluster is not found")
}

func (m *serverlessDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	serverlessParams, bizErr := m.parseServerlessParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	return m.mysqlAdaptor.PreCheckChangefeedParameters(ctx, cluster, serverlessParams.GetMysqlInfo())
}

func (m *serverlessDownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	serverlessParams, bizErr := m.parseServerlessParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	return m.mysqlAdaptor.EncryptPasswordInPlace(ctx, serverlessParams.GetMysqlInfo())
}

func (m *serverlessDownstreamAdaptor) AssembleChangefeedConfig(
	ctx context.Context, cluster *core.Cluster, bizCf *entity.BizChangefeed,
) (*domain.TiCDCChangefeedConfig, bizerr.BizError) {
	serverlessParams, bizErr := m.parseServerlessParamsStr(bizCf.DownstreamParams)
	if bizErr != nil {
		return nil, bizErr
	}
	uriConfig, bizErr := m.mysqlAdaptor.assembleURIConfigData(ctx, serverlessParams.GetMysqlInfo())
	if bizErr != nil {
		return nil, bizErr.Chain("assemble uri config")
	}
	changefeedConfig, bizErr := m.mysqlAdaptor.assembleChangefeedConfigData(
		ctx, cluster, serverlessParams.GetMysqlInfo())
	if bizErr != nil {
		return nil, bizErr.Chain("assemble changefeed config")
	}
	// mysql changefeed config force setting EnableOldValue true. we should reset it.
	changefeedConfig.ReplicaConfig.EnableOldValue = nil
	return &domain.TiCDCChangefeedConfig{
		URIConfig: uriConfig,
		Config:    changefeedConfig,
	}, nil
}

func (m *serverlessDownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editServerlessParams, err := m.parseServerlessParams(editDownstreamParams)
	if err != nil {
		return err
	}
	currentServerlessParams, err := m.parseServerlessParamsStr(currentDownstreamParams)
	if err != nil {
		return err
	}

	editPLEndpointID := editServerlessParams.GetMysqlInfo().GetPrivatelinkInfo().GetDfPrivatelinkEndpointId()
	currentPLEndpointID := currentServerlessParams.GetMysqlInfo().GetPrivatelinkInfo().GetDfPrivatelinkEndpointId()
	if editPLEndpointID != currentPLEndpointID {
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrInvalidParam, "DfPrivatelinkEndpointID", editPLEndpointID)
	}
	editConn := editServerlessParams.GetMysqlInfo().GetConnection()
	currentConn := currentServerlessParams.GetMysqlInfo().GetConnection()
	editConn.Endpoint = currentConn.GetEndpoint()
	if editConn.GetPassword() == "" {
		editConn.Password = currentConn.GetPassword()
	} else {
		if err = m.encryptor.EncryptBase64Inplace(&editConn.Password); err != nil {
			return err.Chain("encrypt password")
		}
	}
	return nil
}

func (m *serverlessDownstreamAdaptor) GetPrivateLinkEndpointID(downstreamParams interface{}) entity.ID {
	serverlessParams, bizErr := m.parseServerlessParams(downstreamParams)
	if bizErr != nil {
		return entity.ID(0)
	}
	return entity.ID(serverlessParams.GetMysqlInfo().GetPrivatelinkInfo().GetDfPrivatelinkEndpointId())
}

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

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

func (m *serverlessDownstreamAdaptor) parseServerlessParams(
	downstreamParams interface{},
) (*dataflow.ServerlessTidbInfo, bizerr.BizError) {
	serverlessParams, ok := downstreamParams.(*dataflow.ServerlessTidbInfo)
	if !ok || serverlessParams == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstreamParams is not *dataflow.ServerlessTidbInfo type")
	}
	return serverlessParams, nil
}

func (m *serverlessDownstreamAdaptor) parseServerlessParamsStr(
	downstreamParams string,
) (*dataflow.ServerlessTidbInfo, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	serverlessParams := new(dataflow.ServerlessTidbInfo)
	if err := decoder.Decode(serverlessParams); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal serverless tidb params error")
	}
	return serverlessParams, nil
}