package service

import (
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strconv"

	"github.com/tidbcloud/config-center/internal/db"
	"github.com/tidbcloud/config-center/pkg/utils"
	pb "github.com/tidbcloud/config-center/proto/config_center"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/timestamppb"
	"gopkg.in/yaml.v3"
)

// new create request
func CreateClusterConfigFile(d db.DB, req *pb.CreateClusterConfigFileRequest) (*pb.ConfigurationFile, error) {
	if err := ValidateFormat(req.ClusterType, req.ConfigurationType, req.Format); err != nil {
		return nil, err
	}

	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	configFile := &pb.ConfigurationFile{
		ClusterType:           req.GetClusterType(),
		ConfigurationFilename: req.GetConfigurationFilename(),
		ConfigurationType:     req.GetConfigurationType(),
		BelongToComponent:     req.GetBelongToComponent(),
		Format:                req.GetFormat(),
		//TemplateId:            configFile.GetTemplateId(),
		//CustomParams:          configFile.GetCustomParams(),
		//FullParams:            fullParams,
		//Text:                  text,
	}

	var systemTemplate *pb.ConfigurationTemplate
	switch source := req.SystemTemplateSource.(type) {
	case *pb.CreateClusterConfigFileRequest_FromTemplate:
		systemTemplate, err = GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
			ClusterType:       configFile.ClusterType,
			ConfigurationType: configFile.ConfigurationType,
			TemplateId:        source.FromTemplate.TemplateId,
		})
		if err != nil {
			return nil, status.Errorf(codes.Internal, "cannot get template: %v", err)
		}
	case *pb.CreateClusterConfigFileRequest_FromSku:
		systemTemplate, err = SearchDefaultConfigurationTemplate(tx, &pb.SearchDefaultConfigurationTemplateRequest{
			ClusterType:       configFile.ClusterType,
			ConfigurationType: configFile.ConfigurationType,
			Version:           source.FromSku.Version,
			CloudProvider:     source.FromSku.CloudProvider,
			NodeInstance:      source.FromSku.NodeInstance,
			TenantId:          strconv.FormatUint(source.FromSku.TenantId, 10),
		})
		if err != nil {
			return nil, err
		}
	}
	configFile.TemplateId = systemTemplate.TemplateId

	var customParams map[string]*pb.ParamItem
	switch source := req.CustomParamsSource.(type) {
	case *pb.CreateClusterConfigFileRequest_FromCustomTemplate:
		customTemplate, err := GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
			ClusterType:       configFile.ClusterType,
			ConfigurationType: configFile.ConfigurationType,
			TemplateId:        source.FromCustomTemplate.TemplateId,
		})
		if err != nil {
			return nil, status.Errorf(codes.Internal, "cannot get template: %v", err)
		}
		customParams = customTemplate.Data
	case *pb.CreateClusterConfigFileRequest_FromInputValues:
		customParams = source.FromInputValues.GetCustomParams()
	case *pb.CreateClusterConfigFileRequest_FromTenant:
		t, err := d.ListTenantInitialCustomParams(req.GetClusterType().String(), req.GetConfigurationType(), source.FromTenant.GetTenantId())
		if err != nil {
			return nil, err
		}
		customParams = make(map[string]*pb.ParamItem)
		for _, row := range t {
			v, err := pb.NewValue(row.Value.Raw)
			if err != nil {
				return nil, err
			}
			customParams[row.Key] = &pb.ParamItem{
				Value:   v,
				Comment: row.Comment,
			}
		}
	}
	configFile.CustomParams = customParams

	configFile.FullParams, configFile.Text, err = calculateConfigurationFile(configFile.GetFormat().String(), systemTemplate.GetData(), customParams)
	if err != nil {
		return nil, err
	}

	if !req.GetValidateOnly() {
		_, err = tx.InsertClusterConfigInfo(db.ClusterConfigInfoRow{
			ClusterType: configFile.ClusterType.String(),
			ClusterID:   req.GetClusterId(),
			Filename:    configFile.ConfigurationFilename,
			ConfigType:  configFile.ConfigurationType,
			BelongTo:    configFile.GetBelongToComponent(),
			Format:      configFile.Format.String(),
			TemplateID:  configFile.GetTemplateId(),
		})
		if err != nil {
			return nil, err
		}

		if len(configFile.GetCustomParams()) > 0 {
			var params []db.ConfigurationItem
			for k, v := range configFile.CustomParams {
				params = append(params, db.ConfigurationItem{
					Key:     k,
					Value:   db.NewAnyValue(v.GetValue().AsInterface()),
					Comment: v.GetComment(),
				})
			}
			err = tx.InsertClusterConfigParams(req.GetClusterType().String(), req.GetClusterId(), configFile.ConfigurationFilename, params)
			if err != nil {
				return nil, err
			}
		}

		err = tx.CreateClusterConfigText(req.GetClusterType().String(), req.GetClusterId(), configFile.ConfigurationFilename, configFile.Text)
		if err != nil {
			return nil, err
		}
	}
	return configFile, tx.Commit()
}

