package repo_imp

import (
	"context"
	"errors"
	"strings"
	"sync"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/tracing"
	"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/repo"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

type backupJobRepoImp struct {
	dbProvider db.Provider
}

func NewBackupJobRepoImp(dbProvider db.Provider) repo.BackupJobRepo {
	return &backupJobRepoImp{dbProvider}
}

func (b *backupJobRepoImp) Delete(ctx context.Context, expireDay int) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupJobRepo.Clean")
	defer func() { tracing.EndSpan(span, err) }()

	curTime := time.Now()
	cleanTime := curTime.AddDate(0, 0, -expireDay)
	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Where("start_time < ?", cleanTime).Delete(&entity.BackupJob{}).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to clean backup jobs")
	}
	return nil
}

type backupScheduleRepoImp struct {
	dbProvider db.Provider
}

func NewBackupScheduleRepoImp(dbProvider db.Provider) repo.BackupScheduleRepo {
	return &backupScheduleRepoImp{dbProvider}
}

func (b *backupScheduleRepoImp) List(
	ctx context.Context,
	options entity.ListBackupScheduleOptions,
) (backupSchedules []*entity.BackupSchedule, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepo.List")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx)
	rdb = buildBackupScheduleFilter(options, rdb)

	if err := rdb.Find(&backupSchedules).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "failed to list backup schedules")
	}
	return backupSchedules, nil
}

func buildBackupScheduleFilter(options entity.ListBackupScheduleOptions, db *gorm.DB) *gorm.DB {
	if options.IncludeDeleted {
		db = db.Unscoped()
	}
	if options.ClusterId != "" {
		db = db.Where("cluster_id = ?", options.ClusterId)
	}
	if len(options.ClustersIn) > 0 {
		db = db.Where("cluster_id in ?", options.ClustersIn)
	}
	if options.IncompleteSnapshot {
		db = db.Where(`
		(source_token IS NULL OR source_token = '' OR
		 enhanced_encryption_enabled IS NULL OR
		 zone_redundant_enabled IS NULL OR
		 service_plan IS NULL OR service_plan = '')
	`)
	}
	if options.IncompleteSnapshotBatchSize > 0 {
		db = db.Order("id ASC").Limit(options.IncompleteSnapshotBatchSize)
	}
	return db
}

func (b *backupScheduleRepoImp) Create(ctx context.Context, schedule *entity.BackupSchedule) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepo.Create")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Create(schedule).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to create schedule")
	}
	return nil
}

func (b *backupScheduleRepoImp) DeleteByClusterID(ctx context.Context, clusterID string) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepo.DeleteByClusterID")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Where("cluster_id = ?", clusterID).Delete(&entity.BackupSchedule{}).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to delete schedule")
	}
	return nil
}

func (b *backupScheduleRepoImp) UpdateRetentionByClusterID(
	ctx context.Context,
	retention int,
	clusterID string,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepo.UpdateRetentionByClusterID")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	result := wdb.Model(&entity.BackupSchedule{}).Where("cluster_id = ?", clusterID).Update("retention", retention)
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to update schedule retention")
	}
	return nil
}

func (b *backupScheduleRepoImp) Save(
	ctx context.Context,
	schedule *entity.BackupSchedule,
) (_ *entity.BackupSchedule, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepo.Save")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Save(schedule).Error; err != nil {
		return schedule, bizerr.WrapGormError(err, "fail to update backup schedule")
	}
	return schedule, nil
}

func (b *backupScheduleRepoImp) CleanDeletedWithoutBackup(
	ctx context.Context,
	expireTimeGte time.Time,
	limit int,
) (count int64, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepo.CleanDeletedWithoutBackup")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	subQuery := wdb.Unscoped().
		Model(&entity.BackupSchedule{}).
		Select("backup_schedules.id").
		Where("backup_schedules.deleted_at IS NOT NULL").
		Where(
			`NOT EXISTS (
				SELECT 1
				FROM user_backups ub
				WHERE ub.cluster_id = backup_schedules.cluster_id
				  AND ub.expire_time >= ?
				  AND ub.state = ?
			)`,
			expireTimeGte,
			constant.BackupStateSucceeded,
		)
	// Soft-deleted schedules are still required by restore-to-new for virtual
	// backups. History cleanup is driven by resolved physical backup times, so a
	// remaining history row is the source of truth that the schedule is still
	// needed even when there is no legacy user_backups row.
	subQuery = subQuery.Where(
		`NOT EXISTS (
			SELECT 1
			FROM backup_schedule_histories bsh
			WHERE bsh.cluster_id = backup_schedules.cluster_id
		)`,
	)
	if limit > 0 {
		subQuery = subQuery.Limit(limit)
	}

	// Query IDs first, then delete by IDs to avoid MySQL Error 1235:
	// "This version of MySQL doesn't yet support 'LIMIT & IN/ALL/ANY/SOME subquery'"
	var ids []int64
	if err := subQuery.Pluck("id", &ids).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to find deleted backup schedules to clean")
	}
	if len(ids) == 0 {
		return 0, nil
	}

	result := wdb.Unscoped().Where("id IN ?", ids).Delete(&entity.BackupSchedule{})
	if result.Error != nil {
		return 0, bizerr.WrapGormError(result.Error, "fail to clean deleted backup schedules")
	}
	return result.RowsAffected, nil
}

