package auditlog

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/url"
	"path"
	"strconv"
	"strings"

	"github.com/samber/lo"
	clusterdedicatedv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"golang.org/x/mod/semver"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/clusterproxy"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	QueryKernelAuditLogConfigStmt = `SELECT @@GLOBAL.tidb_audit_enabled          AS audit_enabled,
       @@GLOBAL.tidb_audit_log_redacted     AS log_redacted,
       @@GLOBAL.tidb_audit_log_format       AS log_format,
       @@GLOBAL.tidb_audit_log_max_filesize AS max_filesize,
       @@GLOBAL.tidb_audit_log_max_lifetime AS max_lifetime;`
	SetAuditEnabledStmt         = "SET @@GLOBAL.tidb_audit_enabled = '%s'"
	SetAuditLogRedactedStmt     = "SET @@GLOBAL.tidb_audit_log_redacted = '%s'"
	SetAuditLogFormatStmt       = "SET @@GLOBAL.tidb_audit_log_format = '%s'"
	SetAuditLogMaxFilesizeStmt  = "SET @@GLOBAL.tidb_audit_log_max_filesize = %d"
	SetAuditLogMaxLifetimeStmt  = "SET @@GLOBAL.tidb_audit_log_max_lifetime = %d"
	SetAuditLogFilePathStmt     = "SET @@GLOBAL.tidb_audit_log = '/var/log/auditlog/tidb-audit.log'"
	SetAuditLogFilePathJsonStmt = "SET @@GLOBAL.tidb_audit_log = '/var/log/auditlog/tidb-audit'"

	CheckAuditAdminPrivilegeStmt = "SELECT 1 FROM mysql.global_grants WHERE USER='cloud_admin' AND HOST='%' AND PRIV='RESTRICTED_AUDIT_ADMIN'"
	GrantAuditAdminPrivilegeStmt = `INSERT INTO mysql.global_grants (USER, HOST, PRIV, WITH_GRANT_OPTION)