func CreateClusterConfigurationFile(d db.DB, req *pb.CreateClusterConfigurationFileRequest) (*pb.ConfigurationFile, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	configFile := req.GetConfigurationFile()

	configType := configFile.GetConfigurationType()
	if configType == "" {
		configType = configFile.GetConfigurationFilename()
	}

	// todo: make format from configType have higher priority
	format := configFile.GetFormat()
	if format == pb.Format_FORMAT_UNSPECIFIED {
		format = req.GetClusterType().GetConfigTypeFormat(configType)
	} else if format != req.GetClusterType().GetConfigTypeFormat(configType) {
		slog.Error("format not match", "req format", format, "code format", req.GetClusterType().GetConfigTypeFormat(configType))
	}

	template, err := GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: configType,
		TemplateId:        configFile.GetTemplateId(),
	})
	if err != nil {
		tx.Rollback()
		return nil, err
	}

	fullParams, text, err := calculateConfigurationFile(configFile.GetFormat().String(), template.GetData(), configFile.GetCustomParams())
	if err != nil {
		tx.Rollback()
		return nil, err
	}

	if !req.GetValidateOnly() {
		_, err = tx.InsertClusterConfigInfo(db.ClusterConfigInfoRow{
			ClusterType: req.GetClusterType().String(),
			ClusterID:   req.GetClusterId(),
			Filename:    configFile.GetConfigurationFilename(),
			ConfigType:  configType,
			BelongTo:    configFile.GetBelongToComponent(),
			Format:      format.String(),
			TemplateID:  configFile.GetTemplateId(),
		})
		if err != nil {
			tx.Rollback()
			return nil, err
		}

		if len(configFile.GetCustomParams()) > 0 {
			var params []db.ConfigurationItem
			for k, v := range configFile.GetCustomParams() {
				params = append(params, db.ConfigurationItem{
					Key:     k,
					Value:   db.NewAnyValue(v.GetValue().AsInterface()),
					Comment: v.GetComment(),
				})
			}
			err = tx.InsertClusterConfigParams(req.GetClusterType().String(), req.GetClusterId(), configFile.GetConfigurationFilename(), params)
			if err != nil {
				tx.Rollback()
				return nil, err
			}
		}

		err = tx.CreateClusterConfigText(req.GetClusterType().String(), req.GetClusterId(), configFile.GetConfigurationFilename(), text)
		if err != nil {
			tx.Rollback()
			return nil, err
		}
	}
	return &pb.ConfigurationFile{
		ClusterType:           req.GetClusterType(),
		ConfigurationFilename: configFile.GetConfigurationFilename(),
		ConfigurationType:     configType,
		BelongToComponent:     configFile.GetBelongToComponent(),
		Format:                format,
		TemplateId:            configFile.GetTemplateId(),
		CustomParams:          configFile.GetCustomParams(),
		FullParams:            fullParams,
		Text:                  text,
	}, tx.Commit()
}

