package auto

import (
	"fmt"
	"time"

	"github.com/Masterminds/semver/v3"
	retry "github.com/avast/retry-go/v4"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	crrfModel "github.com/tidbcloud/dataflow-service/app/models/crrf"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"go.uber.org/zap"
)

var RetryFunc = action.InvokeRunAfterWithExponentialRetryFactory(15*time.Second, 5*time.Minute, 0)

// ── Enable base ───────────────────────────────────────────────────────────

// autoCRRFEnableBase holds the shared enable/poll logic embedded by both enable action types.
type autoCRRFEnableBase struct {
	bsRepo    backupRepo.BackupSettingRepo
	mgmtCli   grpcli.MgmtSvcGrpcClient
	crrfModel crrfModel.CRRFModel
}

const autoCRRFEnableTriggeredKey = "auto-crrf-enable-triggered"

func markAutoCRRFEnableTriggered(eCtx run.ExecuteContext) {
	eCtx.ShareData().Set(autoCRRFEnableTriggeredKey, "true")
}

func isAutoCRRFEnableTriggered(eCtx run.ExecuteContext) bool {
	val, _ := eCtx.ShareData().Get(autoCRRFEnableTriggeredKey)
	return val == "true"
}

// run checks all pre-conditions and calls CRRFModel.EnableCRRF when appropriate.
// Returns nil for any condition that should skip silently, or an error on failure.
//
// Steps:
//  1. Load BackupSetting
//  2. Guard: AWS/EKS only
//  3. Feature flag: autoCRRF must be enabled in crrfOverrides
//  4. PiTR + dual-region both enabled
//  5. Cluster version >= CRRFEnableMinVerConstraint
//  6. Idempotent: skip if already enabling/enabled
//  7. Extract secondary region and call EnableCRRF
func (b *autoCRRFEnableBase) run(eCtx run.ExecuteContext, clusterID uint64) error {
	ctx := eCtx.Context()
	logger := action.GetLogger(eCtx)
	// 1. Load BackupSetting.
	bs, err := b.bsRepo.GetByFilter(ctx, &backupRepo.BackupSettingFilter{ClusterID: &clusterID})
	if err != nil {
		return err
	}
	if bs == nil {
		logger.Error("backup setting not found, skipping auto-CRRF enable", zap.Uint64("clusterID", clusterID))
		return fmt.Errorf("backup setting not found for clusterID=%d", clusterID)
	}

	// 2. Guard: CRRF is only supported on AWS/EKS clusters.
	if bs.InfraProviderType != entity.InfraProviderTypeAWS && bs.InfraProviderType != entity.InfraProviderTypeEKS {
		logger.Info("non-AWS provider, skipping auto-CRRF enable",
			zap.String("provider", bs.InfraProviderType),
			zap.Uint64("clusterID", clusterID))
		return nil
	}

	cluster, clErr := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if clErr != nil {
		logger.Error("failed to get cluster for version check", zap.Error(clErr), zap.Uint64("clusterID", clusterID))
		return clErr
	}

	// 3. Feature flag: auto-CRRF must be explicitly enabled in crrfOverrides config.
	cfg := internal_config.Global().BackupConfig
	override, _ := cfg.CRRFOverrides.GetMatched(bs.TenantID.Uint64(), bs.ProjectID.Uint64(), clusterID, cluster.GetCloudProvider(), cluster.GetRegion())
	if !override.AutoCRRF {
		logger.Info("auto-CRRF not enabled by feature flag, skipping", zap.Uint64("clusterID", clusterID))
		return nil
	}

	// 4. Both PiTR and dual-region must be enabled.
	if bs.PiTR == nil || !bs.PiTR.Enable {
		logger.Info("PiTR not enabled, skipping auto-CRRF enable", zap.Uint64("clusterID", clusterID))
		return nil
	}
	if bs.Bucket == nil || !bs.Bucket.DualRegionEnable {
		logger.Info("dual-region not enabled, skipping auto-CRRF enable", zap.Uint64("clusterID", clusterID))
		return nil
	}

	// 5. Check cluster version.
	dc := cluster.GetDedicatedCluster()
	v, parseErr := semver.NewVersion(utils.NormalizeVersion(dc.GetVersion()))
	if parseErr != nil {
		logger.Error("failed to parse cluster version", zap.String("version", dc.GetVersion()), zap.Error(parseErr))
		return fmt.Errorf("parse cluster version: %w", parseErr)
	}
	if !cfg.CRRFEnableVersionConstraint.Check(v) {
		logger.Info("cluster version below CRRF minimum, skipping auto-CRRF enable",
			zap.String("version", dc.GetVersion()),
			zap.String("constraint", cfg.CRRFEnableVersionConstraintStr))
		return nil
	}

	// 6. Idempotent: skip if already enabling/enabled.
	info, qErr := b.crrfModel.QueryCRRF(ctx, clusterID)
	if qErr != nil {
		return qErr
	}
	if info.Status == entity.CRRFSettingStatusEnabled || info.Status == entity.CRRFSettingStatusEnabling {
		logger.Info("CRRF already enabling/enabled, skipping auto-CRRF enable",
			zap.String("crrfStatus", string(info.Status)),
			zap.Uint64("clusterID", clusterID))
		return nil
	}

	// 7. Extract secondary region and call EnableCRRF.
	secondaryRegions := bs.Bucket.SecondaryRegions()
	if len(secondaryRegions) == 0 {
		logger.Warn("secondary region not found in backup setting, skipping auto-CRRF enable",
			zap.Uint64("clusterID", clusterID))
		return nil
	}
	// There should only be one secondary region, but take the first just in case.
	secondaryRegion := secondaryRegions[0]
	logger.Info("auto-enabling CRRF", zap.Uint64("clusterID", clusterID), zap.String("secondaryRegion", secondaryRegion))
	req := &domain.EnableCRRFReq{
		ClusterID:                   clusterID,
		SecondaryRegion:             secondaryRegion,
		ProjectID:                   uint64(bs.ProjectID),
		TenantID:                    uint64(bs.TenantID),
		SkipEnablePiTRAndDualRegion: true,
	}
	if bizErr := b.crrfModel.EnableCRRF(ctx, req); bizErr != nil {
		logger.Error("auto EnableCRRF failed", zap.Error(bizErr))
		return bizErr
	}
	markAutoCRRFEnableTriggered(eCtx)
	logger.Info("auto-CRRF enable triggered, will poll for completion", zap.Uint64("clusterID", clusterID))
	return nil
}

