# PiTR Compaction Async Deletion 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:** Split compaction file deletion into sync `migrations/` deletion (blocking restore) and async `compactions/compact-{ID}/` deletion (fire-and-forget), so PiTR restore reaches `available` without waiting for large compaction file deletes.

**Architecture:** Add `SyncDeleteMigrations` and `DeleteCompactionsSubDir` to `CompactBackupModel` interface; migrate storage resolution logic from `PurgeCompactionStorageAction` into the model impl; `DeleteCompactionForPiTR` calls `SyncDeleteMigrations` synchronously (fail-fast) then `AsyncDelete` as fire-and-forget and drops its `RunAfter`; `PurgeCompactionStorageAction` delegates to `DeleteCompactionsSubDir`.

**Tech Stack:** Go, fastflow (workflow engine), testify/suite mocks, `pkg/blob` for object storage, `make generate4test` for mock regeneration.

---

## File Map

| File | Change |
|------|--------|
| `app/models/backup/interface.go` | Add two new methods to `CompactBackupModel` interface |
| `app/models/backup/impl/compact_backup_model.go` | Add `bucketRepo`/`encryptor` fields; implement new methods + private `deleteStorageSubDir`/`resolveStorage` |
| `app/models/backup/mock/compact_backup_model.go` | **Regenerated** — do not edit manually |
| `app/workflow/action/backup/compact/purge_compaction_storage.go` | Inject `CompactBackupModel`; delegate to `DeleteCompactionsSubDir`; remove storage logic |
| `app/workflow/action/restore/pitr/create/delete_compaction.go` | Sync `SyncDeleteMigrations` + fire-and-forget `AsyncDelete`; remove `RunAfter`, `dagWaiter`, `compactBackupRepo` |
| `app/workflow/dag/backup/compact/delete.go` | Update `NewPurgeCompactionStorageAction` constructor args |
| `app/workflow/dag/restore/restore_pitr.go` | Simplify `NewDeleteCompactionForPiTR` call; remove `MustNewActionWrapperV2` wrapper |
| `app/workflow/dag/restore/recycle_restore_pitr.go` | Same as above |
| `app/container/container.go` | Add `BucketRepo` + `Encryptor` to `NewCompactBackupModel` call |
| `app/workflow/action/backup/compact/purge_compaction_storage_test.go` | Update constructor call + add mock-based test |
| `app/workflow/action/restore/pitr/create/delete_compaction_test.go` | Remove `TestRunAfter`; update `TestRun` for new behavior |

---

### Task 1: Add interface methods + compile stubs

**Files:**
- Modify: `app/models/backup/interface.go`
- Modify: `app/models/backup/impl/compact_backup_model.go`

- [ ] **Step 1: Add two methods to the `CompactBackupModel` interface**

In `app/models/backup/interface.go`, replace:
```go
type CompactBackupModel interface {
	AsyncCreate(
		ctx context.Context, req *domain.AsyncCreateCompactBackupReq,
	) (*domain.AsyncCreateCompactBackupResp, bizerr.BizError)
	AsyncDelete(
		ctx context.Context, req *domain.AsyncDeleteCompactBackupReq,
	) (*domain.AsyncDeleteCompactBackupResp, bizerr.BizError)
}
```
with:
```go
type CompactBackupModel interface {
	AsyncCreate(
		ctx context.Context, req *domain.AsyncCreateCompactBackupReq,
	) (*domain.AsyncCreateCompactBackupResp, bizerr.BizError)
	AsyncDelete(
		ctx context.Context, req *domain.AsyncDeleteCompactBackupReq,
	) (*domain.AsyncDeleteCompactBackupResp, bizerr.BizError)
	// SyncDeleteMigrations synchronously deletes the v1/migrations/ directory for the
	// given compact backup from object storage. Returns an error on failure so the
	// caller can propagate it for retry.
	SyncDeleteMigrations(ctx context.Context, compactBackupID uint64) bizerr.BizError
	// DeleteCompactionsSubDir synchronously deletes the compaction-specific subdirectory
	// v1/compactions/compact-{ID}/ from object storage.
	DeleteCompactionsSubDir(ctx context.Context, compactBackupID uint64) bizerr.BizError
}
```

- [ ] **Step 2: Add stub implementations to `CompactBackupModel` impl so it compiles**

At the bottom of `app/models/backup/impl/compact_backup_model.go`, add:
```go
func (c *CompactBackupModel) SyncDeleteMigrations(_ context.Context, _ uint64) bizerr.BizError {
	panic("not yet implemented")
}

func (c *CompactBackupModel) DeleteCompactionsSubDir(_ context.Context, _ uint64) bizerr.BizError {
	panic("not yet implemented")
}
```

