package impl

import (
	"context"
	"encoding/json"
	"net/url"
	"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"
	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"
)

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

func NewAzureBlobStorageDownstreamAdaptor(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient,
	encryptor encryptor_wrapper.Encryptor,
) biz_changefeed.DownstreamAdaptor {
	return &azureBlobStorageDownstreamAdaptor{
		cdcPreCheckCli: cdcPreCheckCli,
		encryptor:      encryptor,
	}
}

func (m *azureBlobStorageDownstreamAdaptor) TestConnection(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	azBlobParams, bizErr := m.parseAzureBlobStorageParams(downstreamParams)
	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))),
	}
	networkReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: clusterIDStr,
		Type:      string(entity.TiCDCDownStreamTypeAzureBlobStorage),
		AzureBlobStorageParam: &ticdc_v1.TestNetworkConnectionRequest_AzureBlobStorageParameter{
			BlobUri:  azBlobParams.GetEndpoint().GetBlobUri(),
			SasToken: azBlobParams.GetEndpoint().GetSasToken(),
		},
	}
	if err := m.cdcPreCheckCli.TestNetworkConnection(ctx, region, networkReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_AZURE_BLOB_STORAGE],
			err.ErrMsg())
	}
	return nil
}

//nolint:dupl
func (m *azureBlobStorageDownstreamAdaptor) PreCheckChangefeedParameters(
	ctx context.Context, cluster *core.Cluster, downstreamParams interface{},
) bizerr.BizError {
	azBlobParams, bizErr := m.parseAzureBlobStorageParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	startTSO, bizErr := parseStartTSO(azBlobParams.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.TiCDCDownStreamTypeAzureBlobStorage),
		StartTso:  startTSO,
	}
	if err := m.cdcPreCheckCli.ValidateChangefeedParameters(ctx, region, changefeedParamReq); err != nil {
		return bizerr.New(config.ErrCodeCDCFailedToConnectErr,
			config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ConnectorType_AZURE_BLOB_STORAGE],
			err.ErrMsg())
	}
	return nil
}

func (m *azureBlobStorageDownstreamAdaptor) EncryptPasswordInPlace(
	ctx context.Context, downstreamParams interface{},
) bizerr.BizError {
	azBlobParams, bizErr := m.parseAzureBlobStorageParams(downstreamParams)
	if bizErr != nil {
		return bizErr
	}
	if azBlobParams.GetEndpoint().GetSasToken() == "" {
		return nil
	}

	return m.encryptor.EncryptBase64Inplace(&azBlobParams.Endpoint.SasToken)
}

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

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

func (m *azureBlobStorageDownstreamAdaptor) assembleURIConfig(
	azBlobParams *dataflow.AzureBlobStorageInfo,
) (*domain.TiCDCChangefeedAzureBlobStorageURIConfigData, bizerr.BizError) {
	u, err := url.Parse(azBlobParams.GetEndpoint().GetBlobUri())
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(
			err, config.ErrCodeCDCInvalidParam, "parse azure blob storage uri error")
	}
	i := strings.Index(u.Hostname(), ".")
	if i == -1 {
		return nil, bizerr.New(config.ErrCodeCDCInvalidParam, "can't find storage account name from uri")
	}
	storageAccount := u.Hostname()[:i]
	blobPath := strings.Trim(u.Path, "/")

	uriConfig := &domain.TiCDCChangefeedAzureBlobStorageURIConfigData{
		BlobPath:             blobPath,
		AccountName:          storageAccount,
		ForcePathStyle:       pointer.ToBool(true),
		FlushInterval:        azBlobParams.GetStorageFlushConfig().GetFlushInterval(),
		FileSize:             azBlobParams.GetStorageFlushConfig().GetFileSize(),
		TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
	}
	if azBlobParams.GetEndpoint().GetSasToken() != "" {
		sasToken, err := m.encryptor.DecryptBase64(azBlobParams.GetEndpoint().GetSasToken())
		if err != nil {
			return nil, err.Chain("decrypt sas token")
		}
		uriConfig.SASToken = pointer.ToString(sasToken)
	}
	if azBlobParams.GetDataFormat().GetProtocol() == dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON &&
		azBlobParams.GetDataFormat().GetEnableTidbExtension() {
		uriConfig.EnableTiDBExtension = pointer.ToBool(true)
	}
	return uriConfig, nil
}

func (m *azureBlobStorageDownstreamAdaptor) CompleteEditParams(
	ctx context.Context, editDownstreamParams interface{}, currentDownstreamParams string,
) bizerr.BizError {
	editAzBlobParams, err := m.parseAzureBlobStorageParams(editDownstreamParams)
	if err != nil {
		return err
	}
	currentAzBlobParams, err := m.parseAzureBlobStorageParamsStr(currentDownstreamParams)
	if err != nil {
		return err
	}

	editAzBlobParams.Endpoint.BlobUri = currentAzBlobParams.Endpoint.BlobUri
	if editAzBlobParams.Endpoint.SasToken == "" {
		editAzBlobParams.Endpoint.SasToken = currentAzBlobParams.Endpoint.SasToken
	} else {
		if err = m.encryptor.EncryptBase64Inplace(&editAzBlobParams.Endpoint.SasToken); err != nil {
			return err.Chain("encrypt sas token")
		}
	}
	return nil
}

func (m *azureBlobStorageDownstreamAdaptor) GetPrivateLinkEndpointID(_ interface{}) entity.ID {
	// azure blob storage downstream doesn't support private link
	return entity.ID(0)
}

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

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

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

func (m *azureBlobStorageDownstreamAdaptor) parseAzureBlobStorageParamsStr(
	downstreamParams string,
) (*dataflow.AzureBlobStorageInfo, bizerr.BizError) {
	decoder := json.NewDecoder(strings.NewReader(downstreamParams))
	decoder.DisallowUnknownFields()
	azBlobParams := new(dataflow.AzureBlobStorageInfo)
	if err := decoder.Decode(azBlobParams); err != nil {
		return nil, bizerr.WrapErrorWithCode(
			err, bizerr.ErrCodeUnknown, "unmarshal azure blob storage params error")
	}
	if _, bizErr := normalizeCloudStorageDateSeparator(azBlobParams.GetDataFormat()); bizErr != nil {
		return nil, bizErr
	}
	return azBlobParams, nil
}