package impl

import (
	"context"
	"strconv"

	"github.com/AlekSi/pointer"
	"github.com/BurntSushi/toml"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type ticdcClusterConfigCenterImpl struct {
	configCenterGrpcCli grpcli.ConfigCenterGrpcClient
}

func NewTiCDCClusterConfigCenterManager(
	configCenterGrpcCli grpcli.ConfigCenterGrpcClient,
) config_manager.TiCDCClusterConfigManager {
	return &ticdcClusterConfigCenterImpl{
		configCenterGrpcCli: configCenterGrpcCli,
	}
}

func (m *ticdcClusterConfigCenterImpl) Create(
	ctx context.Context, req *domain.CreateTiCDCClusterConfigReq,
) (*domain.CreateTiCDCClusterConfigResp, bizerr.BizError) {
	template, tenantParams, bizErr := m.getTemplateAndTenantParams(ctx, req)
	if bizErr != nil {
		return nil, bizErr
	}

	clusterIDStr := strconv.FormatUint(req.TiCDCClusterID, 10)
	componentStr := string(domain.ConfigComponentTiCDC)
	configFormat := config_center.Format_TOML
	clusterMetaReq := &config_center.UpdateClusterMetaRequest{
		ClusterType: config_center.ClusterType_TICDC,
		ClusterId:   clusterIDStr,
		ConfigurationFileRequests: []*config_center.UpdateClusterConfigurationFileRequest{
			{
				ConfigurationFilename: componentStr,
				ConfigurationType:     &componentStr,
				BelongToComponent:     &componentStr,
				Format:                &configFormat,
				TemplateId:            &template.TemplateId,
				CustomParams:          tenantParams,
				AllowMissing:          true,
			},
		},
		ComponentRequests: []*config_center.UpdateClusterComponentRequest{
			{
				ComponentName: componentStr,
				Version:       &req.TiDBClusterVersion,
				CloudProvider: pointer.ToString(string(req.CloudProvider)),
				AllowMissing:  true,
			},
		},
	}
	_, bizErr = m.configCenterGrpcCli.UpdateClusterMeta(ctx, clusterMetaReq)
	if bizErr != nil {
		return nil, bizErr.Chain("create cluster configuration and component")
	}
	return &domain.CreateTiCDCClusterConfigResp{}, nil
}

func (m *ticdcClusterConfigCenterImpl) getTemplateAndTenantParams(
	ctx context.Context, req *domain.CreateTiCDCClusterConfigReq,
) (*config_center.ConfigurationTemplate, map[string]*config_center.ParamItem, bizerr.BizError) {
	templateReq := &config_center.SearchDefaultConfigurationTemplateRequest{
		ClusterType:       config_center.ClusterType_TICDC,
		ConfigurationType: string(domain.ConfigComponentTiCDC),
		Version:           req.TiDBClusterVersion,
	}
	template, bizErr := m.configCenterGrpcCli.SearchDefaultConfigurationTemplate(ctx, templateReq)
	if bizErr != nil {
		return nil, nil, bizErr.Chain("search ticdc cluster configuration template")
	}

	tenantParamsReq := &config_center.GetTenantInitialCustomParamsRequest{
		// reuse tenant init params of tidb cluster level
		ClusterType:       config_center.ClusterType_DEDICATED_TIDB,
		ConfigurationType: string(domain.ConfigComponentTiCDC),
		TenantId:          req.TenantID,
	}
	tenantParams, bizErr := m.configCenterGrpcCli.GetTenantInitialCustomParams(ctx, tenantParamsReq)
	if bizErr != nil {
		return nil, nil, bizErr.Chain("get ticdc cluster tenant initial custom params")
	}

	return template, tenantParams.InitialParams, nil
}

//nolint:dupl
func (m *ticdcClusterConfigCenterImpl) Update(
	ctx context.Context, req *domain.UpdateTiCDCClusterConfigReq,
) bizerr.BizError {
	clusterIDStr := strconv.FormatUint(req.TiCDCClusterID, 10)
	clusterMetaReq := &config_center.UpdateClusterMetaRequest{
		ClusterType: config_center.ClusterType_TICDC,
		ClusterId:   clusterIDStr,
		ConfigurationFileRequests: []*config_center.UpdateClusterConfigurationFileRequest{
			{
				ConfigurationFilename: string(domain.ConfigComponentTiCDC),
				CustomParams:          customParamsDataflow2ConfigCenter(req.CustomParams),
			},
		},
	}
	// if the ticdc cluster is upgraded, the template id will be changed
	if req.TemplateID != nil {
		clusterMetaReq.ConfigurationFileRequests[0].TemplateId = pointer.ToInt64(int64(*req.TemplateID))
	}

	if req.TiDBClusterVersion != nil || req.CustomImageTag != nil {
		clusterMetaReq.ComponentRequests = []*config_center.UpdateClusterComponentRequest{
			{
				ComponentName:     string(domain.ConfigComponentTiCDC),
				Version:           req.TiDBClusterVersion,
				OverwriteImageTag: req.CustomImageTag,
			},
		}
	}

	if _, bizErr := m.configCenterGrpcCli.UpdateClusterMeta(ctx, clusterMetaReq); bizErr != nil {
		return bizErr.Chain("update ticdc cluster configuration and component")
	}
	return nil
}

func (m *ticdcClusterConfigCenterImpl) Get(
	ctx context.Context, req *domain.GetTiCDCClusterConfigReq,
) (*domain.GetTiCDCClusterConfigResp, bizerr.BizError) {
	clusterIDStr := strconv.FormatUint(req.TiCDCClusterID, 10)
	componentReq := &config_center.GetClusterComponentRequest{
		ClusterType:   config_center.ClusterType_TICDC,
		ClusterId:     clusterIDStr,
		ComponentName: string(domain.ConfigComponentTiCDC),
	}
	component, bizErr := m.configCenterGrpcCli.GetClusterComponent(ctx, componentReq)
	if bizErr != nil {
		return nil, bizErr.Chain("get ticdc cluster component")
	}

	configReq := &config_center.GetClusterConfigurationFileRequest{
		ClusterType:           config_center.ClusterType_TICDC,
		ClusterId:             clusterIDStr,
		ConfigurationFilename: string(domain.ConfigComponentTiCDC),
	}
	config, bizErr := m.configCenterGrpcCli.GetClusterConfigurationFile(ctx, configReq)
	if bizErr != nil {
		return nil, bizErr.Chain("get ticdc cluster configuration")
	}

	ticdcConfig := make(map[string]interface{})
	if err := toml.Unmarshal([]byte(config.Text), &ticdcConfig); err != nil {
		return nil, bizerr.WrapErrorWithCode(
			err, bizerr.ErrCodeUnknown, "unmarshal ticdc cluster configuration: %s", config.Text)
	}
	configData := &domain.TiCDCConfigData{
		BaseImage: &component.BaseImage,
		Tag:       &component.EffectiveImageTag,
		Config:    ticdcConfig,
	}
	return &domain.GetTiCDCClusterConfigResp{
		Configuration: configData,
	}, nil
}

func (m *ticdcClusterConfigCenterImpl) GetVersionAndImageTag(
	ctx context.Context, req *domain.GetTiCDCClusterVersionAndImageTagReq,
) (*domain.GetTiCDCClusterVersionAndImageTagResp, bizerr.BizError) {
	component, bizErr := m.configCenterGrpcCli.GetClusterComponent(ctx, &config_center.GetClusterComponentRequest{
		ClusterType:   config_center.ClusterType_TICDC,
		ClusterId:     strconv.FormatUint(req.TiCDCClusterID, 10),
		ComponentName: string(domain.ConfigComponentTiCDC),
	})
	if bizErr != nil {
		return nil, bizErr.Chain("get ticdc cluster component")
	}
	if component == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "ticdc cluster component not found")
	}
	return &domain.GetTiCDCClusterVersionAndImageTagResp{
		Version:  component.GetVersion(),
		ImageTag: component.GetEffectiveImageTag(),
	}, nil
}

