feat(worker): cohort skip-guards — no-op every team-iterating job for is_test_cohort teams (PR-1 follow-up) (#90)

* feat(worker): cohort skip-guards — no-op every team-iterating job for is_test_cohort teams (PR-1 follow-up)

Worker-side follow-up to api #246 (W0): teams.is_test_cohort (migration 067,
already on master) tags durable synthetic test-cohort teams. This makes every
background job that charges / churns / emails / quota-nudges a team SKIP an
is_test_cohort team, so continuous synthetic monitoring never pollutes the real
funnel / billing / email surface.

Plan: docs/sessions/2026-06-04/TEST-ACCOUNTS-AND-NR-SYNTHETICS-PLAN.md §1.6.

Mechanism — two skip surfaces, both UNCONDITIONAL + inert (no real team is
is_test_cohort=true today, so behaviour is unchanged for all real teams; this
is a safety guard, not a feature, and does NOT depend on FLOW_SYNTHETIC_ENABLED):
  - SQL scans add `AND NOT [t.]is_test_cohort` (teams-joined) or a NULL-safe
    NOT EXISTS subselect (testCohortNotExistsClause, for team_id-carrying scans).
  - Per-team Go loops can call isTestCohort/skipIfTestCohort (shared helper).

Guarded (§1.6): quota.go (suspend/unsuspend/redis-eviction scans),
quota_wall_nudge.go, churn_predictor.go, expire_imminent.go, expiry_reminder.go,
billing_reconciler.go (primary + orphan sweep), checkout_reconcile.go,
payment_grace_reminder.go, payment_grace_terminator.go, email.go (weekly_digest).

Deliberately NOT guarded (documented in test_cohort_registry_test.go): expire.go
+ expire_stacks.go (TTL REAPERS — must reap synthetic resources/stacks so they
never leak; no customer email), lifecycle_emails.go (pure render layer, gated by
the guarded trigger jobs upstream).

Tests:
  - test_cohort_test.go: sqlmock unit coverage of the helper (flagged /
    not-flagged / row-vanished / DB-error fail-safe-for-real-teams). 100% of
    test_cohort.go.
  - test_cohort_registry_test.go: rule-18 enumeration net —
    TestAllTeamIteratingJobs_FilterTestCohort fails if any enumerated job loses
    its guard, plus TestNotGuardedJobs_AreStillPresent pins the scope boundary.
  - test_cohort_integration_test.go: real-Postgres round-trips proving each
    SQL-driven job skips a cohort team while processing a normal team
    (quota_wall_nudge, churn_predictor, expiry_reminder, expire_imminent,
    weekly_digest, checkout_reconcile, payment_grace_reminder,
    billing_reconciler). Gated by SetupTestDB (skips no-DB lanes).

testhelpers: add is_test_cohort + the users/pending_checkouts/
payment_grace_periods subset tables + cohort seed/read helpers. The two-gate
note (ci.yml + deploy.yml) holds: the harness change is idempotent and DB-gated
so both -short, no-DB workflows stay green.

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

