# PiTR Purge Compaction — 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:** After a compaction-accelerated PiTR restore finishes, automatically clean up the compaction files (`v1/compactions`, `v1/migrations` under the log backup prefix) that were created on cloud storage, and track the lifecycle via new entity statuses and Prometheus metrics.

**Architecture:** A thin fire-and-forget `ActionDeleteCompactionForPiTR` is added to the PiTR restore DAG (never blocks restore — always returns nil). It calls `compactBackupModel.AsyncDelete()` which launches `DagDeleteCompactBackup`. That DAG is extended with two new sequential steps: `ActionPurgeCompactionStorage` (deletes the two subdirectories from cloud storage) and `ActionDeleteCompactBackupFinalStep` (marks entity status = `deleted`). Two new entity statuses (`deleting`, `deleted`) and Prometheus metrics track the state.

**Tech Stack:** Go, GORM, fastflow DAG engine, prometheus/client_golang, testify suites

---

## File Map

| File | Action |
|------|--------|
| `app/repo/entity/constant.go` | Add `CompactBackupStatusDeleting`, `CompactBackupStatusDeleted`; update `IsFinishedStatus()` |
| `app/repo/entity/entity.go` | Extend GORM ENUM tag for `CompactBackup.Status` |
| `app/repo/entity/entity_hook_manager.go` | Add `CompactBackupDisplayType` constant |
| `app/repo/entity/entity_hook.go` | Add `CompactBackup.BeforeSave` + `AfterSave` |
| `migrate/record_202605181000_compact_backup_status_deleting.go` | DB migration: extend ENUM |
| `app/workflow/constants.go` | Add 3 new action constants |
| `app/workflow/dag/register.go` | Register `DagDeleteCompactBackup` (was missing — critical bug fix) |
| `app/repo/backup/backup.go` | Add `Statuses []entity.CompactBackupStatus` and `IDs []entity.ID` to `CompactBackupFilter` |
| `app/repo/backup/backup_impl.go` | Update `buildTxCompactBackupFilter` for `Statuses` and `IDs` |
| `app/models/backup/impl/compact_backup_model.go` | Add `status=deleting` + idempotency guard in `AsyncDelete` |
| `app/workflow/action/restore/pitr/create/delete_compaction.go` | New — thin fire-and-forget `ActionDeleteCompactionForPiTR` |
| `app/workflow/action/backup/compact/purge_compaction_storage.go` | New — deletes `v1/compactions` + `v1/migrations` from cloud storage |
| `app/workflow/action/backup/compact/delete_compact_backup_final_step.go` | New — marks entity `status=deleted` |
| `app/workflow/dag/backup/compact/create.go` | Extend `GetDeleteDag` with 2 new tasks + action constructors |
| `app/workflow/dag/restore/restore_pitr.go` | Add `ActionDeleteCompactionForPiTR` task; update `ActionDeleteClusterJobForRestoring.DependOn` |
| `app/metrics/compact_backup/metric/metric.go` | New — GaugeVec + status int map |
| `app/metrics/compact_backup/hooks/status.go` | New — AfterSave metrics hooker |
| `app/metrics/compact_backup/collector/collector.go` | New — Prometheus collector |
| `app/metrics/register/register.go` | Register new gauge + collector |
| `app/container/container.go` | Register compact backup metrics hooker in `initDBHooks` |

---

### Task 1: Entity status constants + GORM enum

**Files:**
- Modify: `app/repo/entity/constant.go:188-199`
- Modify: `app/repo/entity/entity.go:294`

- [ ] **Step 1: Add two new statuses and update `IsFinishedStatus`**

In `app/repo/entity/constant.go`, replace:
```go
const (
	CompactBackupStatusPending CompactBackupStatus = "pending"
	CompactBackupStatusRunning CompactBackupStatus = "running"
	CompactBackupStatusSuccess CompactBackupStatus = "success"
	CompactBackupStatusFailed  CompactBackupStatus = "failed"
)

func (s CompactBackupStatus) IsFinishedStatus() bool {
	return s == CompactBackupStatusSuccess || s == CompactBackupStatusFailed
}
```
with:
```go
const (
	CompactBackupStatusPending  CompactBackupStatus = "pending"
	CompactBackupStatusRunning  CompactBackupStatus = "running"
	CompactBackupStatusSuccess  CompactBackupStatus = "success"
	CompactBackupStatusFailed   CompactBackupStatus = "failed"
	CompactBackupStatusDeleting CompactBackupStatus = "deleting"
	CompactBackupStatusDeleted  CompactBackupStatus = "deleted"
)

func (s CompactBackupStatus) IsFinishedStatus() bool {
	return s == CompactBackupStatusSuccess ||
		s == CompactBackupStatusFailed ||
		s == CompactBackupStatusDeleted
}
```

- [ ] **Step 2: Extend GORM ENUM tag**

In `app/repo/entity/entity.go` at line 294, change:
```go
Status      CompactBackupStatus `gorm:"type:ENUM('pending','running','success','failed');not null"`
```
to:
```go
Status      CompactBackupStatus `gorm:"type:ENUM('pending','running','success','failed','deleting','deleted');not null"`
```

- [ ] **Step 3: Build to verify**

```bash
make build-dataflow-darwin   # macOS
# or
make build-dataflow          # Linux
```
Expected: compiles with no errors.

- [ ] **Step 4: Commit**

```bash
git add app/repo/entity/constant.go app/repo/entity/entity.go
git commit -m "feat(compact): add deleting/deleted status constants and extend GORM enum

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

---

### Task 2: DB migration

**Files:**
- Create: `migrate/record_202605181000_compact_backup_status_deleting.go`

- [ ] **Step 1: Create migration file**

```go
package migrate

import "gorm.io/gorm"

