package service

import (
	"context"
	"encoding/json"
	"reflect"

	"github.com/imdario/mergo"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/grpc/grpcli"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protowire"
)

const (
	essentialChangefeedConfigFileName      = "changefeed"
	essentialChangefeedURIFileName         = "changefeed-uri"
	essentialChangefeedConfigTypePrefix    = "changefeed-"
	essentialChangefeedURIConfigTypePrefix = "changefeed-uri-"
	clearCustomParamsFieldNumber           = protowire.Number(12)
)

func (s *cdcService) createChangefeedConfigurations(
	ctx context.Context, changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	return createChangefeedConfigurationsWithClient(ctx, s.configCenter, changefeed)
}

func createChangefeedConfigurationsWithClient(
	ctx context.Context, configCenter grpcli.ConfigCenterClient, changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	downstreamType, err := changefeedDownstreamType(changefeed)
	if err != nil {
		return err
	}
	changefeedID := changefeed.ID
	if err := createChangefeedConfigurationWithClient(
		ctx, configCenter, changefeedID,
		essentialChangefeedConfigTypePrefix+downstreamType, essentialChangefeedConfigFileName,
	); err != nil {
		return err.Chain("create changefeed config")
	}
	if err := createChangefeedConfigurationWithClient(
		ctx, configCenter, changefeedID,
		essentialChangefeedURIConfigTypePrefix+downstreamType, essentialChangefeedURIFileName,
	); err != nil {
		return err.Chain("create changefeed uri config")
	}
	return nil
}

func createChangefeedConfigurationWithClient(
	ctx context.Context, configCenter grpcli.ConfigCenterClient, changefeedID, configType, filename string,
) bizerr.BizError {
	template, err := configCenter.SearchDefaultConfigurationTemplate(
		ctx,
		&config_center.SearchDefaultConfigurationTemplateRequest{
			ClusterType:       grpcli.EssentialChangefeedClusterType,
			ConfigurationType: configType,
		},
	)
	if err != nil {
		return err.Chain("search configuration template")
	}
	format := config_center.Format_JSON
	_, err = configCenter.UpdateClusterMeta(ctx, &config_center.UpdateClusterMetaRequest{
		ClusterType: grpcli.EssentialChangefeedClusterType,
		ClusterId:   changefeedID,
		ConfigurationFileRequests: []*config_center.UpdateClusterConfigurationFileRequest{
			{
				ConfigurationFilename: filename,
				ConfigurationType:     &configType,
				Format:                &format,
				TemplateId:            &template.TemplateId,
				AllowMissing:          true,
			},
		},
	})
	if err != nil {
		return err.Chain("create configuration")
	}
	return nil
}

func getChangefeedConfigurationWithClient(
	ctx context.Context, configCenter grpcli.ConfigCenterClient, changefeed *entity.CDCChangefeed,
) (primaryConfig, uriConfig *config_center.ConfigurationFile, err bizerr.BizError) {
	if changefeed == nil || changefeed.ID == "" {
		return nil, nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}
	primaryConfig, err = getChangefeedConfigurationFileWithClient(
		ctx, configCenter, changefeed.ID, essentialChangefeedConfigFileName,
	)
	if err != nil && err.ErrCode() == bizerr.ErrCodeRecordNotFound {
		if createErr := createChangefeedConfigurationsWithClient(ctx, configCenter, changefeed); createErr != nil {
			return nil, nil, createErr
		}
		primaryConfig, err = getChangefeedConfigurationFileWithClient(
			ctx, configCenter, changefeed.ID, essentialChangefeedConfigFileName,
		)
	}
	if err != nil {
		return nil, nil, err.Chain("get changefeed config")
	}
	uriConfig, err = getChangefeedConfigurationFileWithClient(
		ctx, configCenter, changefeed.ID, essentialChangefeedURIFileName,
	)
	if err != nil && err.ErrCode() == bizerr.ErrCodeRecordNotFound {
		if createErr := createChangefeedConfigurationsWithClient(ctx, configCenter, changefeed); createErr != nil {
			return nil, nil, createErr
		}
		uriConfig, err = getChangefeedConfigurationFileWithClient(
			ctx, configCenter, changefeed.ID, essentialChangefeedURIFileName,
		)
	}
	if err != nil {
		return nil, nil, err.Chain("get changefeed uri config")
	}
	return primaryConfig, uriConfig, nil
}