- [ ] **Step 3: Verify it compiles**

```bash
cd /path/to/worktree && make build-dataflow-darwin 2>&1 | tail -5
# or on Linux:
make build-dataflow 2>&1 | tail -5
```
Expected: build succeeds (no interface satisfaction error).

---

### Task 2: Regenerate mocks

**Files:**
- Modify: `app/models/backup/mock/compact_backup_model.go` (generated)

- [ ] **Step 1: Regenerate mocks**

```bash
make generate4test 2>&1 | tail -10
```
Expected: exits 0, `app/models/backup/mock/compact_backup_model.go` now contains `SyncDeleteMigrations` and `DeleteCompactionsSubDir` mock methods.

- [ ] **Step 2: Verify the mock contains the two new methods**

```bash
grep -n "SyncDeleteMigrations\|DeleteCompactionsSubDir" app/models/backup/mock/compact_backup_model.go
```
Expected: two function definitions, one for each method.

---

### Task 3: Implement `deleteStorageSubDir`, `SyncDeleteMigrations`, `DeleteCompactionsSubDir` in model

**Files:**
- Modify: `app/models/backup/impl/compact_backup_model.go`

- [ ] **Step 1: Add new fields and update `NewCompactBackupModel` constructor**

Add `bucketRepo` and `encryptor` fields to the struct and constructor in `app/models/backup/impl/compact_backup_model.go`.

Replace the existing struct:
```go
type CompactBackupModel struct {
	// db
	dbProvider        db.TransProvider
	compactBackupRepo backup_repo.CompactBackupRepo
	logBackupRepo     backup_repo.LogBackupRepo

	// cluster job for locking
	job cluster_job.Model

	// workflow
	workflowCli workflow.Workflow

	// grpc
	coreGrpcClient grpcli.CoreGrpcClient
	mgmtSvcCli     grpcli.MgmtSvcGrpcClient
}
```
with:
```go
type CompactBackupModel struct {
	// db
	dbProvider        db.TransProvider
	compactBackupRepo backup_repo.CompactBackupRepo
	logBackupRepo     backup_repo.LogBackupRepo

	// cluster job for locking
	job cluster_job.Model

	// workflow
	workflowCli workflow.Workflow

	// grpc
	coreGrpcClient grpcli.CoreGrpcClient
	mgmtSvcCli     grpcli.MgmtSvcGrpcClient

	// storage (used for sync deletion helpers)
	bucketRepo repo.BucketRepo
	encryptor  encryptor_wrapper.Encryptor
}
```

Replace the existing constructor:
```go
func NewCompactBackupModel(
	dbProvider db.TransProvider,
	compactBackupRepo backup_repo.CompactBackupRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	job cluster_job.Model,
	workflowCli workflow.Workflow,
	coreGrpcClient grpcli.CoreGrpcClient,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
) *CompactBackupModel {
	return &CompactBackupModel{
		dbProvider:        dbProvider,
		compactBackupRepo: compactBackupRepo,
		logBackupRepo:     logBackupRepo,
		job:               job,
		workflowCli:       workflowCli,
		coreGrpcClient:    coreGrpcClient,
		mgmtSvcCli:        mgmtSvcCli,
	}
}
```
with:
```go
func NewCompactBackupModel(
	dbProvider db.TransProvider,
	compactBackupRepo backup_repo.CompactBackupRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	job cluster_job.Model,
	workflowCli workflow.Workflow,
	coreGrpcClient grpcli.CoreGrpcClient,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
	bucketRepo repo.BucketRepo,
	encryptor encryptor_wrapper.Encryptor,
) *CompactBackupModel {
	return &CompactBackupModel{
		dbProvider:        dbProvider,
		compactBackupRepo: compactBackupRepo,
		logBackupRepo:     logBackupRepo,
		job:               job,
		workflowCli:       workflowCli,
		coreGrpcClient:    coreGrpcClient,
		mgmtSvcCli:        mgmtSvcCli,
		bucketRepo:        bucketRepo,
		encryptor:         encryptor,
	}
}
```

Add the missing imports to `app/models/backup/impl/compact_backup_model.go`:
```go
"github.com/pingcap/errors"
"github.com/tidbcloud/dataflow-service/app/repo"
"github.com/tidbcloud/dataflow-service/internal/config"
"github.com/tidbcloud/dataflow-service/pkg/azblob"
"github.com/tidbcloud/dataflow-service/pkg/blob"
encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
"github.com/tidbcloud/dataflow-service/pkg/s3"
"github.com/tidbcloud/pkgv2/logid"
```

