# Auto-CRRF Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Automatically enable/disable CRRF when a Dedicated cluster has both PiTR and dual-region backup simultaneously active (cluster version ≥ v8.5.6), by inserting auto-CRRF steps into the four PiTR/dual-region DAGs.

**Architecture:** New `app/workflow/action/crrf/auto/` package with two actions (`enable.go`, `disable.go`) using the **Run/RunAfter pattern**. `Run()` starts the CRRF DAG (via `CRRFModel.EnableCRRF`/`DisableCRRF`). `RunAfter()` polls `QueryCRRF` until the triggered DAG reaches terminal state, using `invokeRunAfterWithSlowExponentialRetry` (copied from `enable_pitr.go`). The actions are inserted as the second-to-last step in each of the four PiTR/dual-region enable/disable DAGs, so the backup setting remains locked (status ≠ `normal`) while CRRF is being toggled. The `EnableCRRF` status check is relaxed to also accept `editing_pitr` and `editing_region` so it does not reject calls made mid-DAG.

**Tech Stack:** Go, fastflow DAG, bizerr, semver, testify/suite mocks

---

## File Map

| File | Change |
|------|--------|
| `internal/config/config.go` | **Modify**: add `CRRFEnableMinVer` + `CRRFEnableMinVerConstraint` to `BackupConfig`; parse in `Init()` |
| `app/models/crrf/impl/crrf.go` | **Modify**: relax `backupSetting.Status` check in `EnableCRRF`; add PiTR+dual-region preconditions |
| `app/workflow/constants.go` | **Modify**: add 4 new action name constants |
| `app/workflow/action/crrf/auto/enable.go` | **Create**: `autoCRRFEnableAction` — calls `EnableCRRF` if conditions met |
| `app/workflow/action/crrf/auto/disable.go` | **Create**: `autoCRRFDisableAction` — calls `DisableCRRF` if conditions met |
| `app/workflow/action/crrf/auto/enable_test.go` | **Create**: unit tests for auto-enable action |
| `app/workflow/action/crrf/auto/disable_test.go` | **Create**: unit tests for auto-disable action |
| `app/models/crrf/impl/crrf_test.go` | **Modify**: add tests for relaxed status check + PiTR/dual-region preconditions |
| `app/workflow/dag/crrf/enable.go` | **Modify**: remove `enable_dual_region` + `enable_pitr` tasks; remove `KeySecondaryRegion` var |
| `app/workflow/dag/backup/pit/enbale.go` | **Modify**: insert `AutoCRRFEnableFromPit` task; update `ActionUpdatePitBackupSettingStatusToNormal` DependOn |
| `app/workflow/dag/backup/pit/disable.go` | **Modify**: insert `AutoCRRFDisableFromPit` task; update `ActionUpdatePitBackupSettingStatusToNormal` DependOn |
| `app/workflow/dag/backup_setting/aws/enable_dual_region.go` | **Modify**: insert `AutoCRRFEnableFromDualRegion` task; update `ActionEnableDualRegionBucketFinalStep` DependOn |
| `app/workflow/dag/backup_setting/aws/disable_dual_region.go` | **Modify**: insert `AutoCRRFDisableFromDualRegion` task; update `ActionDisableDualRegionBucketFinalStep` DependOn |
| `app/models/backup/impl/backup_setting.go` | **Modify**: remove CRRF-active guard in `preCheckModifyPitBackupSetting()` |

---

## Task 1: Config — add `CRRFEnableMinVer` constraint

**Files:**
- Modify: `internal/config/config.go:736-885`

- [ ] **Step 1.1: Add `CRRFEnableMinVer` field and parsed constraint to `BackupConfig`**

  In `BackupConfig` struct (after `LogBackupEnableMinVerConstraint` at ~line 746), add:

  ```go
  CRRFEnableMinVer             string              `mapstructure:"crrfEnableMinVer" default:"v8.5.6"`
  CRRFEnableMinVerConstraint   *semver.Constraints `mapstructure:"-"`
  ```

- [ ] **Step 1.2: Parse the constraint in `BackupConfig.Init()` (~line 862)**

  After the `LogBackupEnableMinVerConstraint` block, add:

  ```go
  constraint, err = semver.NewConstraint(fmt.Sprintf(">=%s", bc.CRRFEnableMinVer))
  if err != nil {
      return errors.Annotatef(err, "new constraint: %s", bc.CRRFEnableMinVer)
  }
  bc.CRRFEnableMinVerConstraint = constraint
  ```

- [ ] **Step 1.3: Compile check**

  ```bash
  cd /home/nil/pingcap/workbench/repos/dataflow-service.worktrees/auto-crrf
  go build ./internal/config/...
  ```

  Expected: no errors.

- [ ] **Step 1.4: Commit**

  ```bash
  git add internal/config/config.go
  git commit -m "feat(config): add CRRFEnableMinVer constraint to BackupConfig"
  ```

---

## Task 2: Model — relax `EnableCRRF` status check + add PiTR/dual-region preconditions

**Files:**
- Modify: `app/models/crrf/impl/crrf.go:65-68` (status check relaxation)
- Modify: `app/models/crrf/impl/crrf.go:87-93` (add PiTR + dual-region precondition validation)
- Modify: `app/models/crrf/impl/crrf_test.go` (add tests)

### 2a: Relax status check

- [ ] **Step 2.1: Write failing test for relaxed status check**

  In `app/models/crrf/impl/crrf_test.go`, add to the suite:

  ```go
  func (s *crrfModelTestSuite) Test_EnableCRRF_BackupSettingEditingPitr_Allowed() {
      clusterID := uint64(101)
      bs := awsBackupSetting()
      bs.Status = entity.BackupSettingStatusEditingPitr
      bs.PiTR = &entity.PiTRSetting{Enable: true}
      bs.Bucket = &entity.BucketSetting{DualRegionEnable: true, Buckets: []entity.Bucket{{Master: false, Regions: []string{"us-west-2"}}}}
      s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once()
      s.crrfSettingRepo.On("GetByClusterID", mock.Anything, clusterID).Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")).Once()
      s.crrfSettingRepo.On("Create", mock.Anything, mock.Anything).Return(nil).Once()
      s.wfClient.On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).Return(fakeDagIns(), nil).Once()

      req := &domain.EnableCRRFReq{ClusterID: clusterID, SecondaryRegion: "us-west-2"}
      err := s.model.EnableCRRF(context.Background(), req)
      s.Nil(err)
  }

  func (s *crrfModelTestSuite) Test_EnableCRRF_BackupSettingEditingRegion_Allowed() {
      clusterID := uint64(102)
      bs := awsBackupSetting()
      bs.Status = entity.BackupSettingStatusEditingRegion
      bs.PiTR = &entity.PiTRSetting{Enable: true}
      bs.Bucket = &entity.BucketSetting{DualRegionEnable: true, Buckets: []entity.Bucket{{Master: false, Regions: []string{"us-west-2"}}}}
      s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once()
      s.crrfSettingRepo.On("GetByClusterID", mock.Anything, clusterID).Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")).Once()
      s.crrfSettingRepo.On("Create", mock.Anything, mock.Anything).Return(nil).Once()
      s.wfClient.On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).Return(fakeDagIns(), nil).Once()

      req := &domain.EnableCRRFReq{ClusterID: clusterID, SecondaryRegion: "us-west-2"}
      err := s.model.EnableCRRF(context.Background(), req)
      s.Nil(err)
  }
  ```

