package impl

import (
	"context"
	"fmt"
	"strconv"

	"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/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
)

const (
	changefeedConfigFileName = "changefeed"
	changefeedURIFileName    = "changefeed-uri"
)

type changefeedConfigCenterImpl struct {
	configCenterGrpcCli grpcli.ConfigCenterGrpcClient
}

func NewChangefeedConfigCenterManager(
	configCenterGrpcCli grpcli.ConfigCenterGrpcClient,
) config_manager.ChangefeedConfigManager {
	return &changefeedConfigCenterImpl{
		configCenterGrpcCli: configCenterGrpcCli,
	}
}

func (m *changefeedConfigCenterImpl) Create(
	ctx context.Context, req *domain.CreateChangefeedConfigReq,
) (*domain.CreateChangefeedConfigResp, bizerr.BizError) {
	changefeedIDStr := strconv.FormatUint(req.ChangefeedID, 10)
	changefeedConfigType := genChangefeedConfigType(req.DownstreamType)
	if err := m.createChangefeedConfig(
		ctx, changefeedIDStr, changefeedConfigType, changefeedConfigFileName); err != nil {
		return nil, err.Chain("create changefeed config")
	}
	changefeedURIConfigType := genChangefeedURIConfigType(req.DownstreamType)
	if err := m.createChangefeedConfig(
		ctx, changefeedIDStr, changefeedURIConfigType, changefeedURIFileName); err != nil {
		return nil, err.Chain("create changefeed uri config")
	}
	return &domain.CreateChangefeedConfigResp{}, nil
}

func (m *changefeedConfigCenterImpl) createChangefeedConfig(
	ctx context.Context, changefeedID, configType, configFileName string,
) bizerr.BizError {
	configTemplateReq := &config_center.SearchDefaultConfigurationTemplateRequest{
		ClusterType:       config_center.ClusterType_CHANGEFEED,
		ConfigurationType: configType,
	}
	template, bizErr := m.configCenterGrpcCli.SearchDefaultConfigurationTemplate(ctx, configTemplateReq)
	if bizErr != nil {
		return bizErr.Chain("search configuration template")
	}
	configFormat := config_center.Format_JSON
	clusterMetaReq := &config_center.UpdateClusterMetaRequest{
		ClusterType: config_center.ClusterType_CHANGEFEED,
		ClusterId:   changefeedID,
		ConfigurationFileRequests: []*config_center.UpdateClusterConfigurationFileRequest{
			{
				ConfigurationFilename: configFileName,
				ConfigurationType:     &configType,
				Format:                &configFormat,
				TemplateId:            &template.TemplateId,
				AllowMissing:          true,
			},
		},
	}
	_, bizErr = m.configCenterGrpcCli.UpdateClusterMeta(ctx, clusterMetaReq)
	if bizErr != nil {
		return bizErr.Chain("create configuration")
	}
	return nil
}

func (m *changefeedConfigCenterImpl) Update(
	ctx context.Context, req *domain.UpdateChangefeedConfigReq,
) bizerr.BizError {
	changefeedIDStr := strconv.FormatUint(req.ChangefeedID, 10)
	if req.Configuration != nil {
		if err := m.updateChangefeedConfig(ctx, changefeedIDStr, changefeedConfigFileName, req.Configuration); err != nil {
			return err.Chain("update changefeed config")
		}
	}

	if req.URIConfiguration != nil {
		if err := m.updateChangefeedConfig(
			ctx, changefeedIDStr, changefeedURIFileName, req.URIConfiguration); err != nil {
			return err.Chain("update changefeed uri config")
		}
	}
	return nil
}

func (m *changefeedConfigCenterImpl) updateChangefeedConfig(
	ctx context.Context, changefeedID string, configFileName string, configParams map[string]*dataflow.ParamItem,
) bizerr.BizError {
	clusterMetaReq := &config_center.UpdateClusterMetaRequest{
		ClusterType: config_center.ClusterType_CHANGEFEED,
		ClusterId:   changefeedID,
		ConfigurationFileRequests: []*config_center.UpdateClusterConfigurationFileRequest{
			{
				ConfigurationFilename: configFileName,
				CustomParams:          customParamsDataflow2ConfigCenter(configParams),
			},
		},
	}
	if _, bizErr := m.configCenterGrpcCli.UpdateClusterMeta(ctx, clusterMetaReq); bizErr != nil {
		return bizErr.Chain("update configuration")
	}
	return nil
}

func (m *changefeedConfigCenterImpl) Get(
	ctx context.Context, req *domain.GetChangefeedConfigReq,
) (*domain.GetChangefeedConfigResp, bizerr.BizError) {
	changefeedIDStr := strconv.FormatUint(req.ChangefeedID, 10)
	changefeedConfig, err := m.getChangefeedConfig(ctx, changefeedIDStr, changefeedConfigFileName)
	if err != nil {
		return nil, err.Chain("get changefeed config")
	}

	changefeedURIConfig, err := m.getChangefeedConfig(ctx, changefeedIDStr, changefeedURIFileName)
	if err != nil {
		return nil, err.Chain("get changefeed uri config")
	}
	return &domain.GetChangefeedConfigResp{
		Configuration:    changefeedConfig,
		URIConfiguration: changefeedURIConfig,
	}, nil
}

func (m *changefeedConfigCenterImpl) getChangefeedConfig(
	ctx context.Context, changefeedID string, configFileName string,
) (string, bizerr.BizError) {
	configReq := &config_center.GetClusterConfigurationFileRequest{
		ClusterType:           config_center.ClusterType_CHANGEFEED,
		ClusterId:             changefeedID,
		ConfigurationFilename: configFileName,
	}
	config, bizErr := m.configCenterGrpcCli.GetClusterConfigurationFile(ctx, configReq)
	if bizErr != nil {
		return "", bizErr.Chain("get configuration")
	}
	return config.Text, nil
}

func genChangefeedConfigType(downstreamType entity.TiCDCChangefeedDownStreamType) string {
	return fmt.Sprintf("changefeed-%s", downstreamType)
}

func genChangefeedURIConfigType(downstreamType entity.TiCDCChangefeedDownStreamType) string {
	return fmt.Sprintf("changefeed-uri-%s", downstreamType)
}