- [ ] **Step 2: Replace stub implementations with real code**

Remove the two stub implementations added in Task 1 and replace with the full implementation below.

Add the following private helpers and public methods at the bottom of `app/models/backup/impl/compact_backup_model.go`:

```go
// resolveStorage returns the storage URI and bucket for the given compact backup.
// Migrated from PurgeCompactionStorageAction.
func (c *CompactBackupModel) resolveStorage(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
) (string, *entity.Bucket, error) {
	var storageURI string
	var bucket *entity.Bucket

	if compactBackup.BucketID == 0 {
		b, bErr := c.bucketRepo.Load(ctx, logBackup.BucketID.Uint64())
		if bErr != nil {
			return "", nil, fmt.Errorf("load bucket %d: %w", logBackup.BucketID.Uint64(), bErr)
		}
		storageURI = logBackup.Storage
		bucket = b
	} else {
		b, bErr := c.bucketRepo.Load(ctx, compactBackup.BucketID.Uint64())
		if bErr != nil {
			return "", nil, fmt.Errorf("load bucket %d: %w", compactBackup.BucketID.Uint64(), bErr)
		}
		uri, err := b.URIInBucket(logBackup.Storage)
		if err != nil {
			return "", nil, fmt.Errorf("resolve URI in bucket %d: %w", compactBackup.BucketID.Uint64(), err)
		}
		storageURI = uri
		bucket = b
	}

	if bucket.InfraProviderType == entity.InfraProviderTypeAKS ||
		bucket.InfraProviderType == entity.InfraProviderTypeAzure {
		azurl, err := azblob.ParseAzblobURL(storageURI)
		if err != nil {
			return "", nil, errors.Annotate(err, "parse azure blob url")
		}
		sasToken, err := bucket.GetSASToken(c.encryptor)
		if err != nil {
			return "", nil, errors.Annotate(err, "get sas token")
		}
		azurl.SetSASToken(sasToken)
		storageURI = azurl.ToInnerURL(true)
	}

	return storageURI, bucket, nil
}

// deleteStorageSubDir loads the compact backup and log backup, resolves storage, then
// deletes the subdirectory returned by getSubDir. Not-found errors are silently skipped.
func (c *CompactBackupModel) deleteStorageSubDir(
	ctx context.Context,
	logger *zap.Logger,
	compactBackupID uint64,
	getSubDir func(*entity.CompactBackup) string,
) bizerr.BizError {
	compactBackup, bErr := c.compactBackupRepo.Load(ctx, compactBackupID)
	if bErr != nil {
		logger.Error("load compact backup failed",
			zap.Uint64("compact_backup_id", compactBackupID), zap.Error(bErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, 0)
	}

	logBackup, bErr := c.logBackupRepo.Load(ctx, compactBackup.LogBackupID.Uint64())
	if bErr != nil {
		logger.Error("load log backup failed",
			zap.Uint64("log_backup_id", compactBackup.LogBackupID.Uint64()), zap.Error(bErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	storageURI, bucket, err := c.resolveStorage(ctx, compactBackup, logBackup)
	if err != nil {
		logger.Error("resolve storage failed", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	opt := &blob.StorageOption{
		ProjectID:     logBackup.ProjectID.Uint64(),
		ClusterRegion: bucket.Regions[0],
	}
	if bucket.CloudAccountID != nil {
		opt.AWSRoleARN = domain.BuildDataflowServiceInfraRoleARN(*bucket.CloudAccountID)
	}

	// Use a background context to avoid workflow cancellation interrupting the delete.
	bgCtx := context.Background()
	bgCtx = context.WithValue(bgCtx, logid.CtxKey, ctx.Value(logid.CtxKey))
	deleteOpt := &blob.DeleteOption{Concurrency: config.Global().BackupConfig.CleanupDeleteConcurrency}

	subDir := getSubDir(compactBackup)
	s, blobErr := blob.NewStorage(bgCtx, storageURI+"/"+subDir, opt)
	if blobErr != nil {
		if blob.IsNotFoundError(blobErr) {
			logger.Info("storage subdir not found, skipping", zap.String("subdir", subDir))
			return nil
		}
		logger.Error("init storage failed", zap.String("subdir", subDir), zap.Error(blobErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}
	deleteErr := s.DeleteSubDir(bgCtx, "", deleteOpt)
	if deleteErr != nil && !s3.IsNotFoundError(deleteErr) && !blob.IsNotFoundError(deleteErr) {
		logger.Error("delete storage subdir failed", zap.String("subdir", subDir), zap.Error(deleteErr))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}
	logger.Info("deleted storage subdir", zap.String("subdir", subDir))
	return nil
}

func (c *CompactBackupModel) SyncDeleteMigrations(ctx context.Context, compactBackupID uint64) bizerr.BizError {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.SyncDeleteMigrations").
		With(zap.Uint64("compact_backup_id", compactBackupID))
	return c.deleteStorageSubDir(ctx, logger, compactBackupID, func(_ *entity.CompactBackup) string {
		return "v1/migrations"
	})
}

func (c *CompactBackupModel) DeleteCompactionsSubDir(ctx context.Context, compactBackupID uint64) bizerr.BizError {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.DeleteCompactionsSubDir").
		With(zap.Uint64("compact_backup_id", compactBackupID))
	return c.deleteStorageSubDir(ctx, logger, compactBackupID, func(cb *entity.CompactBackup) string {
		return "v1/compactions/" + cb.GenerateDownstreamID()
	})
}
```