func getChangefeedConfigurationFileWithClient(
	ctx context.Context, configCenter grpcli.ConfigCenterClient, changefeedID, filename string,
) (*config_center.ConfigurationFile, bizerr.BizError) {
	config, err := configCenter.GetClusterConfigurationFile(ctx, &config_center.GetClusterConfigurationFileRequest{
		ClusterType:           grpcli.EssentialChangefeedClusterType,
		ClusterId:             changefeedID,
		ConfigurationFilename: filename,
	})
	if err != nil {
		return nil, err.Chain("get configuration")
	}
	return config, nil
}

func (s *cdcService) ApplyChangefeedConfiguration(
	ctx context.Context,
	clusterID, changefeedID, configurationFilename string,
	params map[string]*config_center.ParamItem,
	templateID *int64,
) bizerr.BizError {
	var primaryParams, uriParams map[string]*config_center.ParamItem
	switch configurationFilename {
	case essentialChangefeedConfigFileName:
		primaryParams = params
	case essentialChangefeedURIFileName:
		uriParams = params
	default:
		return bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported changefeed configuration file: %s",
			configurationFilename,
		)
	}
	return s.applyChangefeedConfigurationChanges(
		ctx, clusterID, changefeedID, primaryParams, uriParams, templateID, configurationFilename,
	)
}

func (s *cdcService) applyChangefeedConfigurationChanges(
	ctx context.Context, clusterID, changefeedID string,
	params, uriParams map[string]*config_center.ParamItem,
	templateID *int64,
	configurationFilename string,
) (err bizerr.BizError) { //nolint:gocognit
	if clusterID == "" || changefeedID == "" {
		return bizerr.New(bizerr.ErrCodeParams, "cluster_id and changefeed_id are required")
	}
	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		return err.Chain("get changefeed")
	}
	files := make([]*config_center.UpdateClusterConfigurationFileRequest, 0, 2)
	if params != nil {
		files = append(files, newChangefeedConfigUpdateRequest(
			essentialChangefeedConfigFileName, params, templateID,
		))
	}
	if uriParams != nil {
		files = append(files, newChangefeedConfigUpdateRequest(
			essentialChangefeedURIFileName, uriParams, templateID,
		))
	}
	if len(files) == 0 && templateID != nil {
		files = append(files, &config_center.UpdateClusterConfigurationFileRequest{
			ConfigurationFilename: configurationFilename,
			TemplateId:            templateID,
		})
	}
	if len(files) == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "at least one changefeed configuration is required")
	}
	if err = ensureChangefeedConfigurationsWithClient(ctx, s.configCenter, changefeed); err != nil {
		return err.Chain("ensure changefeed configuration")
	}

	return runChangefeedConfigApplyWorkflow(
		ctx,
		clusterID,
		changefeedID,
		func() bizerr.BizError {
			_, stopErr := s.StopChangefeed(ctx, clusterID, changefeedID)
			return stopErr
		},
		func() bizerr.BizError {
			_, updateErr := s.configCenter.UpdateClusterMeta(ctx, &config_center.UpdateClusterMetaRequest{
				ClusterType:               grpcli.EssentialChangefeedClusterType,
				ClusterId:                 changefeedID,
				ConfigurationFileRequests: files,
			})
			return updateErr
		},
		func() bizerr.BizError {
			return s.applyChangefeedConfiguration(ctx, changefeed)
		},
		func() bizerr.BizError {
			_, resumeErr := s.StartChangefeed(ctx, clusterID, changefeedID)
			return resumeErr
		},
	)
}

