billing: make Stripe webhook delivery idempotent and retryable

Webhook delivery was neither idempotent nor order-safe, and every failure
returned 400, which Stripe does not retry. A transient DB or hold error
therefore dropped a subscription change silently and permanently.

  - New stripe_processed_events table: event_id as primary key dedups
    redelivery, and event_created per customer drops stale out-of-order
    deliveries.
  - HandleWebhook distinguishes ErrWebhookSignature (400, no retry) from
    ErrWebhookProcessing (500, Stripe redelivers). The event handlers
    return errors instead of swallowing them. ErrBillingDisabled maps to
    400: the route is mounted but billing is off, so redelivery can never
    succeed and Stripe should stop rather than retry to exhaustion.
  - Refuse to boot when billing is enabled with an empty
    STRIPE_WEBHOOK_SECRET. Stripe HMACs with the empty key, so an
    attacker can reproduce the signature and the endpoint is forgeable.
  - UpdateCrewTierOnAllHolds retries each hold (3 attempts, linear
    backoff, 5s per request) and returns a joined error so the webhook
    can fail and let Stripe redeliver.

The fan-out contacts holds concurrently rather than in sequence. Serially,
one unreachable hold burns the caller's entire 10s budget on its own
retries (3 x 5s plus backoff) and the holds after it are never contacted;
because Stripe redelivers in the same order, a persistently-down first
hold means the rest are never updated at all.

On the hold, the signature-validated sub claim is now the source of truth
for updateCrewTier: a mismatched body userDid is rejected with 403 rather
than retargeting the grant to another DID. "Not crew on this hold" is a
200 no-op, since the appview fans updates out to every managed hold and a
subscriber is not crew everywhere.

That no-op has to be told apart from a storage failure. GetCrewMember
collapsed both into one generic error, so a CAR-store failure read as
"not a member", answered 200, and let the appview record the event as
processed — losing the tier grant permanently, which is exactly the
failure mode this commit exists to prevent. Missing records now carry an
ErrCrewMemberNotFound sentinel, and anything else returns 500 so Stripe
redelivers.

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

pkg/appview/db/migrations/0028_create_stripe_processed_events.yaml

@@ -0,0 +1,9 @@
+description: Stripe webhook idempotency + ordering (dedup events, drop stale out-of-order deliveries)
+query: |
+  CREATE TABLE IF NOT EXISTS stripe_processed_events (
+      event_id      TEXT PRIMARY KEY,
+      customer_id   TEXT,
+      event_created INTEGER NOT NULL,
+      processed_at  TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+  );
+  CREATE INDEX IF NOT EXISTS idx_stripe_events_customer ON stripe_processed_events(customer_id);

pkg/appview/db/queries.go

