package service

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	mapset "github.com/deckarep/golang-set/v2"
	"github.com/pingcap/tidb/pkg/parser/mysql"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	aws2 "github.com/tidbcloud/serverless-service/internal/infra/aws"
	"github.com/tidbcloud/serverless-service/internal/metric"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	FilterRuleQuerySQL    = "SELECT alfr.filter_name,alfr.user,alfr.enabled,alf.content FROM mysql.audit_log_filter_rules alfr left join mysql.audit_log_filters alf on alfr.filter_name = alf.filter_name"
	FilterRuleDeleteSQL   = "SELECT audit_log_remove_rule('%s', '%s')"
	FilterDeleteSQL       = "SELECT audit_log_remove_filter('%s')"
	FilterRuleEnableSQL   = "SELECT audit_log_enable_rule('%s', '%s')"
	FilterRuleDisableSQL  = "SELECT audit_log_disable_rule('%s', '%s')"
	FilterCreateSQL       = "SELECT audit_log_create_filter('%s', '%s');"
	FilterRuleCreateSQL   = "SELECT audit_log_create_rule('%s', '%s');"
	FilterReplaceSQL      = "SELECT audit_log_create_filter('%s', '%s', 1);"
	FilterRuleReplaceSQL  = "SELECT audit_log_create_rule('%s', '%s', 1);"
	EnabledAuditLogSQL    = "SET global tidb_audit_enabled = 1;"
	DisabledAuditLogSQL   = "SET global tidb_audit_enabled = 0;"
	RedactedAuditLogSQL   = "SET global tidb_audit_log_redacted = 1;"
	UnredactedAuditLogSQL = "SET global tidb_audit_log_redacted = 0;"

	MaxFilterRuleCount = 50

	ExecuteMultiSQLMaxRetryTimes = 3

	perSignedUrlExpireTime = 8 * time.Hour
)

type AuditLogService interface {
	ListAuditLogFiles(
		ctx context.Context,
		clusterId, pageToken, date string,
		pageSize int32,
	) ([]*vo.AuditLogFile, string, bizerr.BizError)

	DownloadAuditLogFiles(
		ctx context.Context,
		clusterId string,
		fileNames []string,
	) ([]*vo.AuditLogFile, bizerr.BizError)

	GetAuditLogFilterRule(ctx context.Context, clusterId, ruleId string) (*entity.AuditLogFilterRule, bizerr.BizError)

	ListAuditLogFilterRules(
		ctx context.Context,
		opts *vo.ListFilterRuleOpts,
	) ([]*entity.AuditLogFilterRule, uint32, bizerr.BizError)

	CreateAuditLogFilterRule(
		ctx context.Context,
		clusterId string,
		filterRule *entity.AuditLogFilterRule,
	) (*entity.AuditLogFilterRule, bizerr.BizError)

	DeleteAuditLogFilterRule(
		ctx context.Context,
		clusterId, ruleId string,
	) (*entity.AuditLogFilterRule, bizerr.BizError)

	UpdateAuditLogFilterRule(
		ctx context.Context,
		clusterId string,
		filterRule *entity.AuditLogFilterRule,
	) (*entity.AuditLogFilterRule, bizerr.BizError)

	GetAuditLogConfig(
		ctx context.Context,
		clusterId string,
	) (*entity.AuditLogConfig, bizerr.BizError)

	UpdateAuditLogConfig(
		ctx context.Context,
		auditLogConfig *entity.AuditLogConfig,
	) (*entity.AuditLogConfig, bizerr.BizError)

	PreCheckAuditLogConfig(
		ctx context.Context,
		auditLogConfig *entity.AuditLogConfig,
	) bizerr.BizError
}

type auditLogService struct {
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	accountFacade              facade.AccountFacade
	encryptorFacade            facade.EncryptorFacade
	mgmtSvc                    facade.MgmtSvcFacade
	s3ClientProvider           *aws2.S3ClientProvider
	preSignerProvider          *aws2.PresignerProvider
	auditLogFilterRuleRepo     repo.AuditLogFilterRuleRepo
	transProvider              db.TransProvider
	auditLogConfigRepo         repo.AuditLogConfigRepo
}