- [ ] **Step 3: Update container to pass new args**

In `app/container/container.go`, find the `NewCompactBackupModel` call (around line 861) and add `BucketRepo` and `Encryptor`:

Replace:
```go
container.CompactBackupModel = backup_impl.NewCompactBackupModel(
    container.TransProvider,
    container.CompactBackupRepo,
    container.LogBackupRepo,
    container.ClusterJobModel,
    container.WorkflowModel,
    container.CoreGrpcClient,
    container.MgmtSvcGrpcClient,
)
```
with:
```go
container.CompactBackupModel = backup_impl.NewCompactBackupModel(
    container.TransProvider,
    container.CompactBackupRepo,
    container.LogBackupRepo,
    container.ClusterJobModel,
    container.WorkflowModel,
    container.CoreGrpcClient,
    container.MgmtSvcGrpcClient,
    container.BucketRepo,
    container.Encryptor,
)
```

- [ ] **Step 4: Verify it builds**

```bash
make build-dataflow 2>&1 | tail -5
```
Expected: build succeeds.

- [ ] **Step 5: Run existing model tests to verify no regressions**

```bash
go test ./app/models/backup/impl/... -v -count=1 2>&1 | tail -20
```
Expected: all tests pass (no panics from the old stubs).

- [ ] **Step 6: Commit**

```bash
git add app/models/backup/interface.go \
        app/models/backup/impl/compact_backup_model.go \
        app/models/backup/mock/compact_backup_model.go \
        app/container/container.go
git commit -m "feat(backup): add SyncDeleteMigrations and DeleteCompactionsSubDir to CompactBackupModel

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"
```

---

### Task 4: Refactor `PurgeCompactionStorageAction`

**Files:**
- Modify: `app/workflow/action/backup/compact/purge_compaction_storage.go`
- Modify: `app/workflow/action/backup/compact/purge_compaction_storage_test.go`
- Modify: `app/workflow/dag/backup/compact/delete.go`

- [ ] **Step 1: Write the updated test first**

Check whether `newExecCtxWithVars` is already defined in another `_test.go` file in the `compact` package:
```bash
grep -rn "newExecCtxWithVars" app/workflow/action/backup/compact/
```
If it exists, do not redeclare it. If not, it will be included in the test file below.

Replace the entire content of `app/workflow/action/backup/compact/purge_compaction_storage_test.go` with:

```go
package compact

import (
	"context"
	"testing"

	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	fastflow_utils "github.com/shiningrush/fastflow/pkg/utils"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	mock_backup_model "github.com/tidbcloud/dataflow-service/app/models/backup/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
)

type PurgeCompactionStorageSuite struct {
	suite.Suite
}

func TestPurgeCompactionStorage(t *testing.T) {
	suite.Run(t, new(PurgeCompactionStorageSuite))
}

// newExecCtxWithVars builds a minimal fastflow ExecuteContext with the given DAG run variables.
// Remove this definition if another _test.go in this package already defines it.
func newExecCtxWithVars(vars map[string]string) run.ExecuteContext {
	ctx := fastflow_entity.CtxWithRunningTaskIns(context.Background(),
		&fastflow_entity.TaskInstance{DagInsID: "test-dag"})
	getter := fastflow_utils.KeyValueGetter(func(key string) (string, bool) {
		v, ok := vars[key]
		return v, ok
	})
	return run.NewDefExecuteContext(ctx, nil, nil, getter, nil)
}

func (s *PurgeCompactionStorageSuite) TestActionName() {
	compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())
	compactBackupModel := mock_backup_model.NewCompactBackupModel(s.T())
	act := NewPurgeCompactionStorageAction(compactBackupRepo, compactBackupModel)
	s.Equal("purge_compaction_storage", act.Name())
}

func (s *PurgeCompactionStorageSuite) TestRun() {
	tests := map[string]struct {
		compactBackup        *entity.CompactBackup
		deleteCompactionsErr bool
		wantErr              bool
	}{
		"not deleting state returns error": {
			compactBackup: &entity.CompactBackup{Status: entity.CompactBackupStatusPending},
			wantErr:       true,
		},
		"calls DeleteCompactionsSubDir on deleting state": {
			compactBackup:        &entity.CompactBackup{Status: entity.CompactBackupStatusDeleting},
			deleteCompactionsErr: false,
			wantErr:              false,
		},
		"propagates DeleteCompactionsSubDir error": {
			compactBackup:        &entity.CompactBackup{Status: entity.CompactBackupStatusDeleting},
			deleteCompactionsErr: true,
			wantErr:              true,
		},
	}

	for name, tc := range tests {
		s.Run(name, func() {
			compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())
			compactBackupModel := mock_backup_model.NewCompactBackupModel(s.T())

			tc.compactBackup.ID = 42
			compactBackupRepo.On("Load", mock.Anything, uint64(42)).Return(tc.compactBackup, nil)

			if tc.compactBackup.DeletingOrDeleted() {
				if tc.deleteCompactionsErr {
					compactBackupModel.On("DeleteCompactionsSubDir", mock.Anything, uint64(42)).
						Return(bizerr_wrapper.NewInternalErr("storage error"))
				} else {
					compactBackupModel.On("DeleteCompactionsSubDir", mock.Anything, uint64(42)).
						Return(nil)
				}
			}

			vars := map[string]string{workflow.KeyCompactBackupID: "42"}
			execCtx := newExecCtxWithVars(vars)

			act := NewPurgeCompactionStorageAction(compactBackupRepo, compactBackupModel)
			err := act.Run(execCtx, nil)
			if tc.wantErr {
				s.Error(err)
			} else {
				s.NoError(err)
			}
		})
	}
}
```

- [ ] **Step 2: Run the new test to verify it fails (expected)**

```bash
go test ./app/workflow/action/backup/compact/... -run TestPurgeCompactionStorage -v 2>&1 | tail -20
```
Expected: compilation error or test failure because `NewPurgeCompactionStorageAction` still takes four args.

- [ ] **Step 3: Rewrite `PurgeCompactionStorageAction`**

Replace the entire content of `app/workflow/action/backup/compact/purge_compaction_storage.go` with:

```go
package compact

import (
	"fmt"

	"github.com/shiningrush/fastflow/pkg/entity/run"
	backupModel "github.com/tidbcloud/dataflow-service/app/models/backup"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"go.uber.org/zap"
)

type PurgeCompactionStorageAction struct {
	compactBackupRepo  backup_repo.CompactBackupRepo
	compactBackupModel backupModel.CompactBackupModel
}

func NewPurgeCompactionStorageAction(
	compactBackupRepo backup_repo.CompactBackupRepo,
	compactBackupModel backupModel.CompactBackupModel,
) *PurgeCompactionStorageAction {
	return &PurgeCompactionStorageAction{
		compactBackupRepo:  compactBackupRepo,
		compactBackupModel: compactBackupModel,
	}
}

func (a *PurgeCompactionStorageAction) Name() string {
	return workflow.ActionPurgeCompactionStorage
}

func (a *PurgeCompactionStorageAction) ParameterNew() interface{} {
	return nil
}

func (a *PurgeCompactionStorageAction) Run(ctx run.ExecuteContext, _ interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	logger.Info("action run start")

	compactBackupID, err := action.GetUint64FromRunVar(ctx, workflow.KeyCompactBackupID)
	if err != nil {
		return fmt.Errorf("get compact backup id from run var: %w", err)
	}
	compactBackup, bErr := a.compactBackupRepo.Load(ctx.Context(), compactBackupID)
	if bErr != nil {
		return fmt.Errorf("load compact backup %d: %w", compactBackupID, bErr)
	}
	if !compactBackup.DeletingOrDeleted() {
		logger.Error("compact backup is not being deleted",
			zap.Uint64("id", compactBackupID),
			zap.String("status", string(compactBackup.Status)))
		return bizerr_wrapper.NewInternalErr("compact backup is not being deleted")
	}

	if bErr := a.compactBackupModel.DeleteCompactionsSubDir(ctx.Context(), compactBackupID); bErr != nil {
		return bErr
	}
	logger.Info("action run finished")
	return nil
}
```