func AutoCreateClusterConfigurationFile(d db.DB, req *pb.AutoCreateClusterConfigurationFileRequest) (*pb.ConfigurationFile, error) {
	// 1. find suitable config template
	configType := req.GetConfigurationType()
	if configType == "" {
		configType = req.GetConfigurationFilename()
	}
	tenantID := ""
	if req.TenantId != nil {
		tenantID = strconv.FormatUint(*req.TenantId, 10)
	}
	template, err := SearchDefaultConfigurationTemplate(d, &pb.SearchDefaultConfigurationTemplateRequest{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: configType,
		Version:           req.GetVersion(),
		CloudProvider:     req.GetCloudProvider(),
		NodeInstance:      req.GetNodeInstance(),
		TenantId:          tenantID,
	})
	if err != nil {
		return nil, err
	}

	// 2. auto fill custom configs for some clusters
	customConfig := make(map[string]*pb.ParamItem)
	t, err := d.ListTenantInitialCustomParams(req.GetClusterType().String(), req.GetConfigurationType(), req.GetTenantId())
	if err != nil {
		return nil, err
	}
	for _, row := range t {
		v, err := pb.NewValue(row.Value.Raw)
		if err != nil {
			return nil, err
		}
		customConfig[row.Key] = &pb.ParamItem{
			Value:   v,
			Comment: row.Comment,
		}
	}

	// 3. create config file
	return CreateClusterConfigurationFile(d, &pb.CreateClusterConfigurationFileRequest{
		ClusterType: req.GetClusterType(),
		ClusterId:   req.GetClusterId(),
		ConfigurationFile: &pb.ConfigurationFile{
			ConfigurationFilename: req.GetConfigurationFilename(),
			ConfigurationType:     req.GetConfigurationType(),
			BelongToComponent:     req.GetBelongToComponent(),
			Format:                req.GetFormat(),
			TemplateId:            template.GetTemplateId(),
			CustomParams:          customConfig,
		},
		//Format:                req.GetFormat(),
		//TemplateId:            template.GetTemplateId(),
		//CustomParams:          customConfig,
	})
}

func GetClusterConfigurationFile(d db.DB, req *pb.GetClusterConfigurationFileRequest) (*pb.ConfigurationFile, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}

	// 1. read basic info like format, templateID
	file, err := tx.GetClusterConfigInfo(req.GetClusterType().String(), req.GetClusterId(), req.GetConfigurationFilename())
	if err == db.ErrNotFound {
		tx.Rollback()
		return nil, status.Errorf(codes.NotFound, "ConfigurationFile not found")
	} else if err != nil {
		tx.Rollback()
		return nil, err
	}

	// 2. read custom params
	configParams, err := tx.ListClusterConfigParams(file.ClusterType, file.ClusterID, file.Filename)
	if err != nil {
		tx.Rollback()
		return nil, err
	}
	customParams := make(map[string]*pb.ParamItem)
	for _, param := range configParams {
		v, err := pb.NewValue(param.Value.Raw)
		if err != nil {
			tx.Rollback()
			return nil, err
		}
		customParams[param.Key] = &pb.ParamItem{
			Value:   v,
			Comment: param.Comment,
		}
	}

	// 3. read text from cache to avoid different result between multiple calculation
	text, err := tx.GetClusterConfigText(file.ClusterType, file.ClusterID, file.Filename)
	if err != nil {
		tx.Rollback()
		return nil, err
	}

	// 4. calculate full params
	template, err := GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
		ClusterType:       pb.ClusterType(pb.ClusterType_value[file.ClusterType]),
		ConfigurationType: file.ConfigType,
		TemplateId:        file.TemplateID,
	})
	if err != nil {
		tx.Rollback()
		return nil, err
	}
	fullParams, _, err := calculateConfigurationFile(file.Format, template.Data, customParams)
	if err != nil {
		tx.Rollback()
		return nil, err
	}

	result := &pb.ConfigurationFile{
		ClusterType:           req.GetClusterType(),
		ConfigurationFilename: req.GetConfigurationFilename(),
		ConfigurationType:     file.ConfigType,
		BelongToComponent:     file.BelongTo,
		Format:                pb.Format(pb.Format_value[file.Format]),
		TemplateId:            file.TemplateID,
		CustomParams:          customParams,
		FullParams:            fullParams,
		Text:                  text,
		CreateTime:            timestamppb.New(file.CreateTime),
		UpdateTime:            timestamppb.New(file.UpdateTime),
	}
	if file.DeleteTime != nil {
		result.DeleteTime = timestamppb.New(*file.DeleteTime)
	}
	return result, tx.Commit()

}