var _ = AddDataflowDBMigrateRecord(
	"202605181000",
	func(db *gorm.DB) error {
		return db.Exec(
			"ALTER TABLE compact_backups MODIFY COLUMN status " +
				"ENUM('pending','running','success','failed','deleting','deleted') NOT NULL",
		).Error
	},
)
```

- [ ] **Step 2: Build**

```bash
make build-dataflow-darwin   # macOS
# or
make build-dataflow          # Linux
```
Expected: compiles with no errors.

- [ ] **Step 3: Commit**

```bash
git add migrate/record_202605181000_compact_backup_status_deleting.go
git commit -m "feat(compact): db migration - extend compact_backups status enum

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

---

### Task 3: Workflow constants

**Files:**
- Modify: `app/workflow/constants.go:111-114`

- [ ] **Step 1: Add three new action constants after the existing compact backup block (line 114)**

```go
	// action for CompactBackup
	ActionCreateCompactBackupTask = "create_compact_backup_task"
	ActionSyncCompactBackupStatus = "sync_compact_backup_status"
	ActionCleanupCompactBackup    = "cleanup_compact_backup"
	// delete pipeline (new)
	ActionPurgeCompactionStorage       = "purge_compaction_storage"
	ActionDeleteCompactBackupFinalStep = "delete_compact_backup_final_step"
```

Also, after the `ActionRunCompactBackupForPiTR` constant in the restore-pit-backup block (line 131), add:
```go
	// action for restore pit backup
	ActionRunCompactBackupForPiTR   = "run_compact_backup_for_pitr"
	ActionDeleteCompactionForPiTR   = "delete_compaction_for_pitr"  // new
	ActionRunPitRestoring           = "run_pit_restoring"
	ActionRestorePitBackupFinalStep = "restore_pit_backup_final_step"
```

- [ ] **Step 2: Build**

```bash
make build-dataflow-darwin   # macOS
# or
make build-dataflow          # Linux
```

- [ ] **Step 3: Commit**

```bash
git add app/workflow/constants.go
git commit -m "feat(compact): add workflow action constants for purge compaction pipeline

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

---

### Task 4: Register DagDeleteCompactBackup (critical bug fix)

**Files:**
- Modify: `app/workflow/dag/register.go:102-104`

- [ ] **Step 1: Add `DagDeleteCompactBackup` registration after `DagCreateCompactBackup` at line 102-104**

```go
	if err := workflowModel.RegisterDag(dag_backup_compact.GetCreateDag()); err != nil {
		return err
	}
	if err := workflowModel.RegisterDag(dag_backup_compact.GetDeleteDag()); err != nil {
		return err
	}
```

- [ ] **Step 2: Build**

```bash
make build-dataflow-darwin   # macOS
# or
make build-dataflow          # Linux
```

- [ ] **Step 3: Commit**

```bash
git add app/workflow/dag/register.go
git commit -m "fix(compact): register DagDeleteCompactBackup in workflow engine

Previously AsyncDelete would call RunDagWithTags(DagDeleteCompactBackup, ...)
but the DAG was never registered, causing a silent failure.

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

---

### Task 5: CompactBackupFilter extension

**Files:**
- Modify: `app/repo/backup/backup.go:204-208`
- Modify: `app/repo/backup/backup_impl.go:971-991`

- [ ] **Step 1: Add `Statuses` and `IDs` fields to `CompactBackupFilter`**

In `app/repo/backup/backup.go`, replace:
```go
type CompactBackupFilter struct {
	ClusterID   *uint64
	LogBackupID *uint64
	Status      *entity.CompactBackupStatus
}
```
with:
```go
type CompactBackupFilter struct {
	ClusterID   *uint64
	LogBackupID *uint64
	Status      *entity.CompactBackupStatus
	Statuses    []entity.CompactBackupStatus
	IDs         []entity.ID
}
```

- [ ] **Step 2: Update `buildTxCompactBackupFilter` in `backup_impl.go`**

After the existing `filter.Status != nil` block (line 986-988), add:
```go
	if len(filter.Statuses) > 0 {
		tx = tx.Where("status IN ?", filter.Statuses)
	}

	if len(filter.IDs) > 0 {
		tx = tx.Where("id IN ?", filter.IDs)
	}
```

- [ ] **Step 3: Build**

```bash
make build-dataflow-darwin
```

- [ ] **Step 4: Commit**

```bash
git add app/repo/backup/backup.go app/repo/backup/backup_impl.go
git commit -m "feat(compact): extend CompactBackupFilter with Statuses and IDs fields

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

---

### Task 6: AsyncDelete update — set status + idempotency

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

- [ ] **Step 1: Write the failing test**

Create or add to `app/models/backup/impl/compact_backup_model_test.go`:

```go
package impl_test

import (
	"context"
	"testing"

	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/dataflow-service/app/models/backup/impl"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"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"
	mock_workflow "github.com/tidbcloud/dataflow-service/app/workflow/mock"
	bizerr "github.com/tidbcloud/pkgv2/bizerr"
)

type AsyncDeleteSuite struct {
	suite.Suite
}

func TestAsyncDelete(t *testing.T) {
	suite.Run(t, new(AsyncDeleteSuite))
}

func (s *AsyncDeleteSuite) TestSkipWhenDeleted() {
	compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())
	workflowCli := mock_workflow.NewWorkflow(s.T())

	cb := &entity.CompactBackup{}
	cb.ID = 42
	cb.Status = entity.CompactBackupStatusDeleted

	compactBackupRepo.EXPECT().Load(context.Background(), uint64(42)).Return(cb, nil)
	// workflowCli.RunDagWithTags must NOT be called

	model := impl.NewCompactBackupModel(nil, compactBackupRepo, nil, nil, workflowCli, nil)
	resp, err := model.AsyncDelete(context.Background(), &domain.AsyncDeleteCompactBackupReq{
		CompactBackupID: 42,
	})
	s.NoError(err)
	s.Nil(resp)
}