func (m *ticdcClusterConfigCenterImpl) PatchConfig(
	ctx context.Context, req *domain.PatchTiCDCClusterConfigReq,
) bizerr.BizError {
	clusterIDStr := strconv.FormatUint(req.TiCDCClusterID, 10)
	configReq := &config_center.GetClusterConfigurationFileRequest{
		ClusterType:           config_center.ClusterType_TICDC,
		ClusterId:             clusterIDStr,
		ConfigurationFilename: string(domain.ConfigComponentTiCDC),
	}
	config, bizErr := m.configCenterGrpcCli.GetClusterConfigurationFile(ctx, configReq)
	if bizErr != nil {
		return bizErr.Chain("get ticdc cluster configuration")
	}

	if config.CustomParams == nil {
		config.CustomParams = make(map[string]*config_center.ParamItem)
	}
	for k, param := range req.CustomParams {
		config.CustomParams[k] = &config_center.ParamItem{
			Comment: param.Comment,
			Value:   param.Value,
		}
	}

	updateReq := &config_center.UpdateClusterConfigurationFileRequest{
		ClusterType:           config.ClusterType,
		ClusterId:             clusterIDStr,
		ConfigurationFilename: config.ConfigurationFilename,
		CustomParams:          config.CustomParams,
	}
	if _, bizErr = m.configCenterGrpcCli.UpdateClusterConfigurationFile(ctx, updateReq); bizErr != nil {
		return bizErr.Chain("patch ticdc cluster configuration")
	}
	return nil
}

func (m *ticdcClusterConfigCenterImpl) GetNewClusterImage(
	ctx context.Context, req *domain.GetNewTiCDCClusterImageReq,
) (*domain.GetNewTiCDCClusterImageResp, bizerr.BizError) {
	imageReq := &config_center.GetComponentImageRequest{
		ClusterType:      config_center.ClusterType_TICDC,
		ComponentName:    string(domain.ConfigComponentTiCDC),
		ComponentVersion: req.TiCDCClusterVersion,
	}
	image, bizErr := m.configCenterGrpcCli.GetComponentImage(ctx, imageReq)
	if bizErr != nil {
		return nil, bizErr.Chain("get ticdc cluster image")
	}
	return &domain.GetNewTiCDCClusterImageResp{
		BaseImage: image.BaseImage,
		Tag:       image.Tag,
	}, nil
}