package impl

import (
	"context"
	"encoding/json"
	"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"
	"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 gcsDownstreamAdaptor struct {
	cdcPreCheckCli      regionalapi.TiCDCPreCheckClient
	tidbClusterInfraCli grpcli.TiDBClusterInfraClient
}

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

func (m *gcsDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	gcsParams, bizErr := m.parseGCSParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	clusterRole, bizErr := m.tidbClusterInfraCli.GetClusterNodeRole(ctx, cluster.Id)
	if bizErr != nil {
		return bizErr.Chain("get cluster node role")
	}
	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.TiCDCDownStreamTypeGCS),
		GcsParam: &ticdc_v1.TestNetworkConnectionRequest_GCSParameter{
			GcsUrl:                    gcsParams.GetEndpoint().GetGcsUri(),
			Region:                    cluster.RegionName,
			ImpersonateServiceAccount: clusterRole,
		},
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_GCS],
			err.ErrMsg())
	}
	return nil
}

//nolint:dupl
func (m *gcsDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *core.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.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.CDCDownStreamTypeGCS),
		StartTso:  startTSO,
	}
	if err := m.cdcPreCheckCli.ValidateChangefeedParameters(ctx, region, changefeedParamReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_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 *core.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) GetPrivateLinkEndpointID(_ interface{}) entity.ID {
	// gcs downstream doesn't support private link
	return entity.ID(0)
}

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

	downstreamParamsBytes, err := json.Marshal(gcsParams)
	if err != nil {
		return "", false, bizerr.WrapError(err, "marshal downstream params")
	}
	downstreamParams = string(downstreamParamsBytes)
	return downstreamParams, changed, 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
}