func (s *AsyncDeleteSuite) TestSetsDeletingBeforeLaunchingDAG() {
	compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())
	workflowCli := mock_workflow.NewWorkflow(s.T())

	cb := &entity.CompactBackup{}
	cb.ID = 42
	cb.ClusterID = 100
	cb.Status = entity.CompactBackupStatusSuccess

	compactBackupRepo.EXPECT().Load(context.Background(), uint64(42)).Return(cb, nil)
	// Expect status to be set to deleting and saved
	compactBackupRepo.EXPECT().Save(context.Background(), cb).RunAndReturn(func(_ context.Context, saved *entity.CompactBackup) bizerr.BizError {
		s.Equal(entity.CompactBackupStatusDeleting, saved.Status)
		return nil
	})
	dagInst := &fastflow_entity.DagInstance{BaseInfo: fastflow_entity.BaseInfo{ID: "dag-123"}}
	workflowCli.EXPECT().RunDagWithTags(workflow.DagDeleteCompactBackup, mock.Anything, mock.Anything).Return(dagInst, nil)

	model := impl.NewCompactBackupModel(nil, compactBackupRepo, nil, nil, workflowCli, nil)
	resp, err := model.AsyncDelete(context.Background(), &domain.AsyncDeleteCompactBackupReq{
		CompactBackupID: 42,
	})
	s.NoError(err)
	s.NotNil(resp)
	s.Equal("dag-123", resp.WorkflowID)
}
```

- [ ] **Step 2: Run test to see it fail**

```bash
cd app && go test ./models/backup/impl/... -run TestAsyncDelete -v 2>&1 | tail -20
```
Expected: FAIL (method `AsyncDelete` doesn't set status yet).

- [ ] **Step 3: Update `AsyncDelete` to set status=deleting + guard**

In `app/models/backup/impl/compact_backup_model.go`, replace the entire `AsyncDelete` method:

```go
func (c *CompactBackupModel) AsyncDelete(ctx context.Context,
	req *domain.AsyncDeleteCompactBackupReq,
) (*domain.AsyncDeleteCompactBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("CompactBackupModel.AsyncDelete").
		With(zap.Uint64("compact_backup_id", req.CompactBackupID))

	compactBackup, err := c.compactBackupRepo.Load(ctx, req.CompactBackupID)
	if err != nil {
		logger.Error("failed to load compact backup", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, 0)
	}

	// Idempotency: already fully deleted — nothing to do.
	if compactBackup.Status == entity.CompactBackupStatusDeleted {
		logger.Info("compact backup already deleted, skipping")
		return nil, nil
	}

	// Transition to deleting (skip if already in progress to avoid unnecessary writes,
	// but still re-trigger the DAG so retries can resume where they left off).
	if compactBackup.Status != entity.CompactBackupStatusDeleting {
		compactBackup.Status = entity.CompactBackupStatusDeleting
		if saveErr := c.compactBackupRepo.Save(ctx, compactBackup); saveErr != nil {
			logger.Error("failed to set compact backup status to deleting", zap.Error(saveErr))
			return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
		}
	}

	vars := map[string]string{
		workflow.KeyCompactBackupID: strconv.FormatUint(req.CompactBackupID, 10),
	}
	tags := map[string]string{
		workflow.TagClusterID:       strconv.FormatUint(compactBackup.ClusterID.Uint64(), 10),
		workflow.TagCompactBackupID: strconv.FormatUint(req.CompactBackupID, 10),
	}

	dagInstance, runErr := c.workflowCli.RunDagWithTags(workflow.DagDeleteCompactBackup, vars, tags)
	if runErr != nil {
		logger.Error("failed to start delete compact backup dag", zap.Error(runErr))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, compactBackup.ClusterID.Uint64())
	}

	logger.Info("compact backup delete dag started", zap.String("workflow_id", dagInstance.ID))

	return &domain.AsyncDeleteCompactBackupResp{
		WorkflowID: dagInstance.ID,
	}, nil
}
```

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

```bash
cd app && go test ./models/backup/impl/... -run TestAsyncDelete -v 2>&1 | tail -20
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add app/models/backup/impl/compact_backup_model.go
git add app/models/backup/impl/compact_backup_model_test.go
git commit -m "feat(compact): AsyncDelete sets status=deleting + idempotency guard

- Skip entirely if already deleted
- Set status=deleting before launching DAG (re-trigger if already deleting)

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

---

### Task 7: New action — ActionDeleteCompactionForPiTR

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

This action is a thin fire-and-forget in the PiTR DAG. It calls `compactBackupModel.AsyncDelete()` for the compact backup linked to the restore. If the restore has no `CompactBackupID` (non-compaction PiTR), it skips silently. Errors from `AsyncDelete` are only logged — never returned.

- [ ] **Step 1: Write the failing test**

Create `app/workflow/action/restore/pitr/create/delete_compaction_test.go`:

```go
package create

import (
	"context"
	"testing"

	"github.com/shiningrush/fastflow/pkg/entity/run"
	"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"
	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))
}

func (s *DeleteCompactionSuite) TestSkipsNonCompactionPiTR() {
	restoreRepo := mock_restore_repo.NewRestoreRepo(s.T())
	compactModel := mock_backup_model.NewCompactBackupModel(s.T())

	restore := &entity.Restore{}
	restore.ID = 1
	restore.ExtraInfo = nil // no compaction

	restoreRepo.EXPECT().Load(mock.Anything, uint64(1)).Return(restore, nil)
	// compactModel.AsyncDelete must NOT be called

	action := NewDeleteCompactionForPiTR(restoreRepo, compactModel)
	ctx := run.NewDefExecuteContext(context.Background(), nil, nil, nil, nil)

	err := action.Run(ctx, &DeleteCompactionForPiTRParam{RestoreID: 1})
	s.NoError(err)
}

func (s *DeleteCompactionSuite) TestCallsAsyncDeleteAndReturnsNilOnError() {
	restoreRepo := mock_restore_repo.NewRestoreRepo(s.T())
	compactModel := mock_backup_model.NewCompactBackupModel(s.T())

	restore := &entity.Restore{}
	restore.ID = 2
	restore.ExtraInfo = &entity.RestoreExtra{CompactBackupID: 99}

	restoreRepo.EXPECT().Load(mock.Anything, uint64(2)).Return(restore, nil)
	compactModel.EXPECT().AsyncDelete(mock.Anything, &domain.AsyncDeleteCompactBackupReq{
		CompactBackupID: 99,
	}).Return(nil, bizerr_wrapper.NewInternalErr("internal error", 0))
	// Even though AsyncDelete errors, Run must return nil

	action := NewDeleteCompactionForPiTR(restoreRepo, compactModel)
	ctx := run.NewDefExecuteContext(context.Background(), nil, nil, nil, nil)

	err := action.Run(ctx, &DeleteCompactionForPiTRParam{RestoreID: 2})
	s.NoError(err) // fire-and-forget: always returns nil
}
```