func ListClusterConfigurationFiles(d db.DB, req *pb.ListClusterConfigurationFilesRequest) (*pb.ListClusterConfigurationFilesResponse, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}

	files, err := tx.ListClusterConfigInfo2(req.GetClusterType().String(), req.GetClusterId(), req.GetConfigurationTypeFilter(), req.GetBelongToComponentFilter())
	if err != nil {
		tx.Rollback()
		return nil, status.Error(codes.Internal, err.Error())
	}

	resp := &pb.ListClusterConfigurationFilesResponse{
		Files: make([]*pb.ConfigurationFile, 0, len(files)),
	}

	for _, file := range files {
		params, err := tx.ListClusterConfigParams(file.ClusterType, file.ClusterID, file.Filename)
		if err != nil {
			tx.Rollback()
			return nil, status.Error(codes.Internal, err.Error())
		}

		customConfigurations := make(map[string]*pb.ParamItem)
		for _, param := range params {
			v, err := pb.NewValue(param.Value.Raw)
			if err != nil {
				tx.Rollback()
				return nil, status.Error(codes.Internal, err.Error())
			}
			customConfigurations[param.Key] = &pb.ParamItem{
				Value:   v,
				Comment: param.Comment,
			}
		}

		text, err := tx.GetClusterConfigText(file.ClusterType, file.ClusterID, file.Filename)
		if err != nil {
			tx.Rollback()
			return nil, err
		}

		template, err := GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
			ClusterType:       pb.ClusterType(pb.ClusterType_value[file.ClusterType]),
			ConfigurationType: file.ConfigType,
			TemplateId:        file.TemplateID,
		})
		if err != nil {
			tx.Rollback()
			return nil, status.Error(codes.Internal, err.Error())
		}
		fullParams, _, err := calculateConfigurationFile(file.Format, template.Data, customConfigurations)
		if err != nil {
			tx.Rollback()
			return nil, status.Error(codes.Internal, err.Error())
		}

		configFile := &pb.ConfigurationFile{
			ClusterType:           req.GetClusterType(),
			ConfigurationFilename: file.Filename,
			ConfigurationType:     file.ConfigType,
			BelongToComponent:     file.BelongTo,
			TemplateId:            file.TemplateID,
			Format:                pb.Format(pb.Format_value[file.Format]),
			CustomParams:          customConfigurations,
			FullParams:            fullParams,
			Text:                  text,
			CreateTime:            timestamppb.New(file.CreateTime),
			UpdateTime:            timestamppb.New(file.UpdateTime),
		}
		if file.DeleteTime != nil {
			configFile.DeleteTime = timestamppb.New(*file.DeleteTime)
		}
		resp.Files = append(resp.Files, configFile)

	}
	return resp, tx.Commit()
}

func BatchUpdateClusterConfigurationFiles(d db.DB, req *pb.BatchUpdateClusterConfigurationFilesRequest) (*pb.BatchUpdateClusterConfigurationFilesResponse, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	var resp pb.BatchUpdateClusterConfigurationFilesResponse
	for _, r := range req.GetRequests() {
		if req.GetValidateOnly() {
			r.ValidateOnly = true
		}
		f, err := UpdateClusterConfigurationFile(tx, r)
		if err != nil {
			return nil, err
		}
		resp.Files = append(resp.Files, f)
	}
	return &resp, tx.Commit()
}

func BatchAutoUpdateClusterConfigurationFiles(d db.DB, req *pb.BatchAutoUpdateClusterConfigurationFilesRequest) (*pb.BatchAutoUpdateClusterConfigurationFilesResponse, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}

	var resp pb.BatchAutoUpdateClusterConfigurationFilesResponse
	for _, req := range req.GetRequests() {
		r, err := AutoUpdateClusterConfigurationFile(tx, req)
		if err != nil {
			tx.Rollback()
			return nil, err
		}
		resp.Files = append(resp.Files, r)
	}
	return &resp, tx.Commit()
}