func (b *backupScheduleRepoImp) Count(ctx context.Context) (count int64, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepoImp.Count")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx)
	if err := rdb.Model(&entity.BackupSchedule{}).Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to count backup schedules")
	}
	return count, nil
}

func (b *backupScheduleRepoImp) BatchAdd(
	ctx context.Context,
	schedules []*entity.BackupSchedule,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepoImp.BatchAdd")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Create(schedules).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to batch add backup schedules")
	}
	return nil
}

func (b *backupScheduleRepoImp) UpdateKubernetesByClusterID(
	ctx context.Context,
	clusterId, kubernetes string,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleRepoImp.UpdateKubernetesByClusterID")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.BackupSchedule{}).Where("cluster_id = ?", clusterId).Update("kubernetes", kubernetes).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to update kubernetes")
	}
	return nil
}

type backupScheduleHistoryRepoImp struct {
	dbProvider db.Provider
	cutoverMu  sync.RWMutex
	cutoverAt  *time.Time
}

func NewBackupScheduleHistoryRepoImp(dbProvider db.Provider) repo.BackupScheduleHistoryRepo {
	return &backupScheduleHistoryRepoImp{dbProvider: dbProvider}
}

func (b *backupScheduleHistoryRepoImp) GetVirtualizationCutover(
	ctx context.Context,
) (cutoverAt time.Time, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleHistoryRepo.GetVirtualizationCutover")
	defer func() { tracing.EndSpan(span, err) }()

	b.cutoverMu.RLock()
	if b.cutoverAt != nil {
		cutoverAt = *b.cutoverAt
		b.cutoverMu.RUnlock()
		return cutoverAt, nil
	}
	b.cutoverMu.RUnlock()

	var metadata entity.BackupVirtualizationMetadata
	if dbErr := b.dbProvider.ReadDB(ctx).First(&metadata, 1).Error; dbErr != nil {
		return time.Time{}, bizerr.WrapGormError(dbErr, "fail to get backup virtualization cutover")
	}
	cutoverAt = metadata.CutoverAt.UTC()
	b.cutoverMu.Lock()
	if b.cutoverAt == nil {
		b.cutoverAt = &cutoverAt
	} else {
		cutoverAt = *b.cutoverAt
	}
	b.cutoverMu.Unlock()
	return cutoverAt, nil
}

func (b *backupScheduleHistoryRepoImp) ArchiveCurrentSchedule(
	ctx context.Context,
	schedule *entity.BackupSchedule,
	validTo time.Time,
	deletedAt *time.Time,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleHistoryRepo.ArchiveCurrentSchedule")
	defer func() { tracing.EndSpan(span, err) }()

	cutoverAt, err := b.GetVirtualizationCutover(ctx)
	if err != nil {
		return err
	}
	var latestHistory entity.BackupScheduleHistory
	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Model(&entity.BackupScheduleHistory{}).
		Select("valid_to").
		Where("cluster_id = ? AND valid_to IS NOT NULL", schedule.ClusterID).
		Order("valid_to DESC").
		Limit(1).
		Find(&latestHistory).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to find latest backup schedule history")
	}
	previousHistories := []*entity.BackupScheduleHistory{}
	if latestHistory.ValidTo != nil {
		previousHistories = append(previousHistories, &latestHistory)
	}
	history := entity.NewBackupScheduleHistoryFromSchedule(schedule, cutoverAt, previousHistories, &validTo)
	history.DeletedAt = deletedAt
	if err := wdb.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "schedule_id"}, {Name: "valid_to"}},
		DoNothing: true,
	}).Create(history).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to archive backup schedule history")
	}
	return nil
}