- [ ] **Step 2: Run test to verify it fails**

```bash
cd app && go test ./workflow/action/restore/pitr/create/... -run TestDeleteCompaction -v 2>&1 | tail -20
```
Expected: FAIL (type not defined yet).

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

```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 DeleteCompactionForPiTRParam struct {
	RestoreID uint64 `json:"restore_id"`
}

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 &DeleteCompactionForPiTRParam{}
}

func (a *DeleteCompactionForPiTR) Run(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	input := params.(*DeleteCompactionForPiTRParam)
	logger.Info("action run start", zap.Uint64("restore_id", input.RestoreID))

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

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

	_, asyncErr := a.compactBackupModel.AsyncDelete(ctx.Context(), &domain.AsyncDeleteCompactBackupReq{
		CompactBackupID: restore.ExtraInfo.CompactBackupID,
	})
	if asyncErr != nil {
		logger.Warn("failed to trigger compaction deletion (non-fatal)",
			zap.Uint64("compact_backup_id", restore.ExtraInfo.CompactBackupID),
			zap.Error(asyncErr))
	}

	logger.Info("action run finished")
	return nil // fire-and-forget: never block restore
}
```

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

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

- [ ] **Step 5: Commit**

```bash
git add app/workflow/action/restore/pitr/create/delete_compaction.go
git add app/workflow/action/restore/pitr/create/delete_compaction_test.go
git commit -m "feat(compact): add ActionDeleteCompactionForPiTR - fire-and-forget in PiTR DAG

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

---

### Task 8: New action — ActionPurgeCompactionStorage

**Files:**
- Create: `app/workflow/action/backup/compact/purge_compaction_storage.go`

This action runs inside `DagDeleteCompactBackup`. It loads the `CompactBackup` entity, resolves the cloud storage URI, and deletes the `v1/compactions` and `v1/migrations` subdirectories.

URI resolution rules (same as `CreateCompactBackupTaskAction.getStorageURI`):
- `compactBackup.BucketID == 0`: uri = `logBackup.Storage`, bucket = `logBackup.BucketID`
- `compactBackup.BucketID != 0`: uri = `bucket.URIInBucket(logBackup.Storage)`, bucket = `compactBackup.BucketID`

Azure check: if `bucket.InfraProviderType == AKS || Azure`, convert uri via `azblob`.

- [ ] **Step 1: Write the failing test**

Create `app/workflow/action/backup/compact/purge_compaction_storage_test.go`:

```go
package compact

import (
	"testing"

	"github.com/stretchr/testify/suite"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
)

type PurgeCompactionStorageSuite struct {
	suite.Suite
}

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

// TestActionName verifies the action name constant is correct.
func (s *PurgeCompactionStorageSuite) TestActionName() {
	compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())
	logBackupRepo := repo_mock.NewLogBackupRepo(s.T())
	bucketRepo := repo_mock.NewBucketRepo(s.T())

	action := NewPurgeCompactionStorageAction(compactBackupRepo, logBackupRepo, bucketRepo, nil)
	s.Equal("purge_compaction_storage", action.Name())
}
```

Note: Full blob storage deletion is exercised by integration tests; unit tests here cover construction and name validation. Actual delete logic is tested end-to-end via `make app-unit-test-local`.

Note: Full blob storage mocking is complex due to package-level `blob.NewStorage`. The test primarily validates struct construction and action name. Integration-level testing of actual purge is covered by the `make app-unit-test-local` test suite.

- [ ] **Step 2: Create `purge_compaction_storage.go`**

```go
package compact

import (
	"context"
	"fmt"
	"strconv"

	"github.com/pingcap/errors"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backup_repo "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"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/pkg/azblob"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	"github.com/tidbcloud/dataflow-service/pkg/s3"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	"go.uber.org/zap"
)

// compactionSubDirs are the directories created by TiKV compaction under the log backup prefix.
var compactionSubDirs = []string{"v1/compactions", "v1/migrations"}

type PurgeCompactionStorageAction struct {
	compactBackupRepo backup_repo.CompactBackupRepo
	logBackupRepo     backup_repo.LogBackupRepo
	bucketRepo        repo.BucketRepo
	encryptor         encryptor_wrapper.Encryptor
}

func NewPurgeCompactionStorageAction(
	compactBackupRepo backup_repo.CompactBackupRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	bucketRepo repo.BucketRepo,
	encryptor encryptor_wrapper.Encryptor,
) *PurgeCompactionStorageAction {
	return &PurgeCompactionStorageAction{
		compactBackupRepo: compactBackupRepo,
		logBackupRepo:     logBackupRepo,
		bucketRepo:        bucketRepo,
		encryptor:         encryptor,
	}
}

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")

	compactBackupIDStr, ok := ctx.ShareData().Get(workflow.KeyCompactBackupID)
	if !ok {
		return fmt.Errorf("compact backup id not found in share data")
	}
	compactBackupID, err := strconv.ParseUint(compactBackupIDStr, 10, 64)
	if err != nil {
		return fmt.Errorf("invalid compact backup id: %s", compactBackupIDStr)
	}

	if err := a.purge(ctx.Context(), logger, compactBackupID); err != nil {
		return err
	}
	logger.Info("action run finished")
	return nil
}