func UpsertClusterCustomParams(d db.DB, req *pb.UpsertClusterCustomParamsRequest) (*pb.ConfigurationFile, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	clusterType := req.GetClusterType().String()
	clusterID := req.GetClusterId()
	filename := req.GetConfigurationFilename()

	// 1. Get basic info of existing config file
	file, err := tx.GetClusterConfigInfo(clusterType, clusterID, filename)
	switch err {
	case db.ErrNotFound:
		return nil, status.Errorf(codes.NotFound, "config file %s not found", filename)
	case nil:
	default:
		return &pb.ConfigurationFile{}, err
	}

	// 2. Get template
	template, err := GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: file.ConfigType,
		TemplateId:        file.TemplateID,
	})
	if err != nil {
		return nil, err
	}

	// 3. Get existing custom params and merge with upserted params
	customParams := make(map[string]*pb.ParamItem)
	configParams, err := tx.ListClusterConfigParams(clusterType, clusterID, filename)
	if err != nil {
		return nil, err
	}
	for _, param := range configParams {
		v, err := pb.NewValue(param.Value.Raw)
		if err != nil {
			return nil, err
		}
		customParams[param.Key] = &pb.ParamItem{
			Value:   v,
			Comment: param.Comment,
		}
	}
	for k, v := range req.GetUpsertedParams() {
		customParams[k] = v
	}

	// 4. Calculate full params and text, persist text
	fullParams, text, err := calculateConfigurationFile(file.Format, template.Data, customParams)
	if err != nil {
		return nil, err
	}

	// todo: support validate_only
	// 5. update database
	if true {
		if len(req.GetUpsertedParams()) > 0 {
			var params []db.ConfigurationItem
			for k, v := range req.GetUpsertedParams() {
				params = append(params, db.ConfigurationItem{
					Key:     k,
					Value:   db.NewAnyValue(v.GetValue().AsInterface()),
					Comment: v.GetComment(),
				})
			}
			err = tx.UpsertClusterConfigParams(clusterType, clusterID, filename, params)
			if err != nil {
				return nil, err
			}
		}
		err = tx.UpdateClusterConfigText(clusterType, clusterID, filename, text)
		if err != nil {
			return &pb.ConfigurationFile{}, err
		}

		err = tx.Commit()
	}

	return &pb.ConfigurationFile{
		ClusterType:           req.GetClusterType(),
		ConfigurationFilename: filename,
		ConfigurationType:     file.ConfigType,
		BelongToComponent:     file.BelongTo,
		Format:                pb.Format(pb.Format_value[file.Format]),
		TemplateId:            file.TemplateID,
		CustomParams:          customParams,
		FullParams:            fullParams,
		Text:                  text,
	}, err
}

func BatchUpsertClusterCustomParams(d db.DB, req *pb.BatchUpsertClusterCustomParamsRequest) (*pb.BatchUpsertClusterCustomParamsResponse, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	var resp pb.BatchUpsertClusterCustomParamsResponse
	for _, r := range req.GetRequests() {
		f, err := UpsertClusterCustomParams(tx, r)
		if err != nil {
			return nil, err
		}
		resp.Files = append(resp.Files, f)
	}
	return &resp, tx.Commit()
}