func (b *backupScheduleHistoryRepoImp) MarkDeleted(
	ctx context.Context,
	clusterID string,
	deletedAt time.Time,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleHistoryRepo.MarkDeleted")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	updates := map[string]any{
		"valid_to":   gorm.Expr("COALESCE(valid_to, ?)", deletedAt),
		"deleted_at": deletedAt,
	}
	if err := wdb.Model(&entity.BackupScheduleHistory{}).
		Where("cluster_id = ?", clusterID).
		Updates(updates).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to mark backup schedule history deleted")
	}
	return nil
}

func (b *backupScheduleHistoryRepoImp) List(
	ctx context.Context,
	options *entity.ListBackupScheduleHistoryOptions,
) (histories []*entity.BackupScheduleHistory, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleHistoryRepo.List")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx).Model(&entity.BackupScheduleHistory{})
	rdb = buildBackupScheduleHistoryFilter(options, rdb)
	rdb = rdb.Order("cluster_id asc, valid_from asc")
	if err := rdb.Find(&histories).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list backup schedule histories")
	}
	return histories, nil
}

func buildBackupScheduleHistoryFilter(options *entity.ListBackupScheduleHistoryOptions, db *gorm.DB) *gorm.DB {
	if options == nil {
		return db
	}
	if options.ClusterId != "" {
		db = db.Where("cluster_id = ?", options.ClusterId)
	}
	if options.TenantId != "" {
		db = db.Where("tenant_id = ?", options.TenantId)
	}
	if len(options.ClustersIn) > 0 {
		db = db.Where("cluster_id in ?", options.ClustersIn)
	}
	if options.ValidBeforeOrEqual != nil {
		db = db.Where("valid_from <= ?", *options.ValidBeforeOrEqual)
	}
	if options.ValidAfter != nil && options.DeletedOnly {
		db = db.Where("valid_to > ?", *options.ValidAfter)
	} else if options.ValidAfter != nil {
		db = db.Where("(valid_to IS NULL OR valid_to > ?)", *options.ValidAfter)
	}
	if options.DeletedOnly {
		db = db.Where("deleted_at IS NOT NULL")
	}
	if regionFilter, filterArgs := buildBackupScheduleHistoryRegionFilter(options.RegionPairs); regionFilter != "" {
		db = db.Where(regionFilter, filterArgs...)
	}
	return db
}

func buildBackupScheduleHistoryRegionFilter(regionPairs []entity.ProviderRegion) (string, []any) {
	if len(regionPairs) == 0 {
		return "", nil
	}
	regionFilters := make([]string, 0, len(regionPairs))
	filterArgs := make([]any, 0, len(regionPairs)*2)
	for _, pair := range regionPairs {
		providerFilter := "cloud_provider = ?"
		if pair.Provider == config.ProviderAWS {
			providerFilter = "(cloud_provider = ? OR cloud_provider = '' OR cloud_provider IS NULL)"
		}
		regionFilters = append(regionFilters, "("+providerFilter+" AND region = ?)")
		filterArgs = append(filterArgs, pair.Provider, pair.Region)
	}
	return "(" + strings.Join(regionFilters, " OR ") + ")", filterArgs
}

func buildBackupScheduleHistoryScopeMatch(alias string) (string, []any) {
	defaultKubernetesPrefix, defaultKubernetesSuffix := entity.DefaultKubernetesNameParts()
	return alias + ".cloud_provider = COALESCE(NULLIF(backup_schedule_histories.cloud_provider, ''), ?)" +
			" AND " + alias + ".region = backup_schedule_histories.region" +
			" AND " + alias + ".kubernetes = COALESCE(NULLIF(backup_schedule_histories.kubernetes, ''), " +
			"CONCAT(?, LOWER(backup_schedule_histories.region), ?))",
		[]any{entity.DefaultCloudProvider, defaultKubernetesPrefix, defaultKubernetesSuffix}
}