func (a *PurgeCompactionStorageAction) purge(
	ctx context.Context, logger *zap.Logger, compactBackupID uint64,
) error {
	compactBackup, bErr := a.compactBackupRepo.Load(ctx, compactBackupID)
	if bErr != nil {
		return fmt.Errorf("load compact backup %d: %w", compactBackupID, bErr)
	}

	logBackup, bErr := a.logBackupRepo.Load(ctx, compactBackup.LogBackupID.Uint64())
	if bErr != nil {
		return fmt.Errorf("load log backup %d: %w", compactBackup.LogBackupID.Uint64(), bErr)
	}

	storageURI, bucket, err := a.resolveStorage(ctx, compactBackup, logBackup)
	if err != nil {
		return err
	}

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

	if bucket.InfraProviderType == entity.InfraProviderTypeAKS ||
		bucket.InfraProviderType == entity.InfraProviderTypeAzure {
		storageURI, err = a.getAzureBlobURI(bucket, storageURI)
		if err != nil {
			return fmt.Errorf("get azure blob uri: %w", err)
		}
	}

	deleteOpt := &blob.DeleteOption{Concurrency: config.Global().BackupConfig.CleanupDeleteConcurrency}
	for _, subDir := range compactionSubDirs {
		s, blobErr := blob.NewStorage(ctx, storageURI+"/"+subDir, opt)
		if blobErr != nil {
			if blob.IsNotFoundError(blobErr) {
				logger.Info("compaction subdir storage not found, skipping", zap.String("subdir", subDir))
				continue
			}
			return fmt.Errorf("init storage for subdir %s: %w", subDir, blobErr)
		}
		deleteErr := s.DeleteSubDir(ctx, "", deleteOpt)
		if deleteErr != nil && !s3.IsNotFoundError(deleteErr) && !blob.IsNotFoundError(deleteErr) {
			return fmt.Errorf("delete compaction subdir %s: %w", subDir, deleteErr)
		}
		logger.Info("deleted compaction subdir", zap.String("subdir", subDir))
	}
	return nil
}