func UpdateClusterConfigurationFile(d db.DB, req *pb.UpdateClusterConfigurationFileRequest) (*pb.ConfigurationFile, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	clusterType := req.GetClusterType().String()
	clusterID := req.GetClusterId()
	filename := req.GetConfigurationFilename()

	// 1. [optional] check previous text
	if req.PreviousText != nil {
		text, err := tx.GetClusterConfigText(clusterType, clusterID, filename)
		if err == db.ErrNotFound {
			// make sure it is empty if file not exist
			text = ""
		} else if err != nil {
			return nil, err
		}

		if text != *req.PreviousText {
			return nil, errors.New("previous_text is mismatch with config-center")
		}
	}

	// 2. Update templateID of config file and read template content
	file, err := tx.GetClusterConfigInfo(clusterType, clusterID, filename)
	switch err {
	case db.ErrNotFound:
		if req.GetAllowMissing() == false {
			return nil, status.Errorf(codes.NotFound, "config file %s not found", filename)
		}
		c, err := CreateClusterConfigurationFile(tx, &pb.CreateClusterConfigurationFileRequest{
			ClusterType: req.ClusterType,
			ClusterId:   req.ClusterId,
			ConfigurationFile: &pb.ConfigurationFile{
				ConfigurationFilename: req.ConfigurationFilename,
				ConfigurationType:     req.GetConfigurationType(),
				BelongToComponent:     req.GetBelongToComponent(),
				Format:                req.GetFormat(),
				TemplateId:            req.GetTemplateId(),
				CustomParams:          req.GetCustomParams(),
			},
			ValidateOnly: req.GetValidateOnly(),
		})
		if err == nil {
			if !req.GetValidateOnly() {
				err = tx.Commit()
			}
			return c, err
		} else {
			return nil, err
		}
	case nil:
	default:
		return &pb.ConfigurationFile{}, err
	}

	if req.ConfigurationType != nil {
		file.ConfigType = req.GetConfigurationType()
	}
	if req.BelongToComponent != nil {
		file.BelongTo = req.GetBelongToComponent()
	}
	if req.Format != nil {
		file.Format = req.GetFormat().String()
	}
	if req.TemplateId != nil {
		file.TemplateID = req.GetTemplateId()
	}
	if req.ConfigurationType != nil || req.BelongToComponent != nil || req.TemplateId != nil || req.Format != nil {
		err = tx.UpdateClusterConfigInfo(file)
		if err != nil {
			return nil, err
		}
	}

	template, err := GetConfigurationTemplate(tx, &pb.GetConfigurationTemplateRequest{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: file.ConfigType,
		TemplateId:        file.TemplateID,
	})
	if err != nil {
		return nil, err
	}

	// 3. Update custom params or read unchanged items from db
	customParams := make(map[string]*pb.ParamItem)
	if len(req.CustomParams) > 0 || req.ClearCustomParams {
		customParams = req.GetCustomParams()
	} else {
		configParams, err := tx.ListClusterConfigParams(clusterType, clusterID, filename)
		if err != nil {
			return nil, err
		}
		for _, param := range configParams {
			v, err := pb.NewValue(param.Value.Raw)
			if err != nil {
				return nil, err
			}
			customParams[param.Key] = &pb.ParamItem{
				Value:   v,
				Comment: param.Comment,
			}
		}
	}

	// 4. Calculate full params and text, persist text
	fullParams, text, err := calculateConfigurationFile(file.Format, template.Data, customParams)
	if err != nil {
		return nil, err
	}

	if !req.GetValidateOnly() {
		if len(req.CustomParams) > 0 || req.ClearCustomParams {
			err = tx.DeleteClusterConfigParamsByFile(clusterType, clusterID, filename)
			if err != nil {
				return nil, err
			}
		}

		if len(req.CustomParams) > 0 {
			var params []db.ConfigurationItem
			for k, v := range req.GetCustomParams() {
				params = append(params, db.ConfigurationItem{
					Key:     k,
					Value:   db.NewAnyValue(v.GetValue().AsInterface()),
					Comment: v.GetComment(),
				})
			}
			err = tx.InsertClusterConfigParams(clusterType, clusterID, filename, params)
			if err != nil {
				return nil, err
			}
		}
		err = tx.UpdateClusterConfigText(clusterType, clusterID, filename, text)
		if err != nil {
			return &pb.ConfigurationFile{}, err
		}

		err = tx.Commit()
	}

	return &pb.ConfigurationFile{
		ClusterType:           req.GetClusterType(),
		ConfigurationFilename: filename,
		ConfigurationType:     file.ConfigType,
		BelongToComponent:     file.BelongTo,
		Format:                pb.Format(pb.Format_value[file.Format]),
		TemplateId:            file.TemplateID,
		CustomParams:          customParams,
		FullParams:            fullParams,
		Text:                  text,
	}, err
}