func (b *backupScheduleHistoryRepoImp) DeleteExpired(
	ctx context.Context,
	now time.Time,
	limit int,
) (count int64, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleHistoryRepo.DeleteExpired")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	// A closed history is safe to remove only after the slot synchronizer has
	// advanced past its effective end and every physical slot in the interval is
	// expired according to its actual backup_time. This deliberately places no
	// upper bound on the delay between slot_time and backup_time.
	effectiveEnd := "LEAST(backup_schedule_histories.valid_to, " +
		"COALESCE(backup_schedule_histories.deleted_at, backup_schedule_histories.valid_to))"
	watermarkScopeMatch, watermarkScopeArgs := buildBackupScheduleHistoryScopeMatch("watermark")
	slotScopeMatch, slotScopeArgs := buildBackupScheduleHistoryScopeMatch("slot")
	query := wdb.Model(&entity.BackupScheduleHistory{}).
		Where("backup_schedule_histories.valid_to IS NOT NULL").
		Where(
			"EXISTS (SELECT 1 FROM physical_backup_slots watermark WHERE "+watermarkScopeMatch+
				" AND watermark.slot_time >= "+effectiveEnd+")",
			watermarkScopeArgs...,
		).
		Where(
			"NOT EXISTS (SELECT 1 FROM physical_backup_slots slot WHERE "+slotScopeMatch+
				" AND slot.slot_time >= backup_schedule_histories.valid_from"+
				" AND slot.slot_time < "+effectiveEnd+
				" AND DATE_ADD(slot.backup_time, INTERVAL backup_schedule_histories.retention HOUR) >= ?)",
			append(slotScopeArgs, now.UTC())...,
		)
	if limit > 0 {
		var ids []int64
		if err := query.Model(&entity.BackupScheduleHistory{}).Limit(limit).Pluck("id", &ids).Error; err != nil {
			return 0, bizerr.WrapGormError(err, "fail to find expired backup schedule histories")
		}
		if len(ids) == 0 {
			return 0, nil
		}
		query = wdb.Where("id IN ?", ids)
	}
	result := query.Delete(&entity.BackupScheduleHistory{})
	if result.Error != nil {
		return 0, bizerr.WrapGormError(result.Error, "fail to delete expired backup schedule histories")
	}
	return result.RowsAffected, nil
}

type userBackupRepoImp struct {
	dbProvider db.Provider
}

func NewUserBackupRepoImp(dbProvider db.Provider) repo.UserBackupRepo {
	return &userBackupRepoImp{dbProvider}
}

func (b *userBackupRepoImp) Delete(ctx context.Context, options entity.UserBackupDeleteOptions) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.CleanExpire")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if options.ExpireTimeLt != nil {
		wdb = wdb.Where("expire_time < ?", options.ExpireTimeLt)
	}
	if options.Limit > 0 {
		wdb = wdb.Limit(options.Limit)
	}
	result := wdb.Delete(&entity.UserBackup{})
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to clean expire user backups")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no user backups deleted")
	}
	return nil
}

func (b *userBackupRepoImp) DeleteByBackupId(ctx context.Context, backupId string) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.DeleteByBackupId")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	result := wdb.Where("id = ?", backupId).Delete(&entity.UserBackup{})
	if result.Error != nil {
		return bizerr.WrapGormError(result.Error, "fail to delete user backup")
	}
	if result.RowsAffected == 0 {
		return bizerr.New(bizerr.ErrCodeRecordNotFound, "no user backup deleted")
	}
	return nil
}

func (b *userBackupRepoImp) GetByClusterIdAndBackupId(
	ctx context.Context,
	clusterId, backupId string,
) (userBackup *entity.UserBackup, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.GetByClusterIdAndBackupId")
	defer func() { tracing.EndSpan(span, err) }()
	rdb := b.dbProvider.ReadDB(ctx)
	userBackup = &entity.UserBackup{}
	if err := excludeTombstonedUserBackups(rdb).
		Where("id = ? and cluster_id = ?", backupId, clusterId).
		First(userBackup).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get user backup")
	}
	return userBackup, nil
}

func (b *userBackupRepoImp) List(
	ctx context.Context,
	options *entity.ListUserBackupOptions,
) (backups []*entity.UserBackup, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.List")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx)
	rdb = buildUserBackupFilter(options, rdb)

	if err := rdb.Find(&backups).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list backups")
	}
	return backups, nil
}

func (b *userBackupRepoImp) Count(
	ctx context.Context,
	options *entity.ListUserBackupOptions,
) (count int64, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.Count")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx)
	rdb = buildUserBackupCountFilter(options, rdb)
	if err := rdb.Model(&entity.UserBackup{}).Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to count backups")
	}
	return count, nil
}