func NewAuditLogService(
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	accountFacade facade.AccountFacade,
	encryptorFacade facade.EncryptorFacade,
	mgmtSvc facade.MgmtSvcFacade,
	s3ClientProvider *aws2.S3ClientProvider,
	preSignerProvider *aws2.PresignerProvider,
	auditLogFilterRuleRepo repo.AuditLogFilterRuleRepo,
	transProvider db.TransProvider,
	auditLogConfigRepo repo.AuditLogConfigRepo,
) AuditLogService {
	return &auditLogService{
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		accountFacade:              accountFacade,
		encryptorFacade:            encryptorFacade,
		mgmtSvc:                    mgmtSvc,
		s3ClientProvider:           s3ClientProvider,
		preSignerProvider:          preSignerProvider,
		auditLogFilterRuleRepo:     auditLogFilterRuleRepo,
		transProvider:              transProvider,
		auditLogConfigRepo:         auditLogConfigRepo,
	}
}

// getClusterProxyWithCloudAdmin also checks the cluster status
func (a *auditLogService) getClusterProxyWithCloudAdmin(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (facade.ClusterProxyFacade, *tidbmgmtservicev1.Cluster, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", cluster.ClusterId))

	// get cloud admin user
	pswd, err := a.accountFacade.GetCloudAdminPasswd(ctx, cluster.ClusterId)
	if err != nil {
		logger.Error("get cloud admin password failed", zap.Error(err))
		return nil, nil, bizerr.New(bizerr.ErrCodeUnknown, "get cloud admin password failed")
	}
	cloudAdminPswd, err1 := a.encryptorFacade.DecryptBase64(pswd)
	if err1 != nil {
		logger.Error("decrypt admin password failed", zap.Error(err1))
		return nil, nil, bizerr.New(bizerr.ErrCodeUnknown, "decrypt admin password failed")
	}

	// get cluster proxy
	user := cluster.GetServerlessCluster().GetCluster().GetUserPrefix() + ".cloud_admin"
	operator := &clusterproxy.ClusterOperator{
		Username:   user,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: cloudAdminPswd,
	}
	proxy, err := a.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if err != nil {
		logger.Error("failed to get cluster proxy", zap.Error(err))
		return nil, nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to get cluster proxy")
	}
	return proxy, cluster, nil
}

func deleteFilterRule(
	ctx context.Context,
	clusterProxy facade.ClusterProxyFacade,
	filterRule *entity.AuditLogFilterRule,
	clusterId string,
) bizerr.BizError {
	// delete filter rule
	deleteSQLs := make([]string, 0)
	for _, user := range filterRule.Users {
		deleteSQLs = append(deleteSQLs, fmt.Sprintf(FilterRuleDeleteSQL, user, filterRule.ID))
	}
	deleteSQLs = append(deleteSQLs, fmt.Sprintf(FilterDeleteSQL, filterRule.ID))

	// build rollback sqls
	rollbackDeleteSQLs := make([]string, 0)
	for _, user := range filterRule.Users {
		rollbackDeleteSQLs = append(rollbackDeleteSQLs, fmt.Sprintf(FilterRuleReplaceSQL, user, filterRule.ID))
	}
	if *filterRule.Disabled {
		for _, user := range filterRule.Users {
			rollbackDeleteSQLs = append(rollbackDeleteSQLs, fmt.Sprintf(FilterRuleDisableSQL, user, filterRule.ID))
		}
	}

	if err := executeMultiSQLs(ctx, clusterProxy, deleteSQLs); err != nil {
		rollbackSQLs(ctx, clusterProxy, rollbackDeleteSQLs, clusterId, metric.AuditlogOperationDelete)
		return err
	}
	return nil
}

func createFilterRule(
	ctx context.Context,
	clusterProxy facade.ClusterProxyFacade,
	filterRule *entity.AuditLogFilterRule,
	clusterId string,
) bizerr.BizError {
	logger := log.WithContextL(ctx)

	content, err := filterRule.ToContent()
	if err != nil {
		logger.Error("failed to convert filter rule to content", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeParams, "invalid filters, please check it and try again")
	}

	// create filter rule
	createSQLs := make([]string, 0)
	createSQLs = append(createSQLs, fmt.Sprintf(FilterCreateSQL, filterRule.ID, content))
	for _, user := range filterRule.Users {
		createSQLs = append(createSQLs, fmt.Sprintf(FilterRuleCreateSQL, user, filterRule.ID))
	}

	// build rollback sqls
	rollbackCreateSQLs := make([]string, 0)
	for _, user := range filterRule.Users {
		rollbackCreateSQLs = append(rollbackCreateSQLs, fmt.Sprintf(FilterRuleDeleteSQL, user, filterRule.ID))
	}
	rollbackCreateSQLs = append(rollbackCreateSQLs, fmt.Sprintf(FilterDeleteSQL, filterRule.ID))

	if err := executeMultiSQLs(ctx, clusterProxy, createSQLs); err != nil {
		// TODO: as we will set the unique id as filter name, record exist should not happen? consider to remove it
		if err.ErrCode() != bizerr.ErrCodeRecordExist {
			// not excute rollback, or it will delete exist filter rules
			rollbackSQLs(ctx, clusterProxy, rollbackCreateSQLs, clusterId, metric.AuditlogOperationCreate)
		}
		return err
	}
	return nil
}

func generateUpdateFilterRuleSQLs(
	ctx context.Context,
	filterRule *entity.AuditLogFilterRule,
	oldFilterRule *entity.AuditLogFilterRule,
) (updateSQLs, rollbackUpdateSQLs []string, err bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", oldFilterRule.ClusterID))
	// update filter rule
	updateSQLs, err = buildUpdateSQLs(ctx, filterRule, oldFilterRule)
	if err != nil {
		return nil, nil, err
	}
	logger.Info("update sqls", zap.Any("updateSQLs", updateSQLs))

	// swap old and new to build rollback sqls
	rollbackUpdateSQLs, err = buildUpdateSQLs(ctx, oldFilterRule, filterRule)
	if err != nil {
		return nil, nil, err
	}
	logger.Info("rollback update sqls", zap.Any("rollbackUpdateSQLs", rollbackUpdateSQLs))
	return updateSQLs, rollbackUpdateSQLs, nil
}

// buildUpdateSQLs builds the sqls, the final state is same as filterRule's definition
func buildUpdateSQLs(
	ctx context.Context,
	filterRule *entity.AuditLogFilterRule,
	oldFilterRule *entity.AuditLogFilterRule,
) ([]string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", oldFilterRule.ClusterID))
	updateSQLs := make([]string, 0)
	// filter
	if filterRule.Filters != nil {
		content, err := filterRule.ToContent()
		if err != nil {
			logger.Error("failed to convert filter rule to content", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid filters, please check it and try again")
		}
		updateSQLs = append(updateSQLs, fmt.Sprintf(FilterReplaceSQL, filterRule.ID, content))
	}

	var desiredDisabled bool
	if filterRule.Disabled != nil {
		desiredDisabled = *filterRule.Disabled
	} else {
		// if disabled field is not changed, use the old value
		desiredDisabled = *oldFilterRule.Disabled
	}

	// users
	added, removed, notchanged := usersDiff(oldFilterRule.Users, filterRule.Users)
	for _, user := range removed {
		updateSQLs = append(updateSQLs, fmt.Sprintf(FilterRuleDeleteSQL, user, filterRule.ID))
	}
	for _, user := range added {
		updateSQLs = append(updateSQLs, fmt.Sprintf(FilterRuleReplaceSQL, user, filterRule.ID))
		// newly added user is enabled by default so we need to explicitly disable it if desiredDisabled is true
		if desiredDisabled {
			updateSQLs = append(updateSQLs, fmt.Sprintf(FilterRuleDisableSQL, user, filterRule.ID))
		}
	}
	// do not judge oldFilterRule.Disabled, it may be nil when this method is called by rollback
	for _, user := range notchanged {
		if desiredDisabled {
			updateSQLs = append(updateSQLs, fmt.Sprintf(FilterRuleDisableSQL, user, filterRule.ID))
		} else {
			updateSQLs = append(updateSQLs, fmt.Sprintf(FilterRuleEnableSQL, user, filterRule.ID))
		}
	}
	logger.Info("users diff", zap.Any("added", added), zap.Any("removed", removed), zap.Any("notchanged", notchanged))
	return updateSQLs, nil
}

func usersDiff(oldUsers, newUsers []string) (added, removed, notchanged []string) {
	// not changed when newUsers is empty
	if len(newUsers) == 0 {
		return nil, nil, oldUsers
	}

	oldUserSet := mapset.NewSet[string]()
	for _, user := range oldUsers {
		oldUserSet.Add(user)
	}
	newUserSet := mapset.NewSet[string]()
	for _, user := range newUsers {
		newUserSet.Add(user)
	}

	addedSet := newUserSet.Difference(oldUserSet)
	removedSet := oldUserSet.Difference(newUserSet)
	notchangedSet := oldUserSet.Intersect(newUserSet)

	return addedSet.ToSlice(), removedSet.ToSlice(), notchangedSet.ToSlice()
}

func executeMultiSQLs(
	ctx context.Context,
	clusterProxy facade.ClusterProxyFacade,
	sqls []string,
) (err bizerr.BizError) {
	logger := log.WithContextL(ctx)
	logger.Info("start to execute multi sqls", zap.Any("sqls", sqls))

	if len(sqls) == 0 {
		return nil
	}

	for range ExecuteMultiSQLMaxRetryTimes {
		if err = clusterProxy.ExecuteMultiSQL(ctx, sqls); err == nil {
			return nil
		}
	}
	logger.Error("failed to execute sqls", zap.Error(err))
	return parseSQLError(err)
}

func rollbackSQLs(
	ctx context.Context,
	clusterProxy facade.ClusterProxyFacade,
	sqls []string,
	clusterId string,
	operation metric.AuditlogOperation,
) {
	logger := log.WithContextL(ctx)
	logger.Info("start to execute rollback sqls", zap.Any("sqls", sqls))

	if len(sqls) == 0 {
		return
	}

	status := "success"
	if err := clusterProxy.ExecuteMultiSQL(ctx, sqls); err != nil {
		logger.Info("failed to execute rollback sqls, please handle it manually")
		status = "fail"
	}
	metric.AuditLogRollbackSQLCounter.WithLabelValues(clusterId, string(operation), status).Inc()
}

func parseSQLError(err error) bizerr.BizError {
	var clusterProxyErr *facade.ClusterProxyErr
	if errors.As(err, &clusterProxyErr) {
		if clusterProxyErr.SQLErrNo == mysql.ErrDupEntry {
			return bizerr.New(bizerr.ErrCodeRecordExist, "filter rule already exist")
		}
		if clusterProxyErr.SQLErrNo == mysql.ErrUnknown {
			if strings.Contains(clusterProxyErr.SQLErrMsg, "rule does not exist") {
				return bizerr.New(bizerr.ErrCodeRecordNotFound, "filter rule not find")
			}
			if strings.Contains(clusterProxyErr.SQLErrMsg, "invalid table format") {
				return bizerr.New(bizerr.ErrCodeParams, "invalid filters: the table format is invalid")
			}
			if strings.Contains(clusterProxyErr.SQLErrMsg, "invalid event class name") {
				return bizerr.New(bizerr.ErrCodeParams, "invalid filters: the event class name is invalid")
			}
			if strings.Contains(clusterProxyErr.SQLErrMsg, "illegal user format") {
				return bizerr.New(bizerr.ErrCodeParams, "invalid users: the user format is invalid")
			}
		}
		return clusterProxyErr
	}
	return bizerr.New(bizerr.ErrCodeUnknown, "")
}

func (a *auditLogService) ListAuditLogFiles(
	ctx context.Context,
	clusterId, pageToken, date string,
	pageSize int32,
) ([]*vo.AuditLogFile, string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId))
	logger.Info("start to list audit log files",
		zap.Int32("pageSize", pageSize),
		zap.String("pageToken", pageToken),
		zap.String("date", date),
	)

	if !isValidDate(date) {
		return nil, "", bizerr.New(
			errorcode.ErrCodeInvalidDate,
			"Invalid date %s, please provide the date in yyyy-mm-dd format",
			date,
		)
	}
	cluster, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, "", err
	}

	s3Client, err1 := a.s3ClientProvider.GetS3Client(cluster.CloudProvider, cluster.Region)
	if err1 != nil {
		logger.Error("get s3 client failed", zap.Error(err1))
		return nil, "", bizerr.New(errorcode.ErrCodeInternalError, "")
	}

	// list files
	prefix := getAuditLogPath(cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId)
	if date != "" {
		prefix = fmt.Sprintf("%s/%s", prefix, date)
	}
	input := &s3.ListObjectsV2Input{
		Bucket:  aws.String(getAuditLogBucket(cluster.Region)),
		Prefix:  aws.String(prefix),
		MaxKeys: aws.Int32(pageSize),
	}
	if pageToken != "" {
		input.ContinuationToken = aws.String(pageToken)
	}
	output, err1 := s3Client.ListObjectsV2(ctx, input)
	if err1 != nil {
		logger.Error("list objects failed", zap.Error(err1))
		return nil, "", bizerr.New(errorcode.ErrCodeInternalError, "")
	}

	// build response
	auditLogs := make([]*vo.AuditLogFile, 0)
	for _, obj := range output.Contents {
		objectUri := strings.Split(*obj.Key, "/")
		objectName := objectUri[len(objectUri)-1]
		var fileSize uint32
		if obj.Size != nil {
			fileSize = uint32(*obj.Size)
		}
		auditLog := &vo.AuditLogFile{
			Name:      objectName,
			SizeBytes: fileSize,
			CreatedAt: obj.LastModified,
		}
		auditLogs = append(auditLogs, auditLog)
	}
	nextPageToken := ""
	if output.NextContinuationToken != nil {
		nextPageToken = *output.NextContinuationToken
	}
	return auditLogs, nextPageToken, nil
}