func AutoUpdateClusterConfigurationFile(d db.DB, req *pb.AutoUpdateClusterConfigurationFileRequest) (*pb.ConfigurationFile, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	// 1. find suitable config template
	configType := req.GetConfigurationType()
	if configType == "" {
		configType = req.GetConfigurationFilename()
	}
	tenantID := ""
	if req.TenantId != nil {
		tenantID = strconv.FormatUint(*req.TenantId, 10)
	}
	template, err := SearchDefaultConfigurationTemplate(tx, &pb.SearchDefaultConfigurationTemplateRequest{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: configType,
		Version:           req.GetVersion(),
		CloudProvider:     req.GetCloudProvider(),
		NodeInstance:      req.GetNodeInstance(),
		TenantId:          tenantID,
	})
	if err != nil {
		return nil, err
	}

	// 2. update config file
	f, err := UpdateClusterConfigurationFile(tx, &pb.UpdateClusterConfigurationFileRequest{
		ClusterType:           req.GetClusterType(),
		ClusterId:             req.GetClusterId(),
		ConfigurationFilename: req.GetConfigurationFilename(),
		ConfigurationType:     req.ConfigurationType,
		BelongToComponent:     req.BelongToComponent,
		Format:                req.Format,
		TemplateId:            &template.TemplateId,
		CustomParams:          nil,   // must be nil to avoid overwrite
		AllowMissing:          false, // handle it manully since customParams is nil
	})

	// 3. create if not exist
	if s, ok := status.FromError(err); ok && s.Code() == codes.NotFound {
		f, err = AutoCreateClusterConfigurationFile(tx, &pb.AutoCreateClusterConfigurationFileRequest{
			ClusterType:           req.GetClusterType(),
			ClusterId:             req.GetClusterId(),
			ConfigurationFilename: req.GetConfigurationFilename(),
			ConfigurationType:     req.GetConfigurationType(),
			BelongToComponent:     req.GetBelongToComponent(),
			Format:                req.GetFormat(),
			Version:               req.GetVersion(),
			CloudProvider:         req.GetCloudProvider(),
			NodeInstance:          req.GetNodeInstance(),
			TenantId:              req.TenantId,
		})
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
		return f, nil
	} else if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}

	// workaround update releated component
	if req.GetClusterType().ConfigType2Component(f.GetConfigurationType()) != "" {
		_, err = UpdateClusterComponent(tx, &pb.UpdateClusterComponentRequest{
			ClusterType:                     req.GetClusterType(),
			ClusterId:                       req.GetClusterId(),
			ComponentName:                   req.GetClusterType().ConfigType2Component(f.GetConfigurationType()),
			Version:                         nil,
			CloudProvider:                   &req.CloudProvider,
			NodeInstance:                    &req.NodeInstance,
			UpdateRelatedConfigurationFiles: false, // avoid recursion
		})
		if err != nil {
			slog.Error("update related component failed", "error", err)
			return nil, status.Error(codes.Internal, err.Error())
		}
	}
	return f, tx.Commit()
}

func DeleteClusterConfigurationFile(d db.DB, clusterType, clusterID, filename string) error {
	tx, err := d.Begin()
	if err != nil {
		return err
	}

	// Soft delete cluster_config_info only, keep rows in cluster_config_params and cluster_config_texts
	// err = tx.DeleteClusterConfigParamsByFile(clusterType, clusterID, filename)
	// if err != nil {
	// 	tx.Rollback()
	// 	return err
	// }

	// err = tx.DeleteClusterConfigText(clusterType, clusterID, filename)
	// if err != nil {
	// 	tx.Rollback()
	// 	return err
	// }

	err = tx.DeleteClusterConfigInfoByFile(clusterType, clusterID, filename)
	if err != nil {
		tx.Rollback()
		return err
	}

	return tx.Commit()
}

func GetTenantInitialCustomParams(d db.DB, req *pb.GetTenantInitialCustomParamsRequest) (*pb.TenantInitialCustomPatams, error) {
	paramRows, err := d.ListTenantInitialCustomParams(req.GetClusterType().String(), req.GetConfigurationType(), req.GetTenantId())
	if err != nil {
		return nil, err
	}
	initialParams := make(map[string]*pb.ParamItem)
	for _, row := range paramRows {
		v, err := pb.NewValue(row.Value.Raw)
		if err != nil {
			return nil, err
		}
		initialParams[row.Key] = &pb.ParamItem{
			Value:   v,
			Comment: row.Comment,
		}
	}
	return &pb.TenantInitialCustomPatams{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: req.GetConfigurationType(),
		InitialParams:     initialParams,
	}, nil
}

func UpdateTenantInitialCustomParams(d db.DB, req *pb.UpdateTenantInitialCustomParamsRequest) (*pb.TenantInitialCustomPatams, error) {
	tx, err := d.Begin()
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	err = tx.DeleteTenantInitialCustomParams(req.GetClusterType().String(), req.GetConfigurationType(), req.GetTenantId())
	if err != nil {
		return nil, err
	}

	if len(req.GetInitialParams()) > 0 {
		var params []db.ConfigurationItem
		for k, v := range req.GetInitialParams() {
			params = append(params, db.ConfigurationItem{
				Key:     k,
				Value:   db.NewAnyValue(v.GetValue().AsInterface()),
				Comment: v.GetComment(),
			})
		}

		err = tx.InsertTenantInitialCustomParams(req.GetClusterType().String(), req.GetConfigurationType(), req.GetTenantId(), params)
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
	}

	return &pb.TenantInitialCustomPatams{
		ClusterType:       req.GetClusterType(),
		ConfigurationType: req.GetConfigurationType(),
		InitialParams:     req.GetInitialParams(),
	}, tx.Commit()
}