func (b *userBackupRepoImp) ListDeletedClustersWithBackup(
	ctx context.Context,
	options *entity.ListRecycledInstanceOptions,
) (instances []*entity.RecycledInstance, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.ListDeletedClustersWithBackup")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx).Model(&entity.BackupSchedule{})
	rdb = rdb.Unscoped().Where("backup_schedules.deleted_at IS NOT NULL")
	rdb = buildRecycledInstanceFilter(options, rdb)
	if options != nil && options.Offset != nil {
		rdb = rdb.Offset(int(*options.Offset))
	}
	if options != nil && options.PageSize != nil {
		rdb = rdb.Limit(int(*options.PageSize))
	}
	rdb = rdb.Order("cluster_id desc")
	rdb = rdb.Select(
		"cluster_id",
		"region",
		"COALESCE(backup_schedules.enhanced_encryption_enabled, 0) AS enhanced_encryption_enabled",
		"COALESCE(backup_schedules.zone_redundant_enabled, 0) AS zone_redundant_enabled",
	)

	if err := rdb.Find(&instances).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list deleted clusters with backup")
	}
	return instances, nil
}

func buildUserBackupFilter(options *entity.ListUserBackupOptions, db *gorm.DB) *gorm.DB {
	if options == nil {
		return db
	}
	expireTimeGte := time.Now()
	if options.ExpireTimeGte != nil {
		expireTimeGte = *options.ExpireTimeGte
	}
	db = db.Where("expire_time >= ?", expireTimeGte)
	db = excludeTombstonedUserBackups(db)
	if options.ClusterId != "" {
		db = db.Where("cluster_id = ?", options.ClusterId)
	}
	if options.TenantId != "" {
		db = db.Where("tenant_id = ?", options.TenantId)
	}
	if options.Offset != nil {
		db = db.Offset(int(*options.Offset))
	}
	if options.PageSize != nil {
		db = db.Limit(int(*options.PageSize))
	}
	if options.OrderBy != "" {
		db = db.Order(options.OrderBy)
	}
	if options.BackupTimeGte != nil {
		db = db.Where("backup_time >= ?", *options.BackupTimeGte)
	}
	if len(options.States) > 0 {
		db = db.Where("state IN ?", options.States)
	}
	return db
}

func buildRecycledInstanceFilter(options *entity.ListRecycledInstanceOptions, db *gorm.DB) *gorm.DB {
	if options == nil {
		return db
	}
	expireTimeGte := time.Now()
	if options.ExpireTimeGte != nil {
		expireTimeGte = *options.ExpireTimeGte
	}
	db = db.Where(
		`EXISTS (
			SELECT 1
			FROM user_backups ub
			WHERE ub.cluster_id = backup_schedules.cluster_id
				AND ub.expire_time >= ?
				AND ub.state = ?
				AND NOT EXISTS (
					SELECT 1
					FROM backup_tombstones bt
					WHERE bt.cluster_id = ub.cluster_id
						AND bt.schedule_time = ub.schedule_time
				)
		)`,
		expireTimeGte,
		constant.BackupStateSucceeded,
	)
	if options.TenantId != "" {
		db = db.Where("backup_schedules.tenant_id = ?", options.TenantId)
	}
	if options.ClusterId != "" {
		db = db.Where("backup_schedules.cluster_id = ?", options.ClusterId)
	}
	if len(options.RegionPairs) > 0 {
		conditions := make([]string, 0, len(options.RegionPairs))
		args := make([]any, 0, len(options.RegionPairs)*2)
		for _, pair := range options.RegionPairs {
			conditions = append(conditions, "(backup_schedules.cloud_provider = ? AND backup_schedules.region = ?)")
			args = append(args, pair.Provider, pair.Region)
		}
		db = db.Where(strings.Join(conditions, " OR "), args...)
	}
	return db
}

func buildUserBackupCountFilter(options *entity.ListUserBackupOptions, db *gorm.DB) *gorm.DB {
	if options == nil {
		return db
	}
	curTime := time.Now()
	if options.ExpireTimeGte != nil {
		curTime = *options.ExpireTimeGte
	}
	db = db.Where("expire_time >= ?", curTime)
	db = excludeTombstonedUserBackups(db)
	if options.ClusterId != "" {
		db = db.Where("cluster_id = ?", options.ClusterId)
	}
	if options.TenantId != "" {
		db = db.Where("tenant_id = ?", options.TenantId)
	}
	if options.BackupTimeGte != nil {
		db = db.Where("backup_time >= ?", *options.BackupTimeGte)
	}
	if len(options.States) > 0 {
		db = db.Where("state IN ?", options.States)
	}
	return db
}