func (a *auditLogService) DownloadAuditLogFiles(
	ctx context.Context,
	clusterId string,
	fileNames []string,
) ([]*vo.AuditLogFile, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))

	cluster, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	preSignerClient, err1 := a.preSignerProvider.GetPresignerClient(cluster.CloudProvider, cluster.Region)
	if err1 != nil {
		logger.Error("get presigner client failed", zap.Error(err1))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}

	auditLogs := make([]*vo.AuditLogFile, 0)
	for _, name := range fileNames {
		key := fmt.Sprintf("%s/%s", getAuditLogPath(cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId), name)
		res, err := preSignerClient.PresignGetObject(ctx, &s3.GetObjectInput{
			Bucket: aws.String(getAuditLogBucket(cluster.Region)),
			Key:    aws.String(key),
		}, s3.WithPresignExpires(perSignedUrlExpireTime))
		if err != nil {
			logger.Error("failed to generate pre-signed url", zap.String("key", key), zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
		}
		auditLogs = append(auditLogs, &vo.AuditLogFile{
			Name: name,
			Url:  res.URL,
		})
	}
	return auditLogs, nil
}

func (a *auditLogService) checkCluster(
	ctx context.Context,
	clusterId string,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	clusterIdUint64, err := strconv.ParseUint(clusterId, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	cluster, bizErr := a.mgmtSvc.GetCluster(
		ctx,
		clusterIdUint64,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, bizerr.New(
				bizerr.ErrCodeRecordNotFound,
				"You can not execute this operation, reason: cluster not found.",
			)
		}
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	if cluster.GetServerlessCluster().GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster is not running")
	}
	return cluster, nil
}

