package br

import (
	"context"
	"fmt"

	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver"
	brrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br"
	clusterrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster"
	encryptionkeyrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/encryptionkey"
	regionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool"
	brservice "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/br"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/pkg/config"
)

type Worker interface {
	SyncBackups(ctx context.Context) error
	ExpireBackups(ctx context.Context) error
	SyncMissingBackups(ctx context.Context) error
	UpdateBackupGcSetting(ctx context.Context) error
	UpdateBackupSyncSlots(ctx context.Context) error
	ReportAutoBackupMetering(ctx context.Context) error
	BackfillEssentialV2BackupSettings(ctx context.Context) error
}

type worker struct {
	cfg *config.BackupConfig

	backupSettingRepo brrepo.BackupSettingRepo
	backupRepo        brrepo.BackupRepo
	backupPolicyRepo  brrepo.BackupPolicyRepo
	clusterRepo       clusterrepo.Repo
	regionRepo        regionrepo.Repo
	resourcePoolRepo  resourcepool.Repo
	regionalserverCli regionalserver.Client
	backupSettingSvc  backupSettingService
	pauseChecker      func(context.Context, uint64) (bool, error)
	pauseFencer       pauseFencer
	transaction       db.Transactioner

	// internal state
	backupSyncSlots *backupSyncSlots
}

type pauseFencer func(context.Context, uint64, func(context.Context) error) (bool, error)

func (w *worker) isSharedPoolInPauseWindow(ctx context.Context, sharedPoolID uint64) (bool, error) {
	if w.pauseChecker == nil {
		return false, nil
	}
	return w.pauseChecker(ctx, sharedPoolID)
}

// checkSharedPoolPauseWindow reads Global's ResourcePool lifecycle, which is
// the control-plane source of truth for Global workers. It prevents workers
// from starting new scheduled backup work; reconciliation of a real Regional
// backup and automatic-backup retention cleanup intentionally continue because
// neither can change the operation-owned pre-Pause manual-backup set. A deleted
// pool must no longer retain Pause protection because its backup GC cleanup
// still needs to converge.
func checkSharedPoolPauseWindow(
	ctx context.Context,
	sharedPoolID uint64,
	resourcePoolRepo resourcepool.Repo,
) (bool, error) {
	sharedPool, err := resourcePoolRepo.Load(ctx, sharedPoolID, true)
	if err != nil {
		return false, fmt.Errorf("load resource pool %d: %w", sharedPoolID, err)
	}
	if sharedPool.DeleteTime != nil {
		return false, nil
	}
	switch sharedPool.GetLifecycleState() {
	case globalngv1.ResourcePool_PAUSE_PRECHECKING,
		globalngv1.ResourcePool_PAUSING,
		globalngv1.ResourcePool_PAUSED,
		globalngv1.ResourcePool_RESUMING:
		return true, nil
	default:
		return false, nil
	}
}

func runWithSharedPoolPauseFence(
	ctx context.Context,
	sharedPoolID uint64,
	transaction db.Transactioner,
	resourcePoolRepo resourcepool.Repo,
	mutation func(context.Context) error,
) (skipped bool, err error) {
	err = transaction.Transaction(ctx, func(txCtx context.Context) error {
		lifecycle, berr := resourcePoolRepo.LoadLifecycleForUpdate(txCtx, sharedPoolID, true)
		if berr != nil {
			return fmt.Errorf("lock resource pool %d: %w", sharedPoolID, berr)
		}
		if !lifecycle.IsDeleted {
			switch lifecycle.State {
			case globalngv1.ResourcePool_PAUSE_PRECHECKING,
				globalngv1.ResourcePool_PAUSING,
				globalngv1.ResourcePool_PAUSED,
				globalngv1.ResourcePool_RESUMING:
				skipped = true
				return nil
			}
		}
		return mutation(txCtx)
	})
	return skipped, err
}

func NewWorker(
	cfg *config.Config,
	backupSettingRepo brrepo.BackupSettingRepo,
	backupRepo brrepo.BackupRepo,
	backupPolicyRepo brrepo.BackupPolicyRepo,
	clusterRepo clusterrepo.Repo,
	encryptionKeyRepo encryptionkeyrepo.Repo,
	regionRepo regionrepo.Repo,
	resourcePoolRepo resourcepool.Repo,
	transaction db.Transactioner,
	regionalserverCli regionalserver.Client,
	backupSettingSvc brservice.Service,
) Worker {
	backupEncryptionResolver := brservice.NewResolver(encryptionKeyRepo)
	pauseChecker := func(ctx context.Context, sharedPoolID uint64) (bool, error) {
		return checkSharedPoolPauseWindow(ctx, sharedPoolID, resourcePoolRepo)
	}
	pauseFencer := func(ctx context.Context, sharedPoolID uint64, mutation func(context.Context) error) (bool, error) {
		return runWithSharedPoolPauseFence(ctx, sharedPoolID, transaction, resourcePoolRepo, mutation)
	}
	return &worker{
		cfg:               &cfg.BizConfig.BackupConfig,
		backupSettingRepo: backupSettingRepo,
		backupRepo:        backupRepo,
		backupPolicyRepo:  backupPolicyRepo,
		clusterRepo:       clusterRepo,
		regionRepo:        regionRepo,
		resourcePoolRepo:  resourcePoolRepo,
		regionalserverCli: regionalserverCli,
		backupSettingSvc:  backupSettingSvc,
		pauseChecker:      pauseChecker,
		pauseFencer:       pauseFencer,
		transaction:       transaction,
		backupSyncSlots: newBackupSyncSlots(
			&cfg.BizConfig.BackupConfig.SyncConfig,
			backupSettingRepo,
			backupRepo,
			clusterRepo,
			backupEncryptionResolver,
			regionRepo,
			transaction,
			regionalserverCli,
			pauseWorkerGuard{checker: pauseChecker, fencer: pauseFencer},
		),
	}
}

func (w *worker) UpdateBackupSyncSlots(ctx context.Context) error {
	if w == nil || w.backupSyncSlots == nil {
		tilog.Info(ctx, "skip update backup sync slots, worker is not ready")
		return nil
	}
	if w.cfg != nil && !w.cfg.SyncConfig.Enable {
		tilog.Info(ctx, "skip update backup sync slots, sync backups is disabled")
		return nil
	}
	return w.backupSyncSlots.update(ctx)
}