refactor(jobs): route expiry-stage + reaper status logic through common/resourcestatus

Eliminates the api↔worker expiry-stage predicate divergence the BugBash
flagged. The worker's hand-written 12h/6h/1h schedule and the reaper's
reapable-status set now come from instant.dev/common/resourcestatus.

expiry_reminder.go:
  - reminderStage/reminderSchedule deleted; selectStage delegates to
    resourcestatus.DeriveExpiryStage (the P2-12 "most-imminent bucket
    wins" logic now lives in the shared package)
  - hoursLeft delegates to resourcestatus.HoursUntilExpiry
  - outermostReminderWindow derived from resourcestatus.ExpiryWindow12h

expire.go:
  - reapable-status SQL filter (active/paused/suspended) derived from
    resourcestatus.ReapableStatuses() so the SQL IN(...) can never drift
    from Status.IsReapable; terminal write uses StatusDeleted

resourcestatus_conversion_test.go pins the converted selectStage /
hoursLeft against verbatim copies of the original hand-written
algorithms over every time-bucket boundary — proving identical behaviour.

Cross-repo contract change (CLAUDE.md rule 22); common + api pushed first.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mastermanas805

internal/jobs/expire.go

@@ -5,17 +5,37 @@ import (
 	"database/sql"
 	"fmt"
 	"log/slog"
+	"strings"
 	"time"
 
 	madmin "github.com/minio/madmin-go/v3"
 	minio "github.com/minio/minio-go/v7"
 	"github.com/riverqueue/river"
 	"go.opentelemetry.io/otel"
+	"instant.dev/common/resourcestatus"
 	commonv1 "instant.dev/proto/common/v1"
 	"instant.dev/worker/internal/metrics"
 	"instant.dev/worker/internal/provisioner"
 )
 
+// reapableStatusSQLList is the canonical SQL `IN (...)` body listing the
+// statuses a TTL-expiry sweep may act on (active / paused / suspended),
+// derived from resourcestatus.ReapableStatuses() so the reaper's SQL
+// filter can never drift from the shared Go predicate Status.IsReapable.
+// TTL must win over lifecycle state: a paused/suspended resource past its
+// 24h TTL is still expired and must be deprovisioned.
+//
+// Built once at init from a fixed enum of literals (no caller input), so
+// there is no SQL-injection surface — the values are 'active', 'paused',
+// 'suspended'.
+var reapableStatusSQLList = func() string {
+	quoted := make([]string, 0, 3)
+	for _, s := range resourcestatus.ReapableStatuses() {
+		quoted = append(quoted, "'"+s+"'")
+	}
+	return strings.Join(quoted, ", ")
+}()
+
 // ExpireAnonymousArgs holds the arguments for the ExpireAnonymousJob.
 // No fields are needed — it's a periodic maintenance job.
 type ExpireAnonymousArgs struct{}
@@ -106,7 +126,7 @@ func (w *ExpireAnonymousWorker) Work(ctx context.Context, job *river.Job[ExpireA
 		SELECT id::text, token::text, resource_type, COALESCE(provider_resource_id, '')
 		FROM resources
 		WHERE ((team_id IS NULL AND tier = 'anonymous') OR tier = 'free')
-		  AND status IN ('active', 'paused', 'suspended')
+		  AND status IN (`+reapableStatusSQLList+`)
 		  AND expires_at IS NOT NULL
 		  AND expires_at < now()
 	`)
@@ -195,8 +215,8 @@ func (w *ExpireAnonymousWorker) Work(ctx context.Context, job *river.Job[ExpireA
 		// the SELECT above. Gating on status='active' alone would leave the
 		// paused/suspended rows we just deprovisioned stuck in their old status.
 		if _, err := w.db.ExecContext(ctx,
-			`UPDATE resources SET status = 'deleted'
-			 WHERE id = $1 AND status IN ('active', 'paused', 'suspended')`,
+			`UPDATE resources SET status = '`+resourcestatus.StatusDeleted.String()+`'
+			 WHERE id = $1 AND status IN (`+reapableStatusSQLList+`)`,
 			r.id,
 		); err != nil {
 			slog.Error("jobs.expire_anonymous.mark_deleted_failed",
@@ -214,7 +234,7 @@ func (w *ExpireAnonymousWorker) Work(ctx context.Context, job *river.Job[ExpireA
 	var activeAnon int
 	if scanErr := w.db.QueryRowContext(ctx, `
 		SELECT COUNT(*) FROM resources
-		WHERE team_id IS NULL AND status = 'active' AND expires_at IS NOT NULL
+		WHERE team_id IS NULL AND status = '`+resourcestatus.StatusActive.String()+`' AND expires_at IS NOT NULL
 	`).Scan(&activeAnon); scanErr == nil {
 		metrics.ActiveAnonymousResources.Set(float64(activeAnon))
 	}

internal/jobs/expiry_reminder.go

@@ -60,6 +60,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/riverqueue/river"
 	"go.opentelemetry.io/otel"
+	"instant.dev/common/resourcestatus"
 )
 
 // ExpiryReminderArgs is the River job payload — no fields, runs as a sweep.
@@ -69,20 +70,25 @@ type ExpiryReminderArgs struct{}
 func (ExpiryReminderArgs) Kind() string { return "expiry_reminder" }
 
 // reminderStage describes one bucket in the 12h/6h/1h schedule.
+//
+// As of the resourcestatus unification it is a thin wrapper over
+// instant.dev/common/resourcestatus.ExpiryStage — the 12h/6h/1h
+// thresholds, the index→reminder_index mapping, and the label→stage_label
+// mapping all live in the shared package, so api and worker can never
+// disagree on which window a resource falls in. The struct is retained
+// only so the existing `stage.index` / `stage.label` / `stage.stage`
+// call sites in Work() and emitAnonExpiryWarningAudit() stay unchanged.
 type reminderStage struct {
-	index         int           // 1-based; matches reminder_index in the email
-	expiresWithin time.Duration // resource fires this stage when expires_at <= now + expiresWithin
-	label         string        // logging label, also flows into the email as stage_label
+	stage resourcestatus.ExpiryStage
+	index int    // 1-based; matches reminder_index in the email
+	label string // logging label, also flows into the email as stage_label
 }
 
-// reminderSchedule is the canonical stage table. Ordered most-distant
-// to most-imminent. selectStage picks the MOST IMMINENT window the
-// resource currently falls in (time-bucket-first) — see P2-12 below.
-var reminderSchedule = []reminderStage{
-	{index: 1, expiresWithin: 12 * time.Hour, label: "stage_12h"},
-	{index: 2, expiresWithin: 6 * time.Hour, label: "stage_6h"},
-	{index: 3, expiresWithin: 1 * time.Hour, label: "stage_1h"},
-}
+// outermostReminderWindow is the widest reminder window — anything past
+// it is too far from expiry to be a candidate for any stage. Derived
+// from the shared package's canonical 12h threshold (no second copy of
+// the duration here).
+const outermostReminderWindow = resourcestatus.ExpiryWindow12h
 
 // reminderCooldown is the minimum wall-clock gap between two
 // dispatches on the same resource. Belt-and-braces — the CAS on
@@ -155,7 +161,7 @@ func (w *ExpiryReminderWorker) Work(ctx context.Context, job *river.Job[ExpiryRe
 	// The outermost window is the largest stage threshold — anything
 	// outside this window is too far from expiry to be a candidate
 	// for any stage. Inner windows are checked per-stage below.
-	windowEnd := now.Add(reminderSchedule[0].expiresWithin)
+	windowEnd := now.Add(outermostReminderWindow)
 	cooldownBefore := now.Add(-reminderCooldown)
 
 	// LEFT JOIN users u ... AND u.is_primary = true — the is_primary
@@ -314,72 +320,54 @@ func (w *ExpiryReminderWorker) Work(ctx context.Context, job *river.Job[ExpiryRe
 
 // selectStage picks the stage a row is currently eligible for.
 //
-// P2-12 (BugBash 2026-05-18): the stage is chosen by TIME BUCKET first,
-// then gated on reminders_sent — NOT the other way round. The prior
-// implementation required strict monotonic 0→1→2 progression
-// (`reminders_sent == mustHaveSent`), so a resource created less than
-// 6h before its TTL — already inside the (6h, 1h] or (1h, 0h] window
-// from the very first sweep — still fired stage 1 because that was the
-// only stage whose mustHaveSent matched reminders_sent=0. The email
-// then claimed "12h to go" while only ~50 min actually remained:
-// hours_remaining and stage_label disagreed.
+// The time-bucket classification is delegated to
+// resourcestatus.DeriveExpiryStage — the canonical, shared "most
+// imminent window wins" logic (the P2-12 BugBash fix now lives in the
+// common package, so api and worker can never disagree on the bucket).
+// selectStage layers the worker-specific reminders_sent CAS gate on top:
+// a stage already sent (reminders_sent >= stage index) is skipped.
 //
-// The fix: bucket the resource into the MOST IMMINENT stage window its
-// time-to-expiry falls in (schedule is most-distant → most-imminent, so
-// the last matching entry wins), then fire it only if reminders_sent is
-// still below that stage's index. The CAS in Work() fast-forwards
-// reminders_sent straight to stage.index, consuming any skipped earlier
-// stages. Result: a short-TTL resource gets exactly one correctly
-// labelled reminder ("1h to go"), not a mislabelled "12h" one.
+// P2-12 (BugBash 2026-05-18) recap: the stage is chosen by TIME BUCKET
+// first, then gated on reminders_sent — NOT the other way round. A
+// resource created less than 6h before its TTL is bucketed straight into
+// stage_6h / stage_1h; the CAS in Work() fast-forwards reminders_sent to
+// that stage's index, consuming the skipped earlier stages so exactly one
+// correctly labelled reminder fires.
 //
 // Examples:
-//   remaining 8h, reminders_sent 0 → stage 1 ("12h") — bucket (12h,6h]
-//   remaining 4h, reminders_sent 0 → stage 2 ("6h")  — skips stage 1
-//   remaining 40m, reminders_sent 0 → stage 3 ("1h") — skips 1 and 2
-//   remaining 4h, reminders_sent 2 → no stage (already past stage 2)
+//
+//	remaining 8h, reminders_sent 0 → stage 1 ("12h") — bucket (12h,6h]
+//	remaining 4h, reminders_sent 0 → stage 2 ("6h")  — skips stage 1
+//	remaining 40m, reminders_sent 0 → stage 3 ("1h") — skips 1 and 2
+//	remaining 4h, reminders_sent 2 → no stage (already past stage 2)
 func selectStage(r expiryReminderRow, now time.Time) (reminderStage, bool) {
-	remaining := r.expiresAt.Sub(now)
-	if remaining <= 0 {
-		return reminderStage{}, false
-	}
-	var bucket reminderStage
-	found := false
-	for _, s := range reminderSchedule {
-		if remaining <= s.expiresWithin {
-			// schedule is ordered most-distant → most-imminent, so
-			// later matches overwrite earlier ones — the final match
-			// is the tightest window the resource currently sits in.
-			bucket = s
-			found = true
-		}
-	}
-	if !found {
-		// Outside even the widest (12h) window — too far from expiry.
+	es := resourcestatus.DeriveExpiryStage(r.expiresAt, now)
+	if !es.IsWarning() {
+		// ExpiryStageNone (too far out) or ExpiryStagePastTTL — no
+		// reminder stage applies.
 		return reminderStage{}, false
 	}
+	bucket := reminderStage{stage: es, index: es.Index(), label: es.Label()}
 	if r.remindersSent >= bucket.index {
 		// This stage (or a later one) was already sent — nothing to do.
 		return reminderStage{}, false
 	}
 	return bucket, true
 }
 
-// hoursLeft rounds the gap up to whole hours, with a floor of 1 so
-// the email never says "0 hours". The legacy worker had the same
-// floor (it just always rendered the floor due to a Brevo template bug).
+// hoursLeft rounds the gap up to whole hours, with a floor of 1 so the
+// email never says "0 hours". Delegates to the shared
+// resourcestatus.HoursUntilExpiry — identical floor-of-1 / round-up
+// semantics, now shared so api and worker render the same number.
 func hoursLeft(expires, now time.Time) int {
-	delta := expires.Sub(now)
-	if delta <= time.Hour {
+	h := resourcestatus.HoursUntilExpiry(expires, now)
+	if h < 1 {
+		// HoursUntilExpiry returns 0 for a past-TTL / zero expiry; the
+		// reminder path only ever calls this for a future expiry, but
+		// keep the floor of 1 so the email copy never regresses.
 		return 1
 	}
-	hours := int(delta.Hours())
-	if delta-time.Duration(hours)*time.Hour > 0 {
-		hours++
-	}
-	if hours < 1 {
-		hours = 1
-	}
-	return hours
+	return h
 }
 
 // emitAnonExpiryWarningAudit writes one anon.expiry_warning audit_log

internal/jobs/resourcestatus_conversion_test.go

@@ -0,0 +1,145 @@
+package jobs
+
+// resourcestatus_conversion_test.go — proves the worker expiry-stage
+// helpers converted to instant.dev/common/resourcestatus behave
+// IDENTICALLY to the pre-conversion hand-written logic.
+//
+// This test lives in `package jobs` (not jobs_test) so it can reach the
+// unexported selectStage / hoursLeft helpers directly.
+//
+// Each test re-implements the ORIGINAL algorithm inline and asserts the
+// converted helper agrees for every input. If the shared package ever
+// drifts from the worker's original semantics, these tests fail.
+
+import (
+	"testing"
+	"time"
+
+	"instant.dev/common/resourcestatus"
+)
+
+// legacyReminderStage / legacySchedule reproduce the pre-conversion
+// hand-written 12h/6h/1h table that lived in expiry_reminder.go.
+type legacyReminderStage struct {
+	index         int
+	expiresWithin time.Duration
+}
+
+var legacySchedule = []legacyReminderStage{
+	{index: 1, expiresWithin: 12 * time.Hour},
+	{index: 2, expiresWithin: 6 * time.Hour},
+	{index: 3, expiresWithin: 1 * time.Hour},
+}
+
+// legacySelectStage is the ORIGINAL selectStage body, verbatim, used as
+// the oracle the converted selectStage must match.
+func legacySelectStage(r expiryReminderRow, now time.Time) (int, bool) {
+	remaining := r.expiresAt.Sub(now)
+	if remaining <= 0 {
+		return 0, false
+	}
+	var bucketIndex int
+	found := false
+	for _, s := range legacySchedule {
+		if remaining <= s.expiresWithin {
+			bucketIndex = s.index
+			found = true
+		}
+	}
+	if !found {
+		return 0, false
+	}
+	if r.remindersSent >= bucketIndex {
+		return 0, false
+	}
+	return bucketIndex, true
+}
+
+// legacyHoursLeft is the ORIGINAL hoursLeft body, verbatim.
+func legacyHoursLeft(expires, now time.Time) int {
+	delta := expires.Sub(now)
+	if delta <= time.Hour {
+		return 1
+	}
+	hours := int(delta.Hours())
+	if delta-time.Duration(hours)*time.Hour > 0 {
+		hours++
+	}
+	if hours < 1 {
+		hours = 1
+	}
+	return hours
+}
+
+// TestSelectStage_EquivalentToLegacy proves the converted selectStage
+// (now delegating to resourcestatus.DeriveExpiryStage) returns the same
+// (stage index, eligible) result as the original hand-written version,
+// across every time-bucket boundary and every reminders_sent value.
+func TestSelectStage_EquivalentToLegacy(t *testing.T) {
+	now := time.Date(2026, 5, 19, 12, 0, 0, 0, time.UTC)
+	offsets := []time.Duration{
+		-time.Hour,              // past TTL
+		0,                       // exactly now
+		time.Nanosecond,         // 1ns out
+		40 * time.Minute,        // inside 1h
+		time.Hour,               // exactly 1h
+		time.Hour + time.Minute, // just over 1h
+		4 * time.Hour,           // inside 6h
+		6 * time.Hour,           // exactly 6h
+		8 * time.Hour,           // inside 12h
+		12 * time.Hour,          // exactly 12h
+		13 * time.Hour,          // beyond 12h
+		24 * time.Hour,          // far out
+	}
+	for _, off := range offsets {
+		for remindersSent := 0; remindersSent <= 4; remindersSent++ {
+			r := expiryReminderRow{
+				expiresAt:     now.Add(off),
+				remindersSent: remindersSent,
+			}
+			wantIdx, wantOK := legacySelectStage(r, now)
+			gotStage, gotOK := selectStage(r, now)
+			if gotOK != wantOK {
+				t.Errorf("offset=%v remindersSent=%d: ok=%v, legacy ok=%v",
+					off, remindersSent, gotOK, wantOK)
+				continue
+			}
+			if gotOK && gotStage.index != wantIdx {
+				t.Errorf("offset=%v remindersSent=%d: index=%d, legacy index=%d",
+					off, remindersSent, gotStage.index, wantIdx)
+			}
+			// Label and shared-enum agreement.
+			if gotOK {
+				es := resourcestatus.DeriveExpiryStage(r.expiresAt, now)
+				if gotStage.label != es.Label() || gotStage.stage != es {
+					t.Errorf("offset=%v: stage wrapper out of sync with shared enum", off)
+				}
+			}
+		}
+	}
+}
+
+// TestHoursLeft_EquivalentToLegacy proves the converted hoursLeft (now
+// delegating to resourcestatus.HoursUntilExpiry) matches the original
+// hand-written rounding for every future-expiry input.
+func TestHoursLeft_EquivalentToLegacy(t *testing.T) {
+	now := time.Date(2026, 5, 19, 12, 0, 0, 0, time.UTC)
+	offsets := []time.Duration{
+		time.Nanosecond,
+		30 * time.Minute,
+		time.Hour,
+		time.Hour + time.Minute,
+		2 * time.Hour,
+		10 * time.Hour,
+		10*time.Hour + 30*time.Minute,
+		23 * time.Hour,
+	}
+	for _, off := range offsets {
+		expires := now.Add(off)
+		want := legacyHoursLeft(expires, now)
+		got := hoursLeft(expires, now)
+		if got != want {
+			t.Errorf("offset=%v: hoursLeft=%d, legacy=%d", off, got, want)
+		}
+	}
+}