- [ ] **Step 4: Update the DAG definition to pass two args**

In `app/workflow/dag/backup/compact/delete.go`, replace:
```go
compact.NewPurgeCompactionStorageAction(
    container.Global().CompactBackupRepo,
    container.Global().LogBackupRepo,
    container.Global().BucketRepo,
    container.Global().Encryptor,
),
```
with:
```go
compact.NewPurgeCompactionStorageAction(
    container.Global().CompactBackupRepo,
    container.Global().CompactBackupModel,
),
```

- [ ] **Step 5: Run the tests to verify they pass**

```bash
go test ./app/workflow/action/backup/compact/... -v -count=1 2>&1 | tail -20
```
Expected: all tests pass including `TestPurgeCompactionStorage`.

- [ ] **Step 6: Commit**

```bash
git add app/workflow/action/backup/compact/purge_compaction_storage.go \
        app/workflow/action/backup/compact/purge_compaction_storage_test.go \
        app/workflow/dag/backup/compact/delete.go
git commit -m "refactor(backup): delegate PurgeCompactionStorageAction to CompactBackupModel.DeleteCompactionsSubDir

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"
```

---

### Task 5: Refactor `DeleteCompactionForPiTR`

**Files:**
- Modify: `app/workflow/action/restore/pitr/create/delete_compaction.go`
- Modify: `app/workflow/action/restore/pitr/create/delete_compaction_test.go`

- [ ] **Step 1: Update the test file first**

Replace the entire content of `app/workflow/action/restore/pitr/create/delete_compaction_test.go` with:

```go
package create

import (
	"context"
	"testing"

	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	fastflow_utils "github.com/shiningrush/fastflow/pkg/utils"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	mock_backup_model "github.com/tidbcloud/dataflow-service/app/models/backup/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	mock_restore_repo "github.com/tidbcloud/dataflow-service/app/repo/restore/mock"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
)

type DeleteCompactionSuite struct {
	suite.Suite
}

func TestDeleteCompaction(t *testing.T) {
	suite.Run(t, new(DeleteCompactionSuite))
}

// newExecCtxWithVars builds an ExecuteContext with the given DAG vars for testing.
func newExecCtxWithVars(vars map[string]string) run.ExecuteContext {
	ctx := fastflow_entity.CtxWithRunningTaskIns(context.Background(), &fastflow_entity.TaskInstance{DagInsID: "test-dag"})
	getter := fastflow_utils.KeyValueGetter(func(key string) (string, bool) {
		v, ok := vars[key]
		return v, ok
	})
	return run.NewDefExecuteContext(ctx, nil, nil, getter, nil)
}

func (s *DeleteCompactionSuite) TestRun() {
	tests := map[string]struct {
		restoreExtraInfo      *entity.RestoreExtra
		syncDeleteMigrErr     bool
		asyncDeleteErr        bool
		expectSyncMigrations  bool
		expectAsyncDelete     bool
		wantErr               bool
	}{
		"skips all when no compaction associated": {
			restoreExtraInfo:     nil,
			expectSyncMigrations: false,
			expectAsyncDelete:    false,
			wantErr:              false,
		},
		"skips all when compact backup id is zero": {
			restoreExtraInfo:     &entity.RestoreExtra{CompactBackupID: 0},
			expectSyncMigrations: false,
			expectAsyncDelete:    false,
			wantErr:              false,
		},
		"fail-fast when SyncDeleteMigrations returns error": {
			restoreExtraInfo:     &entity.RestoreExtra{CompactBackupID: 99},
			syncDeleteMigrErr:    true,
			expectSyncMigrations: true,
			expectAsyncDelete:    false,
			wantErr:              true,
		},
		"calls AsyncDelete after SyncDeleteMigrations succeeds, non-fatal on AsyncDelete error": {
			restoreExtraInfo:     &entity.RestoreExtra{CompactBackupID: 99},
			syncDeleteMigrErr:    false,
			asyncDeleteErr:       true,
			expectSyncMigrations: true,
			expectAsyncDelete:    true,
			wantErr:              false,
		},
		"happy path: both calls succeed": {
			restoreExtraInfo:     &entity.RestoreExtra{CompactBackupID: 99},
			syncDeleteMigrErr:    false,
			asyncDeleteErr:       false,
			expectSyncMigrations: true,
			expectAsyncDelete:    true,
			wantErr:              false,
		},
	}

	for name, tc := range tests {
		s.Run(name, func() {
			restoreRepo := mock_restore_repo.NewRestoreRepo(s.T())
			compactModel := mock_backup_model.NewCompactBackupModel(s.T())

			restore := &entity.Restore{}
			restore.ID = 1
			restore.ExtraInfo = tc.restoreExtraInfo
			restoreRepo.On("Load", mock.Anything, uint64(1)).Return(restore, nil)

			if tc.expectSyncMigrations {
				if tc.syncDeleteMigrErr {
					compactModel.On("SyncDeleteMigrations", mock.Anything, uint64(99)).
						Return(bizerr_wrapper.NewInternalErr("storage error"))
				} else {
					compactModel.On("SyncDeleteMigrations", mock.Anything, uint64(99)).
						Return(nil)
				}
			}

			if tc.expectAsyncDelete {
				req := &domain.AsyncDeleteCompactBackupReq{CompactBackupID: 99}
				if tc.asyncDeleteErr {
					compactModel.On("AsyncDelete", mock.Anything, req).
						Return(nil, bizerr_wrapper.NewInternalErr("internal error"))
				} else {
					compactModel.On("AsyncDelete", mock.Anything, req).
						Return(&domain.AsyncDeleteCompactBackupResp{}, nil)
				}
			}

			vars := map[string]string{workflow.KeyRestoreID: "1"}
			execCtx := newExecCtxWithVars(vars)

			act := NewDeleteCompactionForPiTR(restoreRepo, compactModel)
			err := act.Run(execCtx, nil)
			if tc.wantErr {
				s.Error(err)
			} else {
				s.NoError(err)
			}
		})
	}
}
```

