fix(worker): R1 — drive customer-backup cadence off plans.yaml rpo_minutes (#103)

The customer-Postgres backup scheduler chose hourly-vs-daily cadence from a
hardcoded `switch canonicalTier()` plus a hardcoded SQL tier allow-list. The
hourly tiers (pro/growth/team) happened to match their advertised
`rpo_minutes:60`, but the coupling was implicit: changing rpo_minutes in
plans.yaml would NOT have moved the cadence, so the product could silently
over-promise RPO (effective ~24h vs sold 60m).

Make the cadence RPO-driven from plans.yaml (the same field surfaced on
/api/v1/capabilities), so the cadence that MAKES the RPO true is read from the
field that PROMISES it — they can no longer drift:

  rpo_minutes in [1,60]  → hourly  (pro/growth/team)
  rpo_minutes > 60       → once-daily team slot (hobby/hobby_plus = 1440)
  rpo_minutes == 0 OR backup_retention_days == 0 → never (anonymous/free)

- BackupPlanRegistry grows RPOMinutes(tier); the common-plans adapter and the
  test fake implement it. Registry.RPOMinutes already existed in common/plans.
- Scheduler takes the registry (wired from the same `backupPlans` the runner
  uses); nil registry falls back to the legacy hardcoded mapping so a
  misconfigured boot never silently stops paid backups.
- Candidate SELECT now excludes only the never-backup tiers (anonymous/free)
  instead of an explicit paid-tier allow-list, so a NEW paid tier in
  plans.yaml is covered automatically — removing the single-site-list failure
  mode (root rule 18) that once dropped hobby_plus + every _yearly variant.
- Per-row retention==0 guard is a second, independent veto (defence-in-depth)
  so a stray rpo_minutes on a zero-retention tier still never enqueues.
- Idempotency unchanged: the atomic INSERT … WHERE NOT EXISTS (50-min DB
  lookback) still prevents a duplicate enqueue within the hour even across
  River RunOnStart double-ticks (dedupe lives in the DB, not River UniqueOpts).

Tests prove: a pro resource becomes due every hour (all 24 hours), a free
resource is never enqueued (SQL-filtered AND registry-gated if it leaks), no
duplicate enqueue within an hour, the [1,60]→hourly / >60→daily boundary, the
zero-retention veto, the nil-registry legacy fallback, and the adapter's
RPOMinutes delegation against the real embedded plans.yaml. jobs package at
96.0% coverage; all new/changed functions at 100%.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: mastermanas805

internal/jobs/backup_extra_test.go

@@ -121,6 +121,22 @@ func TestCommonPlanRegistryAdapter_Delegates(t *testing.T) {
 	if d := adapter.BackupRetentionDays("pro"); d <= 0 {
 		t.Errorf("BackupRetentionDays(pro) = %d; want > 0", d)
 	}
+	// RPOMinutes delegation — the scheduler cadence gate reads this. Pin the
+	// contract against the real embedded plans.yaml: pro/growth/team promise
+	// a 60-minute RPO (→ hourly cadence) while anonymous promises 0 (→ never
+	// backed up). A plans.yaml edit that breaks either trips here.
+	for _, tier := range []string{"pro", "growth", "team"} {
+		if m := adapter.RPOMinutes(tier); m != 60 {
+			t.Errorf("RPOMinutes(%q) = %d; want 60 (hourly-cadence promise)", tier, m)
+		}
+	}
+	if m := adapter.RPOMinutes("anonymous"); m != 0 {
+		t.Errorf("RPOMinutes(anonymous) = %d; want 0 (never backed up)", m)
+	}
+	// hobby promises a coarser daily RPO (1440) → daily cadence.
+	if m := adapter.RPOMinutes("hobby"); m <= 60 {
+		t.Errorf("RPOMinutes(hobby) = %d; want > 60 (daily cadence)", m)
+	}
 	names := adapter.TierNames()
 	if len(names) == 0 {
 		t.Fatal("TierNames returned empty slice")
@@ -719,8 +735,8 @@ func TestRunner_ProcessBackup_BadAESKey(t *testing.T) {
 	w := &CustomerBackupRunnerWorker{
 		db: db, store: newFakeBackupStore(), pgDump: &fakePgDump{},
 		bucket: "b", prefix: "p",
-		aesKey:  "not-hex-not-valid-please-fail",
-		now:     time.Now, timeout: time.Minute, batchN: backupBatchSize,
+		aesKey: "not-hex-not-valid-please-fail",
+		now:    time.Now, timeout: time.Minute, batchN: backupBatchSize,
 	}
 	if err := w.Work(context.Background(), fakeRunnerJob()); err != nil {
 		t.Fatalf("Work: %v", err)
@@ -922,12 +938,11 @@ func TestCustomerBackupSchedulerArgs_Kind(t *testing.T) {
 	}
 }
 
-// TestScheduler_AnonymousTier_DoesNotInsert — defensive: an anonymous row
-// in resource.tier slips the SQL filter (it shouldn't, but the cadence
-// switch also gates it). canonicalTier returns "anonymous"; the switch
-// has no case for it, so the row proceeds to the dedupe INSERT — which
-// is fine because the SQL filter excludes anonymous-tier rows in the
-// first place. This test pins that contract via the SELECT shape.
+// TestScheduler_AnonymousTier_DoesNotInsert — defensive: anonymous rows are
+// excluded by the SQL WHERE clause (`tier NOT IN ('anonymous','free')`), so
+// the candidate set is empty and no INSERT fires. Even if one leaked through,
+// the registry cadence gate (rpo_minutes:0 → cadenceNever) skips it. This
+// test pins the SQL-exclusion contract via the empty SELECT.
 func TestScheduler_AnonymousTier_DoesNotInsert(t *testing.T) {
 	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
 	if err != nil {
@@ -939,7 +954,7 @@ func TestScheduler_AnonymousTier_DoesNotInsert(t *testing.T) {
 	mock.ExpectQuery(`SELECT r\.id::text`).
 		WillReturnRows(sqlmock.NewRows([]string{"id", "tier", "team_id"}))
 
-	w := NewCustomerBackupSchedulerWorker(db)
+	w := NewCustomerBackupSchedulerWorker(db, schedulerPlans())
 	w.now = func() time.Time { return time.Date(2026, 5, 13, 0, 0, 0, 0, time.UTC) }
 	if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil {
 		t.Fatalf("Work: %v", err)
@@ -961,7 +976,7 @@ func TestScheduler_HobbyMissingTeamID_Skips(t *testing.T) {
 			AddRow(resID, "hobby", nil))
 	// No INSERT expected.
 
-	w := NewCustomerBackupSchedulerWorker(db)
+	w := NewCustomerBackupSchedulerWorker(db, schedulerPlans())
 	w.now = func() time.Time { return time.Date(2026, 5, 13, 0, 0, 0, 0, time.UTC) }
 	if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil {
 		t.Fatalf("Work: %v", err)
@@ -987,7 +1002,7 @@ func TestScheduler_InsertError_LoggedNonFatal(t *testing.T) {
 		WithArgs(uuid.MustParse(resID), "pro").
 		WillReturnError(errors.New("db hiccup"))
 
-	w := NewCustomerBackupSchedulerWorker(db)
+	w := NewCustomerBackupSchedulerWorker(db, schedulerPlans())
 	w.now = func() time.Time { return time.Date(2026, 5, 13, 14, 0, 0, 0, time.UTC) }
 	if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil {
 		t.Errorf("Work: per-row insert error must be non-fatal: %v", err)
@@ -1010,7 +1025,7 @@ func TestScheduler_BadUUIDInRow_Skipped(t *testing.T) {
 			AddRow("not-a-uuid", "pro", teamID))
 	// No INSERT expected — bad UUID short-circuits the per-row body.
 
-	w := NewCustomerBackupSchedulerWorker(db)
+	w := NewCustomerBackupSchedulerWorker(db, schedulerPlans())
 	w.now = func() time.Time { return time.Date(2026, 5, 13, 14, 0, 0, 0, time.UTC) }
 	if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil {
 		t.Fatalf("Work: %v", err)

internal/jobs/backup_s3.go

@@ -4,10 +4,10 @@
 // We keep this as a tiny interface (Upload / Download / Delete / List) instead
 // of passing *minio.Client around directly so:
 //
-//   1. The runner / restore-runner / retention sweep tests can use a fake that
-//      exercises the exact streaming path without dialing a real S3.
-//   2. A future cutover from MinIO to a real DO-Spaces SDK (or AWS SDK v2) is a
-//      one-file change — every consumer of this interface stays the same.
+//  1. The runner / restore-runner / retention sweep tests can use a fake that
+//     exercises the exact streaming path without dialing a real S3.
+//  2. A future cutover from MinIO to a real DO-Spaces SDK (or AWS SDK v2) is a
+//     one-file change — every consumer of this interface stays the same.
 //
 // All four methods take the bucket as an explicit argument so the same client
 // can serve a separate retention sweep on a different bucket later (e.g.
@@ -44,6 +44,16 @@ type BackupPlanRegistry interface {
 	// tier — e.g. a Pro→Free downgrade — cannot stick around past
 	// policy).
 	BackupRetentionDays(tier string) int
+	// RPOMinutes returns the per-tier Recovery Point Objective in minutes
+	// from plans.yaml. This is the SOURCE OF TRUTH the backup SCHEDULER
+	// uses to pick a cadence: a tier promising rpo_minutes<=60 must be
+	// backed up hourly (else the effective RPO is ~24h and the product
+	// over-promises), a tier with rpo_minutes>60 gets the once-daily slot,
+	// and rpo_minutes==0 ("not promised" — anonymous/free) is never
+	// enqueued. Keeping the cadence derived from this value (rather than a
+	// hardcoded tier list) means changing rpo_minutes in plans.yaml
+	// automatically moves the cadence, with no scheduler code change.
+	RPOMinutes(tier string) int
 	// TierNames lists the tier names the sweep should iterate. We sweep
 	// per-tier because the SQL hits a partial index on tier_at_backup;
 	// iterating an explicit list (rather than scanning DISTINCT) keeps
@@ -80,6 +90,13 @@ func (a *commonPlanRegistryAdapter) BackupRetentionDays(tier string) int {
 	return a.reg.BackupRetentionDays(tier)
 }
 
+// RPOMinutes delegates to the common Registry. Returns 0 for unknown tiers
+// (common's Get falls back to "anonymous", whose RPO is 0 / no scheduled
+// backups). The scheduler reads this to choose hourly vs daily cadence.
+func (a *commonPlanRegistryAdapter) RPOMinutes(tier string) int {
+	return a.reg.RPOMinutes(tier)
+}
+
 // TierNames returns every tier name registered in plans.yaml. Sorted
 // is unnecessary — the sweep is order-independent — but stable across
 // process lifetime so log lines per tier read predictably.

internal/jobs/coverage_gaps_test.go

@@ -728,7 +728,7 @@ func TestScheduler_Work_ScanError_SkipsRow(t *testing.T) {
 		WillReturnRows(sqlmock.NewRows([]string{"id", "tier", "team_id"}).
 			AddRow("fffffff0-1111-2222-3333-444444444444", "pro", "not-a-uuid"))
 
-	w := NewCustomerBackupSchedulerWorker(db)
+	w := NewCustomerBackupSchedulerWorker(db, schedulerPlans())
 	w.now = func() time.Time { return time.Date(2026, 5, 13, 14, 0, 0, 0, time.UTC) }
 	if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil {
 		t.Fatalf("Work should skip unscannable row, got %v", err)
@@ -746,7 +746,7 @@ func TestScheduler_Work_RowsError_ReturnsError(t *testing.T) {
 		RowError(0, errors.New("rows boom"))
 	mock.ExpectQuery(`SELECT r.id::text, r.tier, r.team_id`).WillReturnRows(rows)
 
-	w := NewCustomerBackupSchedulerWorker(db)
+	w := NewCustomerBackupSchedulerWorker(db, schedulerPlans())
 	w.now = func() time.Time { return time.Date(2026, 5, 13, 14, 0, 0, 0, time.UTC) }
 	if err := w.Work(context.Background(), fakeSchedulerJob()); err == nil ||
 		!strings.Contains(err.Error(), "rows error") {

internal/jobs/customer_backup_runner_test.go

@@ -450,6 +450,7 @@ func TestBackupObjectKey(t *testing.T) {
 // flips a tier, the test fails on the assertion, not on a moved goalpost.
 type fakeBackupPlanRegistry struct {
 	days  map[string]int
+	rpo   map[string]int // tier→rpo_minutes for the scheduler cadence gate
 	tiers []string
 }
 
@@ -460,6 +461,24 @@ func (f *fakeBackupPlanRegistry) BackupRetentionDays(tier string) int {
 	return 0
 }
 
+// RPOMinutes returns the per-tier RPO the scheduler uses to pick cadence.
+// When the test didn't declare an rpo map (runner tests don't care), fall
+// back to deriving a sane value from retention days so those fakes keep
+// satisfying the interface: any tier that takes backups (days>0) reports a
+// 60-minute RPO (hourly), tiers with 0 retention report 0 (never).
+func (f *fakeBackupPlanRegistry) RPOMinutes(tier string) int {
+	if f.rpo != nil {
+		if m, ok := f.rpo[tier]; ok {
+			return m
+		}
+		return 0
+	}
+	if f.BackupRetentionDays(tier) > 0 {
+		return 60
+	}
+	return 0
+}
+
 func (f *fakeBackupPlanRegistry) TierNames() []string { return f.tiers }
 
 // TestRetentionDaysFromRegistry — per-tier retention read straight from