VALUES ('cloud_admin', '%', 'RESTRICTED_AUDIT_ADMIN', 'Y')`
	FlushPrivilegesStmt = "FLUSH PRIVILEGES"

	// Disable legacy plugin stmts.
	DisableAuditPluginStmt = "admin plugins disable audit"

	MaxAuditLogSizeMiB   = 100     // Maximum size of audit log in MiB.
	MinAuditLogSizeMiB   = 1       // Minimum size of audit log in MiB.
	MaxAuditLogAgeSecond = 2592000 // Maximum age of audit log in seconds (30 days).
)

func (svc *service) GetAuditLogConfig(
	ctx context.Context,
	clusterID uint64,
) (*globaldedicatedv1.AuditLogConfig, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "GetAuditLogConfig")
	defer span.End()

	// Get the cluster and region.
	cluster, err := svc.clusterRepo.Get(ctx, clusterID, false)
	if err != nil {
		return nil, err
	}

	region, err := svc.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	proxy, err := svc.GetCloudAdminClusterProxy(ctx, cluster, region)
	if err != nil {
		return nil, err
	}

	res, _, err := svc.getAuditLogConfig(ctx, cluster, region, proxy)
	if err != nil {
		return nil, err
	}

	return res, nil
}

func (svc *service) getAuditLogConfig(
	ctx context.Context,
	cluster *entity.Cluster,
	region *entity.Region,
	proxy clusterproxy.ClusterProxy,
) (globalConfig *globaldedicatedv1.AuditLogConfig, dbRecordExisted bool, err bizerr.BizError) {
	clusterID := cluster.ClusterId
	// Get the audit log status from the region server.
	regionalConfig, err := svc.regionServerCli.GetAuditLogConfiguration(
		ctx,
		region,
		&clusterdedicatedv1.GetAuditLogConfigurationRequest{ClusterId: strconv.FormatUint(clusterID, 10)},
	)
	if err != nil {
		return nil, false, err
	}

	// Convert the regional config to the central config.
	globalConfig = &globaldedicatedv1.AuditLogConfig{
		BucketUrl: regionalConfig.GetBucketUrl(),
		Region:    regionalConfig.GetBucketRegion(),
		RoleArn:   regionalConfig.GetS3().GetRoleArn(),
		SasToken:  regionalConfig.GetAzBlob().GetSasToken(),
		BucketManager: lo.Ternary(
			regionalConfig.GetIsManagedByPingcap(),
			openapidedicatedv1beta1.AuditLogConfig_PINGCAP.String(),
			openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String(),
		),
		ClusterId: clusterID,
	}

	// Query DB
	dbConfig, err := svc.auditLogConfigRepo.GetByClusterID(ctx, clusterID)
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, false, err
		}
		dbRecordExisted = false
	} else {
		dbRecordExisted = true
	}

	// Merge DB config fields into globalConfig if present
	if dbConfig != nil {
		if dbConfig.Id != 0 {
			globalConfig.Id = dbConfig.Id
		}
		if dbConfig.BucketUrl != "" {
			globalConfig.BucketUrl = dbConfig.BucketUrl
		}
		if dbConfig.Region != "" {
			globalConfig.Region = dbConfig.Region
		}
		if dbConfig.RoleArn != "" {
			globalConfig.RoleArn = dbConfig.RoleArn
		}
		if dbConfig.SasToken != "" {
			globalConfig.SasToken = dbConfig.SasToken
		}
		if dbConfig.BucketManager != "" {
			globalConfig.BucketManager = dbConfig.BucketManager
		}
		if dbConfig.Enabled != nil {
			globalConfig.Enabled = dbConfig.Enabled
		}
		if dbConfig.Unredacted != nil {
			globalConfig.Unredacted = dbConfig.Unredacted
		}
		if dbConfig.Format != "" {
			globalConfig.Format = dbConfig.Format
		}
		if dbConfig.MaxSizeMib != 0 {
			globalConfig.MaxSizeMib = dbConfig.MaxSizeMib
		}
		if dbConfig.MaxAgeSecond != nil {
			globalConfig.MaxAgeSecond = dbConfig.MaxAgeSecond
		}
	}

	// Remove our-added cluster id suffix from the bucket url.
	if strings.HasSuffix(globalConfig.BucketUrl, fmt.Sprint(clusterID)) {
		bucketURL, err := url.Parse(globalConfig.BucketUrl)
		if err != nil {
			return nil, false, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse bucket url")
		}
		bucketURL.Path = path.Dir(bucketURL.Path)
		globalConfig.BucketUrl = bucketURL.String()
	}

	// If the cluster is in the paused, pausing or resuming state, or never use audit log feature previously, return config only.
	if cluster.GetState() != globaldedicatedv1.Cluster_AVAILABLE &&
		cluster.GetState() != globaldedicatedv1.Cluster_SCALING ||
		!dbRecordExisted {
		return globalConfig, dbRecordExisted, nil
	}

	kernelConfig, err := svc.queryKernelAuditLogConfig(ctx, proxy)
	if err != nil {
		tilog.Warn(ctx, "failed to query kernel audit log config", zap.Error(err))
	} else {
		// Log diff between kernel and global config.
		jsonKernelConfig, _ := json.Marshal(kernelConfig)
		tilog.Info(ctx, "kernel audit log config", zap.String("config", string(jsonKernelConfig)))
	}

	return globalConfig, dbRecordExisted, nil
}

func (svc *service) UpdateAuditLogConfig(
	ctx context.Context,
	reqConfig *globaldedicatedv1.AuditLogConfig,
	validateOnly bool,
) (*globaldedicatedv1.AuditLogConfig, bizerr.BizError) {
	ctx, span := tracer.Start(ctx, "UpdateAuditLogConfig")
	defer span.End()

	// Get the cluster and region.
	cluster, err := svc.clusterRepo.Get(ctx, reqConfig.GetClusterId(), false)
	if err != nil {
		return nil, err
	}

	region, err := svc.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if err != nil {
		return nil, err
	}

	// Prepare the cluster proxy.
	proxy, err := svc.GetCloudAdminClusterProxy(ctx, cluster, region)
	if err != nil {
		return nil, err
	}

	// Get current config.
	currentConfig, dbRecordExisted, err := svc.getAuditLogConfig(ctx, cluster, region, proxy)
	if err != nil {
		return nil, err
	}
	currentEnabled := currentConfig.GetEnabled()
	currentFormat := currentConfig.GetFormat()

	if !dbRecordExisted {
		if err := svc.ensurePrerequisites(ctx, proxy); err != nil {
			return nil, err
		}
	}

	// Merge the request config with the current config.
	if reqConfig.BucketUrl != "" {
		currentConfig.BucketUrl = reqConfig.BucketUrl
	}
	if reqConfig.Region != "" && reqConfig.Region != globalcommonv1.Region_NAME_UNSPECIFIED.String() {
		currentConfig.Region = reqConfig.Region
	}
	if reqConfig.RoleArn != "" {
		currentConfig.RoleArn = reqConfig.RoleArn
	}
	if reqConfig.SasToken != "" {
		currentConfig.SasToken = reqConfig.SasToken
	}
	if reqConfig.Unredacted != nil {
		currentConfig.Unredacted = reqConfig.Unredacted
	}
	if reqConfig.Format != "" &&
		reqConfig.Format != openapidedicatedv1beta1.AuditLogConfig_FORMAT_UNSPECIFIED.String() {
		currentConfig.Format = reqConfig.Format
	}
	if reqConfig.MaxSizeMib != 0 {
		currentConfig.MaxSizeMib = reqConfig.MaxSizeMib
	}
	if reqConfig.MaxAgeSecond != nil {
		currentConfig.MaxAgeSecond = reqConfig.MaxAgeSecond
	}
	if reqConfig.BucketManager != "" &&
		reqConfig.BucketManager != openapidedicatedv1beta1.AuditLogConfig_BUCKET_MANAGER_UNSPECIFIED.String() {
		currentConfig.BucketManager = reqConfig.BucketManager
	}
	if reqConfig.Enabled != nil {
		currentConfig.Enabled = reqConfig.Enabled
	}
	reqConfig = currentConfig

	// Set default values for the config if not provided.
	if reqConfig.GetBucketManager() == "" ||
		reqConfig.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_BUCKET_MANAGER_UNSPECIFIED.String() {
		reqConfig.BucketManager = openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String()
	}
	if reqConfig.GetFormat() == "" ||
		reqConfig.GetFormat() == openapidedicatedv1beta1.AuditLogConfig_FORMAT_UNSPECIFIED.String() {
		reqConfig.Format = openapidedicatedv1beta1.AuditLogConfig_TEXT.String()
	}
	if reqConfig.GetMaxSizeMib() == 0 {
		reqConfig.MaxSizeMib = 10
	}
	if reqConfig.MaxAgeSecond == nil {
		reqConfig.MaxAgeSecond = lo.ToPtr(int32(86400)) // Default to 1 day.
	}
	originalReqBucketUrl := reqConfig.GetBucketUrl()
	if reqConfig.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String() {
		// Append cluster to path of bucket url.
		bucketURL, err_ := url.Parse(reqConfig.GetBucketUrl())
		if err_ != nil {
			return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeParams, "failed to parse bucket url")
		}
		if !strings.Contains(bucketURL.Path, fmt.Sprint(reqConfig.GetClusterId())) {
			bucketURL.Path = path.Join(bucketURL.Path, fmt.Sprint(reqConfig.GetClusterId()))
			reqConfig.BucketUrl = bucketURL.String()
		}
	}

	// Check the request parameters.
	if reqConfig.GetMaxSizeMib() < MinAuditLogSizeMiB || reqConfig.GetMaxSizeMib() > MaxAuditLogSizeMiB {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"max size must be between %d and %d MiB",
			MinAuditLogSizeMiB,
			MaxAuditLogSizeMiB,
		)
	}
	if reqConfig.GetMaxAgeSecond() > MaxAuditLogAgeSecond {
		return nil, bizerr.New(bizerr.ErrCodeParams, "max age must be less than %d seconds", MaxAuditLogAgeSecond)
	}
	if reqConfig.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String() {
		if reqConfig.GetBucketUrl() == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "bucket uri is required")
		}
		switch cluster.GetCloudProviderId() {
		case globalcommonv1.CloudProvider_AZURE:
			if reqConfig.GetSasToken() == "" {
				return nil, bizerr.New(bizerr.ErrCodeParams, "azure sas token is required")
			}
		case globalcommonv1.CloudProvider_AWS:
			if reqConfig.GetRoleArn() == "" {
				return nil, bizerr.New(bizerr.ErrCodeParams, "aws role arn is required")
			}
			if reqConfig.GetRegion() == "" || reqConfig.GetRegion() == globalcommonv1.Region_NAME_UNSPECIFIED.String() {
				return nil, bizerr.New(bizerr.ErrCodeParams, "region is required")
			}
		case globalcommonv1.CloudProvider_GCP:
			if reqConfig.GetRegion() == "" || reqConfig.GetRegion() == globalcommonv1.Region_NAME_UNSPECIFIED.String() {
				return nil, bizerr.New(bizerr.ErrCodeParams, "region is required")
			}
		}
	}
	if reqConfig.GetBucketManager() == openapidedicatedv1beta1.AuditLogConfig_CUSTOMER.String() {
		// Validate the audit log config.
		if err := ValidateAuditLogConfig(cluster, reqConfig, V2MinVersion); err != nil {
			return nil, err
		}
		// Check bucket access.
		resp, err := svc.regionServerCli.ValidateAuditLogConfiguration(
			ctx,
			region,
			&clusterdedicatedv1.ValidateAuditLogConfigurationRequest{
				ClusterId:             strconv.FormatUint(reqConfig.GetClusterId(), 10),
				AuditLogConfiguration: CvtToRegionalConfig(reqConfig, cluster.GetCloudProviderId()),
			},
		)
		if err != nil {
			return nil, err
		}
		if !resp.GetIsValid() {
			return nil, BuildPrecheckAuditLogConfigBizError(cluster.GetCloudProviderId())
		}
	}

	if currentEnabled && reqConfig.GetEnabled() && // If was enabled and is still enabled after update,
		reqConfig.GetFormat() != currentFormat { // and format is changed,
		return nil, bizerr.New( // then we cannot change the format to prevent audit log loss.
			bizerr.ErrCodeParams,
			"cannot change format while audit log is enabled, please disable it first",
		)
	}

	if validateOnly {
		// If validate only, return the config without applying it.
		reqConfig.BucketUrl = originalReqBucketUrl // Restore original bucket URL.
		return reqConfig, nil
	}

	// If there is already a running workflow, return an error.
	dagIns, err_ := svc.wf.ListRunningDagInstancesByNamePrefix(
		ctx,
		common.UpdateAuditLogConfig,
		strconv.FormatUint(cluster.ClusterId, 10),
	)
	if err_ != nil {
		return nil, bizerr.WrapError(err_, "failed to list running dag instances")
	}
	if len(dagIns) > 0 {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"the last updating audit log config is still running, please wait for it to finish",
		)
	}

	errx := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Ensure DB record is up-to-date before calling regional or cluster proxy.
		if err := svc.auditLogConfigRepo.Upsert(ctx, reqConfig); err != nil {
			return err
		}

		// Start async reconciliation workflow.
		workFlowErr := svc.wf.RunDagWithBizId(
			ctx,
			common.UpdateAuditLogConfig,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			strconv.FormatUint(cluster.ClusterId, 10),
			map[string]string{
				workflow.KeyTraceId:        workflow.GetTraceId(ctx),
				common.KeyClusterId:        strconv.FormatUint(cluster.ClusterId, 10),
				common.KeyAuditLogConfigId: strconv.FormatUint(reqConfig.Id, 10),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if errx != nil {
		return nil, bizerr.WrapError(errx, "update audit log config failed")
	}

	reqConfig.BucketUrl = originalReqBucketUrl // Restore original bucket URL.
	return reqConfig, nil
}

func (svc *service) ensurePrerequisites(ctx context.Context, proxy clusterproxy.ClusterProxy) bizerr.BizError {
	rawBytes, err := proxy.QuerySQL(ctx, CheckAuditAdminPrivilegeStmt)
	if err != nil {
		return err
	}

	var result map[string]any
	if err_ := json.Unmarshal(rawBytes, &result); err_ != nil {
		return bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeUnknown, "failed to unmarshal privilege check result")
	}

	if len(result) > 0 {
		// Privilege already granted.
		return nil
	}

	if err := proxy.ExecuteSQL(ctx, DisableAuditPluginStmt); err != nil {
		tilog.Error(ctx, "failed to disable audit plugin", zap.Error(err))
	}

	if err := proxy.ExecuteMultiSQL(ctx, []string{
		GrantAuditAdminPrivilegeStmt,
		FlushPrivilegesStmt,
	}); err != nil {
		return err
	}

	return nil
}

type kernelAuditLogConfig struct {
	Enabled       bool
	Redacted      bool
	Format        string
	MaxSizeMiB    int32
	MaxAgeSeconds int32
}

func (svc *service) queryKernelAuditLogConfig(
	ctx context.Context,
	proxy clusterproxy.ClusterProxy,
) (*kernelAuditLogConfig, bizerr.BizError) {
	rawBytes, err := proxy.QuerySQL(ctx, QueryKernelAuditLogConfigStmt)
	if err != nil {
		return nil, err
	}

	type raw struct {
		AuditEnabled []int    `json:"audit_enabled"`
		LogRedacted  []int    `json:"log_redacted"`
		LogFormat    []string `json:"log_format"`
		MaxFilesize  []string `json:"max_filesize"`
		MaxLifetime  []string `json:"max_lifetime"`
	}

	var rawObj raw
	if err := json.Unmarshal(rawBytes, &rawObj); err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "failed to unmarshal json")
	}

	if len(rawObj.AuditEnabled) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "no audit log config found")
	}

	enabled := rawObj.AuditEnabled[0] == 1
	redacted := rawObj.LogRedacted[0] == 1
	formatStr, err_ := base64.StdEncoding.DecodeString(rawObj.LogFormat[0])
	if err_ != nil {
		return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeUnknown, "failed to decode base64")
	}
	maxSizeStr, err_ := base64.StdEncoding.DecodeString(rawObj.MaxFilesize[0])
	if err_ != nil {
		return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeUnknown, "failed to decode base64")
	}
	maxAgeStr, err_ := base64.StdEncoding.DecodeString(rawObj.MaxLifetime[0])
	if err_ != nil {
		return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeUnknown, "failed to decode base64")
	}

	maxSize, err_ := strconv.ParseInt(string(maxSizeStr), 10, 32)
	if err_ != nil {
		return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeUnknown, "failed to parse int")
	}
	maxAge, err_ := strconv.ParseInt(string(maxAgeStr), 10, 32)
	if err_ != nil {
		return nil, bizerr.WrapErrorWithCode(err_, bizerr.ErrCodeUnknown, "failed to parse int")
	}

	return &kernelAuditLogConfig{
		Enabled:       enabled,
		Redacted:      redacted,
		Format:        string(formatStr),
		MaxSizeMiB:    int32(maxSize),
		MaxAgeSeconds: int32(maxAge),
	}, nil
}

func ValidateAuditLogConfig(
	cluster *entity.Cluster,
	config *globaldedicatedv1.AuditLogConfig,
	minVersion string,
) bizerr.BizError {
	if config == nil {
		return bizerr.New(bizerr.ErrCodeParams, "config is nil")
	}

	// Check cluster status.
	if cluster.GetState() != globaldedicatedv1.Cluster_AVAILABLE &&
		cluster.GetState() != globaldedicatedv1.Cluster_SCALING {
		return bizerr.New(bizerr.ErrCodeParams, "the cluster's status is invalid: %s", cluster.GetState())
	}

	u, err := url.Parse(config.BucketUrl)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to parse bucket url")
	}

	switch cluster.GetCloudProviderId() {
	case globalcommonv1.CloudProvider_AWS:
		if u.Scheme != "s3" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be s3, but is: %s", u.Scheme)
		}
		if config.GetRoleArn() == "" {
			return bizerr.New(bizerr.ErrCodeParams, "role ARN is required")
		}
		r_, err := enum.FromString[globalcommonv1.Region_Name](config.GetRegion())
		if err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region: %s", config.GetRegion())
		}
		if r_ == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "region is required")
		}

	case globalcommonv1.CloudProvider_GCP:
		if u.Scheme != "gcs" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be gcs, but is: %s", u.Scheme)
		}
		r_, err := enum.FromString[globalcommonv1.Region_Name](config.GetRegion())
		if err != nil {
			return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region: %s", config.GetRegion())
		}
		if r_ == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "region is required")
		}

	case globalcommonv1.CloudProvider_AZURE:
		if u.Scheme != "https" {
			return bizerr.New(bizerr.ErrCodeParams, "source url schema should be https, but is: %s", u.Scheme)
		}

	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported source type: %s", cluster.GetCloudProviderId())
	}

	// Validate TiDB version.
	ver := cluster.GetVersion()
	if !semver.IsValid(ver) {
		return bizerr.New(bizerr.ErrCodeParams, "invalid version: %s", cluster.GetVersion())
	}
	if semver.Compare(semver.Canonical(ver), minVersion) < 0 {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"TiDB version %s is not supported, please contact support to upgrade to %s or above",
			cluster.GetVersion(),
			minVersion,
		)
	}

	return nil
}

func CvtToRegionalConfig(
	config *globaldedicatedv1.AuditLogConfig,
	cloudProvider globalcommonv1.CloudProvider_Name,
) *clusterdedicatedv1.AuditLogConfiguration {
	c := &clusterdedicatedv1.AuditLogConfiguration{
		BucketUrl: config.GetBucketUrl(),
	}
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		c.Storage = &clusterdedicatedv1.AuditLogConfiguration_S3_{
			S3: &clusterdedicatedv1.AuditLogConfiguration_S3{
				RoleArn: config.GetRoleArn(),
			},
		}
		c.BucketRegion = config.GetRegion()
	case globalcommonv1.CloudProvider_GCP:
		c.Storage = &clusterdedicatedv1.AuditLogConfiguration_Gcs{
			Gcs: &clusterdedicatedv1.AuditLogConfiguration_GCS{},
		}
		c.BucketRegion = config.GetRegion()
	case globalcommonv1.CloudProvider_AZURE:
		c.Storage = &clusterdedicatedv1.AuditLogConfiguration_AzBlob{
			AzBlob: &clusterdedicatedv1.AuditLogConfiguration_AZBlob{
				SasToken: config.GetSasToken(),
			},
		}
	}
	return c
}

func BuildPrecheckAuditLogConfigBizError(cloudProvider globalcommonv1.CloudProvider_Name) bizerr.BizError {
	switch cloudProvider {
	case globalcommonv1.CloudProvider_AWS:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL and role ARN are correct, then try again.",
		)
	case globalcommonv1.CloudProvider_GCP:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL is correct, then try again.",
		)
	case globalcommonv1.CloudProvider_AZURE:
		return bizerr.New(
			bizerr.ErrCodePermission,
			"Bucket access verification failed. Please check the bucket URL and SAS token are correct, then try again.",
		)
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported cloud provider: %s", cloudProvider)
	}
}