func newChangefeedConfigUpdateRequest(
	filename string,
	params map[string]*config_center.ParamItem,
	templateID *int64,
) *config_center.UpdateClusterConfigurationFileRequest {
	req := &config_center.UpdateClusterConfigurationFileRequest{
		ConfigurationFilename: filename,
		CustomParams:          params,
		TemplateId:            templateID,
	}
	if len(params) == 0 {
		// The currently pinned Config Center proto predates clear_custom_params,
		// while the server already supports it as field 12. Preserve compatibility
		// by sending the field through protobuf unknown fields until the dependency
		// can be upgraded together with its gRPC requirements.
		unknown := req.ProtoReflect().GetUnknown()
		unknown = protowire.AppendTag(unknown, clearCustomParamsFieldNumber, protowire.VarintType)
		unknown = protowire.AppendVarint(unknown, 1)
		req.ProtoReflect().SetUnknown(unknown)
	}
	return req
}

func ensureChangefeedConfigurationsWithClient(
	ctx context.Context, configCenter grpcli.ConfigCenterClient, changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	_, _, err := getChangefeedConfigurationWithClient(ctx, configCenter, changefeed)
	return err
}

func (s *cdcService) EnsureChangefeedConfiguration(
	ctx context.Context, clusterID, changefeedID string,
) bizerr.BizError {
	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		return err.Chain("get changefeed")
	}
	return ensureChangefeedConfigurationsWithClient(ctx, s.configCenter, changefeed)
}

func runChangefeedConfigApplyWorkflow(
	ctx context.Context,
	clusterID, changefeedID string,
	pause, updateConfiguration, applyConfiguration, resume func() bizerr.BizError,
) (err bizerr.BizError) {
	if err = pause(); err != nil {
		return err.Chain("pause changefeed")
	}
	defer func() {
		if resumeErr := resume(); resumeErr != nil {
			log.WithContextL(ctx).Error(
				"resume changefeed after applying config failed",
				zap.String("clusterID", clusterID),
				zap.String("changefeedID", changefeedID),
				zap.Error(resumeErr),
			)
			if err == nil {
				err = resumeErr.Chain("resume changefeed")
			}
		}
	}()

	if err = updateConfiguration(); err != nil {
		return err.Chain("update configuration")
	}
	if err = applyConfiguration(); err != nil {
		return err.Chain("update changefeed config")
	}
	return nil
}

func changefeedDownstreamType(changefeed *entity.CDCChangefeed) (string, bizerr.BizError) {
	if changefeed == nil {
		return "", bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}

	switch changefeed.SinkType {
	case constant.ChangefeedTypeKafka:
		return "kafka", nil
	case constant.ChangefeedTypeMySQL:
		return "mysql", nil
	case constant.ChangefeedTypeCloudStorage:
		if changefeed.CloudStorageInfo == nil || changefeed.CloudStorageInfo.CloudStorage == nil {
			return "", bizerr.New(bizerr.ErrCodeParams, "cloud storage sink configuration is required")
		}
		switch changefeed.CloudStorageInfo.CloudStorage.Type {
		case constant.CloudStorageTypeS3, constant.CloudStorageTypeOSS:
			return "s3", nil
		case constant.CloudStorageTypeGCS:
			return "gcs", nil
		case constant.CloudStorageTypeAzureBlob:
			return "azblob", nil
		case constant.CloudStorageTypeTiDBCloud:
			return "serverless_tidb", nil
		default:
			return "", bizerr.New(
				bizerr.ErrCodeParams,
				"unsupported cloud storage sink type: %s",
				changefeed.CloudStorageInfo.CloudStorage.Type,
			)
		}
	default:
		return "", bizerr.New(bizerr.ErrCodeParams, "unsupported changefeed sink type: %s", changefeed.SinkType)
	}
}