func getAuditLogBucket(region string) string {
	return fmt.Sprintf("%s-%s-auditlog", config.Stage(), region)
}

func getAuditLogPath(orgId, projectId, clusterId uint64) string {
	return fmt.Sprintf("%d/%d/%d", orgId, projectId, clusterId)
}

func (a *auditLogService) GetAuditLogFilterRule(
	ctx context.Context,
	clusterId, ruleId string,
) (*entity.AuditLogFilterRule, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId), zap.String("ruleId", ruleId))
	logger.Info("start to get audit log filter rule")

	_, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	filterRule, err := a.auditLogFilterRuleRepo.GetByID(ctx, clusterId, ruleId)
	if err != nil {
		logger.Error("failed to get audit log filter rule", zap.Error(err))
		return nil, err
	}

	return filterRule, nil
}

func (a *auditLogService) ListAuditLogFilterRules(
	ctx context.Context,
	opts *vo.ListFilterRuleOpts,
) ([]*entity.AuditLogFilterRule, uint32, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", opts.ClusterID))
	logger.Info("start to list audit log filter rules", zap.Any("opts", opts))

	_, err := a.checkCluster(ctx, opts.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, 0, err
	}

	filter := entity.AuditLogFilterRuleFilter{
		ClusterID: opts.ClusterID,
		Limit:     &opts.Limit,
		Offset:    &opts.Offset,
	}

	filterRules, err := a.auditLogFilterRuleRepo.ListByFilter(ctx, filter)
	if err != nil {
		logger.Error("failed to list audit log filter rules", zap.Error(err))
		return nil, 0, err
	}

	total, err := a.auditLogFilterRuleRepo.CountByFilter(ctx, filter)
	if err != nil {
		logger.Error("failed to count audit log filter rules", zap.Error(err))
		return nil, 0, err
	}

	return filterRules, uint32(total), nil
}