func (b *userBackupRepoImp) GetByBackupId(
	ctx context.Context,
	backupId string,
) (userBackup *entity.UserBackup, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "userBackupRepo.GetByBackupId")
	defer func() { tracing.EndSpan(span, err) }()
	rdb := b.dbProvider.ReadDB(ctx)
	userBackup = &entity.UserBackup{}
	if err := excludeTombstonedUserBackups(rdb).Where("id = ?", backupId).First(userBackup).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to get user backup")
	}
	return userBackup, nil
}

func excludeTombstonedUserBackups(db *gorm.DB) *gorm.DB {
	return db.Where(
		`NOT EXISTS (
			SELECT 1
			FROM backup_tombstones bt
			WHERE bt.cluster_id = user_backups.cluster_id
				AND bt.schedule_time = user_backups.schedule_time
		)`,
	)
}

type backupTombstoneRepoImp struct {
	dbProvider db.Provider
}

type physicalBackupSlotRepoImp struct{ dbProvider db.Provider }

func NewPhysicalBackupSlotRepoImp(dbProvider db.Provider) repo.PhysicalBackupSlotRepo {
	return &physicalBackupSlotRepoImp{dbProvider: dbProvider}
}

func (r *physicalBackupSlotRepoImp) List(
	ctx context.Context,
	keys []entity.PhysicalBackupSlotKey,
) (_ []*entity.PhysicalBackupSlot, err bizerr.BizError) {
	if len(keys) == 0 {
		return []*entity.PhysicalBackupSlot{}, nil
	}
	values := make([][]any, 0, len(keys))
	for _, key := range keys {
		values = append(values, []any{key.CloudProvider, key.Region, key.Kubernetes, key.SlotTime.UTC()})
	}
	var slots []*entity.PhysicalBackupSlot
	if dbErr := r.dbProvider.ReadDB(ctx).
		Where("(cloud_provider, region, kubernetes, slot_time) IN ?", values).
		Find(&slots).Error; dbErr != nil {
		return nil, bizerr.WrapGormError(dbErr, "fail to list physical backup slots")
	}
	return slots, nil
}

func (r *physicalBackupSlotRepoImp) ListByBackupTime(
	ctx context.Context,
	scopes []entity.PhysicalBackupSlotScope,
	backupTimeGte time.Time,
) (_ []*entity.PhysicalBackupSlot, err bizerr.BizError) {
	if len(scopes) == 0 {
		return []*entity.PhysicalBackupSlot{}, nil
	}
	values := make([][]any, 0, len(scopes))
	for _, scope := range scopes {
		values = append(values, []any{scope.CloudProvider, scope.Region, scope.Kubernetes})
	}
	var slots []*entity.PhysicalBackupSlot
	if dbErr := r.dbProvider.ReadDB(ctx).
		Where("(cloud_provider, region, kubernetes) IN ?", values).
		Where("backup_time >= ?", backupTimeGte.UTC()).
		Find(&slots).Error; dbErr != nil {
		return nil, bizerr.WrapGormError(dbErr, "fail to list physical backup slots by backup time")
	}
	return slots, nil
}

func (r *physicalBackupSlotRepoImp) Get(
	ctx context.Context,
	provider, region, kubernetes string,
	slotTime time.Time,
) (_ *entity.PhysicalBackupSlot, err bizerr.BizError) {
	var slot entity.PhysicalBackupSlot
	dbErr := r.dbProvider.ReadDB(ctx).
		Where("cloud_provider = ? AND region = ? AND kubernetes = ? AND slot_time = ?", provider, region, kubernetes, slotTime.UTC()).
		First(&slot).
		Error
	if dbErr != nil {
		return nil, bizerr.WrapGormError(dbErr, "fail to get physical backup slot")
	}
	return &slot, nil
}

func (r *physicalBackupSlotRepoImp) GetLatest(
	ctx context.Context,
	provider, region, kubernetes string,
) (_ *entity.PhysicalBackupSlot, err bizerr.BizError) {
	var slot entity.PhysicalBackupSlot
	dbErr := r.dbProvider.ReadDB(ctx).
		Where("cloud_provider = ? AND region = ? AND kubernetes = ?", provider, region, kubernetes).
		Order("slot_time DESC").
		First(&slot).Error
	if dbErr != nil {
		return nil, bizerr.WrapGormError(dbErr, "fail to get latest physical backup slot")
	}
	return &slot, nil
}

