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/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type gcsDownstreamAdaptor struct {
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient
}

func NewGCSDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient,
) biz_changefeed.DownstreamAdaptor {
	return &gcsDownstreamAdaptor{
		cdcPreCheckCli: cdcPreCheckCli,
	}
}

func (m *gcsDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *utils.Cluster, downstreamParams interface{},
) bizerr.BizError {
	gcsParams, bizErr := m.parseGCSParams(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.TiCDCDownStreamTypeGCS),
		GcsParam: &ticdc_v1.TestNetworkConnectionRequest_GCSParameter{
			GcsUrl: gcsParams.GetEndpoint().GetGcsUri(),
			Region: cluster.GetRegionName(),
		},
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ChangefeedType_GCS],
			err.ErrMsg())
	}
	return nil
}

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

func (m *gcsDownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	return nil
}

func (m *gcsDownstreamAdaptor) AssembleChangefeedConfig(
	ctx context.Context, cluster *utils.Cluster, bizCf *entity.BizChangefeed,
) (*domain.TiCDCChangefeedConfig, bizerr.BizError) {
	gcsParams, bizErr := m.parseGCSParamsStr(bizCf.DownstreamParams)
	if bizErr != nil {
		return nil, bizErr
	}

	uriConfig, bizErr := m.assembleURIConfig(gcsParams)
	if bizErr != nil {
		return nil, bizErr.Chain("assemble uri config")
	}
	changefeedConfig, bizErr := assembleCloudStorageChangefeedConfigData(
		gcsParams.GetTableConfig(), gcsParams.GetStartPosition(), gcsParams.GetDataFormat())
	if bizErr != nil {
		return nil, bizErr.Chain("assemble changefeed config data")
	}
	return &domain.TiCDCChangefeedConfig{
		URIConfig: uriConfig,
		Config:    changefeedConfig,
	}, nil
}

func (m *gcsDownstreamAdaptor) assembleURIConfig(
	gcsParams *dataflow.GcsInfo,
) (*domain.TiCDCChangefeedGcsURIConfigData, bizerr.BizError) {
	uriConfig := &domain.TiCDCChangefeedGcsURIConfigData{
		GcsURI:               gcsParams.GetEndpoint().GetGcsUri(),
		ForcePathStyle:       pointer.ToBool(true),
		FlushInterval:        gcsParams.GetStorageFlushConfig().GetFlushInterval(),
		FileSize:             gcsParams.GetStorageFlushConfig().GetFileSize(),
		TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
	}
	if gcsParams.GetDataFormat().GetProtocol() == dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON &&
		gcsParams.GetDataFormat().GetEnableTidbExtension() {
		uriConfig.EnableTiDBExtension = pointer.ToBool(true)
	}
	return uriConfig, nil
}

func (m *gcsDownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editGCSParams, err := m.parseGCSParams(editDownstreamParams)
	if err != nil {
		return err
	}
	currentGCSParams, err := m.parseGCSParamsStr(currentDownstreamParams)
	if err != nil {
		return err
	}

	editGCSParams.Endpoint.GcsUri = currentGCSParams.Endpoint.GcsUri
	return nil
}

func (m *gcsDownstreamAdaptor) parseGCSParams(downstreamParams interface{}) (*dataflow.GcsInfo, bizerr.BizError) {
	gcsParams, ok := downstreamParams.(*dataflow.GcsInfo)
	if !ok || gcsParams == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "downstreamParams is not *dataflow.GcsInfo type")
	}
	if _, bizErr := normalizeCloudStorageDateSeparator(gcsParams.GetDataFormat()); bizErr != nil {
		return nil, bizErr
	}
	return gcsParams, nil
}

func (m *gcsDownstreamAdaptor) parseGCSParamsStr(downstreamParams string) (*dataflow.GcsInfo, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	gcsParams := new(dataflow.GcsInfo)
	if err := decoder.Decode(gcsParams); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "unmarshal gcs params error")
	}
	return gcsParams, nil
}