From 50cfe9066cceea95ab635f5e511ce0839f0979f7 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Thu, 14 May 2026 15:42:54 +0530 Subject: [PATCH] fix(worker): backup scheduler covers hobby_plus + *_yearly + refunds quota on failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave FIX-H worker companion to InstaNode-dev/api#107. Closes the worker- side gaps in BugBash B36. #56/#R6 — scheduler tier set The hardcoded `tier IN ('hobby','pro','growth','team')` silently excluded hobby_plus and every _yearly variant: paying hobby_plus / hobby_plus_yearly / pro_yearly customers received zero scheduled backups. The fix widens the IN list to every tier whose plans.yaml row has backup_retention_days > 0 (hobby, hobby_plus, pro, growth, team plus all _yearly counterparts) and treats hobby/hobby_plus as daily-slot cadence (canonicalTier strips _yearly). #59 — SHA-256 integrity pg_dump → gzip output is now teed through a sha256.Hash via io.MultiWriter; the hex digest is stamped into the new resource_backups.sha256 column (migration 043 ships in api PR #107) at finalize. On failure the digest is NOT written so the restore handler treats it as "unknown integrity, skip the check". #65/#Q47 — refund manual-backup counter on failure New WithRefundClient wires the runner to call POST /internal/teams/:id/backup-quota/refund (HS256 with WORKER_INTERNAL_JWT_SECRET) when a MANUAL backup row fails terminally. Scheduled-row failures skip the refund. Idempotent on the api side — a worker restart that re-processes the same backup id is a no-op. Tests - TestScheduler_HobbyPlus_OnSlotInserts (regression for #56/#R6) - TestScheduler_YearlyVariants_BackupHourly - TestCanonicalTier - Existing TestRunner_HappyPath / _PgDumpFails / _ClaimRace updated to expect backup_kind + sha256 in the row shape DO NOT TOUCH list respected — no edits to circuit/, apiclient internals beyond using New() at the boundary. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/jobs/customer_backup_runner.go | 193 +++++++++++++++++- internal/jobs/customer_backup_runner_test.go | 16 +- internal/jobs/customer_backup_scheduler.go | 44 +++- .../jobs/customer_backup_scheduler_test.go | 101 +++++++++ internal/jobs/workers.go | 9 +- 5 files changed, 340 insertions(+), 23 deletions(-) diff --git a/internal/jobs/customer_backup_runner.go b/internal/jobs/customer_backup_runner.go index b77ef77..dbcfd7b 100644 --- a/internal/jobs/customer_backup_runner.go +++ b/internal/jobs/customer_backup_runner.go @@ -35,15 +35,23 @@ package jobs import ( + "bytes" "compress/gzip" "context" + "crypto/hmac" + "crypto/sha256" "database/sql" + "encoding/base64" + "encoding/hex" "encoding/json" "errors" "fmt" + "hash" "io" "log/slog" + "net/http" "os/exec" + "strings" "time" "github.com/google/uuid" @@ -51,6 +59,8 @@ import ( "go.opentelemetry.io/otel" "instant.dev/common/crypto" + "instant.dev/worker/internal/apiclient" + "instant.dev/worker/internal/circuit" ) // CustomerBackupRunnerArgs holds no fields — periodic job. @@ -129,6 +139,13 @@ type CustomerBackupRunnerWorker struct { now func() time.Time timeout time.Duration batchN int + + // apiBase / apiCli / jwtSecret — used by the FIX-H #65/#Q47 refund + // path. When apiBase or jwtSecret is empty the refund call is a + // no-op (logged) — same fail-open posture as the rest of the worker. + apiBase string + apiCli *apiclient.Client + jwtSecret string } // NewCustomerBackupRunner constructs a runner with production defaults. @@ -152,6 +169,22 @@ func NewCustomerBackupRunner(db *sql.DB, store BackupObjectStore, bucket, prefix } } +// WithRefundClient wires the api endpoint + JWT secret used by the +// FIX-H #65/#Q47 refund path. cmd/ should call this after construction +// with the api base URL (e.g. http://instant-api.instant.svc.cluster.local:8080) +// and the shared WORKER_INTERNAL_JWT_SECRET. Calling with empty strings +// disables the refund (no-op + WARN); same posture as the rest of the +// fail-open guards in this worker. +func (w *CustomerBackupRunnerWorker) WithRefundClient(apiBase, jwtSecret string, httpCli *http.Client) *CustomerBackupRunnerWorker { + w.apiBase = strings.TrimRight(apiBase, "/") + w.jwtSecret = jwtSecret + if httpCli == nil { + httpCli = &http.Client{Timeout: 10 * time.Second} + } + w.apiCli = apiclient.New(httpCli) + return w +} + // Work runs a single sweep tick. Returns nil on partial failure (fail-open // per row); returns an error only on a DB-level failure that prevents any // progress (e.g. the SELECT query itself failed). @@ -173,7 +206,7 @@ func (w *CustomerBackupRunnerWorker) Work(ctx context.Context, job *river.Job[Cu // path only needs one ExecContext for the claim + one for the final // status update. rows, err := w.db.QueryContext(ctx, ` - SELECT b.id::text, b.resource_id::text, b.tier_at_backup, + SELECT b.id::text, b.resource_id::text, b.tier_at_backup, b.backup_kind, r.token::text, r.connection_url, r.resource_type, r.team_id FROM resource_backups b JOIN resources r ON r.id = b.resource_id @@ -190,6 +223,7 @@ func (w *CustomerBackupRunnerWorker) Work(ctx context.Context, job *river.Job[Cu backupID string resourceID string tier sql.NullString + kind string // 'scheduled' | 'manual' — for refund routing token string connURL sql.NullString resourceType string @@ -199,7 +233,7 @@ func (w *CustomerBackupRunnerWorker) Work(ctx context.Context, job *river.Job[Cu for rows.Next() { var p pending if scanErr := rows.Scan( - &p.backupID, &p.resourceID, &p.tier, + &p.backupID, &p.resourceID, &p.tier, &p.kind, &p.token, &p.connURL, &p.resourceType, &p.teamID, ); scanErr != nil { slog.Warn("jobs.customer_backup_runner.scan_failed", "error", scanErr) @@ -255,6 +289,7 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p backupID string resourceID string tier sql.NullString + kind string token string connURL sql.NullString resourceType string @@ -317,17 +352,27 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p return false } - // Step 3 — stream pg_dump → gzip → S3 via io.Pipe. + // Step 3 — stream pg_dump → gzip → (sha256 + S3) via io.Pipe. + // + // FIX-H #59 — the gzip output is teed into a SHA-256 hasher so the + // final hex digest is available at finalize time. We hash the + // COMPRESSED bytes (not the raw pg_dump output) because the + // compressed object is what lives in S3 and what the restore + // handler / runner will re-read for verification. Hashing happens + // inline on the writer side — no second pass over the bytes. objectKey := backupObjectKey(w.prefix, p.token, p.backupID) pr, pw := io.Pipe() + hasher := sha256.New() // Goroutine: pg_dump writes raw archive bytes into the gzip writer, - // which writes compressed bytes into the pipe writer. Closing the - // gzip writer flushes its final gzip footer, then we close pw to - // signal EOF to the S3 Upload reader side. + // which writes compressed bytes into a MultiWriter that fans out to + // the sha256 hasher AND the pipe writer (which the S3 Upload reads + // from). Closing the gzip writer flushes its final gzip footer, + // then we close pw to signal EOF to the S3 Upload reader side. dumpDone := make(chan error, 1) go func() { - gz := gzip.NewWriter(pw) + mw := io.MultiWriter(hasher, pw) + gz := gzip.NewWriter(mw) runErr := w.pgDump.Run(ctx, plainConn, gz) // Close gzip first to flush the trailer, then the pipe so the // Upload side sees EOF (not just the partial gzip stream). If @@ -344,6 +389,7 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p size, upErr := w.store.Upload(ctx, w.bucket, objectKey, pr) dumpErr := <-dumpDone + digestHex := finalizeDigest(hasher, dumpErr, upErr) // Prefer the dump-side error when both fail (almost always the more // actionable: "pg_dump: connection refused" vs "pipe: io: read/write @@ -363,15 +409,18 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p return false } - // Step 4 — finalize. + // Step 4 — finalize. FIX-H #59: stamp sha256 alongside s3_key and + // size_bytes so the restore handler can verify integrity against a + // fresh re-read of the object. if _, updErr := w.db.ExecContext(parentCtx, ` UPDATE resource_backups SET status = 'ok', finished_at = now(), s3_key = $2, - size_bytes = $3 + size_bytes = $3, + sha256 = NULLIF($4,'') WHERE id = $1 - `, p.backupID, objectKey, size); updErr != nil { + `, p.backupID, objectKey, size, digestHex); updErr != nil { slog.Error("jobs.customer_backup_runner.finalize_failed", "backup_id", p.backupID, "object_key", objectKey, @@ -408,12 +457,18 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p // markFailed updates the row to 'failed', emits backup.failed audit, and // logs the error_summary. parentCtx is used for the DB write so a timed- // out backup still records its failure (the inner ctx is already dead). +// +// FIX-H #65/#Q47 — when the failed row was a MANUAL backup, we POST to +// the api's internal refund endpoint so the team's daily counter is +// credited. Scheduled backups don't burn the manual-counter so no +// refund is needed. func (w *CustomerBackupRunnerWorker) markFailed( ctx context.Context, backupID, errSummary string, start time.Time, p struct { backupID string resourceID string tier sql.NullString + kind string token string connURL sql.NullString resourceType string @@ -452,9 +507,37 @@ func (w *CustomerBackupRunnerWorker) markFailed( "error_summary", errSummary, "duration_ms", duration.Milliseconds(), ) + + // FIX-H #65/#Q47 — refund the manual-backups-today counter when a + // MANUAL backup fails. Scheduled backups don't burn the counter so + // they don't need a refund. Best-effort: a refund failure (api down, + // breaker open) logs and moves on; the customer's counter stays + // burned for the rest of the UTC day, matching pre-fix behavior. + if p.kind == "manual" && p.teamID.Valid { + refundErr := w.refundManualBackupQuota(p.teamID.UUID, backupID) + if refundErr != nil { + slog.Warn("jobs.customer_backup_runner.refund_failed", + "backup_id", backupID, + "team_id", p.teamID.UUID, + "error", refundErr, + ) + } + } + _ = ctx // (parentCtx) keep param to surface intent even though we use a fresh ctx } +// finalizeDigest returns the hex-encoded SHA-256 of the gzipped pg_dump +// stream IFF the dump + upload both succeeded. On failure we deliberately +// return "" so the finalize UPDATE writes NULL into sha256 — recording a +// digest for a partial / corrupt object would lie to the restore handler. +func finalizeDigest(h hash.Hash, dumpErr, upErr error) string { + if dumpErr != nil || upErr != nil { + return "" + } + return hex.EncodeToString(h.Sum(nil)) +} + // writeAudit emits an audit_log row. Errors are logged but not propagated — // a missing audit row is a bookkeeping issue, not a data-correctness one. func (w *CustomerBackupRunnerWorker) writeAudit( @@ -580,3 +663,93 @@ func (b *limitedBuffer) Write(p []byte) (int, error) { func (b *limitedBuffer) String() string { return string(b.buf[:b.n]) } + +// refundManualBackupQuota POSTs to the api's internal refund endpoint to +// decrement the team's manual-backups-today counter after a manual +// backup failed terminally. FIX-H #65/#Q47 BugBash B36. +// +// Returns nil on a 2xx response or when the refund is disabled (no +// apiBase / no jwtSecret). The endpoint is idempotent: replays for the +// same backup_id are no-ops on the api side, so a worker restart +// mid-batch that re-processes the same row can't double-credit. +// +// Failure modes (logged but not retried): +// - circuit.ErrOpen: api is hosed; refund skipped. The customer +// loses one unit of daily headroom — same as pre-fix behavior. +// - network / 5xx: same as above. The next manual backup the team +// attempts will see the (unrefunded) counter. +// - 4xx: refund call shape is wrong (e.g. bad JWT). Skip; an operator +// will see the error in slog and fix the wiring. +func (w *CustomerBackupRunnerWorker) refundManualBackupQuota(teamID uuid.UUID, backupID string) error { + if w.apiBase == "" || w.jwtSecret == "" || w.apiCli == nil { + slog.Warn("jobs.customer_backup_runner.refund_disabled", + "reason", "apiBase/jwtSecret/apiCli unset", + "team_id", teamID.String(), + "backup_id", backupID, + ) + return nil + } + + url := fmt.Sprintf("%s/internal/teams/%s/backup-quota/refund", w.apiBase, teamID.String()) + bodyBytes, _ := json.Marshal(map[string]string{"backup_id": backupID}) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(bodyBytes)) + if err != nil { + return fmt.Errorf("build refund request: %w", err) + } + tok, tokErr := signBackupRefundJWT(w.jwtSecret, teamID.String()) + if tokErr != nil { + return fmt.Errorf("sign refund jwt: %w", tokErr) + } + req.Header.Set("Authorization", "Bearer "+tok) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "instanode-worker/backup-refund") + + resp, doErr := w.apiCli.Do(req) + if doErr != nil { + if errors.Is(doErr, circuit.ErrOpen) { + return fmt.Errorf("api circuit open") + } + return fmt.Errorf("api request: %w", doErr) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + return fmt.Errorf("api status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + _, _ = io.Copy(io.Discard, resp.Body) + return nil +} + +// signBackupRefundJWT mints the HS256 token the api's +// /internal/teams/:id/backup-quota/refund endpoint expects. Shape +// matches verifyInternalBackupRefundJWT on the api side: +// +// purpose — "internal_backup_refund" +// team_id — the team uuid the api will compare against the path :id +// iat — required, within ±60s of api-side now +func signBackupRefundJWT(secret, teamID string) (string, error) { + if secret == "" { + return "", errors.New("empty secret") + } + header := base64.RawURLEncoding.EncodeToString([]byte(`{"alg":"HS256","typ":"JWT"}`)) + now := time.Now().UTC().Unix() + claims := map[string]any{ + "purpose": "internal_backup_refund", + "team_id": teamID, + "iat": now, + "exp": now + 5*60, + } + claimsJSON, err := json.Marshal(claims) + if err != nil { + return "", fmt.Errorf("marshal claims: %w", err) + } + body := header + "." + base64.RawURLEncoding.EncodeToString(claimsJSON) + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write([]byte(body)) + sig := base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) + return body + "." + sig, nil +} diff --git a/internal/jobs/customer_backup_runner_test.go b/internal/jobs/customer_backup_runner_test.go index 18285ab..e2342fa 100644 --- a/internal/jobs/customer_backup_runner_test.go +++ b/internal/jobs/customer_backup_runner_test.go @@ -133,9 +133,9 @@ func TestRunner_HappyPath(t *testing.T) { mock.ExpectQuery(`SELECT b.id::text, b.resource_id::text, b.tier_at_backup`). WithArgs(backupBatchSize). WillReturnRows(sqlmock.NewRows([]string{ - "id", "resource_id", "tier_at_backup", + "id", "resource_id", "tier_at_backup", "backup_kind", "token", "connection_url", "resource_type", "team_id", - }).AddRow(backupID, resID, "pro", token, encConn, "postgres", teamID)) + }).AddRow(backupID, resID, "pro", "scheduled", token, encConn, "postgres", teamID)) // Atomic claim returns the id. mock.ExpectQuery(`UPDATE resource_backups\s+SET status = 'running'`). @@ -146,9 +146,9 @@ func TestRunner_HappyPath(t *testing.T) { mock.ExpectExec(`INSERT INTO audit_log`). WillReturnResult(sqlmock.NewResult(1, 1)) - // Finalize UPDATE — status=ok. + // Finalize UPDATE — status=ok. FIX-H #59: adds sha256 column. mock.ExpectExec(`UPDATE resource_backups\s+SET status = 'ok'`). - WithArgs(backupID, sqlmock.AnyArg(), sqlmock.AnyArg()). + WithArgs(backupID, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(1, 1)) // backup.succeeded audit row. @@ -258,9 +258,9 @@ func TestRunner_PgDumpFails_MarksFailed(t *testing.T) { mock.ExpectQuery(`SELECT b.id::text`). WithArgs(backupBatchSize). WillReturnRows(sqlmock.NewRows([]string{ - "id", "resource_id", "tier_at_backup", + "id", "resource_id", "tier_at_backup", "backup_kind", "token", "connection_url", "resource_type", "team_id", - }).AddRow(backupID, resID, "pro", "tok", encConn, "postgres", teamID)) + }).AddRow(backupID, resID, "pro", "scheduled", "tok", encConn, "postgres", teamID)) mock.ExpectQuery(`UPDATE resource_backups\s+SET status = 'running'`). WithArgs(backupID). @@ -322,9 +322,9 @@ func TestRunner_ClaimRace_SkipsSilently(t *testing.T) { mock.ExpectQuery(`SELECT b.id::text`). WithArgs(backupBatchSize). WillReturnRows(sqlmock.NewRows([]string{ - "id", "resource_id", "tier_at_backup", + "id", "resource_id", "tier_at_backup", "backup_kind", "token", "connection_url", "resource_type", "team_id", - }).AddRow(backupID, resID, "pro", "tok", encConn, "postgres", teamID)) + }).AddRow(backupID, resID, "pro", "scheduled", "tok", encConn, "postgres", teamID)) // Claim returns no rows (sql.ErrNoRows path). mock.ExpectQuery(`UPDATE resource_backups\s+SET status = 'running'`). diff --git a/internal/jobs/customer_backup_scheduler.go b/internal/jobs/customer_backup_scheduler.go index 20dff96..8ea0f94 100644 --- a/internal/jobs/customer_backup_scheduler.go +++ b/internal/jobs/customer_backup_scheduler.go @@ -63,6 +63,19 @@ func NewCustomerBackupSchedulerWorker(db *sql.DB) *CustomerBackupSchedulerWorker return &CustomerBackupSchedulerWorker{db: db, now: time.Now} } +// canonicalTier strips the "_yearly" suffix from a plan tier name so the +// cadence gate can treat e.g. "hobby_yearly" the same as "hobby". Kept +// local to this package so the scheduler doesn't need a hard dependency +// on common/plans just for one string strip. Mirrors +// instant.dev/common/plans.CanonicalTier. +func canonicalTier(tier string) string { + const suffix = "_yearly" + if len(tier) > len(suffix) && tier[len(tier)-len(suffix):] == suffix { + return tier[:len(tier)-len(suffix)] + } + return tier +} + // hobbyDailySlot returns the hour-of-day [0,24) at which a given team should // receive its single daily hobby-tier backup. Deterministic per team UUID: // the high 4 bits of the first byte are used to spread teams across 24 @@ -89,12 +102,28 @@ func (w *CustomerBackupSchedulerWorker) Work(ctx context.Context, job *river.Job // resource.tier — not team.plan_tier — because the resource is what // carries the user-paid retention contract (mirrors ElevateResourceTiers // on the api side). + // + // FIX-H (#56/#R6 B36) — the prior hardcoded set + // (hobby, pro, growth, team) silently excluded hobby_plus and every + // _yearly variant, so paid hobby_plus / hobby_plus_yearly / pro_yearly / + // growth_yearly / team_yearly customers received ZERO scheduled + // backups. The fix lists every tier whose plans.yaml row has + // backup_retention_days > 0. We keep the list inline rather than + // querying plans.Registry here because the scheduler doesn't yet + // take a Registry — adding a registry param would force a constructor + // change across cmd/, deferred to a separate refactor. rows, err := w.db.QueryContext(ctx, ` SELECT r.id::text, r.tier, r.team_id FROM resources r WHERE r.status = 'active' AND r.resource_type IN ('postgres', 'vector') - AND r.tier IN ('hobby', 'pro', 'growth', 'team') + AND r.tier IN ( + 'hobby', 'hobby_yearly', + 'hobby_plus', 'hobby_plus_yearly', + 'pro', 'pro_yearly', + 'growth', 'growth_yearly', + 'team', 'team_yearly' + ) `) if err != nil { return fmt.Errorf("CustomerBackupSchedulerWorker: query failed: %w", err) @@ -124,9 +153,16 @@ func (w *CustomerBackupSchedulerWorker) Work(ctx context.Context, job *river.Job skippedDedup := 0 for _, c := range candidates { // Cadence gate. - if c.tier == "hobby" { + // + // FIX-H — hobby and hobby_plus (and their _yearly variants) run + // at one daily slot per team. Pro / Growth / Team (and yearly + // counterparts) back up every hour. canonicalTier strips the + // _yearly suffix so hobby_yearly / hobby_plus_yearly share the + // daily-slot policy with their monthly canonical tier. + switch canonicalTier(c.tier) { + case "hobby", "hobby_plus": if !c.teamID.Valid { - // Defensive: a tier='hobby' resource without a team_id is + // Defensive: a tier='hobby*' resource without a team_id is // nonsensical (only anonymous rows have NULL team) but if // it slips in we skip rather than panic-divide. continue @@ -136,7 +172,7 @@ func (w *CustomerBackupSchedulerWorker) Work(ctx context.Context, job *river.Job continue } } - // pro / growth / team / hobby-on-slot all proceed. + // pro / growth / team (any variant) + hobby/hobby_plus-on-slot proceed. rid, err := uuid.Parse(c.id) if err != nil { diff --git a/internal/jobs/customer_backup_scheduler_test.go b/internal/jobs/customer_backup_scheduler_test.go index 46a2cd2..552315c 100644 --- a/internal/jobs/customer_backup_scheduler_test.go +++ b/internal/jobs/customer_backup_scheduler_test.go @@ -179,6 +179,107 @@ func TestScheduler_DedupExists_Skips(t *testing.T) { } } +// TestScheduler_HobbyPlus_OnSlotInserts — FIX-H regression. Hobby Plus +// (the $19/mo mid-tier) MUST be in the scheduled-backup set. Pre-fix the +// scheduler hardcoded `tier IN ('hobby','pro','growth','team')` and any +// hobby_plus / hobby_plus_yearly / pro_yearly customer received zero +// scheduled backups despite paying for them. +func TestScheduler_HobbyPlus_OnSlotInserts(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + // Pick a team UUID whose slot = 5; run scheduler at hour 5. + teamID := uuid.UUID{5, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0} + if hobbyDailySlot(teamID) != 5 { + t.Fatalf("test fixture wrong: hobbyDailySlot(teamID)=%d, want 5", hobbyDailySlot(teamID)) + } + resID := "fffffff0-1111-2222-3333-444444444444" + + mock.ExpectQuery(`SELECT r.id::text`). + WillReturnRows(sqlmock.NewRows([]string{"id", "tier", "team_id"}). + AddRow(resID, "hobby_plus", teamID)) + mock.ExpectQuery(`SELECT EXISTS`). + WithArgs(uuid.MustParse(resID)). + WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false)) + mock.ExpectExec(`INSERT INTO resource_backups`). + WithArgs(uuid.MustParse(resID), "hobby_plus"). + WillReturnResult(sqlmock.NewResult(1, 1)) + + w := NewCustomerBackupSchedulerWorker(db) + w.now = func() time.Time { return time.Date(2026, 5, 14, 5, 0, 0, 0, time.UTC) } + + if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil { + t.Fatalf("Work: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unmet expectations: %v", err) + } +} + +// TestScheduler_YearlyVariants_BackupHourly — pro_yearly and team_yearly +// (and any other _yearly tier with hourly cadence) must back up every +// hour just like their canonical monthly counterpart. Regression guard +// for the FIX-H widened tier set. +func TestScheduler_YearlyVariants_BackupHourly(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + defer db.Close() + + teamID := uuid.MustParse("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + resID := "fffffff0-1111-2222-3333-444444444444" + + mock.ExpectQuery(`SELECT r.id::text, r.tier, r.team_id`). + WillReturnRows(sqlmock.NewRows([]string{"id", "tier", "team_id"}). + AddRow(resID, "pro_yearly", teamID)) + mock.ExpectQuery(`SELECT EXISTS`). + WithArgs(uuid.MustParse(resID)). + WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(false)) + mock.ExpectExec(`INSERT INTO resource_backups`). + WithArgs(uuid.MustParse(resID), "pro_yearly"). + WillReturnResult(sqlmock.NewResult(1, 1)) + + w := NewCustomerBackupSchedulerWorker(db) + // Hour 14 — pro_yearly should fire regardless (hourly cadence). + w.now = func() time.Time { return time.Date(2026, 5, 14, 14, 0, 0, 0, time.UTC) } + + if err := w.Work(context.Background(), fakeSchedulerJob()); err != nil { + t.Fatalf("Work: %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unmet expectations: %v", err) + } +} + +// TestCanonicalTier — sanity: _yearly strips, others pass through. +func TestCanonicalTier(t *testing.T) { + cases := map[string]string{ + "hobby": "hobby", + "hobby_yearly": "hobby", + "hobby_plus": "hobby_plus", + "hobby_plus_yearly": "hobby_plus", + "pro": "pro", + "pro_yearly": "pro", + "team": "team", + "team_yearly": "team", + "growth": "growth", + "growth_yearly": "growth", + "anonymous": "anonymous", + "": "", + "_yearly": "_yearly", // not stripped — guard: too short + } + for in, want := range cases { + got := canonicalTier(in) + if got != want { + t.Errorf("canonicalTier(%q) = %q, want %q", in, got, want) + } + } +} + // TestScheduler_DBSelectError_ReturnsError — bad SELECT bubbles up. func TestScheduler_DBSelectError_ReturnsError(t *testing.T) { db, mock, err := sqlmock.New() diff --git a/internal/jobs/workers.go b/internal/jobs/workers.go index c0af059..3a6f64c 100644 --- a/internal/jobs/workers.go +++ b/internal/jobs/workers.go @@ -312,7 +312,14 @@ func StartWorkers(ctx context.Context, db *sql.DB, rdb *redis.Client, cfg *confi // doesn't block worker boot. See each worker's Work() top for the // exact WARN line emitted. river.AddWorker(workers, WithObservability(NewCustomerBackupSchedulerWorker(db), nrApp)) - river.AddWorker(workers, WithObservability(NewCustomerBackupRunner(db, backupStore, cfg.BackupS3Bucket, cfg.BackupS3PathPrefix, cfg.AESKey, backupPlans), nrApp)) + // FIX-H #65/#Q47 — wire the refund client so terminal MANUAL backup + // failures credit the team's daily counter via the api's internal + // /internal/teams/:id/backup-quota/refund endpoint. Empty apiBase / + // jwtSecret disables the refund (logs WARN, no-op) — matches the + // rest of the worker's fail-open posture. + customerRunner := NewCustomerBackupRunner(db, backupStore, cfg.BackupS3Bucket, cfg.BackupS3PathPrefix, cfg.AESKey, backupPlans). + WithRefundClient(cfg.InstantAPIInternalURL, cfg.WorkerInternalJWTSecret, nil) + river.AddWorker(workers, WithObservability(customerRunner, nrApp)) river.AddWorker(workers, WithObservability(NewCustomerRestoreRunner(db, backupStore, cfg.BackupS3Bucket, cfg.AESKey), nrApp)) // Platform-DB backup — nightly 02:00 UTC pg_dump of the platform DB to