func (r *physicalBackupSlotRepoImp) BatchUpsert(
	ctx context.Context,
	slots []*entity.PhysicalBackupSlot,
) (err bizerr.BizError) {
	if len(slots) == 0 {
		return nil
	}
	if dbErr := r.dbProvider.WriteDB(ctx).Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "cloud_provider"}, {Name: "region"}, {Name: "kubernetes"}, {Name: "slot_time"}},
		DoUpdates: clause.AssignmentColumns([]string{"backup_id", "backup_name", "backup_time"}),
	}).Create(slots).Error; dbErr != nil {
		return bizerr.WrapGormError(dbErr, "fail to batch save physical backup slots")
	}
	return nil
}

func (r *physicalBackupSlotRepoImp) DeleteExpired(
	ctx context.Context,
	before time.Time,
	limit int,
) (int64, bizerr.BizError) {
	db := r.dbProvider.WriteDB(ctx).Where("backup_time < ?", before.UTC())
	if limit > 0 {
		db = db.Limit(limit)
	}
	result := db.Delete(&entity.PhysicalBackupSlot{})
	if result.Error != nil {
		return 0, bizerr.WrapGormError(result.Error, "fail to clean physical backup slots")
	}
	return result.RowsAffected, nil
}

func NewBackupTombstoneRepoImp(dbProvider db.Provider) repo.BackupTombstoneRepo {
	return &backupTombstoneRepoImp{dbProvider}
}

func (b *backupTombstoneRepoImp) Create(
	ctx context.Context,
	tombstone *entity.BackupTombstone,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupTombstoneRepo.Create")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Create(tombstone).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to create backup tombstone")
	}
	return nil
}

func (b *backupTombstoneRepoImp) List(
	ctx context.Context,
	options entity.ListBackupTombstoneOptions,
) (tombstones []*entity.BackupTombstone, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupTombstoneRepo.List")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx).Model(&entity.BackupTombstone{})
	if options.ClusterId != "" {
		rdb = rdb.Where("cluster_id = ?", options.ClusterId)
	}
	if len(options.ClustersIn) > 0 {
		rdb = rdb.Where("cluster_id IN ?", options.ClustersIn)
	}
	if err := rdb.Find(&tombstones).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list backup tombstones")
	}
	return tombstones, nil
}

func (b *backupTombstoneRepoImp) Exists(
	ctx context.Context,
	clusterID string,
	scheduleTime time.Time,
) (exists bool, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupTombstoneRepo.Exists")
	defer func() { tracing.EndSpan(span, err) }()

	var count int64
	rdb := b.dbProvider.ReadDB(ctx).Model(&entity.BackupTombstone{})
	if err := rdb.Where("cluster_id = ? AND schedule_time = ?", clusterID, scheduleTime.UTC()).Count(&count).Error; err != nil {
		return false, bizerr.WrapGormError(err, "fail to check backup tombstone")
	}
	return count > 0, nil
}

func (b *backupTombstoneRepoImp) Delete(
	ctx context.Context,
	options entity.BackupTombstoneDeleteOptions,
) (count int64, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupTombstoneRepo.Delete")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if options.ExpireTimeLt != nil {
		wdb = wdb.Where("expire_time IS NOT NULL AND expire_time < ?", *options.ExpireTimeLt)
	}
	if options.PersistedExpireTimeGte != nil {
		wdb = wdb.Where(
			`NOT EXISTS (
				SELECT 1
				FROM user_backups ub
				WHERE ub.cluster_id = backup_tombstones.cluster_id
					AND ub.schedule_time = backup_tombstones.schedule_time
					AND ub.expire_time >= ?
			)`,
			*options.PersistedExpireTimeGte,
		)
	}
	if options.Limit > 0 {
		wdb = wdb.Limit(options.Limit)
	}
	result := wdb.Delete(&entity.BackupTombstone{})
	if result.Error != nil {
		return 0, bizerr.WrapGormError(result.Error, "fail to clean backup tombstones")
	}
	return result.RowsAffected, nil
}

type restoreJobRepoImp struct {
	dbProvider db.Provider
}

func NewRestoreJobRepoImp(dbProvider db.Provider) repo.RestoreJobRepo {
	return &restoreJobRepoImp{dbProvider}
}

