package br

import (
	"context"
	"fmt"
	"time"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/br"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
)

const (
	ApplyBucketLifecycleRulesActionName         = "apply-bucket-lifecycle-rules"
	ApplyTransitionSafeLifecycleRulesActionName = "apply-transition-safe-lifecycle-rules"
	ApplyPausedBRLifecycleRulesActionName       = "apply-paused-br-lifecycle-rules"
	RestoreBRLifecycleRulesActionName           = "restore-br-lifecycle-rules"
	ApplyDeletingBRLifecycleRulesActionName     = "apply-deleting-br-lifecycle-rules"
)

type applyBucketLifecycleRulesAction struct {
	name                  string
	ignorePauseProtection bool
	guarded               bool
	gcRepo                repo.BackupGcSettingRepo
	sharedPoolRepo        repo.SharedPoolRepo
	osRepo                repo.SharedPoolObjectStoreRepo
	bucketSvc             dplanemgrv1.BucketServiceClient
	dataplaneManager      dataplane.Manager
	transitionSafe        bool
}

// NewApplyTransitionSafeLifecycleRulesAction raises physical TTLs for both
// the currently live and desired archive delays before TiBRGC is updated.
func NewApplyTransitionSafeLifecycleRulesAction(
	gcRepo repo.BackupGcSettingRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
	dataplaneManager dataplane.Manager,
) actions.Action {
	return actions.CreateActionFromAsync(&applyBucketLifecycleRulesAction{
		name: ApplyTransitionSafeLifecycleRulesActionName, gcRepo: gcRepo, sharedPoolRepo: sharedPoolRepo,
		osRepo: osRepo, bucketSvc: bucketSvc, dataplaneManager: dataplaneManager, transitionSafe: true,
	}, time.Minute, time.Minute*10)
}

func NewApplyBucketLifecycleRulesAction(
	gcRepo repo.BackupGcSettingRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
) actions.Action {
	return newApplyBucketLifecycleRulesAction(ApplyBucketLifecycleRulesActionName, false, gcRepo, sharedPoolRepo, osRepo, bucketSvc)
}

// NewApplyPausedBRLifecycleRulesAction has a dedicated name so the Pause DAG
// can track terminal lifecycle-application failures on its operation without
// changing the behaviour of ordinary GC-setting workflows.
func NewApplyPausedBRLifecycleRulesAction(
	gcRepo repo.BackupGcSettingRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
	guard actions.ActionGuard,
) actions.Action {
	return newApplyBucketLifecycleRulesAction(ApplyPausedBRLifecycleRulesActionName, false, gcRepo, sharedPoolRepo, osRepo, bucketSvc, guard)
}

// NewRestoreBRLifecycleRulesAction is intentionally only wired after the
// Resume archive gate. It restores the original lifecycle configuration while
// the persisted pool state is still RESUMING, so regular concurrent applies
// remain protected until this workflow reaches CompleteResume.
func NewRestoreBRLifecycleRulesAction(
	gcRepo repo.BackupGcSettingRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
	guard actions.ActionGuard,
) actions.Action {
	return newApplyBucketLifecycleRulesAction(RestoreBRLifecycleRulesActionName, true, gcRepo, sharedPoolRepo, osRepo, bucketSvc, guard)
}

// NewApplyDeletingBRLifecycleRulesAction deliberately bypasses Pause
// protection. A stable PAUSED pool may be deleted without clearing its pause
// state, so the delete workflow must restore the configured expiration rules
// before TiBRGC is removed and can no longer reconcile them.
func NewApplyDeletingBRLifecycleRulesAction(
	gcRepo repo.BackupGcSettingRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
) actions.Action {
	return newApplyBucketLifecycleRulesAction(ApplyDeletingBRLifecycleRulesActionName, true, gcRepo, sharedPoolRepo, osRepo, bucketSvc)
}