func assembleLegacyChangefeedConfig(
	ctx context.Context,
	configCenter grpcli.ConfigCenterClient,
	changefeed *entity.CDCChangefeed,
	userConfig *serverless.ChangefeedConfig,
	currentReplicaConfig *serverless.ReplicaConfig,
) (*serverless.ChangefeedConfig, bizerr.BizError) {
	primaryConfig, uriConfig, err := getChangefeedConfigurationWithClient(ctx, configCenter, changefeed)
	if err != nil {
		return nil, err
	}
	sinkURI, err := assembleChangefeedSinkURI(changefeed, userConfig, uriConfig.GetText())
	if err != nil {
		return nil, err.Chain("assemble changefeed sink uri")
	}
	replicaConfig, err := mergeLegacyReplicaConfig(
		currentReplicaConfig, userConfig.ReplicaConfig, primaryConfig.GetText(),
	)
	if err != nil {
		return nil, err
	}
	effectiveConfig := *userConfig
	effectiveConfig.SinkURI = &sinkURI
	effectiveConfig.ReplicaConfig = replicaConfig
	return &effectiveConfig, nil
}

func assembleNewArchChangefeedConfig(
	ctx context.Context,
	configCenter grpcli.ConfigCenterClient,
	changefeed *entity.CDCChangefeed,
	userConfig *serverless.ChangefeedConfig,
	currentReplicaConfig *clusterproxy.ReplicaConfig,
) (*clusterproxy.ChangefeedConfig, bizerr.BizError) {
	primaryConfig, uriConfig, err := getChangefeedConfigurationWithClient(ctx, configCenter, changefeed)
	if err != nil {
		return nil, err
	}
	sinkURI, err := assembleChangefeedSinkURI(changefeed, userConfig, uriConfig.GetText())
	if err != nil {
		return nil, err.Chain("assemble changefeed sink uri")
	}
	userConfigV2, rawErr := convertServerlessChangefeedConfigToV2(userConfig)
	if rawErr != nil {
		return nil, bizerr.WrapError(rawErr, "convert user changefeed config")
	}
	replicaConfig, err := mergeNewArchReplicaConfig(
		currentReplicaConfig, userConfigV2.ReplicaConfig, primaryConfig.GetText(),
	)
	if err != nil {
		return nil, err
	}
	userConfigV2.SinkURI = sinkURI
	userConfigV2.ReplicaConfig = replicaConfig
	return userConfigV2, nil
}

func (s *cdcService) applyChangefeedConfiguration(
	ctx context.Context, changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	userConfig, err := changefeed.ToChangefeedForUpdate(ctx, s.encryptor)
	if err != nil {
		return err.Chain("assemble user changefeed config")
	}

	if isNewArchChangefeed(changefeed) {
		return s.applyNewArchChangefeedConfiguration(ctx, changefeed, userConfig)
	}
	return s.applyLegacyChangefeedConfiguration(ctx, changefeed, userConfig)
}

func (s *cdcService) applyLegacyChangefeedConfiguration(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	userConfig *serverless.ChangefeedConfig,
) bizerr.BizError {
	proxy, err := s.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	if err != nil {
		return err.Chain("get cluster proxy")
	}
	kernelChangefeed, err := proxy.GetServerlessChangefeed(ctx, changefeed.KeySpaceID, changefeed.ChangefeedID)
	if err != nil {
		return err.Chain("get kernel changefeed")
	}
	effectiveConfig, mergeErr := assembleLegacyChangefeedConfig(
		ctx, s.configCenter, changefeed, userConfig, kernelChangefeed.Config,
	)
	if mergeErr != nil {
		return mergeErr
	}
	if _, err = proxy.UpdateServerlessChangefeed(
		ctx, changefeed.KeySpaceID, changefeed.ChangefeedID, effectiveConfig,
	); err != nil {
		return err.Chain("update legacy changefeed")
	}
	return nil
}