func (r *restoreJobRepoImp) Create(ctx context.Context, restore *entity.RestoreJob) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "restoreJobRepo.Create")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Create(restore).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to create restore job")
	}
	return nil
}

func (r *restoreJobRepoImp) List(
	ctx context.Context,
	options entity.ListRestoreJobOptions,
) (jobs []*entity.RestoreJob, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "restoreJobRepo.List")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildRestoreJobFilter(options, rdb)
	if err := rdb.Find(&jobs).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to list running restore job")
	}
	return jobs, nil
}

func buildRestoreJobFilter(options entity.ListRestoreJobOptions, db *gorm.DB) *gorm.DB {
	if options.State != "" {
		db = db.Where("state = ?", options.State)
	}
	if options.ClusterId != "" {
		db = db.Where("target_cluster_id = ?", options.ClusterId)
	}
	if options.Offset != nil {
		db = db.Offset(int(*options.Offset))
	}
	if options.PageSize != nil {
		db = db.Limit(int(*options.PageSize))
	}
	if options.OrderBy != "" {
		db = db.Order(options.OrderBy)
	}
	return db
}

func buildRestoreJobCountFilter(options entity.ListRestoreJobOptions, db *gorm.DB) *gorm.DB {
	if options.State != "" {
		db = db.Where("state = ?", options.State)
	}
	if options.ClusterId != "" {
		db = db.Where("target_cluster_id = ?", options.ClusterId)
	}
	return db
}

func (r *restoreJobRepoImp) Save(ctx context.Context, restore *entity.RestoreJob) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "restoreJobRepo.Save")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := r.dbProvider.WriteDB(ctx)
	if err := wdb.Save(restore).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to update restore job")
	}
	return nil
}

func (r *restoreJobRepoImp) Count(
	ctx context.Context,
	options entity.ListRestoreJobOptions,
) (count int64, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "restoreJobRepo.Count")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := r.dbProvider.ReadDB(ctx)
	rdb = buildRestoreJobCountFilter(options, rdb)
	if err := rdb.Model(&entity.RestoreJob{}).Count(&count).Error; err != nil {
		return 0, bizerr.WrapGormError(err, "fail to count restore jobs")
	}
	return count, nil
}

func (r *restoreJobRepoImp) GroupByState(ctx context.Context) (_ []*entity.RestoreJobStateCount, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "restoreJobRepo.GroupByState")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := r.dbProvider.ReadDB(ctx)
	var result []*entity.RestoreJobStateCount
	if err := rdb.Model(&entity.RestoreJob{}).Select("state, count(*) as count").Group("state").Scan(&result).Error; err != nil {
		return nil, bizerr.WrapGormError(err, "fail to count by state")
	}
	return result, nil
}

type backupScheduleCheckJobRepoImp struct {
	dbProvider db.Provider
}

func NewBackupScheduleCheckJobRepoImp(dbProvider db.Provider) repo.BackupScheduleCheckJobRepo {
	return &backupScheduleCheckJobRepoImp{dbProvider}
}

func (b *backupScheduleCheckJobRepoImp) Create(
	ctx context.Context,
	job *entity.BackupScheduleCheckJob,
) (err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleCheckJobRepo.Create")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Create(job).Error; err != nil {
		return bizerr.WrapGormError(err, "fail to create backup schedule check job")
	}
	return nil
}

func (b *backupScheduleCheckJobRepoImp) Save(
	ctx context.Context,
	job *entity.BackupScheduleCheckJob,
) (_ *entity.BackupScheduleCheckJob, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleCheckJobRepo.Save")
	defer func() { tracing.EndSpan(span, err) }()

	wdb := b.dbProvider.WriteDB(ctx)
	if err := wdb.Save(job).Error; err != nil {
		return job, bizerr.WrapGormError(err, "fail to update backup schedule check job")
	}
	return job, nil
}

func (b *backupScheduleCheckJobRepoImp) Get(
	ctx context.Context,
	date string,
) (_ *entity.BackupScheduleCheckJob, err bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "backupScheduleCheckJobRepo.Get")
	defer func() { tracing.EndSpan(span, err) }()

	rdb := b.dbProvider.ReadDB(ctx)
	job := &entity.BackupScheduleCheckJob{}
	if err := rdb.Where("check_date = ?", date).First(job).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, nil
		}
		return nil, bizerr.WrapGormError(err, "fail to get backup schedule check job")
	}
	return job, nil
}