func (a *auditLogService) DeleteAuditLogFilterRule(
	ctx context.Context,
	clusterId, ruleId string,
) (*entity.AuditLogFilterRule, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId), zap.String("ruleId", ruleId))
	logger.Info("start to delete audit log filter rule")

	cluster, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	filterRule, err := a.auditLogFilterRuleRepo.GetByID(ctx, clusterId, ruleId)
	if err != nil {
		logger.Error("failed to get audit log filter rule", zap.Error(err))
		return nil, err
	}

	clusterProxy, _, err := a.getClusterProxyWithCloudAdmin(ctx, cluster)
	if err != nil {
		return nil, err
	}

	// delete filter rule
	if err := a.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err = a.auditLogFilterRuleRepo.Delete(ctx, clusterId, ruleId); err != nil {
			logger.Error("failed to delete audit log filter rule in db", zap.Error(err))
			return err
		}
		if err = deleteFilterRule(ctx, clusterProxy, filterRule, clusterId); err != nil {
			logger.Error("failed to delete audit log filter rule", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	return filterRule, nil
}

func (a *auditLogService) CreateAuditLogFilterRule(
	ctx context.Context,
	clusterId string,
	filterRule *entity.AuditLogFilterRule,
) (*entity.AuditLogFilterRule, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId))
	logger.Info("start to create audit log filter rule", zap.Any("filterRule", filterRule))

	cluster, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	clusterProxy, cluster, err := a.getClusterProxyWithCloudAdmin(ctx, cluster)
	if err != nil {
		return nil, err
	}

	// fill in the audit log filter rule
	filterRule.FillWithCluster(cluster)
	filterRule.ID = entity.GenerateIDWithPrefix(constant.AuditLogFilterRulePrefix)

	// validate filter rule
	userPrefix := cluster.GetServerlessCluster().GetCluster().GetUserPrefix()
	if err = filterRule.Validate(userPrefix); err != nil {
		logger.Error("filter rule validation failed", zap.Error(err))
		return nil, err
	}
	// check number of filter rule
	count, err := a.auditLogFilterRuleRepo.CountByFilter(ctx, entity.AuditLogFilterRuleFilter{
		ClusterID: clusterId,
	})
	if err != nil {
		logger.Error("failed to count audit log filter rules", zap.Error(err))
		return nil, err
	}
	if count >= MaxFilterRuleCount {
		return nil, bizerr.New(errorcode.ErrCodeFilterRuleExceedMaxNumber, "")
	}

	// create filter rule
	if err := a.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err = a.auditLogFilterRuleRepo.Create(ctx, filterRule); err != nil {
			logger.Error("failed to create audit log filter rule in db", zap.Error(err))
			return err
		}
		if err := createFilterRule(ctx, clusterProxy, filterRule, clusterId); err != nil {
			logger.Error("failed to create audit log filter rule", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	return filterRule, nil
}

func (a *auditLogService) UpdateAuditLogFilterRule(
	ctx context.Context,
	clusterId string,
	filterRule *entity.AuditLogFilterRule,
) (*entity.AuditLogFilterRule, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId))
	logger.Info("start to update audit log filter rule", zap.Any("filterRule", filterRule))

	cluster, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	clusterProxy, cluster, err := a.getClusterProxyWithCloudAdmin(ctx, cluster)
	if err != nil {
		return nil, err
	}

	userPrefix := cluster.GetServerlessCluster().GetCluster().GetUserPrefix()
	if err := filterRule.Validate(userPrefix); err != nil {
		return nil, err
	}

	oldFilterRule, err := a.auditLogFilterRuleRepo.GetByID(ctx, clusterId, filterRule.ID)
	if err != nil {
		logger.Error("failed to get old audit log filter rule", zap.Error(err))
		return nil, err
	}

	// generate update sqls
	updateSQLs, rollbackUpdateSQLs, err := generateUpdateFilterRuleSQLs(ctx, filterRule, oldFilterRule)
	if err != nil {
		logger.Error("failed to generate update sqls", zap.Error(err))
		return nil, err
	}

	// build final filter rule
	if filterRule.Filters != nil {
		oldFilterRule.Filters = filterRule.Filters
	}
	// users will not be empty because we judge it before
	if filterRule.Users != nil {
		oldFilterRule.Users = filterRule.Users
	}
	if filterRule.Disabled != nil {
		oldFilterRule.Disabled = filterRule.Disabled
	}
	if filterRule.Name != "" {
		oldFilterRule.Name = filterRule.Name
	}

	// update filter rule
	if err := a.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err = a.auditLogFilterRuleRepo.Save(ctx, oldFilterRule); err != nil {
			logger.Error("failed to create audit log filter rule in db", zap.Error(err))
			return err
		}
		if err := executeMultiSQLs(ctx, clusterProxy, updateSQLs); err != nil {
			rollbackSQLs(ctx, clusterProxy, rollbackUpdateSQLs, clusterId, metric.AuditlogOperationUpdate)
			return err
		}

		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	return oldFilterRule, nil
}