* test(testhelpers): in-package smoke test for cohort.go → 100% (unblock #90 coverage)

cohort.go is exercised only cross-package by internal/jobs integration tests, so
the per-package coverage job reported it 0% (94 missing lines) → 100%-patch gate
red. Add an in-package smoke test (DB-gated via SetupTestDB, runs in coverage.yml)
bringing all 8 seed/query helpers to 100%, mirroring billing_deletion_smoke_test.go.
Recurring testhelpers gotcha (3rd: #87/#89/#90) — see memory.

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

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mastermanas805

internal/jobs/billing_reconciler.go

@@ -822,6 +822,7 @@ func (w *BillingReconcilerWorker) Work(ctx context.Context, job *river.Job[Billi
 		  FROM teams
 		 WHERE stripe_customer_id IS NOT NULL
 		   AND stripe_customer_id != ''
+		   AND NOT is_test_cohort
 		 ORDER BY id
 		 LIMIT $1
 	`, billingReconcilerBatchLimit)
@@ -1204,6 +1205,7 @@ func (w *BillingReconcilerWorker) runOrphanSweep(ctx context.Context) (scanned,
 		  JOIN teams t ON t.id = pc.team_id
 		 WHERE pc.subscription_id IS NOT NULL
 		   AND pc.subscription_id != ''
+		   AND NOT t.is_test_cohort
 		   AND pc.created_at < $1
 		 ORDER BY pc.created_at
 		 LIMIT $2

internal/jobs/checkout_reconcile.go

@@ -148,9 +148,10 @@ func (w *CheckoutReconcileWorker) Work(ctx context.Context, job *river.Job[Check
 
 	rows, err := w.db.QueryContext(ctx, `
 		SELECT subscription_id, team_id, customer_email, plan_tier
-		FROM pending_checkouts
+		FROM pending_checkouts pc
 		WHERE resolved_at IS NULL
 		  AND failure_notified_at IS NULL
+		  AND `+testCohortNotExistsClause("pc.team_id")+`
 		  AND created_at < $1
 		ORDER BY created_at ASC
 		LIMIT $2

internal/jobs/churn_predictor.go

@@ -87,14 +87,14 @@ const churnInactivityWindow = 7 * 24 * time.Hour
 // churnDedupeWindow is the minimum gap between two churn.risk_flagged
 // rows for the same team. 30 days is the brief's window. Rationale:
 //
-//   * Slower than 30d would let a team get re-flagged month after month
+//   - Slower than 30d would let a team get re-flagged month after month
 //     without intervention — operationally noisy and emotionally
 //     repetitive ("you again missed us") for the customer.
-//   * Faster than 30d (e.g. weekly) would risk sending the same
+//   - Faster than 30d (e.g. weekly) would risk sending the same
 //     "we miss you" copy four times in a month to a customer who has
 //     decided not to return; the perception cost (spammy) outweighs
 //     the recall benefit.
-//   * 30d also aligns with monthly billing cycles — most B2B SaaS
+//   - 30d also aligns with monthly billing cycles — most B2B SaaS
 //     reactivation playbooks rate-limit on a monthly cadence for the
 //     same reason.
 //
@@ -210,6 +210,7 @@ func (w *ChurnPredictorWorker) Work(ctx context.Context, job *river.Job[ChurnPre
 			)
 		LEFT JOIN resources r ON r.team_id = t.id AND r.status = 'active'
 		WHERE t.plan_tier != 'team'
+		  AND NOT t.is_test_cohort
 		  AND NOT EXISTS (
 			SELECT 1 FROM audit_log f
 			WHERE f.team_id = t.id
@@ -300,10 +301,10 @@ func (w *ChurnPredictorWorker) Work(ctx context.Context, job *river.Job[ChurnPre
 		// the brief enumerated lives here: tier, last_activity_days_ago,
 		// active_resource_count, email.
 		meta := map[string]any{
-			"tier":                    r.planTier,
-			"last_activity_days_ago":  daysSince,
-			"active_resource_count":   r.activeResourceCount,
-			"email":                   r.ownerEmail.String,
+			"tier":                   r.planTier,
+			"last_activity_days_ago": daysSince,
+			"active_resource_count":  r.activeResourceCount,
+			"email":                  r.ownerEmail.String,
 		}
 		metaBytes, mErr := churnMetaMarshal(meta)
 		if mErr != nil {

internal/jobs/email.go

@@ -120,6 +120,7 @@ func (w *WeeklyDigestWorker) Work(ctx context.Context, job *river.Job[WeeklyDige
 		FROM users u
 		JOIN teams t ON t.id = u.team_id
 		WHERE t.plan_tier != 'anonymous'
+		  AND NOT t.is_test_cohort
 		  AND NOT EXISTS (
 		    SELECT 1 FROM audit_log a
 		    WHERE a.team_id = t.id
@@ -228,11 +229,12 @@ func (w *WeeklyDigestWorker) buildResourceDigestCounts(ctx context.Context, team
 // to if a per-row count is missing.
 //
 // Metadata shape — buildDigestWeekly (event_email_mapping.go) reads these:
-//   email                   — recipient address (also the forwarder's resolver)
-//   team_name               — display name for the email greeting (may be "")
-//   total_active_resources  — sum across all resource_types; 0 means "empty week"
-//   resource_breakdown      — JSON-encoded array of {resource_type, count}
-//                             (templates that don't render a table can ignore)
+//
+//	email                   — recipient address (also the forwarder's resolver)
+//	team_name               — display name for the email greeting (may be "")
+//	total_active_resources  — sum across all resource_types; 0 means "empty week"
+//	resource_breakdown      — JSON-encoded array of {resource_type, count}
+//	                          (templates that don't render a table can ignore)
 func emitWeeklyDigestAudit(ctx context.Context, db *sql.DB, teamID uuid.UUID, email, teamName string, stats []DigestResourceCount) error {
 	var total int64
 	for _, s := range stats {
@@ -268,4 +270,3 @@ func emitWeeklyDigestAudit(ctx context.Context, db *sql.DB, teamID uuid.UUID, em
 	}
 	return nil
 }
-

internal/jobs/expire_imminent.go

@@ -167,6 +167,7 @@ func (w *ExpireImminentWorker) Work(ctx context.Context, job *river.Job[ExpireIm
 		  AND r.expires_at IS NOT NULL
 		  AND r.expires_at > $1
 		  AND r.expires_at < $2
+		  AND `+testCohortNotExistsClause("r.team_id")+`
 		  AND NOT EXISTS (
 			SELECT 1
 			FROM audit_log al
@@ -303,4 +304,3 @@ func (w *ExpireImminentWorker) Work(ctx context.Context, job *river.Job[ExpireIm
 	)
 	return nil
 }
-

internal/jobs/expiry_reminder.go

@@ -181,6 +181,7 @@ func (w *ExpiryReminderWorker) Work(ctx context.Context, job *river.Job[ExpiryRe
 		WHERE r.team_id IS NOT NULL
 		  AND r.tier = 'free'
 		  AND r.status = 'active'
+		  AND `+testCohortNotExistsClause("r.team_id")+`
 		  AND r.expires_at IS NOT NULL
 		  AND r.expires_at > $1
 		  AND r.expires_at <= $2

internal/jobs/payment_grace_reminder.go

@@ -113,10 +113,11 @@ func (w *PaymentGraceReminderWorker) Work(ctx context.Context, job *river.Job[Pa
 	// first so a backlog drains in FIFO order.
 	rows, err := w.db.QueryContext(ctx, `
 		SELECT id, team_id, expires_at
-		FROM payment_grace_periods
+		FROM payment_grace_periods pgp
 		WHERE status = 'active'
 		  AND expires_at > $1
 		  AND (last_reminder_at IS NULL OR last_reminder_at < $2)
+		  AND `+testCohortNotExistsClause("pgp.team_id")+`
 		ORDER BY last_reminder_at ASC NULLS FIRST, started_at ASC
 		LIMIT $3
 	`, now, cutoff, paymentGraceReminderBatchLimit)
@@ -186,9 +187,9 @@ func (w *PaymentGraceReminderWorker) Work(ctx context.Context, job *river.Job[Pa
 
 		hoursRemaining := math.Round(r.expiresAt.Sub(now).Hours()*10) / 10
 		meta := map[string]any{
-			"grace_id":         r.id.String(),
-			"hours_remaining":  hoursRemaining,
-			"grace_ends_at":    r.expiresAt.UTC().Format(time.RFC3339),
+			"grace_id":        r.id.String(),
+			"hours_remaining": hoursRemaining,
+			"grace_ends_at":   r.expiresAt.UTC().Format(time.RFC3339),
 		}
 		metaBytes, mErr := graceReminderMetaMarshal(meta)
 		if mErr != nil {

internal/jobs/payment_grace_terminator.go

@@ -149,9 +149,10 @@ func (w *PaymentGraceTerminatorWorker) Work(ctx context.Context, job *river.Job[
 	// Candidate query: active grace rows whose clock has expired.
 	rows, err := w.db.QueryContext(ctx, `
 		SELECT id, team_id, expires_at
-		FROM payment_grace_periods
+		FROM payment_grace_periods pgp
 		WHERE status = 'active'
 		  AND expires_at < $1
+		  AND `+testCohortNotExistsClause("pgp.team_id")+`
 		ORDER BY expires_at ASC
 		LIMIT $2
 	`, now, paymentGraceTerminatorBatchLimit)
@@ -308,14 +309,15 @@ func (w *PaymentGraceTerminatorWorker) terminate(ctx context.Context, teamID uui
 // HS256 algorithm and the claim set below.
 //
 // Claims:
-//   sub  — the team id being terminated (allows the api to assert the
-//          path :id matches the JWT subject; prevents a stolen token
-//          from terminating a different team)
-//   iss  — "instanode-worker" so the api can route on issuer
-//   iat  — issued-at (UTC seconds)
-//   exp  — issued-at + 5 minutes (short window — single-shot use)
-//   aud  — "internal-teams-terminate" so the same secret can't be
-//          re-used by an attacker against a different internal route
+//
+//	sub  — the team id being terminated (allows the api to assert the
+//	       path :id matches the JWT subject; prevents a stolen token
+//	       from terminating a different team)
+//	iss  — "instanode-worker" so the api can route on issuer
+//	iat  — issued-at (UTC seconds)
+//	exp  — issued-at + 5 minutes (short window — single-shot use)
+//	aud  — "internal-teams-terminate" so the same secret can't be
+//	       re-used by an attacker against a different internal route
 func signWorkerInternalJWT(secret, teamID string) (string, error) {
 	if secret == "" {
 		return "", errors.New("empty secret")

internal/jobs/quota.go

@@ -241,9 +241,10 @@ func (w *EnforceStorageQuotaWorker) runRedisEvictionLoop(ctx context.Context) (i
 	for {
 		rows, err := w.db.QueryContext(ctx, `
 			SELECT id, token, tier, storage_bytes
-			FROM resources
+			FROM resources r
 			WHERE status = $1
 			  AND resource_type = 'redis'
+			  AND `+testCohortNotExistsClause("r.team_id")+`
 			  AND id::text > $2
 			ORDER BY id::text ASC
 			LIMIT $3
@@ -362,9 +363,10 @@ func (w *EnforceStorageQuotaWorker) runSuspendLoop(ctx context.Context) ([]strin
 			SELECT id, token, resource_type, tier, storage_bytes,
 			       COALESCE(provider_resource_id, ''),
 			       team_id, COALESCE(name, '')
-			FROM resources
+			FROM resources r
 			WHERE status = $1
 			  AND resource_type IN ('postgres', 'redis', 'mongodb')
+			  AND `+testCohortNotExistsClause("r.team_id")+`
 			  AND id::text > $2
 			ORDER BY id::text ASC
 			LIMIT $3
@@ -528,9 +530,10 @@ func (w *EnforceStorageQuotaWorker) runUnsuspendLoop(ctx context.Context, skipID
 			SELECT id, token, resource_type, tier, storage_bytes,
 			       COALESCE(provider_resource_id, ''),
 			       team_id, COALESCE(name, '')
-			FROM resources
+			FROM resources r
 			WHERE status = $1
 			  AND resource_type IN ('postgres', 'redis', 'mongodb')
+			  AND `+testCohortNotExistsClause("r.team_id")+`
 			  AND id::text > $2
 			ORDER BY id::text ASC
 			LIMIT $3

internal/jobs/quota_wall_nudge.go

@@ -145,6 +145,7 @@ func (w *QuotaWallNudgeWorker) Work(ctx context.Context, job *river.Job[QuotaWal
 			SELECT id, plan_tier
 			FROM teams
 			WHERE plan_tier NOT IN ('team', 'anonymous', 'free')
+			  AND NOT is_test_cohort
 			  AND id::text > $1
 			ORDER BY id::text ASC
 			LIMIT $2