- [ ] **Step 2.2: Run tests to confirm they fail**

  ```bash
  cd /home/nil/pingcap/workbench/repos/dataflow-service.worktrees/auto-crrf
  go test ./app/models/crrf/impl/... -run "Test_EnableCRRF_BackupSettingEditing" -v 2>&1 | tail -20
  ```

  Expected: FAIL — status check rejects `editing_pitr`/`editing_region`.

- [ ] **Step 2.3: Relax the status check in `EnableCRRF` (crrf.go:65-68)**

  Replace:
  ```go
  if backupSetting.Status != entity.BackupSettingStatusNormal {
      return bizerr_wrapper.NewInvalidInputError(
          "backup setting is in %s state, must be normal to enable CRRF", backupSetting.Status)
  }
  ```

  With:
  ```go
  if backupSetting.Status != entity.BackupSettingStatusNormal &&
      backupSetting.Status != entity.BackupSettingStatusEditingPitr &&
      backupSetting.Status != entity.BackupSettingStatusEditingRegion {
      return bizerr_wrapper.NewInvalidInputError(
          "backup setting is in %s state, must be normal, editing_pitr, or editing_region to enable CRRF",
          backupSetting.Status)
  }
  ```

### 2b: Add PiTR + dual-region preconditions

- [ ] **Step 2.4: Write failing test for PiTR-not-enabled precondition**

  ```go
  func (s *crrfModelTestSuite) Test_EnableCRRF_PiTRNotEnabled_Error() {
      clusterID := uint64(103)
      bs := awsBackupSetting()
      // PiTR is nil (not enabled)
      bs.Bucket = &entity.BucketSetting{DualRegionEnable: true, Buckets: []entity.Bucket{{Master: false, Regions: []string{"us-west-2"}}}}
      s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once()

      req := &domain.EnableCRRFReq{ClusterID: clusterID, SecondaryRegion: "us-west-2"}
      err := s.model.EnableCRRF(context.Background(), req)
      s.NotNil(err)
      s.Contains(err.Error(), "PiTR")
  }

  func (s *crrfModelTestSuite) Test_EnableCRRF_DualRegionNotEnabled_Error() {
      clusterID := uint64(104)
      bs := awsBackupSetting()
      bs.PiTR = &entity.PiTRSetting{Enable: true}
      // DualRegionEnable is false (default)
      s.backupSettingRepo.On("GetByFilter", mock.Anything, mock.Anything).Return(bs, nil).Once()

      req := &domain.EnableCRRFReq{ClusterID: clusterID, SecondaryRegion: "us-west-2"}
      err := s.model.EnableCRRF(context.Background(), req)
      s.NotNil(err)
      s.Contains(err.Error(), "dual-region")
  }
  ```

- [ ] **Step 2.5: Run tests to confirm they fail**

  ```bash
  go test ./app/models/crrf/impl/... -run "Test_EnableCRRF_PiTRNot|Test_EnableCRRF_DualRegion" -v 2>&1 | tail -20
  ```

  Expected: FAIL — no precondition checks exist yet.

- [ ] **Step 2.6: Add PiTR+dual-region preconditions in `EnableCRRF` (after AWS check ~line 74)**

  After the AWS/EKS check block, add:

  ```go
  // Require PiTR to be enabled (or being enabled mid-DAG).
  if backupSetting.PiTR == nil || !backupSetting.PiTR.Enable {
      return bizerr_wrapper.NewInvalidInputError(
          "PiTR must be enabled to enable CRRF for cluster %d", req.ClusterID)
  }

  // Require dual-region backup to be enabled (or being enabled mid-DAG).
  if backupSetting.Bucket == nil || !backupSetting.Bucket.DualRegionEnable {
      return bizerr_wrapper.NewInvalidInputError(
          "dual-region backup must be enabled to enable CRRF for cluster %d", req.ClusterID)
  }
  ```

- [ ] **Step 2.7: Run all model tests**

  ```bash
  go test ./app/models/crrf/impl/... -v 2>&1 | tail -30
  ```

  Expected: ALL PASS.

- [ ] **Step 2.8: Commit**

  ```bash
  git add app/models/crrf/impl/crrf.go app/models/crrf/impl/crrf_test.go
  git commit -m "feat(crrf): relax EnableCRRF status check and add PiTR+dual-region preconditions"
  ```

---

## Task 3: Workflow constants — add 4 auto-CRRF action names

**Files:**
- Modify: `app/workflow/constants.go` (~line 169, after `ActionCRRFTogglePeriodicBackup`)

- [ ] **Step 3.1: Add 4 action constants**

  After `ActionCRRFTogglePeriodicBackup = "crrf_toggle_periodic_cluster_settings_backup"`, add:

  ```go
  // Auto-CRRF action names — inserted into PiTR and dual-region DAGs.
  ActionAutoCRRFEnableFromPit         = "auto_crrf_enable_from_pit"
  ActionAutoCRRFDisableFromPit        = "auto_crrf_disable_from_pit"
  ActionAutoCRRFEnableFromDualRegion  = "auto_crrf_enable_from_dual_region"
  ActionAutoCRRFDisableFromDualRegion = "auto_crrf_disable_from_dual_region"
  ```

- [ ] **Step 3.2: Compile check**

  ```bash
  go build ./app/workflow/...
  ```

  Expected: no errors.

- [ ] **Step 3.3: Commit**

  ```bash
  git add app/workflow/constants.go
  git commit -m "feat(workflow): add auto-CRRF action name constants"
  ```

---

## Task 4: New action `app/workflow/action/crrf/auto/enable.go`

**Files:**
- Create: `app/workflow/action/crrf/auto/enable.go`
- Create: `app/workflow/action/crrf/auto/enable_test.go`

The auto-enable action uses the **Run/RunAfter pattern**:
- `Run()`: checks conditions (PiTR on, dual-region on, cluster version ≥ min, CRRF not already enabling/enabled). On skip → `action.StopRetry`. On proceed → calls `EnableCRRF` → returns `nil`.
- `RunAfter()`: polls `QueryCRRF` until status is `"enabled"` (StopRetry). If status is `"disabled"` with a non-empty `ErrorMessage` → return error (CRRF DAG failed). Otherwise retry.
- Constructor uses `action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry)` (15s→5min exponential backoff, unlimited attempts, stops on `StopRetry`).
- `invokeRunAfterWithSlowExponentialRetry` is copied from `enable_pitr.go` into the `auto/` package.

**How to get ClusterID from a dual-region DAG**: param is `BackupSettingId` (loaded from `workflow.KeyBackupSettingId`). Load BackupSetting, read `ClusterID`.

**How to get ClusterID from a PiTR DAG**: param is `PitBackupSwitchID` (loaded from `workflow.KeyPitBackupSwitchID`). Load PitBackupSwitch, read `ClusterID`.

Both actions share the same underlying logic; the difference is the param struct.

- [ ] **Step 4.1: Write failing tests (compile will fail until Step 4.2)**

  Create `app/workflow/action/crrf/auto/enable_test.go` with the full content below. Tests will fail to compile because `runAutoCRRFEnable` and `pollCRRFEnabled` don't exist yet — that's expected.

  ```bash
  go test ./app/workflow/action/crrf/auto/... -v 2>&1 | head -10
  ```
  Expected: `undefined: runAutoCRRFEnable` compile error.

