# CRPiTR Per-Region Checkpoint for `ListPiTRTimeRanges`

## Summary

Add `groupPiTRTimeRangesByRegionV2` so that secondary-region buckets (`bucket.Master == false`)
use the actual cross-region checkpoint (read from `resume-state.json` in S3/blob storage) instead
of the main region's DB checkpoint. Primary-region buckets (`bucket.Master == true`) continue to
use the DB checkpoint unchanged.

---

## Background

`ListPiTRTimeRanges` currently assigns the same `[startTS, endTS]` to every region a logbackup
covers — the `endTS` always comes from `logBackup.CheckpointTS` stored in the database.

For Cross-Region PiTR (CRPiTR), the CRRF daemon running in the secondary region writes a
`resume-state.json` file directly into the logbackup's S3 storage path. This file records the
latest checkpoint that has been replicated to the secondary region's bucket. The secondary
region's recoverable time range ends at this checkpoint, not at the main region's checkpoint.

The `BucketSettingSnapshot.Buckets` list records each backup destination with a `Master` flag:
`Master == true` for the primary region, `false` for every secondary (CRPiTR) region.

---

## Design

### Affected call chain

```
ListPiTRTimeRanges
  └─ mergePiTRTimeRanges
       └─ groupPiTRTimeRangesByRegion   (V1 — unchanged, kept for compatibility)
            (future: replaced by groupPiTRTimeRangesByRegionV2 once wiring is added)

groupPiTRTimeRangesByRegionV2 (NEW — iterates all logbackups uniformly)
  └─ for each logBackup → for each bucket in BucketSettingSnapshot:
       ├─ bucket.Master == true  → endTS = logBackupTimeRange.EndTS  (unchanged)
       └─ bucket.Master == false → getCheckpointForCRPiTR (injectable)
              found  → endTS = checkpoint
              !found → skip bucket (no time-range entry)
              error  → log error, skip bucket
```

### `pitBackupModel` struct change

Add `bucketRepo`, `getStorageFunc`, and `getCheckpointForCRPiTR` to the struct.

```go
type pitBackupModel struct {
    // db
    dbProvider          db.TransProvider
    pitBackupSwitchRepo backupRepo.PitBackupSwitchRepo
    logBackupRepo       backupRepo.LogBackupRepo
    backupRepo          backupRepo.BackupRepo
    scheduleRepo        backupRepo.BackupScheduleRepo
    restoreRepo         restoreRepo.RestoreRepo
    bucketRepo          bucketRepo.BucketRepo  // NEW

    // sub modules
    snapshotBackupModel backup.SnapshotBackupModel
    backupSettingModel  backup.BackupSettingModel

    // internal helpers
    workflowCli        workflow.Workflow
    coreCli            grpcli.CoreGrpcClient
    switchStatusMatrix status_matrix.StatusMatrix
    featureGate        feature_gate.FeatureGate

    // injectable for CRPiTR checkpoint and testing
    getStorageFunc         blob.GetStorageFunc  // NEW; default: blob.NewStorage
    getCheckpointForCRPiTR func(                // NEW; default: bound method m.GetCheckpointForCRPiTR
        ctx            context.Context,
        logBackup      entity.LogBackup,
        region         string,
        getStorageFunc blob.GetStorageFunc,
    ) (int64, bool, error)
}
```

`NewPitBackupModel` signature adds `bucketRepo` as the 6th parameter (after `restoreRepo`).
The constructor uses a two-step pattern to allow binding the method:

```go
func NewPitBackupModel(..., bucketRepo bucketRepo.BucketRepo, ...) backup.PitBackupModel {
    m := &pitBackupModel{
        ...,
        bucketRepo:     bucketRepo,
        getStorageFunc: blob.NewStorage,
    }
    m.getCheckpointForCRPiTR = m.GetCheckpointForCRPiTR
    return m
}
```

### New file: `pit_backup_checkpoint.go`

```go
const crPiTRResumeStateFile = "resume-state.json"

// PersistentState mirrors the checkpoint file written by the CRPiTR replicator.
type PersistentState struct {
    LastCheckpoint uint64 `json:"last_checkpoint"`
}

// GetCheckpointForCRPiTR reads resume-state.json from the bucket covering the
// given region and returns the physical timestamp (seconds) for cross-region PiTR.
//
// Return values:
//   checkpointTS, true,  nil   — success; use checkpointTS as EndTS
//   0,            false, nil   — bucket/file not found (CRRF not started); caller must skip region
//   0,            false, err   — read/parse failure; caller must log and skip region
//
// getStorageFunc is injected to allow test overrides (default: blob.NewStorage via struct field).
func (b *pitBackupModel) GetCheckpointForCRPiTR(
    ctx context.Context,
    logBackup entity.LogBackup,
    region string,
    getStorageFunc blob.GetStorageFunc,
) (int64, bool, error)
```

**Implementation steps:**
1. `bucketSnapshot, found := logBackup.BucketSettingSnapshot.GetBucketByRegion(region)` (on `*entity.BucketSetting`)
   - `!found` → return `(0, false, nil)`
2. `bucket, err := b.bucketRepo.Load(ctx, bucketSnapshot.BucketId)`
3. `opt := BuildStorageOption(bucket, region)` — builds `blob.StorageOption` (AWS/EKS sets `AWSRoleARN`; other providers leave it unset)
4. `storage, err := getStorageFunc(ctx, logBackup.Storage, opt)`
   - `blob.IsNotFoundError(err)` → return `(0, false, nil)`
   - other error → log warn, return `(0, false, err)`