func calculateConfigurationFile(format string, templateParams, customParams map[string]*pb.ParamItem) (map[string]*pb.ParamItem, string, error) {
	// LINES format has different merge logic
	if format == pb.Format_LINES.String() {
		return calculateLinesFormat(templateParams, customParams)
	}

	//todo: check conflict
	fullParams := make(map[string]*pb.ParamItem)
	for k, v := range templateParams {
		fullParams[k] = v
	}
	// it is by-design not to expand "v" for full_params if it is an struct
	for k, v := range customParams {
		fullParams[k] = v
	}

	templateKV := make(map[string]any)
	customKV := make(map[string]any)
	for k, v := range templateParams {
		templateKV[k] = v.Value.AsInterface()
	}
	for k, v := range customParams {
		customKV[k] = v.Value.AsInterface()
	}
	merged := utils.MergeConfig(templateKV, customKV)

	var text string
	var err error
	switch format {
	case pb.Format_TOML.String():
		utils.Convert2Int(merged)
		var b []byte
		b, err = utils.MarshalTOML(merged)
		text = string(b)
	case pb.Format_JSON.String():
		var b []byte
		b, err = json.Marshal(merged)
		text = string(b)
	case pb.Format_YAML.String():
		var b []byte
		b, err = yaml.Marshal(merged)
		text = string(b)
	default:
		err = fmt.Errorf("unknown format: %s", format)
	}

	return fullParams, text, err
}

func calculateLinesFormat(templateParams, customParams map[string]*pb.ParamItem) (map[string]*pb.ParamItem, string, error) {
	resultMap := make(map[string]*pb.ParamItem)
	text := ""

	templateLines := pb.LinesMapToSlice(templateParams)
	customLines := pb.LinesMapToSlice(customParams)
	mergedLines := append(templateLines, customLines...)

	for id, line := range mergedLines {
		resultMap[strconv.FormatInt(int64(id), 10)] = line
		text += line.Value.AsInterface().(string) + "\n"
	}

	return resultMap, text, nil

}

func buildConfigurationFile(info db.ClusterConfigInfoRow, params []db.ClusterConfigParamRow, template pb.ConfigurationTemplate) (pb.ConfigurationFile, error) {
	customParams := make(map[string]*pb.ParamItem)
	for _, param := range params {
		v, err := pb.NewValue(param.Value.Raw)
		if err != nil {
			return pb.ConfigurationFile{}, err
		}
		customParams[param.Key] = &pb.ParamItem{
			Value:   v,
			Comment: param.Comment,
		}
	}

	fullParams, text, err := calculateConfigurationFile(info.Format, template.Data, customParams)
	if err != nil {
		return pb.ConfigurationFile{}, err
	}

	result := pb.ConfigurationFile{
		ClusterType:           pb.ClusterType(pb.ClusterType_value[info.ClusterType]),
		ConfigurationFilename: info.Filename,
		ConfigurationType:     info.ConfigType,
		BelongToComponent:     info.BelongTo,
		Format:                pb.Format(pb.Format_value[info.Format]),
		TemplateId:            info.TemplateID,
		CustomParams:          customParams,
		FullParams:            fullParams,
		Text:                  text,
		CreateTime:            timestamppb.New(info.CreateTime),
		UpdateTime:            timestamppb.New(info.UpdateTime),
	}
	if info.DeleteTime != nil {
		result.DeleteTime = timestamppb.New(*info.DeleteTime)
	}
	return result, nil
}

func ValidateFormat(clusterType pb.ClusterType, configType string, inputFormat pb.Format) error {
	recordedFormat := clusterType.GetConfigTypeFormat(configType)
	// only unknown(new added) config type can pass the check
	if inputFormat != recordedFormat && recordedFormat != pb.Format_FORMAT_UNSPECIFIED {
		return status.Errorf(codes.InvalidArgument, "format not match with input format %s and recorded format %s", inputFormat, recordedFormat)
	}
	return nil
}