@@ -3,6 +3,7 @@ package db
 import (
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/url"
 	"regexp"
@@ -1510,6 +1511,52 @@ func UpdateFirehoseCursor(db DBTX, cursor int64) error {
 	return err
 }
 
+// StripeEventSeen reports whether a Stripe event has already been successfully
+// processed (idempotency guard for webhook redelivery).
+func StripeEventSeen(db DBTX, eventID string) (bool, error) {
+	var one int
+	err := db.QueryRow("SELECT 1 FROM stripe_processed_events WHERE event_id = ?", eventID).Scan(&one)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// LatestStripeEventCreatedForCustomer returns the highest event_created
+// timestamp recorded for a customer's successfully processed events, or 0 if
+// none. Used to drop stale out-of-order deliveries.
+func LatestStripeEventCreatedForCustomer(db DBTX, customerID string) (int64, error) {
+	if customerID == "" {
+		return 0, nil
+	}
+	var created sql.NullInt64
+	err := db.QueryRow(
+		"SELECT MAX(event_created) FROM stripe_processed_events WHERE customer_id = ?",
+		customerID,
+	).Scan(&created)
+	if errors.Is(err, sql.ErrNoRows) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+	return created.Int64, nil
+}
+
+// RecordStripeEvent records a successfully processed Stripe event. Idempotent:
+// a duplicate event_id is ignored rather than erroring.
+func RecordStripeEvent(db DBTX, eventID, customerID string, created int64) error {
+	_, err := db.Exec(`
+        INSERT INTO stripe_processed_events (event_id, customer_id, event_created)
+        VALUES (?, ?, ?)
+        ON CONFLICT(event_id) DO NOTHING
+    `, eventID, customerID, created)
+	return err
+}
+
 // GetChildManifestPlatform returns the platform info for a manifest that is
 // referenced as a child of a manifest list. Returns nil if the digest is not
 // a child of any manifest list (i.e., it's a top-level single-arch manifest).

pkg/appview/db/schema.sql

@@ -193,6 +193,17 @@ CREATE TABLE IF NOT EXISTS jetstream_cursor (
     updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
 );
 
+-- Stripe webhook idempotency + ordering. One row per successfully processed
+-- Stripe event. Dedups redelivered events (event_id PRIMARY KEY) and lets us
+-- drop stale out-of-order deliveries (compare event_created per customer).
+CREATE TABLE IF NOT EXISTS stripe_processed_events (
+    event_id      TEXT PRIMARY KEY,
+    customer_id   TEXT,
+    event_created INTEGER NOT NULL,
+    processed_at  TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_stripe_events_customer ON stripe_processed_events(customer_id);
+
 CREATE TABLE IF NOT EXISTS stars (
     starrer_did TEXT NOT NULL,
     owner_did TEXT NOT NULL,

pkg/appview/db/stripe_events_test.go

@@ -0,0 +1,91 @@
+package db
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+)
+
+func TestStripeProcessedEvents_DedupAndOrdering(t *testing.T) {
+	safeName := strings.ReplaceAll(t.Name(), "/", "_")
+	d, err := InitDB(fmt.Sprintf("file:%s?mode=memory&cache=shared", safeName), LibsqlConfig{})
+	if err != nil {
+		t.Fatalf("init db: %v", err)
+	}
+	d.SetMaxOpenConns(1)
+	defer d.Close()
+
+	// Fresh DB: nothing seen, no latest timestamp.
+	seen, err := StripeEventSeen(d, "evt_1")
+	if err != nil {
+		t.Fatalf("seen empty: %v", err)
+	}
+	if seen {
+		t.Fatal("evt_1 should not be seen on a fresh DB")
+	}
+
+	latest, err := LatestStripeEventCreatedForCustomer(d, "cus_a")
+	if err != nil {
+		t.Fatalf("latest empty: %v", err)
+	}
+	if latest != 0 {
+		t.Errorf("latest = %d, want 0", latest)
+	}
+
+	// Record an event, then it should be seen and bump the customer's latest.
+	if err := RecordStripeEvent(d, "evt_1", "cus_a", 1000); err != nil {
+		t.Fatalf("record evt_1: %v", err)
+	}
+	seen, err = StripeEventSeen(d, "evt_1")
+	if err != nil {
+		t.Fatalf("seen evt_1: %v", err)
+	}
+	if !seen {
+		t.Fatal("evt_1 should be seen after recording")
+	}
+	latest, err = LatestStripeEventCreatedForCustomer(d, "cus_a")
+	if err != nil {
+		t.Fatalf("latest cus_a: %v", err)
+	}
+	if latest != 1000 {
+		t.Errorf("latest = %d, want 1000", latest)
+	}
+
+	// Duplicate record (webhook redelivery) is a no-op, not an error.
+	if err := RecordStripeEvent(d, "evt_1", "cus_a", 1000); err != nil {
+		t.Fatalf("duplicate record should not error: %v", err)
+	}
+
+	// A newer event raises the latest; ordering uses MAX, not insertion order.
+	if err := RecordStripeEvent(d, "evt_3", "cus_a", 3000); err != nil {
+		t.Fatalf("record evt_3: %v", err)
+	}
+	if err := RecordStripeEvent(d, "evt_2", "cus_a", 2000); err != nil {
+		t.Fatalf("record evt_2: %v", err)
+	}
+	latest, err = LatestStripeEventCreatedForCustomer(d, "cus_a")
+	if err != nil {
+		t.Fatalf("latest after more: %v", err)
+	}
+	if latest != 3000 {
+		t.Errorf("latest = %d, want 3000", latest)
+	}
+
+	// Latest is scoped per customer.
+	latest, err = LatestStripeEventCreatedForCustomer(d, "cus_b")
+	if err != nil {
+		t.Fatalf("latest cus_b: %v", err)
+	}
+	if latest != 0 {
+		t.Errorf("cus_b latest = %d, want 0", latest)
+	}
+
+	// Empty customer id short-circuits to 0.
+	latest, err = LatestStripeEventCreatedForCustomer(d, "")
+	if err != nil {
+		t.Fatalf("latest empty customer: %v", err)
+	}
+	if latest != 0 {
+		t.Errorf("empty-customer latest = %d, want 0", latest)
+	}
+}

pkg/appview/holdclient/tier_update.go

@@ -5,17 +5,28 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
 	"strings"
+	"sync"
+	"time"
 
 	"atcr.io/pkg/atproto"
 	"atcr.io/pkg/auth"
 	"github.com/bluesky-social/indigo/atproto/atcrypto"
 )
 
+// tierUpdateMaxAttempts bounds per-hold retries when pushing a tier update.
+const tierUpdateMaxAttempts = 3
+
+// tierUpdateHTTPClient bounds each hold request so a hung hold can't block the
+// Stripe webhook indefinitely (http.DefaultClient has no timeout). The caller
+// also imposes an overall context deadline across all holds and retries.
+var tierUpdateHTTPClient = &http.Client{Timeout: 5 * time.Second}
+
 // UpdateCrewTierOnHold calls io.atcr.hold.updateCrewTier on a specific hold.
 // It signs a short-lived JWT with the appview's P-256 key and sends the tier update request.
 func UpdateCrewTierOnHold(ctx context.Context, holdDID, holdURL, userDID string, tierRank int, privateKey *atcrypto.PrivateKeyP256, appviewDID string) error {
@@ -44,7 +55,7 @@ func UpdateCrewTierOnHold(ctx context.Context, holdDID, holdURL, userDID string,
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Authorization", "Bearer "+token)
 
-	resp, err := http.DefaultClient.Do(req)
+	resp, err := tierUpdateHTTPClient.Do(req)
 	if err != nil {
 		return fmt.Errorf("failed to call updateCrewTier on %s: %w", holdDID, err)
 	}
@@ -58,33 +69,71 @@ func UpdateCrewTierOnHold(ctx context.Context, holdDID, holdURL, userDID string,
 	return nil
 }
 
-// UpdateCrewTierOnAllHolds pushes a tier update to all managed holds.
-// It resolves each hold DID to a URL and calls updateCrewTier.
-// Errors are logged but do not cause the function to fail — best effort.
-func UpdateCrewTierOnAllHolds(ctx context.Context, managedHolds []string, userDID string, tierRank int, privateKey *atcrypto.PrivateKeyP256, appviewDID string) {
-	for _, holdDID := range managedHolds {
-		holdURL, err := atproto.ResolveHoldDIDToURL(ctx, holdDID)
-		if err != nil {
-			slog.Warn("Could not resolve hold DID to URL, skipping",
-				"holdDID", holdDID,
-				"error", err,
-			)
-			continue
-		}
+// UpdateCrewTierOnAllHolds pushes a tier update to all managed holds, retrying
+// each hold a bounded number of times. It returns a joined error covering every
+// hold that could not be updated (after retries), so callers — notably the
+// Stripe webhook — can fail the request and let Stripe redeliver. Returns nil
+// only when every managed hold was updated successfully.
+// Holds are contacted concurrently. Serially, one unreachable hold burns the
+// caller's whole deadline on its own retries (3 attempts x 5s plus backoff) and
+// the holds after it are never contacted at all — and since the caller retries
+// in the same order, a persistently-down first hold means later holds are never
+// updated. Fanning out gives every hold the same shot at the budget.
+func UpdateCrewTierOnAllHolds(ctx context.Context, managedHolds []string, userDID string, tierRank int, privateKey *atcrypto.PrivateKeyP256, appviewDID string) error {
+	errs := make([]error, len(managedHolds))
+
+	var wg sync.WaitGroup
+	for i, holdDID := range managedHolds {
+		wg.Go(func() {
+			holdURL, err := atproto.ResolveHoldDIDToURL(ctx, holdDID)
+			if err != nil {
+				slog.Warn("Could not resolve hold DID to URL",
+					"holdDID", holdDID,
+					"error", err,
+				)
+				errs[i] = fmt.Errorf("resolve hold %s: %w", holdDID, err)
+				return
+			}
+
+			if err := updateCrewTierWithRetry(ctx, holdDID, holdURL, userDID, tierRank, privateKey, appviewDID); err != nil {
+				slog.Error("Failed to update crew tier on hold",
+					"holdDID", holdDID,
+					"userDID", userDID,
+					"tierRank", tierRank,
+					"error", err,
+				)
+				errs[i] = err
+				return
+			}
 
-		if err := UpdateCrewTierOnHold(ctx, holdDID, holdURL, userDID, tierRank, privateKey, appviewDID); err != nil {
-			slog.Error("Failed to update crew tier on hold",
-				"holdDID", holdDID,
-				"userDID", userDID,
-				"tierRank", tierRank,
-				"error", err,
-			)
-		} else {
 			slog.Info("Updated crew tier on hold",
 				"holdDID", holdDID,
 				"userDID", userDID,
 				"tierRank", tierRank,
 			)
+		})
+	}
+	wg.Wait()
+
+	return errors.Join(errs...)
+}
+
+// updateCrewTierWithRetry calls UpdateCrewTierOnHold with bounded retries and a
+// small linear backoff, aborting early if the context is cancelled.
+func updateCrewTierWithRetry(ctx context.Context, holdDID, holdURL, userDID string, tierRank int, privateKey *atcrypto.PrivateKeyP256, appviewDID string) error {
+	var lastErr error
+	for attempt := 1; attempt <= tierUpdateMaxAttempts; attempt++ {
+		lastErr = UpdateCrewTierOnHold(ctx, holdDID, holdURL, userDID, tierRank, privateKey, appviewDID)
+		if lastErr == nil {
+			return nil
+		}
+		if attempt < tierUpdateMaxAttempts {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(time.Duration(attempt) * 200 * time.Millisecond):
+			}
 		}
 	}
+	return fmt.Errorf("update crew tier on %s after %d attempts: %w", holdDID, tierUpdateMaxAttempts, lastErr)
 }

pkg/appview/holdclient/tier_update_test.go

@@ -0,0 +1,84 @@
+package holdclient
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"atcr.io/pkg/atproto"
+	"github.com/bluesky-social/indigo/atproto/atcrypto"
+)
+
+func TestUpdateCrewTierWithRetry_SucceedsAfterTransientFailures(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+
+	var calls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Fail the first two attempts, succeed on the third.
+		if calls.Add(1) < 3 {
+			http.Error(w, "temporarily unavailable", http.StatusInternalServerError)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{"tierName":"bosun"}`))
+	}))
+	defer srv.Close()
+
+	err = updateCrewTierWithRetry(context.Background(), "did:web:hold", srv.URL, "did:plc:user", 1, priv, "did:web:appview")
+	if err != nil {
+		t.Fatalf("expected success after retries, got %v", err)
+	}
+	if got := calls.Load(); got != 3 {
+		t.Errorf("expected 3 attempts, got %d", got)
+	}
+}
+
+func TestUpdateCrewTierWithRetry_FailsAfterMaxAttempts(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+
+	var calls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls.Add(1)
+		http.Error(w, "down", http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+
+	err = updateCrewTierWithRetry(context.Background(), "did:web:hold", srv.URL, "did:plc:user", 1, priv, "did:web:appview")
+	if err == nil {
+		t.Fatal("expected error after exhausting retries")
+	}
+	if got := calls.Load(); got != int32(tierUpdateMaxAttempts) {
+		t.Errorf("expected %d attempts, got %d", tierUpdateMaxAttempts, got)
+	}
+}
+
+// Ensure the URL builder matches the expected hold endpoint, guarding against
+// accidental path drift (the retry test relies on hitting the test server).
+func TestUpdateCrewTierOnHold_PostsToEndpoint(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+
+	var gotPath string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	if err := UpdateCrewTierOnHold(context.Background(), "did:web:hold", srv.URL, "did:plc:user", 0, priv, "did:web:appview"); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if gotPath != atproto.HoldUpdateCrewTier {
+		t.Errorf("posted to %q, want %q", gotPath, atproto.HoldUpdateCrewTier)
+	}
+}

pkg/appview/server.go

@@ -266,6 +266,7 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer,
 		appviewDID,
 		cfg.Server.ManagedHolds,
 		baseURL,
+		s.Database,
 	)
 	// Allow hold captains to bypass billing feature gates
 	if len(cfg.Server.ManagedHolds) > 0 {
@@ -277,6 +278,11 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer,
 		})
 	}
 	if s.BillingManager.Enabled() {
+		// Fail closed: an empty Stripe webhook secret makes webhooks forgeable
+		// (Stripe HMACs with the empty key, which an attacker can reproduce).
+		if !s.BillingManager.WebhookConfigured() {
+			return nil, fmt.Errorf("billing is enabled but STRIPE_WEBHOOK_SECRET is not set; refusing to start with a forgeable webhook endpoint")
+		}
 		slog.Info("Billing enabled", "appview_did", appviewDID, "managed_holds", len(cfg.Server.ManagedHolds))
 		go s.BillingManager.RefreshHoldTiers()
 	}

pkg/billing/billing.go

@@ -4,7 +4,9 @@ package billing
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -14,6 +16,7 @@ import (
 	"sync"
 	"time"
 
+	"atcr.io/pkg/appview/db"
 	"atcr.io/pkg/appview/holdclient"
 	"github.com/bluesky-social/indigo/atproto/atcrypto"
 
@@ -36,6 +39,9 @@ type Manager struct {
 	stripeKey     string
 	webhookSecret string
 
+	// db is the writable appview database, used for Stripe webhook idempotency.
+	db *sql.DB
+
 	// Captain checker: bypasses billing for hold owners
 	captainChecker CaptainChecker
 
@@ -72,7 +78,8 @@ const priceCacheTTL = 1 * time.Hour
 
 // New creates a new billing manager with Stripe integration.
 // Env vars STRIPE_SECRET_KEY and STRIPE_WEBHOOK_SECRET take precedence over config values.
-func New(cfg *Config, privateKey *atcrypto.PrivateKeyP256, appviewDID string, managedHolds []string, baseURL string) *Manager {
+// database is the writable appview DB, used for Stripe webhook idempotency.
+func New(cfg *Config, privateKey *atcrypto.PrivateKeyP256, appviewDID string, managedHolds []string, baseURL string, database *sql.DB) *Manager {
 	stripeKey := os.Getenv("STRIPE_SECRET_KEY")
 	if stripeKey == "" {
 		stripeKey = cfg.StripeSecretKey
@@ -94,6 +101,7 @@ func New(cfg *Config, privateKey *atcrypto.PrivateKeyP256, appviewDID string, ma
 		baseURL:       baseURL,
 		stripeKey:     stripeKey,
 		webhookSecret: webhookSecret,
+		db:            database,
 		customerCache: make(map[string]*cachedCustomer),
 		priceCache:    make(map[string]*cachedPrice),
 		holdTierCache: make(map[string]*cachedHoldTiers),
@@ -110,6 +118,12 @@ func (m *Manager) isCaptain(userDID string) bool {
 	return m.captainChecker != nil && userDID != "" && m.captainChecker(userDID)
 }
 
+// WebhookConfigured reports whether a Stripe webhook signing secret is set.
+// Used at startup to fail closed: an empty secret makes webhooks forgeable.
+func (m *Manager) WebhookConfigured() bool {
+	return m.webhookSecret != ""
+}
+
 // Enabled returns true if billing is properly configured.
 func (m *Manager) Enabled() bool {
 	return m.cfg != nil && m.stripeKey != "" && len(m.cfg.Tiers) > 0
@@ -125,7 +139,6 @@ func (m *Manager) GetWebhookLimits(userDID string) (int, bool) {
 	if !m.Enabled() {
 		return 1, false
 	}
-
 	info, err := m.GetSubscriptionInfo(userDID)
 	if err != nil || info == nil {
 		return m.cfg.Tiers[0].MaxWebhooks, m.cfg.Tiers[0].WebhookAllTriggers
@@ -148,7 +161,6 @@ func (m *Manager) HasAIAdvisor(userDID string) bool {
 	if !m.Enabled() {
 		return false
 	}
-
 	info, err := m.GetSubscriptionInfo(userDID)
 	if err != nil || info == nil {
 		return m.cfg.Tiers[0].AIAdvisor
@@ -172,7 +184,6 @@ func (m *Manager) GetSupporterBadge(userDID string) string {
 	if !m.Enabled() {
 		return ""
 	}
-
 	info, err := m.GetSubscriptionInfo(userDID)
 	if err != nil || info == nil {
 		return ""
@@ -214,10 +225,9 @@ func (m *Manager) GetSubscriptionInfo(userDID string) (*SubscriptionInfo, error)
 	}
 
 	info := &SubscriptionInfo{
-		UserDID:         userDID,
-		PaymentsEnabled: true,
-		CurrentTier:     m.cfg.Tiers[0].Name, // default to lowest
-		TierRank:        0,
+		UserDID:     userDID,
+		CurrentTier: m.cfg.Tiers[0].Name, // default to lowest
+		TierRank:    0,
 	}
 
 	// Build tier list with live Stripe prices
@@ -271,7 +281,8 @@ func (m *Manager) GetSubscriptionInfo(userDID string) (*SubscriptionInfo, error)
 	params.Filters.AddFilter("status", "", "active")
 	iter := subscription.List(params)
 
-	for iter.Next() {
+	// Use the first active subscription only.
+	if iter.Next() {
 		sub := iter.Subscription()
 		info.SubscriptionID = sub.ID
 
@@ -292,7 +303,6 @@ func (m *Manager) GetSubscriptionInfo(userDID string) (*SubscriptionInfo, error)
 				}
 			}
 		}
-		break
 	}
 
 	// Mark current tier
@@ -388,69 +398,157 @@ func (m *Manager) GetBillingPortalURL(userDID, returnURL string) (*BillingPortal
 	return &BillingPortalResponse{PortalURL: s.URL}, nil
 }
 
+// ErrWebhookSignature wraps signature/parse failures. These are client errors:
+// the caller (not Stripe) sent a bad request, so the HTTP handler returns 400
+// and Stripe should not retry.
+var ErrWebhookSignature = errors.New("webhook signature verification failed")
+
+// ErrWebhookProcessing wraps transient processing failures (DB, tier push to a
+// hold). The HTTP handler returns 5xx so Stripe retries; idempotency makes the
+// retry safe.
+var ErrWebhookProcessing = errors.New("webhook processing failed")
+
 // HandleWebhook processes a Stripe webhook event.
 // On subscription changes, it pushes tier updates to all managed holds.
+//
+// Returns ErrWebhookSignature (-> 400, no retry) for bad signatures/parse, or a
+// wrapped ErrWebhookProcessing (-> 5xx, Stripe retries) when applying the event
+// failed. On success the event is recorded for idempotency so redeliveries are
+// no-ops.
 func (m *Manager) HandleWebhook(r *http.Request) error {
 	if !m.Enabled() {
 		return ErrBillingDisabled
 	}
 
+	// Fail closed: an empty secret makes signatures forgeable (Stripe HMACs with
+	// the empty key). Startup also guards this; this is defense in depth.
+	if m.webhookSecret == "" {
+		return fmt.Errorf("%w: webhook secret not configured", ErrWebhookProcessing)
+	}
+
 	body, err := io.ReadAll(r.Body)
 	if err != nil {
-		return fmt.Errorf("failed to read webhook body: %w", err)
+		return fmt.Errorf("%w: read body: %v", ErrWebhookProcessing, err)
 	}
 
 	event, err := webhook.ConstructEvent(body, r.Header.Get("Stripe-Signature"), m.webhookSecret)
 	if err != nil {
-		return fmt.Errorf("webhook signature verification failed: %w", err)
+		return fmt.Errorf("%w: %v", ErrWebhookSignature, err)
 	}
 
+	// Idempotency: skip events we've already successfully processed so a Stripe
+	// redelivery (or replay of a validly-signed event) is a no-op.
+	if m.db != nil && event.ID != "" {
+		seen, err := db.StripeEventSeen(m.db, event.ID)
+		if err != nil {
+			return fmt.Errorf("%w: idempotency check: %v", ErrWebhookProcessing, err)
+		}
+		if seen {
+			slog.Debug("Ignoring already-processed Stripe event", "eventID", event.ID, "type", event.Type)
+			return nil
+		}
+	}
+
+	var procErr error
 	switch event.Type {
 	case EventCheckoutSessionCompleted:
-		m.handleCheckoutCompleted(event)
+		procErr = m.handleCheckoutCompleted(event)
 	case EventSubscriptionCreated,
 		EventSubscriptionUpdated,
 		EventSubscriptionDeleted,
 		EventSubscriptionPaused,
 		EventSubscriptionResumed:
-		m.handleSubscriptionChange(event)
+		procErr = m.handleSubscriptionChange(event)
 	case EventInvoicePaymentFailed:
-		m.handleInvoicePaymentFailed(event)
+		procErr = m.handleInvoicePaymentFailed(event)
 	case EventChargeDisputeCreated:
-		m.handleChargeDisputeCreated(event)
+		procErr = m.handleChargeDisputeCreated(event)
 	default:
 		slog.Debug("Ignoring Stripe event", "type", event.Type)
 	}
+	if procErr != nil {
+		// Not recorded as processed: Stripe will retry and we'll re-apply.
+		return fmt.Errorf("%w: %v", ErrWebhookProcessing, procErr)
+	}
+
+	// Record success (idempotency + per-customer ordering). Best-effort: a record
+	// failure shouldn't fail an already-applied event, but log it.
+	if m.db != nil && event.ID != "" {
+		if err := db.RecordStripeEvent(m.db, event.ID, subscriptionEventCustomerID(event), int64(event.Created)); err != nil {
+			slog.Warn("Failed to record processed Stripe event", "eventID", event.ID, "error", err)
+		}
+	}
 
 	return nil
 }
 
+// subscriptionEventCustomerID extracts the Stripe customer ID from a
+// subscription lifecycle event, or "" for other event types. Only subscription
+// events use per-customer ordering, so other types are recorded with no
+// customer (dedup still works on event_id).
+func subscriptionEventCustomerID(event stripe.Event) string {
+	switch event.Type {
+	case EventSubscriptionCreated,
+		EventSubscriptionUpdated,
+		EventSubscriptionDeleted,
+		EventSubscriptionPaused,
+		EventSubscriptionResumed:
+		var sub stripe.Subscription
+		if err := json.Unmarshal(event.Data.Raw, &sub); err == nil && sub.Customer != nil {
+			return sub.Customer.ID
+		}
+	}
+	return ""
+}
+
 // handleCheckoutCompleted processes a checkout.session.completed event.
-func (m *Manager) handleCheckoutCompleted(event stripe.Event) {
+func (m *Manager) handleCheckoutCompleted(event stripe.Event) error {
 	var cs stripe.CheckoutSession
 	if err := json.Unmarshal(event.Data.Raw, &cs); err != nil {
 		slog.Error("Failed to parse checkout session", "error", err)
-		return
+		// Malformed payload: don't ask Stripe to retry a bad event.
+		return nil
 	}
 
 	slog.Info("Checkout completed", "customerID", cs.Customer.ID, "subscriptionID", cs.Subscription.ID)
 
 	// The subscription.created event will handle the tier update
+	return nil
 }
 
 // handleSubscriptionChange processes subscription lifecycle events.
-func (m *Manager) handleSubscriptionChange(event stripe.Event) {
+func (m *Manager) handleSubscriptionChange(event stripe.Event) error {
 	var sub stripe.Subscription
 	if err := json.Unmarshal(event.Data.Raw, &sub); err != nil {
 		slog.Error("Failed to parse subscription", "error", err)
-		return
+		// Malformed payload: don't ask Stripe to retry a bad event.
+		return nil
+	}
+
+	// Ordering guard: ignore an event older than the newest one we've already
+	// applied for this customer (Stripe does not guarantee delivery order, so a
+	// stale active could otherwise re-grant a canceled tier).
+	if m.db != nil && sub.Customer != nil && sub.Customer.ID != "" {
+		latest, err := db.LatestStripeEventCreatedForCustomer(m.db, sub.Customer.ID)
+		if err != nil {
+			return fmt.Errorf("ordering check: %w", err)
+		}
+		if latest > 0 && int64(event.Created) < latest {
+			slog.Warn("Ignoring stale out-of-order subscription event",
+				"customerID", sub.Customer.ID,
+				"subscriptionID", sub.ID,
+				"eventCreated", event.Created,
+				"latestApplied", latest,
+			)
+			return nil
+		}
 	}
 
 	// Get user DID from customer metadata
 	userDID := m.getCustomerDID(sub.Customer.ID)
 	if userDID == "" {
 		slog.Warn("No user DID found for Stripe customer", "customerID", sub.Customer.ID)
-		return
+		return nil
 	}
 
 	// Determine new tier from subscription
@@ -473,7 +571,7 @@ func (m *Manager) handleSubscriptionChange(event stripe.Event) {
 			"subscriptionID", sub.ID,
 			"customerID", sub.Customer.ID,
 		)
-		return
+		return nil
 
 	case stripe.SubscriptionStatusUnpaid,
 		stripe.SubscriptionStatusCanceled,
@@ -491,16 +589,16 @@ func (m *Manager) handleSubscriptionChange(event stripe.Event) {
 			"userDID", userDID,
 			"subscriptionID", sub.ID,
 		)
-		return
+		return nil
 
 	default:
 		slog.Debug("Ignoring subscription status", "status", sub.Status, "subscriptionID", sub.ID)
-		return
+		return nil
 	}
 
 	if tierName == "" {
 		slog.Warn("Could not resolve tier from subscription", "priceID", sub.Items.Data[0].Price.ID)
-		return
+		return nil
 	}
 
 	slog.Info("Pushing tier update to managed holds",
@@ -510,31 +608,43 @@ func (m *Manager) handleSubscriptionChange(event stripe.Event) {
 		"event", event.Type,
 	)
 
-	// Push tier update to all managed holds
-	go holdclient.UpdateCrewTierOnAllHolds(
-		context.Background(),
+	// Push tier update to all managed holds synchronously. If any hold fails
+	// (after retries) we return an error so the webhook 5xxs and Stripe retries
+	// the whole event; idempotency makes that retry safe. This closes the
+	// "paid but never received tier" hole the old fire-and-forget goroutine had.
+	//
+	// Bound the whole push so a hung hold can't block the webhook past Stripe's
+	// delivery timeout (which would pile up goroutines and trigger redelivery).
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := holdclient.UpdateCrewTierOnAllHolds(
+		ctx,
 		m.managedHolds,
 		userDID,
 		tierRank,
 		m.privateKey,
 		m.appviewDID,
-	)
+	); err != nil {
+		return fmt.Errorf("push tier to managed holds: %w", err)
+	}
 
 	// Invalidate customer cache
 	m.customerCacheMu.Lock()
 	delete(m.customerCache, userDID)
 	m.customerCacheMu.Unlock()
+
+	return nil
 }
 
 // handleInvoicePaymentFailed logs a failed invoice payment.
 // No tier change: handleSubscriptionChange reacts to the subsequent
 // past_due → unpaid transition. Stripe Smart Retries handle retry cadence
 // and customer-facing dunning emails.
-func (m *Manager) handleInvoicePaymentFailed(event stripe.Event) {
+func (m *Manager) handleInvoicePaymentFailed(event stripe.Event) error {
 	var inv stripe.Invoice
 	if err := json.Unmarshal(event.Data.Raw, &inv); err != nil {
 		slog.Error("Failed to parse invoice", "error", err)
-		return
+		return nil
 	}
 
 	customerID := ""
@@ -559,6 +669,7 @@ func (m *Manager) handleInvoicePaymentFailed(event stripe.Event) {
 		"attemptCount", inv.AttemptCount,
 		"nextAttempt", nextAttempt,
 	)
+	return nil
 }
 
 // handleChargeDisputeCreated logs a new chargeback.
@@ -566,11 +677,11 @@ func (m *Manager) handleInvoicePaymentFailed(event stripe.Event) {
 // than waiting for resolution. Stripe emails the account owner by default
 // (Dashboard → Settings → Team & security → Notifications → "Disputes and
 // inquiries").
-func (m *Manager) handleChargeDisputeCreated(event stripe.Event) {
+func (m *Manager) handleChargeDisputeCreated(event stripe.Event) error {
 	var dispute stripe.Dispute
 	if err := json.Unmarshal(event.Data.Raw, &dispute); err != nil {
 		slog.Error("Failed to parse dispute", "error", err)
-		return
+		return nil
 	}
 
 	customerID := ""
@@ -596,6 +707,7 @@ func (m *Manager) handleChargeDisputeCreated(event stripe.Event) {
 		"status", dispute.Status,
 		"evidenceDueBy", evidenceDueBy,
 	)
+	return nil
 }
 
 // subscriptionIDFromInvoice returns the subscription ID on an invoice, or "".

pkg/billing/billing_stub.go

@@ -3,6 +3,7 @@
 package billing
 
 import (
+	"database/sql"
 	"net/http"
 
 	"github.com/bluesky-social/indigo/atproto/atcrypto"
@@ -15,7 +16,7 @@ type Manager struct {
 }
 
 // New creates a no-op billing manager.
-func New(_ *Config, _ *atcrypto.PrivateKeyP256, _ string, _ []string, _ string) *Manager {
+func New(_ *Config, _ *atcrypto.PrivateKeyP256, _ string, _ []string, _ string, _ *sql.DB) *Manager {
 	return &Manager{}
 }
 
@@ -24,6 +25,9 @@ func (m *Manager) SetCaptainChecker(fn CaptainChecker) {
 	m.captainChecker = fn
 }
 
+// WebhookConfigured returns false when billing is not compiled in.
+func (m *Manager) WebhookConfigured() bool { return false }
+
 // Enabled returns false when billing is not compiled in.
 func (m *Manager) Enabled() bool { return false }
 

pkg/billing/events.go

@@ -5,10 +5,10 @@ package billing
 // Stripe webhook event types the appview subscribes to.
 //
 // When adding or removing an entry here, update four places:
-//   1. The switch in HandleWebhook (billing.go)
-//   2. The SubscribedEvents slice below
-//   3. The Stripe Dashboard endpoint subscription list
-//   4. docs/BILLING.md "Webhook Events" table
+//  1. The switch in HandleWebhook (billing.go)
+//  2. The SubscribedEvents slice below
+//  3. The Stripe Dashboard endpoint subscription list
+//  4. docs/BILLING.md "Webhook Events" table
 const (
 	// EventCheckoutSessionCompleted fires when a customer finishes Stripe Checkout.
 	// No-op handler: customer.subscription.created does the actual tier work.

pkg/billing/handlers.go

@@ -4,6 +4,7 @@ package billing
 
 import (
 	"encoding/json"
+	"errors"
 	"log/slog"
 	"net/http"
 
@@ -25,10 +26,26 @@ func (m *Manager) RegisterRoutes(r chi.Router) {
 }
 
 // handleStripeWebhook processes incoming Stripe webhook events.
+//
+// Status codes matter for Stripe's retry behavior: a bad signature is a client
+// error (400, no retry), while a transient processing failure returns 500 so
+// Stripe redelivers. Event idempotency makes redelivery safe.
 func (m *Manager) handleStripeWebhook(w http.ResponseWriter, r *http.Request) {
 	if err := m.HandleWebhook(r); err != nil {
 		slog.Error("Stripe webhook error", "error", err)
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		switch {
+		case errors.Is(err, ErrWebhookSignature):
+			http.Error(w, "invalid signature", http.StatusBadRequest)
+		case errors.Is(err, ErrBillingDisabled):
+			// Permanent: this deployment has the route mounted but billing off,
+			// so no amount of redelivery will make it processable. 400 tells
+			// Stripe to stop rather than retry the event to exhaustion.
+			http.Error(w, "billing not enabled", http.StatusBadRequest)
+		default:
+			// ErrWebhookProcessing and anything unclassified: assume transient
+			// and let Stripe retry. Event idempotency makes redelivery safe.
+			http.Error(w, "processing error", http.StatusInternalServerError)
+		}
 		return
 	}
 

pkg/hold/pds/crew.go

@@ -9,10 +9,17 @@ import (
 	"time"
 
 	"atcr.io/pkg/atproto"
+	"github.com/bluesky-social/indigo/mst"
 	"github.com/bluesky-social/indigo/repo"
 	"github.com/ipfs/go-cid"
 )
 
+// ErrCrewMemberNotFound reports that this hold's repo has no crew record for the
+// member. It is deliberately distinct from a storage failure: callers that treat
+// "not crew" as a benign no-op must not swallow a CAR-store error the same way,
+// or a failed lookup becomes indistinguishable from a genuine non-member.
+var ErrCrewMemberNotFound = errors.New("crew member not found")
+
 // AddCrewMember adds a new crew member to the hold and commits to carstore
 // Uses deterministic rkey based on member DID hash for O(1) lookups and automatic deduplication
 // If the member already exists, updates their record (upsert behavior)
@@ -42,6 +49,9 @@ func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *atp
 	// Use repomgr.GetRecord - our types are registered in init()
 	recordCID, val, err := p.repomgr.GetRecord(ctx, p.uid, atproto.CrewCollection, rkey, cid.Undef)
 	if err != nil {
+		if errors.Is(err, mst.ErrNotFound) {
+			return cid.Undef, nil, fmt.Errorf("%w: %s", ErrCrewMemberNotFound, rkey)
+		}
 		return cid.Undef, nil, fmt.Errorf("failed to get crew record: %w", err)
 	}
 

pkg/hold/pds/crew_test.go

@@ -2,6 +2,7 @@ package pds
 
 import (
 	"bytes"
+	"errors"
 	"strings"
 	"testing"
 
@@ -122,10 +123,14 @@ func TestGetCrewMember_NotFound(t *testing.T) {
 		t.Fatal("Expected error when getting non-existent crew member")
 	}
 
-	// Verify error message
-	errMsg := err.Error()
-	if !strings.Contains(errMsg, "failed to get crew record") {
-		t.Errorf("Expected 'failed to get crew record' in error, got: %s", errMsg)
+	// A missing record must be distinguishable from a storage failure. Callers
+	// that treat "not crew" as a benign no-op (updateCrewTier) would otherwise
+	// swallow a CAR-store error as a non-member and silently drop a tier grant.
+	if !errors.Is(err, ErrCrewMemberNotFound) {
+		t.Errorf("Expected ErrCrewMemberNotFound, got: %v", err)
+	}
+	if !strings.Contains(err.Error(), "nonexistent-rkey") {
+		t.Errorf("Expected the rkey in the error, got: %s", err)
 	}
 }
 

pkg/hold/pds/xrpc.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	"atcr.io/pkg/atproto"
@@ -1760,10 +1761,15 @@ func (h *XRPCHandler) HandleUpdateCrewTier(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	// Verify the userDid in the body matches the sub claim
+	// The signature-validated sub claim is the source of truth for which user
+	// this tier applies to. If the body also carries a userDid, it must match;
+	// reject a mismatch rather than letting the unauthenticated body retarget
+	// the grant to a different DID.
 	if req.UserDID != "" && req.UserDID != userDID {
-		// Use the body's userDid (the sub claim was the appview-specified user)
-		userDID = req.UserDID
+		slog.Warn("updateCrewTier body userDid does not match token sub",
+			"bodyUserDid", req.UserDID, "tokenSub", userDID)
+		http.Error(w, "userDid does not match authenticated subject", http.StatusForbidden)
+		return
 	}
 
 	if _, err := syntax.ParseDID(userDID); err != nil {
@@ -1778,6 +1784,27 @@ func (h *XRPCHandler) HandleUpdateCrewTier(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
+	// A tier only applies to crew members. The appview fans a tier update out to
+	// ALL managed holds, but a user is only crew on holds they've actually pushed
+	// to (and a brand-new subscriber may be crew nowhere yet). Treat "not crew on
+	// this hold" as a successful no-op rather than an error — otherwise the Stripe
+	// webhook would 500 and retry forever on holds where there's nothing to do.
+	// Only a genuinely missing record counts as "not crew"; see below.
+	if _, _, err := h.pds.GetCrewMemberByDID(r.Context(), userDID); err != nil {
+		if !errors.Is(err, ErrCrewMemberNotFound) {
+			// A storage failure is not the same as "not a member". Answering 200
+			// here would let the appview mark the Stripe event processed and drop
+			// the tier grant for good, so fail loudly and let Stripe redeliver.
+			slog.Error("Failed to look up crew membership for tier update",
+				"userDid", userDID, "tierName", tierName, "error", err)
+			http.Error(w, "failed to look up crew membership", http.StatusInternalServerError)
+			return
+		}
+		slog.Info("Skipping tier update; user is not crew on this hold", "userDid", userDID, "tierName", tierName)
+		render.JSON(w, r, map[string]any{"tierName": tierName, "applied": false})
+		return
+	}
+
 	// Update the crew member's tier
 	if err := h.pds.UpdateCrewMemberTier(r.Context(), userDID, tierName); err != nil {
 		slog.Error("Failed to update crew tier", "userDid", userDID, "tier", tierName, "error", err)
@@ -1787,7 +1814,7 @@ func (h *XRPCHandler) HandleUpdateCrewTier(w http.ResponseWriter, r *http.Reques
 
 	slog.Info("Updated crew tier via appview", "userDid", userDID, "tierRank", req.TierRank, "tierName", tierName)
 
-	render.JSON(w, r, map[string]string{"tierName": tierName})
+	render.JSON(w, r, map[string]any{"tierName": tierName, "applied": true})
 }
 
 // resolveTierByRank maps a 0-based rank index to a tier name from the quota config.

test/stripe-integration/manager_test.go

@@ -44,7 +44,7 @@ func newManager(t *testing.T, env stripeEnv) *billing.Manager {
 			},
 		},
 	}
-	return billing.New(cfg, nil, "did:web:test-appview.local", nil, "http://test-appview.local")
+	return billing.New(cfg, nil, "did:web:test-appview.local", nil, "http://test-appview.local", nil)
 }
 
 func TestManagerEnabled(t *testing.T) {