func (a *PurgeCompactionStorageAction) resolveStorage(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
) (string, *entity.Bucket, error) {
	if compactBackup.BucketID == 0 {
		bucket, bErr := a.bucketRepo.Load(ctx, logBackup.BucketID.Uint64())
		if bErr != nil {
			return "", nil, fmt.Errorf("load bucket %d: %w", logBackup.BucketID.Uint64(), bErr)
		}
		return logBackup.Storage, bucket, nil
	}

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

func (a *PurgeCompactionStorageAction) getAzureBlobURI(bucket *entity.Bucket, storageURL string) (string, error) {
	azurl, err := azblob.ParseAzblobURL(storageURL)
	if err != nil {
		return "", errors.Annotate(err, "parse azure blob url")
	}
	sasToken, err := bucket.GetSASToken(a.encryptor)
	if err != nil {
		return "", errors.Annotate(err, "get sas token")
	}
	azurl.SetSASToken(sasToken)
	return azurl.ToInnerURL(true), nil
}
```

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

```bash
cd app && go test ./workflow/action/backup/compact/... -run TestPurgeCompactionStorage -v 2>&1 | tail -20
```
Expected: PASS (struct construction test).

- [ ] **Step 4: Build**

```bash
make build-dataflow-darwin
```

- [ ] **Step 5: Commit**

```bash
git add app/workflow/action/backup/compact/purge_compaction_storage.go
git add app/workflow/action/backup/compact/purge_compaction_storage_test.go
git commit -m "feat(compact): add ActionPurgeCompactionStorage - deletes v1/compactions + v1/migrations

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

---

### Task 9: New action — ActionDeleteCompactBackupFinalStep

**Files:**
- Create: `app/workflow/action/backup/compact/delete_compact_backup_final_step.go`

This action is the last step in `DagDeleteCompactBackup`. It sets `CompactBackup.Status = deleted` and records the finished time.

- [ ] **Step 1: Write the failing test**

Create `app/workflow/action/backup/compact/delete_compact_backup_final_step_test.go`:

```go
package compact

import (
	"context"
	"testing"

	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
)

// Note: mockExecuteContext and mockShareData are defined in sync_compact_status_action_test.go
// and are shared across all test files in package compact.

type DeleteCompactBackupFinalStepSuite struct {
	suite.Suite
}

func TestDeleteCompactBackupFinalStep(t *testing.T) {
	suite.Run(t, new(DeleteCompactBackupFinalStepSuite))
}

func (s *DeleteCompactBackupFinalStepSuite) TestSetsStatusDeleted() {
	compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())

	cb := &entity.CompactBackup{}
	cb.ID = 42
	cb.Status = entity.CompactBackupStatusDeleting

	ctx := context.Background()
	ctx = fastflow_entity.CtxWithRunningTaskIns(ctx, &fastflow_entity.TaskInstance{DagInsID: "test-dag"})
	execCtx := newMockExecuteContext(ctx)
	execCtx.SetShareData(workflow.KeyCompactBackupID, "42")

	compactBackupRepo.EXPECT().Load(mock.Anything, uint64(42)).Return(cb, nil)
	compactBackupRepo.EXPECT().Save(mock.Anything, mock.MatchedBy(func(saved *entity.CompactBackup) bool {
		return saved.Status == entity.CompactBackupStatusDeleted && saved.FinishedAt != nil
	})).Return(nil)

	action := NewDeleteCompactBackupFinalStep(compactBackupRepo)
	s.Equal("delete_compact_backup_final_step", action.Name())

	err := action.Run(execCtx, nil)
	s.NoError(err)
}

func (s *DeleteCompactBackupFinalStepSuite) TestIdempotentWhenAlreadyDeleted() {
	compactBackupRepo := repo_mock.NewCompactBackupRepo(s.T())

	cb := &entity.CompactBackup{}
	cb.ID = 42
	cb.Status = entity.CompactBackupStatusDeleted

	ctx := context.Background()
	ctx = fastflow_entity.CtxWithRunningTaskIns(ctx, &fastflow_entity.TaskInstance{DagInsID: "test-dag"})
	execCtx := newMockExecuteContext(ctx)
	execCtx.SetShareData(workflow.KeyCompactBackupID, "42")

	compactBackupRepo.EXPECT().Load(mock.Anything, uint64(42)).Return(cb, nil)
	// Save must NOT be called when already deleted

	action := NewDeleteCompactBackupFinalStep(compactBackupRepo)
	err := action.Run(execCtx, nil)
	s.NoError(err)
}
```

- [ ] **Step 2: Run test to verify it fails**

```bash
cd app && go test ./workflow/action/backup/compact/... -run TestDeleteCompactBackupFinalStep -v 2>&1 | tail -20
```

- [ ] **Step 3: Create `delete_compact_backup_final_step.go`**

```go
package compact

import (
	"fmt"
	"strconv"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	backup_repo "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"
	"go.uber.org/zap"
)

type DeleteCompactBackupFinalStep struct {
	compactBackupRepo backup_repo.CompactBackupRepo
}

func NewDeleteCompactBackupFinalStep(
	compactBackupRepo backup_repo.CompactBackupRepo,
) *DeleteCompactBackupFinalStep {
	return &DeleteCompactBackupFinalStep{compactBackupRepo: compactBackupRepo}
}

func (a *DeleteCompactBackupFinalStep) Name() string {
	return workflow.ActionDeleteCompactBackupFinalStep
}

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

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

	compactBackupIDStr, ok := ctx.ShareData().Get(workflow.KeyCompactBackupID)
	if !ok {
		return fmt.Errorf("compact backup id not found in share data")
	}
	compactBackupID, err := strconv.ParseUint(compactBackupIDStr, 10, 64)
	if err != nil {
		return fmt.Errorf("invalid compact backup id: %s", compactBackupIDStr)
	}

	compactBackup, bErr := a.compactBackupRepo.Load(ctx.Context(), compactBackupID)
	if bErr != nil {
		logger.Error("failed to load compact backup", zap.Uint64("id", compactBackupID), zap.Error(bErr))
		return bErr
	}

	if compactBackup.Status == entity.CompactBackupStatusDeleted {
		logger.Info("compact backup already deleted, idempotent skip",
			zap.Uint64("id", compactBackupID))
		return nil
	}

	compactBackup.Status = entity.CompactBackupStatusDeleted
	compactBackup.FinishedAt = pointer.To(time.Now())
	if saveErr := a.compactBackupRepo.Save(ctx.Context(), compactBackup); saveErr != nil {
		logger.Error("failed to save compact backup as deleted", zap.Error(saveErr))
		return saveErr
	}

	logger.Info("compact backup marked as deleted", zap.Uint64("id", compactBackupID))
	return nil
}
```

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

```bash
cd app && go test ./workflow/action/backup/compact/... -run TestDeleteCompactBackupFinalStep -v 2>&1 | tail -20
```
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add app/workflow/action/backup/compact/delete_compact_backup_final_step.go
git add app/workflow/action/backup/compact/delete_compact_backup_final_step_test.go
git commit -m "feat(compact): add ActionDeleteCompactBackupFinalStep - marks entity as deleted

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

---

### Task 10: Extend GetDeleteDag + wire PiTR DAG

**Files:**
- Modify: `app/workflow/dag/backup/compact/create.go:70-98`
- Modify: `app/workflow/dag/restore/restore_pitr.go`

- [ ] **Step 1: Extend `GetDeleteDag` in `create.go`**

Add the two new action constructors and tasks. Replace the `GetDeleteDag` function:

```go
func GetDeleteDag() (*entity.Dag, []run.Action) {
	actions := []run.Action{
		compact.NewCleanupCompactBackupAction(
			container.Global().TransProvider,
			container.Global().CompactBackupRepo,
			container.Global().CompactRegionalFacade,
			container.Global().CoreGrpcClient,
		),
		compact.NewPurgeCompactionStorageAction(
			container.Global().CompactBackupRepo,
			container.Global().LogBackupRepo,
			container.Global().BucketRepo,
			container.Global().Encryptor,
		),
		compact.NewDeleteCompactBackupFinalStep(
			container.Global().CompactBackupRepo,
		),
	}

	dag := &entity.Dag{
		BaseInfo: entity.BaseInfo{
			ID: workflow.DagDeleteCompactBackup,
		},
		Name: workflow.DagDeleteCompactBackup,
		Vars: entity.DagVars{
			workflow.KeyCompactBackupID: {},
		},
		Status: entity.DagStatusNormal,
		Tasks: []entity.Task{
			{
				ID:          workflow.ActionCleanupCompactBackup,
				ActionName:  workflow.ActionCleanupCompactBackup,
				TimeoutSecs: 3 * 60,
			},
			{
				ID:          workflow.ActionPurgeCompactionStorage,
				ActionName:  workflow.ActionPurgeCompactionStorage,
				DependOn:    []string{workflow.ActionCleanupCompactBackup},
				TimeoutSecs: 10 * 60,
			},
			{
				ID:          workflow.ActionDeleteCompactBackupFinalStep,
				ActionName:  workflow.ActionDeleteCompactBackupFinalStep,
				DependOn:    []string{workflow.ActionPurgeCompactionStorage},
				TimeoutSecs: 60,
			},
		},
	}
	return dag, actions
}
```

- [ ] **Step 2: Add `ActionDeleteCompactionForPiTR` task to PiTR DAG**

In `app/workflow/dag/restore/restore_pitr.go`:

1. Add the constructor to the `actions` slice (after `restoreCreate.NewRestoreSqlBindings(...)`):
```go
		pitrCreate.NewDeleteCompactionForPiTR(
			container.Global().RestoreRepo,
			container.Global().CompactBackupModel,
		),
```

2. Add the new task between `ActionRestoreSqlBindings` (line ~162) and `ActionDeleteClusterJobForRestoring`:
```go
		{
			ID:         workflow.ActionDeleteCompactionForPiTR,
			ActionName: workflow.ActionDeleteCompactionForPiTR,
			Params: map[string]interface{}{
				workflow.KeyRestoreID: fmt.Sprintf("{{%v}}", workflow.KeyRestoreID),
			},
			DependOn:    []string{workflow.ActionRunPitRestoring},
			TimeoutSecs: 30,
		},
```

3. Update `ActionDeleteClusterJobForRestoring.DependOn` to include both parallel steps:
```go
		{
			ID:         workflow.ActionDeleteClusterJobForRestoring,
			ActionName: workflow.ActionDeleteClusterJobForRestoring,
			Params: map[string]interface{}{
				workflow.KeyRestoreID: fmt.Sprintf("{{%v}}", workflow.KeyRestoreID),
			},
			DependOn:    []string{workflow.ActionRestoreSqlBindings, workflow.ActionDeleteCompactionForPiTR},
			TimeoutSecs: 30,
		},
```

- [ ] **Step 3: Build**

```bash
make build-dataflow-darwin
```
Expected: compiles with no errors.

- [ ] **Step 4: Commit**

```bash
git add app/workflow/dag/backup/compact/create.go
git add app/workflow/dag/restore/restore_pitr.go
git commit -m "feat(compact): extend DagDeleteCompactBackup + wire ActionDeleteCompactionForPiTR in PiTR DAG

- DagDeleteCompactBackup: add PurgeCompactionStorage + DeleteCompactBackupFinalStep
- DagPitRestore: add ActionDeleteCompactionForPiTR (parallel to ActionRestoreSqlBindings)
- ActionDeleteClusterJobForRestoring now waits for both parallel branches

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

---

### Task 11: Entity hooks for CompactBackup

**Files:**
- Modify: `app/repo/entity/entity_hook_manager.go:14-20`
- Modify: `app/repo/entity/entity_hook.go`

- [ ] **Step 1: Add `CompactBackupDisplayType` constant**

In `app/repo/entity/entity_hook_manager.go`, add to the first `const` block:
```go
const (
	BackupDisplayType          DataflowEventType = "BackupCluster"
	BackupExportDisplayType    DataflowEventType = "ExportBackup"
	RestoreDisplayType         DataflowEventType = "RestoreFromBackup"
	ChangefeedDisplayType      DataflowEventType = "RunChangefeed"
	ImportDisplayType          DataflowEventType = "ImportData"
	CompactBackupDisplayType   DataflowEventType = "CompactBackup"
)
```

- [ ] **Step 2: Add GORM hooks in `entity_hook.go`**

Append to `app/repo/entity/entity_hook.go`:
```go
func (c *CompactBackup) BeforeSave(tx *gorm.DB) error {
	GetEventHookerManager().RunBeforeSave(CompactBackupDisplayType, tx, c)
	return nil
}

func (c *CompactBackup) AfterSave(tx *gorm.DB) error {
	GetEventHookerManager().RunAfterSave(CompactBackupDisplayType, tx, c)
	return nil
}
```

- [ ] **Step 3: Build**

```bash
make build-dataflow-darwin
```

- [ ] **Step 4: Commit**

```bash
git add app/repo/entity/entity_hook_manager.go app/repo/entity/entity_hook.go
git commit -m "feat(compact): add GORM hooks for CompactBackup entity

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

---

### Task 12: Compact backup metrics

**Files:**
- Create: `app/metrics/compact_backup/metric/metric.go`
- Create: `app/metrics/compact_backup/hooks/status.go`
- Create: `app/metrics/compact_backup/collector/collector.go`
- Modify: `app/metrics/register/register.go`
- Modify: `app/container/container.go`

- [ ] **Step 1: Create metric definitions**

Create `app/metrics/compact_backup/metric/metric.go`:

```go
package metric

import (
	"fmt"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
)

const (
	NS                   = "dataflow"
	SUBSYSTEM_COMPACT    = "compact_backup"
	ClusterIDLabel       = "cluster_id"
	CompactBackupIDLabel = "compact_backup_id"

	WaitMetricScrapeInterval = time.Minute * 2
)

var commonLabels = []string{ClusterIDLabel, CompactBackupIDLabel}

var (
	//nolint:gomnd
	CompactBackupStatusIntMap = map[entity.CompactBackupStatus]int{
		entity.CompactBackupStatusPending:  1,
		entity.CompactBackupStatusRunning:  2,
		entity.CompactBackupStatusSuccess:  3,
		entity.CompactBackupStatusFailed:   4,
		entity.CompactBackupStatusDeleting: 5,
		entity.CompactBackupStatusDeleted:  6,
	}

	CompactBackupStatusDesc = prometheus.NewDesc(
		prometheus.BuildFQName(NS, SUBSYSTEM_COMPACT, "status_desc"),
		"the status of compact backup jobs",
		commonLabels, nil)

	CompactBackupStatus = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Namespace: NS,
			Subsystem: SUBSYSTEM_COMPACT,
			Name:      "status",
			Help:      "compact backup status: 1-pending, 2-running, 3-success, 4-failed, 5-deleting, 6-deleted",
		}, commonLabels)
)

func UpdateCompactBackupStatusMetric(cb *entity.CompactBackup) {
	CompactBackupStatus.WithLabelValues(
		fmt.Sprintf("%d", cb.ClusterID),
		fmt.Sprintf("%d", cb.ID),
	).Set(float64(CompactBackupStatusIntMap[cb.Status]))
}
```

- [ ] **Step 2: Create the AfterSave metrics hooker**

Create `app/metrics/compact_backup/hooks/status.go`:

```go
package hook

import (
	"fmt"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	compact_metric "github.com/tidbcloud/dataflow-service/app/metrics/compact_backup/metric"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"gorm.io/gorm"
)

type compactBackupMetricsHooker struct{}

var _ entity.DataflowEventHooker = (*compactBackupMetricsHooker)(nil)

func NewCompactBackupMetricsHooker() *compactBackupMetricsHooker {
	return &compactBackupMetricsHooker{}
}

func (h *compactBackupMetricsHooker) Type() entity.DataflowEventType {
	return entity.CompactBackupDisplayType
}

func (h *compactBackupMetricsHooker) BeforeSave(_ *gorm.DB, _ interface{}) {}

func (h *compactBackupMetricsHooker) AfterSave(_ *gorm.DB, v interface{}) {
	cb, ok := v.(*entity.CompactBackup)
	if !ok {
		return
	}
	compact_metric.UpdateCompactBackupStatusMetric(cb)
	if cb.DeletedAt != nil || cb.Status == entity.CompactBackupStatusDeleted ||
		cb.Status == entity.CompactBackupStatusSuccess || cb.Status == entity.CompactBackupStatusFailed {
		go func() {
			time.Sleep(compact_metric.WaitMetricScrapeInterval)
			compact_metric.CompactBackupStatus.DeletePartialMatch(
				prometheus.Labels{compact_metric.CompactBackupIDLabel: fmt.Sprintf("%d", cb.ID)})
		}()
	}
}
```

- [ ] **Step 3: Create the Prometheus collector**

Create `app/metrics/compact_backup/collector/collector.go`:

```go
package collector

import (
	"context"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/app/container"
	compact_metric "github.com/tidbcloud/dataflow-service/app/metrics/compact_backup/metric"
	"github.com/tidbcloud/dataflow-service/app/metrics/base"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
)

func NewCompactBackupMetricsCollector() prometheus.Collector {
	app := container.Global()
	activeStatuses := []entity.CompactBackupStatus{
		entity.CompactBackupStatusPending,
		entity.CompactBackupStatusRunning,
		entity.CompactBackupStatusDeleting,
	}

	return base.NewMetricsCollector(&base.Target[*entity.CompactBackup]{
		Name:            "compact_backup",
		StatusCollector: compact_metric.CompactBackupStatusDesc,
		ListUnfinished: func(ctx context.Context) ([]*entity.CompactBackup, error) {
			result, bErr := app.CompactBackupRepo.ListByFilter(ctx, &backup_repo.CompactBackupFilter{
				Statuses: activeStatuses,
			})
			return result, bErr
		},
		ListByIDs: func(ctx context.Context, ids []entity.ID) ([]*entity.CompactBackup, error) {
			if len(ids) == 0 {
				return nil, nil
			}
			result, bErr := app.CompactBackupRepo.ListByFilter(ctx, &backup_repo.CompactBackupFilter{
				IDs: ids,
			})
			return result, bErr
		},
		UpdateStatusMetrics: func(_ *prometheus.Desc, obj *entity.CompactBackup) {
			compact_metric.UpdateCompactBackupStatusMetric(obj)
		},
	})
}
```

- [ ] **Step 4: Register gauge + collector**

In `app/metrics/register/register.go`, add imports and registration:

```go
import (
	// ... existing imports ...
	compact_backup_collector "github.com/tidbcloud/dataflow-service/app/metrics/compact_backup/collector"
	compact_backup_metric "github.com/tidbcloud/dataflow-service/app/metrics/compact_backup/metric"
)

func RegisterMetrics() {
	// ... existing ...
	prometheus.MustRegister(compact_backup_metric.CompactBackupStatus)
	prometheus.MustRegister(compact_backup_collector.NewCompactBackupMetricsCollector())
}
```

- [ ] **Step 5: Register hooker in `container.go`**

In `app/container/container.go`, add import and hooker registration in `initDBHooks`:
```go
import (
    // ... existing ...
    compact_backup_metric_hooks "github.com/tidbcloud/dataflow-service/app/metrics/compact_backup/hooks"
)

func initDBHooks(container *GlobalContainer) {
    // ... existing ...
    container.DataflowEventHookerManager.Register(compact_backup_metric_hooks.NewCompactBackupMetricsHooker())
}
```

- [ ] **Step 6: Build**

```bash
make build-dataflow-darwin
```

- [ ] **Step 7: Commit**

```bash
git add app/metrics/compact_backup/
git add app/metrics/register/register.go
git add app/container/container.go
git commit -m "feat(compact): add Prometheus metrics for compact backup status lifecycle

- GaugeVec dataflow_compact_backup_status with cluster_id + compact_backup_id labels
- Collector queries pending/running/deleting at scrape time
- AfterSave hook updates metric + schedules cleanup on terminal status

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

---

### Task 13: Full verification

- [ ] **Step 1: Run all app unit tests**

```bash
make app-unit-test-local
```
Expected: all tests pass with no failures.

- [ ] **Step 2: Verify build**

```bash
make build-dataflow-darwin
```
Expected: clean build.

- [ ] **Step 3: Verify generate-entity still works**

```bash
make generate-entity
```
Then `git diff` — the generated entity files must reflect the updated ENUM. If any drift, commit.

- [ ] **Step 4: Final commit (if any generate-entity drift)**

```bash
git add -A
git commit -m "chore: regenerate entity after compact_backup status enum extension

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

---

## Summary of DAG changes

### `DagPitRestore` (final task graph)

```
ActionCheckAndCreateClusterForRestoring
        ↓                    ↓
ActionInitRestoreConfig   ActionAddBucketPolicyForRestoring
                                      ↓
                          ActionRunCompactBackupForPiTR
                                      ↓
                            ActionRunPitRestoring
                              ↙              ↘
        ActionRestoreSqlBindings    ActionDeleteCompactionForPiTR
                              ↘              ↙
                     ActionDeleteClusterJobForRestoring
                                      ↓
                         ActionRestorePitBackupFinalStep
```

### `DagDeleteCompactBackup` (extended)

```
ActionCleanupCompactBackup
        ↓
ActionPurgeCompactionStorage
        ↓
ActionDeleteCompactBackupFinalStep
```