func (a *auditLogService) GetAuditLogConfig(
	ctx context.Context,
	clusterId string,
) (*entity.AuditLogConfig, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", clusterId))
	logger.Info("start to get audit log config")

	_, err := a.checkCluster(ctx, clusterId)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	alConfig, err := a.auditLogConfigRepo.Get(ctx, clusterId)
	if err != nil {
		logger.Error("failed to get audit log config", zap.Error(err))
		return nil, err
	}
	return alConfig, nil
}

func (a *auditLogService) UpdateAuditLogConfig(
	ctx context.Context,
	auditLogConfig *entity.AuditLogConfig,
) (*entity.AuditLogConfig, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", auditLogConfig.ClusterID))
	logger.Info("start to update audit log config", zap.Any("auditLogConfig", auditLogConfig))

	// do checks
	cluster, err := a.checkCluster(ctx, auditLogConfig.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}
	if err = auditLogConfig.Validate(cluster); err != nil {
		logger.Error("filter rule validation failed", zap.Error(err))
		return nil, err
	}
	// build updated audit log config
	updatedAuditLogConfig, changed, err := a.buildUpdatedAuditLogConfig(ctx, auditLogConfig)
	if err != nil {
		return nil, err
	}
	if !changed {
		return updatedAuditLogConfig, nil
	}
	// build regional audit log config
	regionalAuditLogConfig := auditLogConfig.ToRegional()
	if err = a.preCheckCloudStorage(ctx, regionalAuditLogConfig.CloudStorage, auditLogConfig.ClusterID); err != nil {
		logger.Error("pre check cloud storage failed", zap.Error(err))
		return nil, err
	}

	// save to db and update regional config
	if err := a.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err = a.createOrUpdateAuditLogConfig(ctx, updatedAuditLogConfig, cluster); err != nil {
			logger.Error("failed to save audit log config in db", zap.Error(err))
			return err
		}
		if err = a.updateRegionalAuditLogConfig(ctx, cluster, regionalAuditLogConfig); err != nil {
			logger.Error("failed to update regional audit log config", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	return updatedAuditLogConfig, nil
}

func (a *auditLogService) updateRegionalAuditLogConfig(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	regionalAuditLogConfig *regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AuditLogConfig,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.Uint64("clusterId", cluster.ClusterId))
	logger.Info("start to update regional audit log config")
	// This logical will be put into serverless regional svc in the future
	if regionalAuditLogConfig.Enabled != nil || regionalAuditLogConfig.Unredacted != nil {
		clusterProxy, _, err := a.getClusterProxyWithCloudAdmin(ctx, cluster)
		if err != nil {
			logger.Error("failed to get cluster proxy", zap.Error(err))
			return err
		}
		if regionalAuditLogConfig.Enabled != nil {
			sql := DisabledAuditLogSQL
			if *regionalAuditLogConfig.Enabled {
				sql = EnabledAuditLogSQL
			}
			if err := clusterProxy.ExecuteSQL(ctx, sql); err != nil {
				logger.Error("failed to enable/disable audit log", zap.String("sql", sql), zap.Error(err))
				return err
			}
		}
		if regionalAuditLogConfig.Unredacted != nil {
			sql := RedactedAuditLogSQL
			if *regionalAuditLogConfig.Unredacted {
				sql = UnredactedAuditLogSQL
			}
			if err := clusterProxy.ExecuteSQL(ctx, sql); err != nil {
				logger.Error("failed to redact/unredact audit log", zap.String("sql", sql), zap.Error(err))
				return err
			}
		}
	}
	// update regional config
	if err := a.mgmtSvc.UpdateAuditLogConfig(ctx, strconv.FormatUint(cluster.ClusterId, 10), regionalAuditLogConfig); err != nil {
		logger.Error("failed to update audit log config", zap.Error(err))
		return err
	}
	return nil
}

func (a *auditLogService) createOrUpdateAuditLogConfig(
	ctx context.Context,
	auditLogConfig *entity.AuditLogConfig,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	auditLogConfig.HideSecret()
	// If the tenant ID is empty, it means it is the first time to set the audit log config
	if auditLogConfig.TenantID == "" {
		auditLogConfig.FillWithCluster(cluster)
		return a.auditLogConfigRepo.Create(ctx, auditLogConfig)
	}
	return a.auditLogConfigRepo.Save(ctx, auditLogConfig)
}

// buildUpdatedAuditLogConfig return the updated audit log config, and auditLogConfig will be set to nil for the unchanged fields
func (a *auditLogService) buildUpdatedAuditLogConfig(
	ctx context.Context,
	auditLogConfig *entity.AuditLogConfig,
) (*entity.AuditLogConfig, bool, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", auditLogConfig.ClusterID))
	// get changed audit log config
	alfEntity, err := a.auditLogConfigRepo.Get(ctx, auditLogConfig.ClusterID)
	if err != nil {
		logger.Error("failed to get audit log config", zap.Error(err))
		return nil, false, err
	}
	if !buildNewAuditLogConfig(ctx, alfEntity, auditLogConfig) {
		logger.Info("no changes in audit log config, skip update")
		return alfEntity, false, nil
	}
	return alfEntity, true, nil
}

func buildNewAuditLogConfig(
	ctx context.Context,
	oldConfig, newConfig *entity.AuditLogConfig,
) bool {
	logger := log.WithContextL(ctx).
		With(zap.String("oldConfig", oldConfig.ClusterID), zap.String("newConfig", newConfig.ClusterID))
	logger.Info("compare audit log config", zap.Any("oldConfig", oldConfig), zap.Any("newConfig", newConfig))
	// update oldConfig for the changed fields, set newConfig to nil if the value is not changed
	somethingChanged := false
	if newConfig.Enabled != nil {
		if *newConfig.Enabled != *oldConfig.Enabled {
			oldConfig.Enabled = newConfig.Enabled
			somethingChanged = true
		} else {
			newConfig.Enabled = nil
		}
	}
	if newConfig.Unredacted != nil {
		if *newConfig.Unredacted != *oldConfig.Unredacted {
			oldConfig.Unredacted = newConfig.Unredacted
			somethingChanged = true
		} else {
			newConfig.Unredacted = nil
		}
	}
	if newConfig.CloudStorageType != "" {
		somethingChanged = true
		oldConfig.CloudStorage = newConfig.CloudStorage
		oldConfig.CloudStorageType = newConfig.CloudStorageType
	}
	if newConfig.RotationIntervalSeconds != 0 {
		if newConfig.RotationIntervalSeconds != oldConfig.RotationIntervalSeconds {
			oldConfig.RotationIntervalSeconds = newConfig.RotationIntervalSeconds
			somethingChanged = true
		} else {
			newConfig.RotationIntervalSeconds = 0
		}
	}
	if newConfig.RotationSizeBytes != 0 {
		if newConfig.RotationSizeBytes != oldConfig.RotationSizeBytes {
			oldConfig.RotationSizeBytes = newConfig.RotationSizeBytes
			somethingChanged = true
		} else {
			newConfig.RotationSizeBytes = 0
		}
	}
	return somethingChanged
}

func (a *auditLogService) PreCheckAuditLogConfig(
	ctx context.Context,
	auditLogConfig *entity.AuditLogConfig,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterId", auditLogConfig.ClusterID))
	logger.Info("start to pre check audit log config", zap.Any("auditLogConfig", auditLogConfig))

	_, err := a.checkCluster(ctx, auditLogConfig.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return err
	}

	regionalcs := auditLogConfig.CloudStorage.ToRegionalCloudStorage()
	if err := a.preCheckCloudStorage(ctx, regionalcs, auditLogConfig.ClusterID); err != nil {
		logger.Error("pre check cloud storage failed", zap.Error(err))
		return err
	}

	return nil
}

func (a *auditLogService) preCheckCloudStorage(
	ctx context.Context,
	cs *regionalserverlessv1.CloudStorage,
	clusterId string,
) bizerr.BizError {
	if cs == nil || cs.CloudStorageType == regionalserverlessv1.CloudStorageType_INTERNAL {
		return nil
	}
	request := &globalserverlessv1.CheckCloudStoragePermissionRequest{
		CloudStorage: cs,
		ClusterId:    clusterId,
		FeatureName:  globalserverlessv1.CheckCloudStoragePermissionRequest_FEATURE_NAME_AUDITLOG,
	}
	if err := a.mgmtSvc.CheckCloudStoragePermission(ctx, request); err != nil {
		return err
	}
	return nil
}

func isValidDate(dateStr string) bool {
	if dateStr == "" {
		return true
	}
	_, err := time.Parse(time.DateOnly, dateStr)
	return err == nil
}