func newApplyBucketLifecycleRulesAction(
	name string,
	ignorePauseProtection bool,
	gcRepo repo.BackupGcSettingRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	osRepo repo.SharedPoolObjectStoreRepo,
	bucketSvc dplanemgrv1.BucketServiceClient,
	guards ...actions.ActionGuard,
) actions.Action {
	asyncAction := &applyBucketLifecycleRulesAction{
		name:                  name,
		ignorePauseProtection: ignorePauseProtection,
		guarded:               len(guards) > 0,
		gcRepo:                gcRepo,
		sharedPoolRepo:        sharedPoolRepo,
		osRepo:                osRepo,
		bucketSvc:             bucketSvc,
	}
	if len(guards) > 0 {
		return actions.CreateGuardedActionFromAsync(asyncAction, time.Minute, time.Minute*10, guards[0])
	}
	return actions.CreateActionFromAsync(asyncAction, time.Minute, time.Minute*10)
}

func (a *applyBucketLifecycleRulesAction) Name() string {
	return a.name
}

func (a *applyBucketLifecycleRulesAction) Start(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	if a.guarded {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, actions.GuardedWriteTimeout)
		defer cancel()
	}
	sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		tilog.Error(ctx, "failed to get shared pool id", zap.Error(err))
		return fmt.Errorf("failed to get shared pool id: %w", err)
	}

	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, true)
	if err != nil {
		tilog.Error(ctx, "failed to load shared pool", zap.Error(err))
		return fmt.Errorf("failed to load shared pool: %w", err)
	}

	sharedPoolObjectStore, err := a.osRepo.Load(ctx, sharedPoolId, schema.SharedPoolObjectStoreTypeData, true)
	if err != nil {
		tilog.Error(ctx, "failed to load sharedPoolObjectStore", zap.Error(err))
		return fmt.Errorf("failed to load sharedPoolObjectStore: %w", err)
	}

	bucketId := sharedPoolObjectStore.StoreBucket
	if bucketId == "" {
		bucketId = entity.GetDataBucketId(sharedPoolId, sharedPoolObjectStore.StoreRegion)
		tilog.Warn(ctx, "bucket id is empty in sharedPoolObjectStore, use generated bucket id", zap.String("sharedPoolId", sharedPoolId), zap.String("bucketId", bucketId))
	}

	backupGcSetting, err := a.gcRepo.Load(ctx, sharedPoolId)
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Info(ctx, "backup gc setting not found, it should be early created sharedpool without gc setting", zap.String("shared_pool_id", sharedPoolId))
			return nil
		}
		tilog.Error(ctx, "failed to load backup gc setting", zap.Error(err))
		return fmt.Errorf("failed to load backup gc setting: %w", err)
	}
	if a.transitionSafe && sharedPool.DeletedAt == nil {
		remoteClient, err := a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
		if err != nil {
			return fmt.Errorf("get dataplane client for transition-safe lifecycle rules: %w", err)
		}
		tibrgc := &v1alpha1br.TiBRGC{}
		key := types.NamespacedName{Name: tioperator.DefaultTiBRGCRName, Namespace: sharedPool.GetDataPlaneNamespace()}
		getErr := remoteClient.Get(ctx, key, tibrgc)
		transitionDelay := backupGcSetting.MinT3DataAgeDays
		switch {
		case getErr == nil:
			liveDelay := uint32(0)
			for _, strategy := range tibrgc.Spec.GCStrategy.TieredStrategies {
				if strategy.Name == v1alpha1br.TieredStorageStrategyNameToT3Storage {
					liveDelay = strategy.TimeThresholdDays
					break
				}
			}
			if liveDelay == 0 {
				return fmt.Errorf("live TiBRGC has no valid T3 archive delay")
			}
			transitionDelay = max(liveDelay, transitionDelay)
		case apierrors.IsNotFound(getErr):
			tilog.Info(ctx, "live TiBRGC not found, use desired T3 archive delay for transition-safe lifecycle rules",
				zap.String("shared_pool_id", sharedPoolId), zap.Uint32("desired_delay_days", transitionDelay))
		default:
			return fmt.Errorf("get live TiBRGC for transition-safe lifecycle rules: %w", getErr)
		}
		if transitionDelay == 0 {
			return fmt.Errorf("desired backup GC setting has no valid T3 archive delay")
		}
		backupGcSetting = br.ArchiveSafeBackupGcSettingForDelay(backupGcSetting, transitionDelay)
	} else if a.transitionSafe {
		tilog.Info(ctx, "skip live TiBRGC probe for deleted shared pool",
			zap.String("shared_pool_id", sharedPoolId))
	}

	_, err = a.bucketSvc.GetBucket(ctx, &dplanemgrv1.GetBucketRequest{
		DataplaneId: sharedPool.DataplaneName,
		BucketId:    bucketId,
	})
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Info(ctx, "bucket not found, skip applying lifecycle rules", zap.String("bucketId", bucketId))
			return nil
		}
		tilog.Error(ctx, "failed to get bucket before applying lifecycle rules", zap.String("bucketId", bucketId), zap.Error(err))
		return fmt.Errorf("failed to get bucket before applying lifecycle rules: %w", err)
	}

	rules := br.MakeLifecycleRules(backupGcSetting, sharedPool)
	if a.ignorePauseProtection {
		rules = br.MakeLifecycleRulesWithoutPauseProtection(backupGcSetting, sharedPool)
	}
	req := &dplanemgrv1.ApplyLifecycleRulesRequest{
		DataplaneId: sharedPool.DataplaneName,
		BucketId:    bucketId,
		Rules:       rules,
	}
	resp, err := a.bucketSvc.ApplyLifecycleRules(ctx, req)
	if err != nil {
		tilog.Error(ctx, "failed to apply lifecycle rules", zap.Any("request", req), zap.Error(err))
		return fmt.Errorf("failed to apply lifecycle rules: %w", err)
	}
	tilog.Info(ctx, "successfully apply lifecycle rules", zap.Any("response", resp))

	return nil
}

func (a *applyBucketLifecycleRulesAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	ctx := utils.GenerateContext(args)
	sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		tilog.Error(ctx, "failed to get shared pool id", zap.Error(err))
		return false, fmt.Errorf("failed to get shared pool id: %w", err)
	}

	_, err = a.gcRepo.Load(ctx, sharedPoolId)
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Info(ctx, "backup gc setting not found, it should be early created sharedpool without gc setting", zap.String("shared_pool_id", sharedPoolId))
			return true, nil
		}
		tilog.Error(ctx, "failed to load backup gc setting", zap.Error(err))
		return false, fmt.Errorf("failed to load backup gc setting: %w", err)
	}

	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, true)
	if err != nil {
		tilog.Error(ctx, "failed to load shared pool", zap.Error(err))
		return false, fmt.Errorf("failed to load shared pool: %w", err)
	}

	sharedPoolObjectStore, err := a.osRepo.Load(ctx, sharedPoolId, schema.SharedPoolObjectStoreTypeData, true)
	if err != nil {
		tilog.Error(ctx, "failed to load sharedPoolObjectStore", zap.Error(err))
		return false, fmt.Errorf("failed to load sharedPoolObjectStore: %w", err)
	}

	bucketId := sharedPoolObjectStore.StoreBucket
	if bucketId == "" {
		bucketId = entity.GetDataBucketId(sharedPoolId, sharedPoolObjectStore.StoreRegion)
		tilog.Warn(ctx, "bucket id is empty in sharedPoolObjectStore, use generated bucket id", zap.String("sharedPoolId", sharedPoolId), zap.String("bucketId", bucketId))
	}

	_, err = a.bucketSvc.GetBucket(ctx, &dplanemgrv1.GetBucketRequest{
		DataplaneId: sharedPool.DataplaneName,
		BucketId:    bucketId,
	})
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Info(ctx, "bucket not found, skip checking lifecycle rules", zap.String("bucketId", bucketId))
			return true, nil
		}
		tilog.Error(ctx, "failed to get bucket before checking lifecycle rules", zap.String("bucketId", bucketId), zap.Error(err))
		return false, fmt.Errorf("failed to get bucket before checking lifecycle rules: %w", err)
	}

	req := &dplanemgrv1.GetLifecycleRulesRequest{
		DataplaneId: sharedPool.DataplaneName,
		BucketId:    bucketId,
	}
	resp, err := a.bucketSvc.GetLifecycleRules(ctx, req)
	if err != nil {
		tilog.Error(ctx, "failed to get lifecycle rules", zap.Any("request", req), zap.Error(err))
		return false, fmt.Errorf("failed to get lifecycle rules: %w", err)
	}
	if resp.Status != dplanemgrv1.GetLifecycleRulesResponse_Synced {
		tilog.Info(ctx, "lifecycle rules not synced", zap.Any("response", resp))
		return false, nil
	}
	tilog.Info(ctx, "lifecycle rules is synced", zap.Any("response", resp))
	return true, nil
}