- [ ] **Step 2: Run the test to verify it fails (expected)**

```bash
go test ./app/workflow/action/restore/pitr/create/... -run TestDeleteCompaction -v 2>&1 | tail -20
```
Expected: compilation error because `NewDeleteCompactionForPiTR` still takes four args and `RunAfter` still exists.

- [ ] **Step 3: Rewrite `delete_compaction.go`**

Replace the entire content of `app/workflow/action/restore/pitr/create/delete_compaction.go` with:

```go
package create

import (
	"github.com/shiningrush/fastflow/pkg/entity/run"
	backupModel "github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	restoreRepo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"go.uber.org/zap"
)

type DeleteCompactionForPiTR struct {
	restoreRepo        restoreRepo.RestoreRepo
	compactBackupModel backupModel.CompactBackupModel
}

func NewDeleteCompactionForPiTR(
	restoreRepo restoreRepo.RestoreRepo,
	compactBackupModel backupModel.CompactBackupModel,
) *DeleteCompactionForPiTR {
	return &DeleteCompactionForPiTR{
		restoreRepo:        restoreRepo,
		compactBackupModel: compactBackupModel,
	}
}

func (a *DeleteCompactionForPiTR) Name() string {
	return workflow.ActionDeleteCompactionForPiTR
}

func (a *DeleteCompactionForPiTR) ParameterNew() interface{} {
	return nil
}

func (a *DeleteCompactionForPiTR) Run(ctx run.ExecuteContext, _ interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())

	restoreID, err := action.GetUint64FromRunVar(ctx, workflow.KeyRestoreID)
	if err != nil {
		return err
	}
	logger.Info("action run start", zap.Uint64("restore_id", restoreID))

	restore, bErr := a.restoreRepo.Load(ctx.Context(), restoreID)
	if bErr != nil {
		logger.Warn("failed to load restore, skipping compaction deletion", zap.Error(bErr))
		return nil
	}

	if restore.ExtraInfo == nil || restore.ExtraInfo.CompactBackupID == 0 {
		logger.Info("no compaction associated with this restore, skipping")
		return nil
	}

	compactBackupID := restore.ExtraInfo.CompactBackupID

	// Step 1: synchronously delete migrations/ (fast, must complete before restore becomes available).
	if bErr := a.compactBackupModel.SyncDeleteMigrations(ctx.Context(), compactBackupID); bErr != nil {
		logger.Error("failed to delete migrations, will retry",
			zap.Uint64("compact_backup_id", compactBackupID), zap.Error(bErr))
		return bErr
	}

	// Step 2: fire-and-forget deletion of compactions/ subdirectory (may be large).
	_, asyncErr := a.compactBackupModel.AsyncDelete(ctx.Context(), &domain.AsyncDeleteCompactBackupReq{
		CompactBackupID: compactBackupID,
	})
	if asyncErr != nil {
		logger.Warn("failed to trigger compaction deletion (non-fatal)",
			zap.Uint64("compact_backup_id", compactBackupID),
			zap.Error(asyncErr))
	}

	logger.Info("action run finished")
	return nil
}
```