5. `data, err := storage.ReadFile(ctx, "resume-state.json")`
   - `blob.IsNotFoundError(err)` → return `(0, false, nil)`
   - other error → log warn, return `(0, false, err)`
6. `json.Unmarshal(data, &state)` → `util.TransToTS(state.LastCheckpoint)` → return `(ts, true, nil)`

**`BuildStorageOption`** (also in `pit_backup_checkpoint.go`):

```go
// BuildStorageOption builds a blob.StorageOption for the given bucket and target region.
// Note: function name has a typo ("Stoage") to match the existing codebase convention.
func BuildStorageOption(bucket *entity.Bucket, region string) *blob.StorageOption {
    opt := &blob.StorageOption{
        ClusterRegion: region,
        ProjectID:     bucket.Uint64ProjectID(),
    }
    if bucket.InfraProviderType == entity.InfraProviderTypeAWS ||
        bucket.InfraProviderType == entity.InfraProviderTypeEKS {
        if bucket.CloudAccountID != nil {
            opt.AWSRoleARN = domain.BuildDataflowServiceInfraRoleARN(*bucket.CloudAccountID)
        }
    }
    return opt
}
```

### New `groupPiTRTimeRangesByRegionV2`

**New function** added alongside the unchanged V1. V2 takes `cluster` and `cloudRegions` as
pre-fetched parameters (easier to unit-test). It iterates all logbackups uniformly and uses
`bucket.Master` to distinguish primary from secondary regions:

```go
func (b *pitBackupModel) groupPiTRTimeRangesByRegionV2(
    ctx context.Context,
    logBackupTimeRanges []*domain.PiTRTimeRange,
    cluster *core.Cluster,
    cloudRegions map[string]map[string]*core.CloudRegion,
) ([]*domain.PiTRRegionTimeRange, error) {
    regions := cloudRegions[cluster.CloudProviderName]
    timeRangeMap := make(map[string]*domain.PiTRRegionTimeRange)

    for _, logBackupTimeRange := range logBackupTimeRanges {
        for _, bucket := range logBackupTimeRange.LogBackup.BucketSettingSnapshot.Buckets {
            var endTs int64
            if bucket.Master {
                endTs = logBackupTimeRange.EndTS
            } else {
                // Non-master (secondary) region: use CRPiTR checkpoint
                checkpoint, found, err := b.getCheckpointForCRPiTR(
                    ctx, logBackupTimeRange.LogBackup, bucket.Regions[0], b.getStorageFunc)
                if err != nil { logger.Error(...); continue }
                if !found { continue }
                endTs = checkpoint
            }
            // Append time range for this bucket's first region
            region := bucket.Regions[0]
            // ... add to timeRangeMap, build RegionInfo from cloudRegions if available
        }
    }
    // Flatten map to slice
    return timeRanges, nil
}
```

> **Note:** V1 (`groupPiTRTimeRangesByRegion`) is kept unchanged. `mergePiTRTimeRanges` still
> calls V1; wiring V2 is a follow-on step.

### `PersistentState.LastCheckpoint` conversion

`last_checkpoint` in `resume-state.json` is a raw TSO `uint64`.
Use `util.TransToTS(state.LastCheckpoint)` to convert to physical timestamp in seconds.

---

## Files Changed

| File | Change |
|------|--------|
| `app/models/backup/impl/pit_backup.go` | Add `bucketRepo`, `getStorageFunc`, `getCheckpointForCRPiTR` fields; update `NewPitBackupModel` (bucketRepo at position 6, two-step init); add `groupPiTRTimeRangesByRegionV2` |
| `app/models/backup/impl/pit_backup_checkpoint.go` | **New**: `PersistentState`, `GetCheckpointForCRPiTR` method, `BuildStorageOption` |
| `app/models/backup/impl/pit_backup_testhelper.go` | **New**: `NewPitBackupModelWithStorageFunc` (non-test file exporting test helper) |
| `app/models/backup/impl/pit_backup_test.go` | **New**: unit tests for `mergeSingleRegionPiTRTimeRanges` and `groupPiTRTimeRangesByRegionV2` (package `impl`) |
| `app/container/container.go` | Pass `container.BucketRepo` as 6th arg to `NewPitBackupModel` |

---

## Error Handling

| Scenario | Behavior |
|----------|----------|
| `resume-state.json` not found (CRRF not started) | `found=false` → skip bucket, no time-range entry |
| Bucket not found for region in snapshot | `found=false` → skip bucket |
| Read/parse error | `err!=nil` → log error, skip bucket |
| `bucket.Master == true` (primary region) | Always use `logBackupTimeRange.EndTS` (unchanged) |
| Non-AWS/EKS bucket (e.g. Azure) | No `AWSRoleARN` set; `GetCheckpointForCRPiTR` proceeds but S3 auth may fail → error path handles it |

---

## Not In Scope

- Changes to `genPiTRTimeRanges`, `mergeSingleRegionPiTRTimeRanges`, or `genPiTRTips`
- Changes to the `genLogBackupValidRange` path
- Wiring `groupPiTRTimeRangesByRegionV2` into `mergePiTRTimeRanges` (follow-on step)