// poll queries CRRF status until it reaches "enabled". Returns nil on success (stops the retry loop).
// Returns retry.Unrecoverable on terminal DAG failure, or a transient error to be retried.
func (b *autoCRRFEnableBase) poll(eCtx run.ExecuteContext, clusterID uint64) error {
	ctx := eCtx.Context()
	logger := action.GetLogger(eCtx)
	if !isAutoCRRFEnableTriggered(eCtx) {
		logger.Info("auto-CRRF enable was not triggered in run, skipping polling", zap.Uint64("clusterID", clusterID))
		return nil
	}

	info, err := b.crrfModel.QueryCRRF(ctx, clusterID)
	if err != nil {
		return err
	}
	switch info.Status {
	case entity.CRRFSettingStatusEnabled:
		logger.Info("CRRF enabled successfully", zap.Uint64("clusterID", clusterID))
		return nil
	case entity.CRRFSettingStatusDisabled:
		if info.ErrorMessage != "" {
			return retry.Unrecoverable(fmt.Errorf("CRRF enable DAG failed: %s", info.ErrorMessage))
		}
		// Disabled without an error: either we skipped or CRRF was reset cleanly.
		logger.Info("CRRF is disabled (no error), treating as no-op success", zap.Uint64("clusterID", clusterID))
		return nil
	default:
		return fmt.Errorf("CRRF not yet enabled, status=%s, clusterID=%d", info.Status, clusterID)
	}
}

// ── Disable base ──────────────────────────────────────────────────────────

// autoCRRFDisableBase holds the shared disable/poll logic embedded by both disable action types.
type autoCRRFDisableBase struct {
	crrfModel crrfModel.CRRFModel
}

// run checks CRRF status and calls DisableCRRF if needed.
func (b *autoCRRFDisableBase) run(eCtx run.ExecuteContext, clusterID uint64) error {
	ctx := eCtx.Context()
	logger := action.GetLogger(eCtx)
	info, err := b.crrfModel.QueryCRRF(ctx, clusterID)
	if err != nil {
		return err
	}
	switch info.Status {
	case entity.CRRFSettingStatusDisabled:
		logger.Info("CRRF already disabled, skipping auto-CRRF disable", zap.Uint64("clusterID", clusterID))
		return nil
	case entity.CRRFSettingStatusDisabling:
		logger.Info("CRRF already disabling, waiting for completion", zap.Uint64("clusterID", clusterID))
		return nil
	}

	logger.Info("auto-disabling CRRF",
		zap.Uint64("clusterID", clusterID),
		zap.String("crrfStatus", string(info.Status)))
	if bizErr := b.crrfModel.DisableCRRF(ctx, &domain.DisableCRRFReq{ClusterID: clusterID}); bizErr != nil {
		logger.Error("auto DisableCRRF failed", zap.Error(bizErr))
		return bizErr
	}
	logger.Info("auto-CRRF disable triggered, will poll for completion", zap.Uint64("clusterID", clusterID))
	return nil
}

// poll queries CRRF status until it reaches "disabled". Returns nil on success (stops the retry loop).
// Returns retry.Unrecoverable if the disable DAG reached a terminal failure state.
func (b *autoCRRFDisableBase) poll(eCtx run.ExecuteContext, clusterID uint64) error {
	ctx := eCtx.Context()
	logger := action.GetLogger(eCtx)
	info, err := b.crrfModel.QueryCRRF(ctx, clusterID)
	if err != nil {
		return err
	}
	switch info.Status {
	case entity.CRRFSettingStatusDisabled:
		logger.Info("CRRF disabled successfully", zap.Uint64("clusterID", clusterID))
		return nil
	case entity.CRRFSettingStatusFailed:
		return retry.Unrecoverable(fmt.Errorf("CRRF disable DAG failed: %s, clusterID=%d", info.ErrorMessage, clusterID))
	default:
		return fmt.Errorf("CRRF not yet disabled, status=%s, clusterID=%d", info.Status, clusterID)
	}
}