- [ ] **Step 4: Run the test to verify it passes**

```bash
go test ./app/workflow/action/restore/pitr/create/... -run TestDeleteCompaction -v 2>&1 | tail -20
```
Expected: all test cases pass.

- [ ] **Step 5: Commit**

```bash
git add app/workflow/action/restore/pitr/create/delete_compaction.go \
        app/workflow/action/restore/pitr/create/delete_compaction_test.go
git commit -m "feat(restore): DeleteCompactionForPiTR syncs migrations then fires async compactions delete

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"
```

---

### Task 6: Update DAG wiring for `DeleteCompactionForPiTR`

**Files:**
- Modify: `app/workflow/dag/restore/restore_pitr.go`
- Modify: `app/workflow/dag/restore/recycle_restore_pitr.go`

- [ ] **Step 1: Update `restore_pitr.go`**

Find the `NewDeleteCompactionForPiTR` call (around line 72) and replace:
```go
action.MustNewActionWrapperV2(
    pitrCreate.NewDeleteCompactionForPiTR(
        container.Global().RestoreRepo,
        container.Global().CompactBackupModel,
        container.Global().CompactBackupRepo,
        container.Global().WorkflowModel,
    ),
    action.SetRunAfterRetryFun(action.InvokeRunAfterWithExponentialRetryFactory(15*time.Second, 1*time.Minute, 0)),
),
```
with:
```go
pitrCreate.NewDeleteCompactionForPiTR(
    container.Global().RestoreRepo,
    container.Global().CompactBackupModel,
),
```

Remove the `"time"` import from `restore_pitr.go` if it is no longer used after this change. Use `goimports` or check manually.

- [ ] **Step 2: Update `recycle_restore_pitr.go`**

Find the same pattern (around line 46) and apply the same replacement:
```go
action.MustNewActionWrapperV2(
    pitrCreate.NewDeleteCompactionForPiTR(
        container.Global().RestoreRepo,
        container.Global().CompactBackupModel,
        container.Global().CompactBackupRepo,
        container.Global().WorkflowModel,
    ),
    action.SetRunAfterRetryFun(action.InvokeRunAfterWithExponentialRetryFactory(15*time.Second, 1*time.Minute, 0)),
),
```
with:
```go
pitrCreate.NewDeleteCompactionForPiTR(
    container.Global().RestoreRepo,
    container.Global().CompactBackupModel,
),
```

Remove unused imports (`"time"`, `action.SetRunAfterRetryFun` etc.) from the file as needed.

- [ ] **Step 3: Verify the build**

```bash
make build-dataflow 2>&1 | tail -5
```
Expected: build succeeds with no unused import errors.

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

```bash
go test ./app/workflow/... -count=1 2>&1 | tail -20
```
Expected: all tests pass.

- [ ] **Step 5: Commit**

```bash
git add app/workflow/dag/restore/restore_pitr.go \
        app/workflow/dag/restore/recycle_restore_pitr.go
git commit -m "chore(dag): remove RunAfter wrapper from DeleteCompactionForPiTR wiring

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"
```

---

### Task 7: Final verification

- [ ] **Step 1: Run the full unit test suite**

```bash
make app-unit-test-local 2>&1 | tail -30
```
Expected: all tests pass, no compilation errors.

- [ ] **Step 2: Verify no dead code left in purge action old imports**

```bash
grep -n "logBackupRepo\|bucketRepo\|encryptor\|logid\|azblob\|blob\|compactionSubDirs" \
    app/workflow/action/backup/compact/purge_compaction_storage.go
```
Expected: no matches (all storage logic has moved to the model).

- [ ] **Step 3: Verify `DeleteCompactionForPiTR` has no `RunAfter`, `dagWaiter`, `compactBackupRepo`**

```bash
grep -n "RunAfter\|dagWaiter\|compactBackupRepo\|DagCanceler\|worflow_helper" \
    app/workflow/action/restore/pitr/create/delete_compaction.go
```
Expected: no matches.

- [ ] **Step 4: Final commit if any formatting fixes needed**

```bash
make fmt
git add -u
git commit -m "chore: apply gofmt fixes

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>" --allow-empty
```