func (s *cdcService) applyNewArchChangefeedConfiguration(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	userConfig *serverless.ChangefeedConfig,
) bizerr.BizError {
	client, err := s.getNewArchTiCDCClient(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	if err != nil {
		return err.Chain("get newarch ticdc client")
	}
	clusterName := s.loadNewArchShardForChangefeed(ctx, changefeed).ClusterName
	keyspaceName, err := s.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
	if err != nil {
		return err
	}
	kernelChangefeed, rawErr := client.GetChangefeed(ctx, clusterName, keyspaceName, changefeed.ChangefeedID)
	if rawErr != nil {
		return bizerr.WrapError(rawErr, "get newarch kernel changefeed")
	}
	effectiveConfig, mergeErr := assembleNewArchChangefeedConfig(
		ctx, s.configCenter, changefeed, userConfig, kernelChangefeed.Config,
	)
	if mergeErr != nil {
		return mergeErr
	}
	if _, rawErr = client.UpdateChangefeed(
		ctx, clusterName, keyspaceName, changefeed.ChangefeedID, effectiveConfig,
	); rawErr != nil {
		return bizerr.WrapError(rawErr, "update newarch changefeed")
	}
	return nil
}

func mergeLegacyReplicaConfig(
	current, userDefined *serverless.ReplicaConfig, configCenterText string,
) (*serverless.ReplicaConfig, bizerr.BizError) {
	configCenterDefined := new(serverless.ReplicaConfig)
	if err := json.Unmarshal([]byte(configCenterText), configCenterDefined); err != nil {
		return nil, bizerr.WrapError(err, "unmarshal changefeed replica config")
	}
	if err := mergeChangefeedConfig(configCenterDefined, userDefined, false); err != nil {
		return nil, err
	}
	dst, err := cloneReplicaConfig(current)
	if err != nil {
		return nil, err
	}
	if err := mergeChangefeedConfig(dst, configCenterDefined, true); err != nil {
		return nil, err
	}
	return dst, nil
}

func mergeNewArchReplicaConfig(
	current, userDefined *clusterproxy.ReplicaConfig, configCenterText string,
) (*clusterproxy.ReplicaConfig, bizerr.BizError) {
	configCenterDefined := new(clusterproxy.ReplicaConfig)
	if err := json.Unmarshal([]byte(configCenterText), configCenterDefined); err != nil {
		return nil, bizerr.WrapError(err, "unmarshal changefeed replica config")
	}
	if err := mergeChangefeedConfig(configCenterDefined, userDefined, false); err != nil {
		return nil, err
	}
	dst, err := cloneReplicaConfig(current)
	if err != nil {
		return nil, err
	}
	if err := mergeChangefeedConfig(dst, configCenterDefined, true); err != nil {
		return nil, err
	}
	return dst, nil
}

func cloneReplicaConfig[T any](src *T) (*T, bizerr.BizError) {
	if src == nil {
		return new(T), nil
	}
	data, err := json.Marshal(src)
	if err != nil {
		return nil, bizerr.WrapError(err, "marshal current changefeed replica config")
	}
	dst := new(T)
	if err := json.Unmarshal(data, dst); err != nil {
		return nil, bizerr.WrapError(err, "unmarshal current changefeed replica config")
	}
	return dst, nil
}

func mergeChangefeedConfig(dst, src any, overrideEmptySlice bool) bizerr.BizError {
	options := []func(*mergo.Config){mergo.WithOverride, mergo.WithTransformers(&boolMergoTransformer{})}
	if overrideEmptySlice {
		options = append(options, mergo.WithOverrideEmptySlice)
	}
	if err := mergo.Merge(dst, src, options...); err != nil {
		return bizerr.WrapError(err, "merge changefeed replica config")
	}
	return nil
}

type boolMergoTransformer struct{}

func (*boolMergoTransformer) Transformer(typ reflect.Type) func(dst, src reflect.Value) error {
	if typ != reflect.TypeOf((*bool)(nil)) {
		return nil
	}
	return func(dst, src reflect.Value) error {
		if !src.IsNil() {
			dst.Set(src)
		}
		return nil
	}
}