- [ ] **Step 4.2: Create `app/workflow/action/crrf/auto/enable.go`**

  ```go
  package auto

  import (
      "context"
      "fmt"
      "time"

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

  // invokeRunAfterWithSlowExponentialRetry uses exponential backoff starting at 15s up to 5min,
  // suitable for long-running waits like CRRF enable/disable DAGs.
  //
  //nolint:gomnd
  func invokeRunAfterWithSlowExponentialRetry(ctx context.Context, logger *zap.Logger, run func() error) error {
      return retry.Do(func() error {
          return run()
      },
          retry.Context(ctx),
          retry.DelayType(retry.BackOffDelay),
          retry.Delay(15*time.Second),
          retry.MaxDelay(5*time.Minute),
          retry.Attempts(0),
          retry.LastErrorOnly(true),
          retry.OnRetry(func(n uint, err error) {
              logger.Warn("action run error", zap.Uint("retry_times", n), zap.Error(err))
          }),
          retry.RetryIf(func(err error) bool {
              return err != action.StopRetry
          }),
      )
  }

  // ── Enable from PiTR DAG ──────────────────────────────────────────────────

  // AutoCRRFEnablePitParams are injected by the PiTR enable DAG via KeyPitBackupSwitchID.
  type AutoCRRFEnablePitParams struct {
      PitBackupSwitchID uint64 `json:"pit_backup_switch_id"`
  }

  type autoCRRFEnableFromPitAction struct {
      backupSettingRepo backupRepo.BackupSettingRepo
      pitSwitchRepo     backupRepo.PitBackupSwitchRepo
      coreCli           grpcli.CoreGrpcClient
      crrfModel         crrfModel.CRRFModel
  }

  // NewAutoCRRFEnableFromPitAction creates the auto-enable action for insertion into the PiTR enable DAG.
  func NewAutoCRRFEnableFromPitAction(
      backupSettingRepo backupRepo.BackupSettingRepo,
      pitSwitchRepo backupRepo.PitBackupSwitchRepo,
      coreCli grpcli.CoreGrpcClient,
      crrfModel crrfModel.CRRFModel,
  ) *action.Wrapper {
      return action.MustNewActionWrapper(
          workflow.ActionAutoCRRFEnableFromPit,
          &autoCRRFEnableFromPitAction{
              backupSettingRepo: backupSettingRepo,
              pitSwitchRepo:     pitSwitchRepo,
              coreCli:           coreCli,
              crrfModel:         crrfModel,
          },
          action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry),
      )
  }

  func (a *autoCRRFEnableFromPitAction) Params() interface{} {
      return &AutoCRRFEnablePitParams{}
  }

  func (a *autoCRRFEnableFromPitAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFEnablePitParams)

      pitSwitch, err := a.pitSwitchRepo.Load(ctx, p.PitBackupSwitchID)
      if err != nil {
          return err
      }
      clusterID := uint64(pitSwitch.ClusterID)

      return runAutoCRRFEnable(ctx, logger, clusterID, a.backupSettingRepo, a.coreCli, a.crrfModel)
  }

  // RunAfter polls QueryCRRF until the enable DAG reaches terminal state.
  func (a *autoCRRFEnableFromPitAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFEnablePitParams)
      pitSwitch, err := a.pitSwitchRepo.Load(ctx, p.PitBackupSwitchID)
      if err != nil {
          return err
      }
      return pollCRRFEnabled(ctx, logger, uint64(pitSwitch.ClusterID), a.crrfModel)
  }

  // ── Enable from dual-region DAG ───────────────────────────────────────────

  // AutoCRRFEnableDualRegionParams are injected by the dual-region enable DAG via KeyBackupSettingId.
  type AutoCRRFEnableDualRegionParams struct {
      BackupSettingID uint64 `json:"backup_setting_id"`
  }

  type autoCRRFEnableFromDualRegionAction struct {
      backupSettingRepo backupRepo.BackupSettingRepo
      coreCli           grpcli.CoreGrpcClient
      crrfModel         crrfModel.CRRFModel
  }

  // NewAutoCRRFEnableFromDualRegionAction creates the auto-enable action for the dual-region enable DAG.
  func NewAutoCRRFEnableFromDualRegionAction(
      backupSettingRepo backupRepo.BackupSettingRepo,
      coreCli grpcli.CoreGrpcClient,
      crrfModel crrfModel.CRRFModel,
  ) *action.Wrapper {
      return action.MustNewActionWrapper(
          workflow.ActionAutoCRRFEnableFromDualRegion,
          &autoCRRFEnableFromDualRegionAction{
              backupSettingRepo: backupSettingRepo,
              coreCli:           coreCli,
              crrfModel:         crrfModel,
          },
          action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry),
      )
  }

  func (a *autoCRRFEnableFromDualRegionAction) Params() interface{} {
      return &AutoCRRFEnableDualRegionParams{}
  }

  func (a *autoCRRFEnableFromDualRegionAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFEnableDualRegionParams)

      bs, err := a.backupSettingRepo.GetByID(ctx, p.BackupSettingID)
      if err != nil {
          return err
      }
      clusterID := uint64(bs.ClusterID)

      return runAutoCRRFEnable(ctx, logger, clusterID, a.backupSettingRepo, a.coreCli, a.crrfModel)
  }

  // RunAfter polls QueryCRRF until the enable DAG reaches terminal state.
  func (a *autoCRRFEnableFromDualRegionAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFEnableDualRegionParams)
      bs, err := a.backupSettingRepo.GetByID(ctx, p.BackupSettingID)
      if err != nil {
          return err
      }
      return pollCRRFEnabled(ctx, logger, uint64(bs.ClusterID), a.crrfModel)
  }

  // ── Shared logic ──────────────────────────────────────────────────────────

  // pollCRRFEnabled polls until CRRF status is "enabled". Returns action.StopRetry on success.
  // Returns an error (to be retried) if still in-progress, or a fatal error if the DAG failed.
  func pollCRRFEnabled(ctx context.Context, logger *zap.Logger, clusterID uint64, cm crrfModel.CRRFModel) error {
      info, err := cm.QueryCRRF(ctx, clusterID)
      if err != nil {
          return err
      }
      switch info.Status {
      case string(entity.CRRFSettingStatusEnabled):
          logger.Info("CRRF enabled successfully", zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      case string(entity.CRRFSettingStatusDisabled):
          if info.ErrorMessage != "" {
              return fmt.Errorf("CRRF enable DAG failed: %s", info.ErrorMessage)
          }
          return fmt.Errorf("CRRF returned to disabled unexpectedly, clusterID=%d", clusterID)
      default:
          return fmt.Errorf("CRRF not yet enabled, status=%s, clusterID=%d", info.Status, clusterID)
      }
  }

  // runAutoCRRFEnable checks conditions and calls CRRFModel.EnableCRRF.
  // Returns action.StopRetry for any condition that should skip (not retry) silently.
  // Returns nil on success → Wrapper will call RunAfter to poll for completion.
  func runAutoCRRFEnable(
      ctx context.Context,
      logger *zap.Logger,
      clusterID uint64,
      bsRepo backupRepo.BackupSettingRepo,
      coreCli grpcli.CoreGrpcClient,
      cm crrfModel.CRRFModel,
  ) error {
      // 1. Load BackupSetting to check PiTR + dual-region state.
      bs, err := bsRepo.GetByFilter(ctx, &backupRepo.BackupSettingFilter{ClusterID: &clusterID})
      if err != nil {
          return err
      }
      if bs == nil {
          logger.Warn("backup setting not found, skipping auto-CRRF enable", zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      }

      // 2. Conditions: 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 action.StopRetry
      }
      if bs.Bucket == nil || !bs.Bucket.DualRegionEnable {
          logger.Info("dual-region not enabled, skipping auto-CRRF enable", zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      }

      // 3. Check cluster version.
      cluster, clErr := coreCli.GetAnyClusterByID(ctx, clusterID)
      if clErr != nil {
          logger.Error("failed to get cluster for version check", zap.Error(clErr), zap.Uint64("clusterID", clusterID))
          return clErr
      }
      v, parseErr := semver.NewVersion(cluster.Version)
      if parseErr != nil {
          logger.Error("failed to parse cluster version", zap.String("version", cluster.Version), zap.Error(parseErr))
          return fmt.Errorf("parse cluster version: %w", parseErr)
      }
      cfg := internal_config.Global().BackupConfig
      if !cfg.CRRFEnableMinVerConstraint.Check(v) {
          logger.Info("cluster version below CRRF minimum, skipping auto-CRRF enable",
              zap.String("version", cluster.Version),
              zap.String("minVer", cfg.CRRFEnableMinVer))
          return action.StopRetry
      }

      // 4. Extract secondary region.
      secondaryRegion := extractSecondaryRegion(bs)
      if secondaryRegion == "" {
          logger.Warn("secondary region not found in backup setting, skipping auto-CRRF enable",
              zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      }

      // 5. Check if CRRF is already enabled or enabling — idempotent skip.
      info, qErr := cm.QueryCRRF(ctx, clusterID)
      if qErr != nil {
          return qErr
      }
      if info.Status == string(entity.CRRFSettingStatusEnabled) || info.Status == string(entity.CRRFSettingStatusEnabling) {
          logger.Info("CRRF already enabling/enabled, skipping auto-CRRF enable",
              zap.String("crrfStatus", info.Status),
              zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      }

      // 6. Call EnableCRRF — returns nil so Wrapper calls RunAfter to poll.
      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),
      }
      if bizErr := cm.EnableCRRF(ctx, req); bizErr != nil {
          logger.Error("auto EnableCRRF failed", zap.Error(bizErr))
          return bizErr
      }
      logger.Info("auto-CRRF enable triggered, will poll for completion", zap.Uint64("clusterID", clusterID))
      return nil // proceed to RunAfter
  }

  // extractSecondaryRegion returns the secondary region from the backup setting bucket list.
  func extractSecondaryRegion(bs *entity.BackupSetting) string {
      if bs.Bucket == nil {
          return ""
      }
      for _, b := range bs.Bucket.Buckets {
          if !b.Master && len(b.Regions) > 0 {
              return b.Regions[0]
          }
      }
      return ""
  }
  ```

  ```go
  package auto

  import (
      "context"
      "testing"

      "github.com/Masterminds/semver/v3"
      "github.com/stretchr/testify/mock"
      "github.com/stretchr/testify/suite"
      crrfMock "github.com/tidbcloud/dataflow-service/app/models/crrf/mock"
      "github.com/tidbcloud/dataflow-service/app/models/domain"
      backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
      repoMock "github.com/tidbcloud/dataflow-service/app/repo/mock"
      "github.com/tidbcloud/dataflow-service/app/repo/entity"
      "github.com/tidbcloud/dataflow-service/app/workflow/action"
      internal_config "github.com/tidbcloud/dataflow-service/internal/config"
      coreMock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
      corev1proto "github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
      "github.com/tidbcloud/pkgv2/bizerr"
  )

  // --- Helpers ---

  func setupCRRFMinVer(t *testing.T, ver string) {
      c, err := semver.NewConstraint(">=" + ver)
      if err != nil {
          t.Fatal(err)
      }
      internal_config.Global().BackupConfig.CRRFEnableMinVer = ver
      internal_config.Global().BackupConfig.CRRFEnableMinVerConstraint = c
  }

  func awsBS(clusterID uint64, pitrOn, dualOn bool) *entity.BackupSetting {
      bs := &entity.BackupSetting{InfraProviderType: "AWS"}
      bs.ClusterID = entity.ID(clusterID)
      bs.ProjectID = entity.ID(1)
      bs.TenantID = entity.ID(1)
      if pitrOn {
          bs.PiTR = &entity.PiTRSetting{Enable: true}
      }
      if dualOn {
          bs.Bucket = &entity.BucketSetting{
              DualRegionEnable: true,
              Buckets: []entity.Bucket{
                  {Master: true},
                  {Master: false, Regions: []string{"us-west-2"}},
              },
          }
      }
      return bs
  }

  func fakeCluster(ver string) *corev1proto.Cluster { return &corev1proto.Cluster{Version: ver} }

  // --- Suite ---

  type enableSuite struct {
      suite.Suite
      bsRepo    *repoMock.BackupSettingRepo
      coreCli   *coreMock.CoreGrpcClient
      crrfModel *crrfMock.CRRFModel
  }

  func TestAutoCRRFEnableSuite(t *testing.T) { suite.Run(t, new(enableSuite)) }

  func (s *enableSuite) SetupTest() {
      s.bsRepo = repoMock.NewBackupSettingRepo(s.T())
      s.coreCli = coreMock.NewCoreGrpcClient(s.T())
      s.crrfModel = crrfMock.NewCRRFModel(s.T())
      setupCRRFMinVer(s.T(), "v8.5.6")
  }

  func (s *enableSuite) TearDownTest() {
      s.bsRepo.AssertExpectations(s.T())
      s.coreCli.AssertExpectations(s.T())
      s.crrfModel.AssertExpectations(s.T())
  }

  func (s *enableSuite) runEnable(clusterID uint64) error {
      return runAutoCRRFEnable(context.Background(), zap.NewNop(), clusterID, s.bsRepo, s.coreCli, s.crrfModel)
  }

  // Cluster version too low → StopRetry (skip).
  func (s *enableSuite) Test_VersionTooLow_StopRetry() {
      clusterID := uint64(1)
      s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
          Return(awsBS(clusterID, true, true), nil).Once()
      s.coreCli.On("GetAnyClusterByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.5"), nil).Once()

      err := runAutoCRRFEnable(context.Background(), zap.NewNop(), clusterID, s.bsRepo, s.coreCli, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // PiTR not enabled → StopRetry (skip).
  func (s *enableSuite) Test_PiTRNotEnabled_StopRetry() {
      clusterID := uint64(2)
      s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
          Return(awsBS(clusterID, false, true), nil).Once()

      err := runAutoCRRFEnable(context.Background(), zap.NewNop(), clusterID, s.bsRepo, s.coreCli, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // Dual-region not enabled → StopRetry (skip).
  func (s *enableSuite) Test_DualRegionNotEnabled_StopRetry() {
      clusterID := uint64(3)
      s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
          Return(awsBS(clusterID, true, false), nil).Once()

      err := runAutoCRRFEnable(context.Background(), zap.NewNop(), clusterID, s.bsRepo, s.coreCli, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // CRRF already enabling → StopRetry (idempotent).
  func (s *enableSuite) Test_CRRFAlreadyEnabling_StopRetry() {
      clusterID := uint64(4)
      s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
          Return(awsBS(clusterID, true, true), nil).Once()
      s.coreCli.On("GetAnyClusterByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "enabling"}, nil).Once()

      err := runAutoCRRFEnable(context.Background(), zap.NewNop(), clusterID, s.bsRepo, s.coreCli, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // Happy path: all conditions met → EnableCRRF called, returns nil (to proceed to RunAfter).
  func (s *enableSuite) Test_HappyPath_EnableCRRFCalled() {
      clusterID := uint64(5)
      s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
          Return(awsBS(clusterID, true, true), nil).Once()
      s.coreCli.On("GetAnyClusterByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "disabled"}, nil).Once()
      s.crrfModel.On("EnableCRRF", mock.Anything, mock.MatchedBy(func(req *domain.EnableCRRFReq) bool {
          return req.ClusterID == clusterID && req.SecondaryRegion == "us-west-2"
      })).Return(nil).Once()

      err := runAutoCRRFEnable(context.Background(), zap.NewNop(), clusterID, s.bsRepo, s.coreCli, s.crrfModel)
      s.Nil(err) // nil → Wrapper proceeds to RunAfter
  }

  // RunAfter: CRRF enabled → StopRetry (done).
  func (s *enableSuite) Test_RunAfter_Enabled_StopRetry() {
      clusterID := uint64(6)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "enabled"}, nil).Once()

      err := pollCRRFEnabled(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // RunAfter: CRRF still enabling → retry.
  func (s *enableSuite) Test_RunAfter_StillEnabling_Retry() {
      clusterID := uint64(7)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "enabling"}, nil).Once()

      err := pollCRRFEnabled(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.NotNil(err)
      s.NotEqual(action.StopRetry, err)
  }

  // RunAfter: CRRF disabled with error → fatal error (no retry).
  func (s *enableSuite) Test_RunAfter_DisabledWithError_FatalError() {
      clusterID := uint64(8)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "disabled", ErrorMessage: "some dag failure"}, nil).Once()

      err := pollCRRFEnabled(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.NotNil(err)
      s.Contains(err.Error(), "CRRF enable DAG failed")
  }

  > **Note on logger param:** All test calls pass `zap.NewNop()`. No nil-guarding needed.

- [ ] **Step 4.3: Run tests**

  ```bash
  cd /home/nil/pingcap/workbench/repos/dataflow-service.worktrees/auto-crrf
  go test ./app/workflow/action/crrf/auto/... -run "TestAutoCRRFEnableSuite" -v 2>&1 | tail -30
  ```

  Expected: ALL PASS.

- [ ] **Step 4.4: Commit**

  ```bash
  git add app/workflow/action/crrf/auto/enable.go app/workflow/action/crrf/auto/enable_test.go
  git commit -m "feat(crrf/auto): add auto-CRRF enable action with version+condition checks"
  ```

---

## Task 5: New action `app/workflow/action/crrf/auto/disable.go`

**Files:**
- Create: `app/workflow/action/crrf/auto/disable.go`
- Create: `app/workflow/action/crrf/auto/disable_test.go`

The auto-disable action uses the **Run/RunAfter pattern**:
- `Run()`: calls `QueryCRRF` first. If `"disabled"` → `action.StopRetry`. If `"disabling"` → return `nil` (skip calling DisableCRRF to avoid duplicate DAG launch; proceed to RunAfter to wait). Otherwise (`"enabled"`, `"enabling"`, `"failed"`) → call `DisableCRRF` → return `nil`.
- `RunAfter()`: polls `QueryCRRF` until status is `"disabled"` (StopRetry). Otherwise retry.
- No `coreCli` needed — `DisableCRRFReq` has `TenantID`/`ProjectID` fields but the model implementation ignores them; only `ClusterID` is actually used.
- Constructor uses `action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry)`.

- [ ] **Step 5.1: Create `app/workflow/action/crrf/auto/disable.go`**

  ```go
  package auto

  import (
      "context"
      "fmt"

      backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
      "github.com/tidbcloud/dataflow-service/app/repo/entity"
      "github.com/tidbcloud/dataflow-service/app/workflow"
      "github.com/tidbcloud/dataflow-service/app/workflow/action"
      crrfModel "github.com/tidbcloud/dataflow-service/app/models/crrf"
      "github.com/tidbcloud/dataflow-service/app/models/domain"
      "go.uber.org/zap"
  )

  // ── Disable from PiTR DAG ─────────────────────────────────────────────────

  type AutoCRRFDisablePitParams struct {
      PitBackupSwitchID uint64 `json:"pit_backup_switch_id"`
  }

  type autoCRRFDisableFromPitAction struct {
      pitSwitchRepo backupRepo.PitBackupSwitchRepo
      crrfModel     crrfModel.CRRFModel
  }

  func NewAutoCRRFDisableFromPitAction(
      pitSwitchRepo backupRepo.PitBackupSwitchRepo,
      crrfModel crrfModel.CRRFModel,
  ) *action.Wrapper {
      return action.MustNewActionWrapper(
          workflow.ActionAutoCRRFDisableFromPit,
          &autoCRRFDisableFromPitAction{
              pitSwitchRepo: pitSwitchRepo,
              crrfModel:     crrfModel,
          },
          action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry),
      )
  }

  func (a *autoCRRFDisableFromPitAction) Params() interface{} {
      return &AutoCRRFDisablePitParams{}
  }

  func (a *autoCRRFDisableFromPitAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFDisablePitParams)
      pitSwitch, err := a.pitSwitchRepo.Load(ctx, p.PitBackupSwitchID)
      if err != nil {
          return err
      }
      return runAutoCRRFDisable(ctx, logger, uint64(pitSwitch.ClusterID), a.crrfModel)
  }

  func (a *autoCRRFDisableFromPitAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFDisablePitParams)
      pitSwitch, err := a.pitSwitchRepo.Load(ctx, p.PitBackupSwitchID)
      if err != nil {
          return err
      }
      return pollCRRFDisabled(ctx, logger, uint64(pitSwitch.ClusterID), a.crrfModel)
  }

  // ── Disable from dual-region DAG ──────────────────────────────────────────

  type AutoCRRFDisableDualRegionParams struct {
      BackupSettingID uint64 `json:"backup_setting_id"`
  }

  type autoCRRFDisableFromDualRegionAction struct {
      backupSettingRepo backupRepo.BackupSettingRepo
      crrfModel         crrfModel.CRRFModel
  }

  func NewAutoCRRFDisableFromDualRegionAction(
      backupSettingRepo backupRepo.BackupSettingRepo,
      crrfModel crrfModel.CRRFModel,
  ) *action.Wrapper {
      return action.MustNewActionWrapper(
          workflow.ActionAutoCRRFDisableFromDualRegion,
          &autoCRRFDisableFromDualRegionAction{
              backupSettingRepo: backupSettingRepo,
              crrfModel:         crrfModel,
          },
          action.SetRunAfterRetryFun(invokeRunAfterWithSlowExponentialRetry),
      )
  }

  func (a *autoCRRFDisableFromDualRegionAction) Params() interface{} {
      return &AutoCRRFDisableDualRegionParams{}
  }

  func (a *autoCRRFDisableFromDualRegionAction) Run(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFDisableDualRegionParams)
      bs, err := a.backupSettingRepo.GetByID(ctx, p.BackupSettingID)
      if err != nil {
          return err
      }
      return runAutoCRRFDisable(ctx, logger, uint64(bs.ClusterID), a.crrfModel)
  }

  func (a *autoCRRFDisableFromDualRegionAction) RunAfter(ctx context.Context, logger *zap.Logger, params interface{}) error {
      p := params.(*AutoCRRFDisableDualRegionParams)
      bs, err := a.backupSettingRepo.GetByID(ctx, p.BackupSettingID)
      if err != nil {
          return err
      }
      return pollCRRFDisabled(ctx, logger, uint64(bs.ClusterID), a.crrfModel)
  }

  // ── Shared logic ──────────────────────────────────────────────────────────

  // runAutoCRRFDisable checks CRRF status and calls DisableCRRF if needed.
  // - "disabled" → StopRetry (nothing to do).
  // - "disabling" → nil (a disable DAG is already running; RunAfter will wait).
  // - otherwise → call DisableCRRF → nil (proceed to RunAfter).
  func runAutoCRRFDisable(
      ctx context.Context,
      logger *zap.Logger,
      clusterID uint64,
      cm crrfModel.CRRFModel,
  ) error {
      info, err := cm.QueryCRRF(ctx, clusterID)
      if err != nil {
          return err
      }
      switch entity.CRRFSettingStatus(info.Status) {
      case entity.CRRFSettingStatusDisabled:
          logger.Info("CRRF already disabled, skipping auto-CRRF disable",
              zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      case entity.CRRFSettingStatusDisabling:
          // A disable DAG is already running — don't launch a second one.
          // Proceed to RunAfter to wait for it to complete.
          logger.Info("CRRF already disabling, waiting for completion",
              zap.Uint64("clusterID", clusterID))
          return nil
      }

      // Status is enabling/enabled/failed — call DisableCRRF.
      logger.Info("auto-disabling CRRF",
          zap.Uint64("clusterID", clusterID),
          zap.String("crrfStatus", info.Status))
      req := &domain.DisableCRRFReq{
          ClusterID: clusterID,
          // TenantID and ProjectID are unused by DisableCRRF model implementation.
      }
      if bizErr := cm.DisableCRRF(ctx, req); 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 // proceed to RunAfter
  }

  // pollCRRFDisabled polls until CRRF status is "disabled". Returns action.StopRetry on success.
  func pollCRRFDisabled(ctx context.Context, logger *zap.Logger, clusterID uint64, cm crrfModel.CRRFModel) error {
      info, err := cm.QueryCRRF(ctx, clusterID)
      if err != nil {
          return err
      }
      if info.Status == string(entity.CRRFSettingStatusDisabled) {
          logger.Info("CRRF disabled successfully", zap.Uint64("clusterID", clusterID))
          return action.StopRetry
      }
      return fmt.Errorf("CRRF not yet disabled, status=%s, clusterID=%d", info.Status, clusterID)
  }
  ```

- [ ] **Step 5.2: Write tests for auto-disable**

  Create `app/workflow/action/crrf/auto/disable_test.go`:

  ```go
  package auto

  import (
      "context"
      "testing"

      "github.com/stretchr/testify/mock"
      "github.com/stretchr/testify/suite"
      crrfMock "github.com/tidbcloud/dataflow-service/app/models/crrf/mock"
      "github.com/tidbcloud/dataflow-service/app/models/domain"
      "github.com/tidbcloud/dataflow-service/app/workflow/action"
      "go.uber.org/zap"
  )

  type disableSuite struct {
      suite.Suite
      crrfModel *crrfMock.CRRFModel
  }

  func TestAutoCRRFDisableSuite(t *testing.T) { suite.Run(t, new(disableSuite)) }

  func (s *disableSuite) SetupTest() {
      s.crrfModel = crrfMock.NewCRRFModel(s.T())
  }

  func (s *disableSuite) TearDownTest() {
      s.crrfModel.AssertExpectations(s.T())
  }

  // CRRF already disabled → StopRetry.
  func (s *disableSuite) Test_AlreadyDisabled_StopRetry() {
      clusterID := uint64(10)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "disabled"}, nil).Once()

      err := runAutoCRRFDisable(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // CRRF already disabling → nil (RunAfter will wait; avoid duplicate DAG).
  func (s *disableSuite) Test_Disabling_Nil() {
      clusterID := uint64(11)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "disabling"}, nil).Once()

      err := runAutoCRRFDisable(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.Nil(err) // nil → Wrapper proceeds to RunAfter
  }

  // CRRF enabled → DisableCRRF called, returns nil.
  func (s *disableSuite) Test_Enabled_DisableCRRFCalled() {
      clusterID := uint64(12)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "enabled"}, nil).Once()
      s.crrfModel.On("DisableCRRF", mock.Anything, mock.MatchedBy(func(req *domain.DisableCRRFReq) bool {
          return req.ClusterID == clusterID
      })).Return(nil).Once()

      err := runAutoCRRFDisable(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.Nil(err) // nil → Wrapper proceeds to RunAfter
  }

  // CRRF enabling → DisableCRRF called.
  func (s *disableSuite) Test_Enabling_DisableCRRFCalled() {
      clusterID := uint64(13)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "enabling"}, nil).Once()
      s.crrfModel.On("DisableCRRF", mock.Anything, mock.MatchedBy(func(req *domain.DisableCRRFReq) bool {
          return req.ClusterID == clusterID
      })).Return(nil).Once()

      err := runAutoCRRFDisable(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.Nil(err)
  }

  // RunAfter: CRRF disabled → StopRetry (done).
  func (s *disableSuite) Test_RunAfter_Disabled_StopRetry() {
      clusterID := uint64(14)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "disabled"}, nil).Once()

      err := pollCRRFDisabled(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.Equal(action.StopRetry, err)
  }

  // RunAfter: CRRF still disabling → retry.
  func (s *disableSuite) Test_RunAfter_StillDisabling_Retry() {
      clusterID := uint64(15)
      s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
          Return(&domain.CRRFInfo{Status: "disabling"}, nil).Once()

      err := pollCRRFDisabled(context.Background(), zap.NewNop(), clusterID, s.crrfModel)
      s.NotNil(err)
      s.NotEqual(action.StopRetry, err)
  }
  ```

- [ ] **Step 5.3: Run tests**

  ```bash
  go test ./app/workflow/action/crrf/auto/... -v 2>&1 | tail -30
  ```

  Expected: ALL PASS.

- [ ] **Step 5.4: Commit**

  ```bash
  git add app/workflow/action/crrf/auto/disable.go app/workflow/action/crrf/auto/disable_test.go
  git commit -m "feat(crrf/auto): add auto-CRRF disable action"
  ```

---

## Task 6: CRRF enable DAG simplification

**Files:**
- Modify: `app/workflow/dag/crrf/enable.go`

Remove tasks `ActionCRRFEnableDualRegion` and `ActionCRRFEnablePitr` (these are now handled by the trigger DAGs), and remove `KeySecondaryRegion` DAG var. The first task in the simplified DAG becomes `ActionCRRFModifyReplicationRTC`.

> **Keep the action files** `enable_dual_region.go` and `enable_pitr.go` — they are not deleted, just not referenced by the CRRF enable DAG anymore.

- [ ] **Step 6.1: Remove the two tasks and the unused DAG var in `enable.go`**

  The new `GetEnableCRRFDag()` actions list and DAG definition:

  ```go
  func GetEnableCRRFDag() (*entity.Dag, []run.Action) {
      c := container.Global()

      actions := []run.Action{
          crrf_common.NewCRRFModifyReplicationRTCAction(c.BackupSettingModel),
          crrf_enable.NewEnableCRRFDaemonAction(c.BackupSettingRepo, c.LogBackupRepo, c.CRRFSettingRepo, c.BucketRepo, c.CRRFFacade),
          crrf_common.NewTogglePeriodicClusterSettingsBackupAction(c.DedicatedGlobalGrpcClient),
          crrf_enable.NewMarkEnabledAction(c.CRRFSettingRepo),
      }

      dag := &entity.Dag{
          BaseInfo: entity.BaseInfo{ID: workflow.DagEnableCRRF},
          Name:     workflow.DagEnableCRRF,
          Vars: entity.DagVars{
              workflow.KeyCRRFSettingID:   {},
              workflow.KeyBackupSettingId: {},
              workflow.ClusterID:          {},
          },
          Status: entity.DagStatusNormal,
          Tasks: []entity.Task{
              {
                  ID:          workflow.ActionCRRFModifyReplicationRTC + "_enable",
                  ActionName:  workflow.ActionCRRFModifyReplicationRTC,
                  Params:      map[string]interface{}{workflow.ClusterID: fmt.Sprintf("{{%v}}", workflow.ClusterID), "enable": true},
                  TimeoutSecs: 5 * 60,
              },
              {
                  ID:          workflow.ActionCRRFEnableDaemon,
                  ActionName:  workflow.ActionCRRFEnableDaemon,
                  Params:      map[string]interface{}{workflow.ClusterID: fmt.Sprintf("{{%v}}", workflow.ClusterID)},
                  DependOn:    []string{workflow.ActionCRRFModifyReplicationRTC + "_enable"},
                  TimeoutSecs: 15 * 60,
              },
              {
                  ID:          workflow.ActionCRRFTogglePeriodicBackup + "_enable",
                  ActionName:  workflow.ActionCRRFTogglePeriodicBackup,
                  Params:      map[string]interface{}{workflow.ClusterID: fmt.Sprintf("{{%v}}", workflow.ClusterID), "enable": true},
                  DependOn:    []string{workflow.ActionCRRFEnableDaemon},
                  TimeoutSecs: 5 * 60,
              },
              {
                  ID:          workflow.ActionCRRFMarkEnabled,
                  ActionName:  workflow.ActionCRRFMarkEnabled,
                  Params:      map[string]interface{}{workflow.KeyCRRFSettingID: fmt.Sprintf("{{%v}}", workflow.KeyCRRFSettingID), workflow.ClusterID: fmt.Sprintf("{{%v}}", workflow.ClusterID)},
                  DependOn:    []string{workflow.ActionCRRFTogglePeriodicBackup + "_enable"},
                  TimeoutSecs: 5 * 60,
              },
          },
      }
      return dag, actions
  }
  ```

  Also remove the import for `crrf_enable.NewEnableDualRegionAction` and `crrf_enable.NewEnablePitrAction` if they were only referenced in the actions slice (keep the `crrf_enable` import if `NewEnableCRRFDaemonAction` and `NewMarkEnabledAction` still use it).

- [ ] **Step 6.2: Compile and verify**

  ```bash
  go build ./app/workflow/dag/crrf/...
  ```

  Expected: no errors.

- [ ] **Step 6.3: Commit**

  ```bash
  git add app/workflow/dag/crrf/enable.go
  git commit -m "refactor(crrf): simplify enable DAG — remove enable_dual_region and enable_pitr tasks"
  ```

---

## Task 7: Insert auto-CRRF steps into the four trigger DAGs

**Files:**
- Modify: `app/workflow/dag/backup/pit/enbale.go`
- Modify: `app/workflow/dag/backup/pit/disable.go`
- Modify: `app/workflow/dag/backup_setting/aws/enable_dual_region.go`
- Modify: `app/workflow/dag/backup_setting/aws/disable_dual_region.go`

### 7a: PiTR enable DAG

- [ ] **Step 7.1: Update `pit/enbale.go`**

  1. Import the new `auto` package at the top.
  2. Add `NewAutoCRRFEnableFromPitAction` to `actions` slice.
  3. Insert new task between `ActionUpdatePitBackupSwitchStatusFromEnablingToEnabled` and `ActionUpdatePitBackupSettingStatusToNormal`, and update `DependOn`.

  New actions slice addition (after `pit_enable.NewUpdateSwitchStatusToEnabledAction`):
  ```go
  crrf_auto.NewAutoCRRFEnableFromPitAction(
      container.Global().BackupSettingRepo,
      container.Global().PitBackupSwitchRepo,
      container.Global().CoreGrpcClient,
      container.Global().CRRFModel,
  ),
  ```

  New task (insert before `ActionUpdatePitBackupSettingStatusToNormal`):
  ```go
  {
      ID:         workflow.ActionAutoCRRFEnableFromPit,
      ActionName: workflow.ActionAutoCRRFEnableFromPit,
      Params: map[string]interface{}{
          workflow.KeyPitBackupSwitchID: fmt.Sprintf("{{%v}}", workflow.KeyPitBackupSwitchID),
      },
      DependOn:    []string{workflow.ActionUpdatePitBackupSwitchStatusFromEnablingToEnabled},
      TimeoutSecs: 60 * 60,
  },
  ```

  Update `ActionUpdatePitBackupSettingStatusToNormal` task:
  ```go
  {
      ID:          workflow.ActionUpdatePitBackupSettingStatusToNormal,
      ActionName:  workflow.ActionUpdatePitBackupSettingStatusToNormal,
      DependOn:    []string{workflow.ActionAutoCRRFEnableFromPit},  // changed from ActionUpdatePitBackupSwitchStatusFromEnablingToEnabled
      TimeoutSecs: 5 * 60,
  },
  ```

  Add the import:
  ```go
  crrf_auto "github.com/tidbcloud/dataflow-service/app/workflow/action/crrf/auto"
  ```

  And add `"fmt"` if not already imported (it is not in the current `enbale.go`):
  ```go
  "fmt"
  ```

### 7b: PiTR disable DAG

- [ ] **Step 7.2: Update `pit/disable.go`**

  Add `NewAutoCRRFDisableFromPitAction` to actions slice, insert task, update `DependOn`.

  Action addition (after `pit_disable.NewUpdateSwitchStatusToDisabledAction`):
  ```go
  crrf_auto.NewAutoCRRFDisableFromPitAction(
      container.Global().PitBackupSwitchRepo,
      container.Global().CRRFModel,
  ),
  ```

  New task:
  ```go
  {
      ID:         workflow.ActionAutoCRRFDisableFromPit,
      ActionName: workflow.ActionAutoCRRFDisableFromPit,
      Params: map[string]interface{}{
          workflow.KeyPitBackupSwitchID: fmt.Sprintf("{{%v}}", workflow.KeyPitBackupSwitchID),
      },
      DependOn:    []string{workflow.ActionUpdatePitBackupSwitchStatusToDisabled},
      TimeoutSecs: 40 * 60,
  },
  ```

  Update `ActionUpdatePitBackupSettingStatusToNormal`:
  ```go
  DependOn: []string{workflow.ActionAutoCRRFDisableFromPit},
  ```

  Imports to add: `"fmt"`, `crrf_auto "..."`.

### 7c: Dual-region enable DAG

- [ ] **Step 7.3: Update `aws/enable_dual_region.go`**

  Action addition (after `enable_dual_region.NewCreateBucketReplicationAction`):
  ```go
  crrf_auto.NewAutoCRRFEnableFromDualRegionAction(
      container.Global().BackupSettingRepo,
      container.Global().CoreGrpcClient,
      container.Global().CRRFModel,
  ),
  ```

  New task (between `ActionCreateBucketReplication` and `ActionEnableDualRegionBucketFinalStep`):
  ```go
  {
      ID:         workflow.ActionAutoCRRFEnableFromDualRegion,
      ActionName: workflow.ActionAutoCRRFEnableFromDualRegion,
      Params: map[string]interface{}{
          workflow.KeyBackupSettingId: fmt.Sprintf("{{%v}}", workflow.KeyBackupSettingId),
      },
      DependOn:    []string{workflow.ActionCreateBucketReplication},
      TimeoutSecs: 60 * 60,
  },
  ```

  Update `ActionEnableDualRegionBucketFinalStep`:
  ```go
  DependOn: []string{workflow.ActionAutoCRRFEnableFromDualRegion},
  ```

### 7d: Dual-region disable DAG

- [ ] **Step 7.4: Update `aws/disable_dual_region.go`**

  Action addition (after `create.NewAddNodeRoleToBucketPolicyAction`):
  ```go
  crrf_auto.NewAutoCRRFDisableFromDualRegionAction(
      container.Global().BackupSettingRepo,
      container.Global().CRRFModel,
  ),
  ```

  New task (between `ActionAddNodeRoleToBucketPolicy` and `ActionDisableDualRegionBucketFinalStep`):
  ```go
  {
      ID:         workflow.ActionAutoCRRFDisableFromDualRegion,
      ActionName: workflow.ActionAutoCRRFDisableFromDualRegion,
      Params: map[string]interface{}{
          workflow.KeyBackupSettingId: fmt.Sprintf("{{%v}}", workflow.KeyBackupSettingId),
      },
      DependOn:    []string{workflow.ActionAddNodeRoleToBucketPolicy},
      TimeoutSecs: 40 * 60,
  },
  ```

  Update `ActionDisableDualRegionBucketFinalStep`:
  ```go
  DependOn: []string{workflow.ActionAutoCRRFDisableFromDualRegion},
  ```

- [ ] **Step 7.5: Compile and verify all four DAG files**

  ```bash
  go build ./app/workflow/dag/...
  ```

  Expected: no errors.

- [ ] **Step 7.6: Commit**

  ```bash
  git add app/workflow/dag/backup/pit/enbale.go \
          app/workflow/dag/backup/pit/disable.go \
          app/workflow/dag/backup_setting/aws/enable_dual_region.go \
          app/workflow/dag/backup_setting/aws/disable_dual_region.go
  git commit -m "feat(dag): insert auto-CRRF enable/disable steps into PiTR and dual-region DAGs"
  ```

---

## Task 8: Remove CRRF-active guard from `preCheckModifyPitBackupSetting`

**Files:**
- Modify: `app/models/backup/impl/backup_setting.go:842-853`

This guard (which blocks PiTR disable when CRRF is active) is no longer needed — the auto-CRRF disable step will handle teardown as part of the DAG.

- [ ] **Step 8.1: Write failing test demonstrating guard removal**

  Find the existing test for this guard in `app/models/backup/impl/backup_setting_test.go` (or add a new one). The test should confirm that disabling PiTR while CRRF is active is now allowed:

  ```go
  func (s *backupSettingModelTestSuite) Test_DisablePiTR_WithCRRFActive_Allowed() {
      // ... set up cluster, backup setting, crrf setting (IsActive = true)
      // ... expect pitBackupSwitcher.Disable to be called successfully
      // ... expect no error from ModifyPitBackupSetting
  }
  ```

- [ ] **Step 8.2: Run test to confirm it fails**

  ```bash
  go test ./app/models/backup/impl/... -run "Test_DisablePiTR_WithCRRFActive" -v 2>&1 | tail -10
  ```

  Expected: FAIL (guard returns error).

- [ ] **Step 8.3: Remove the guard block in `backup_setting.go`**

  Remove lines 842–853:
  ```go
  crrfSetting, crrfErr := m.crrfSettingRepo.GetByClusterID(ctx, req.ClusterID)
  if crrfErr != nil && crrfErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
      logger.Warn("failed to query CRRF setting for PiTR guard", zap.Error(crrfErr))
      return crrfErr
  }
  if crrfSetting != nil && crrfSetting.IsActive() {
      logger.Warn("PiTR disable blocked: CRRF is active", ...)
      return bizerr_wrapper.NewInvalidInputError("cannot disable PiTR while CRRF is active; disable CRRF first")
  }
  ```

  The else branch now becomes:
  ```go
  } else {
      _, err := m.pitBackupSwitcher.Disable(
          ctx, &domain.DisablePitBackupReq{SwitchPitBackupReq: switchReq})
      if err != nil {
          logger.Warn("disable pit backup failed", zap.Error(err))
          return err
      }
  }
  ```

  Remove unused import of `crrfSettingRepo` from the method if it's only used by the removed guard. Check whether `m.crrfSettingRepo` is still used elsewhere in `backup_setting.go`.

- [ ] **Step 8.4: Run all backup setting model tests**

  ```bash
  go test ./app/models/backup/impl/... -v 2>&1 | tail -30
  ```

  Expected: ALL PASS.

- [ ] **Step 8.5: Commit**

  ```bash
  git add app/models/backup/impl/backup_setting.go
  git commit -m "feat(backup): remove CRRF-active guard from PiTR disable — auto-CRRF handles teardown"
  ```

---

## Task 9: Final build + test pass

- [ ] **Step 9.1: Full build**

  ```bash
  cd /home/nil/pingcap/workbench/repos/dataflow-service.worktrees/auto-crrf
  go build ./...
  ```

  Expected: no errors.

- [ ] **Step 9.2: Run all affected tests**

  ```bash
  go test \
    ./internal/config/... \
    ./app/models/crrf/... \
    ./app/models/backup/... \
    ./app/workflow/action/crrf/... \
    ./app/workflow/dag/... \
    -v 2>&1 | grep -E "^(PASS|FAIL|---)" | tail -50
  ```

  Expected: all PASS.

- [ ] **Step 9.3: Run the linter**

  ```bash
  golangci-lint run ./... 2>&1 | head -40
  ```

  Fix any lint errors before proceeding.

- [ ] **Step 9.4: Final commit (if any cleanup needed)**

  ```bash
  git add -A
  git commit -m "chore(auto-crrf): cleanup after full build and lint pass"
  ```

---

## Key Decisions and Watch-Outs

| Area | Decision | Rationale |
|------|----------|-----------|
| Placement | Second-to-last step in each DAG | Keeps backup setting locked (≠ `normal`) during CRRF enable/disable, preventing concurrent user requests |
| Status check relaxation | Allow `editing_pitr`/`editing_region` in `EnableCRRF` | Auto-CRRF is called mid-DAG when status is exactly these values |
| Version check | `CRRFEnableMinVer` default `v8.5.6`, parsed in `BackupConfig.Init()` | Follows existing `LogBackupEnableMinVer` pattern |
| Fire-and-forget vs RunAfter | Auto actions use Run/RunAfter polling pattern | `Run()` triggers the CRRF DAG; `RunAfter()` polls until terminal state, ensuring the parent DAG doesn't proceed until CRRF is done |
| Guard removal | Remove CRRF-active PiTR-disable guard | Auto-CRRF disable is now embedded in the disable DAG |
| `enable_dual_region.go` / `enable_pitr.go` | Keep files, remove from CRRF enable DAG | May be needed for manual CRRF flows in the future |
| `DisableCRRFReq` TenantID/ProjectID | Model implementation ignores these fields | Only `ClusterID` is used; `coreCli` not needed in disable action |
| `disabling` in disable Run() | Skip DisableCRRF call; proceed to RunAfter | Avoids launching duplicate disable DAG when one is already running |
| PitBackupSwitchID param | PiTR DAG vars include `KeyPitBackupSwitchID` | Auto actions use this to load `PitBackupSwitch.ClusterID` |
| BackupSettingID param | Dual-region DAG vars include `KeyBackupSettingId` | Auto actions use this to load `BackupSetting.ClusterID` |