at-container-registry/pkg/billing/webhook_retry_test.go

//go:build billing

package billing

import (
	"database/sql"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	appdb "atcr.io/pkg/appview/db"
	"atcr.io/pkg/atproto"
	"github.com/bluesky-social/indigo/atproto/atcrypto"
	"github.com/stripe/stripe-go/v84"
	"github.com/stripe/stripe-go/v84/webhook"
)

// newTestManager builds a Manager with a REAL database.
//
// test/stripe-integration builds its manager with a nil database, so every
// `m.db != nil` branch — which is all of the idempotency and ordering work
// 12c55ed added — is skipped there. A suite that never touches those branches
// cannot notice them regressing.
func newTestManager(t *testing.T, secret string) (*Manager, *sql.DB) {
	t.Helper()

	// A file rather than ":memory:", which is per-connection in libsql and
	// produces "no such table" once the pool opens a second one.
	database, err := appdb.InitDB(filepath.Join(t.TempDir(), "ui.db"), appdb.LibsqlConfig{})
	if err != nil {
		t.Fatalf("init db: %v", err)
	}
	t.Cleanup(func() { _ = database.Close() })

	// New() reads these first, so a stray value in the environment would
	// otherwise decide the test.
	t.Setenv("STRIPE_SECRET_KEY", "sk_test_fake")
	t.Setenv("STRIPE_WEBHOOK_SECRET", secret)

	cfg := &Config{
		StripeSecretKey: "sk_test_fake",
		WebhookSecret:   secret,
		// Rank is positional (0-based by list order), not a field.
		Tiers: []BillingTierConfig{
			{Name: "free"},
			{Name: "supporter", StripePriceMonthly: "price_monthly_test"},
		},
	}
	m := New(cfg, nil, "did:web:test-appview.local", nil, "http://test-appview.local", database)
	if !m.Enabled() {
		t.Fatal("manager should be enabled")
	}
	return m, database
}

// stripeAPIReturning points the global stripe-go backend at a stub that answers
// every API call with the given status and body, so a Stripe-side failure can
// be simulated without network access.
// Returns a counter of API calls made. Counting is what makes the idempotency
// and ordering tests real: both guards short-circuit BEFORE the customer
// lookup, so "did Stripe get called again" is the only observable difference.
// Row counts are not — RecordStripeEvent is an idempotent upsert, so removing
// the guards entirely leaves the table looking identical.
func stripeAPIReturning(t *testing.T, status int, body string) *atomic.Int64 {
	t.Helper()
	var calls atomic.Int64
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls.Add(1)
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(status)
		_, _ = w.Write([]byte(body))
	}))
	t.Cleanup(srv.Close)

	prev := stripe.GetBackend(stripe.APIBackend)
	stripe.SetBackend(stripe.APIBackend, stripe.GetBackendWithConfig(stripe.APIBackend, &stripe.BackendConfig{
		URL:               stripe.String(srv.URL),
		MaxNetworkRetries: stripe.Int64(0),
	}))
	t.Cleanup(func() { stripe.SetBackend(stripe.APIBackend, prev) })
	return &calls
}

func signedSubscriptionEvent(t *testing.T, secret, eventID, customerID string, created int64) []byte {
	t.Helper()
	// api_version is required: ConstructEvent rejects a payload whose version
	// does not match what stripe-go expects, before any of the logic under test
	// runs, and the resulting error looks like a signature failure.
	payload := fmt.Sprintf(`{
		"id": %q,
		"object": "event",
		"api_version": %q,
		"type": "customer.subscription.updated",
		"created": %d,
		"data": {"object": {
			"id": "sub_test",
			"object": "subscription",
			"status": "active",
			"customer": %q,
			"items": {"object":"list","data":[{"price":{"id":"price_monthly_test"}}]}
		}}
	}`, eventID, stripe.APIVersion, created, customerID)
	return []byte(payload)
}

func postWebhook(t *testing.T, m *Manager, secret string, payload []byte) error {
	t.Helper()
	signed := webhook.GenerateTestSignedPayload(&webhook.UnsignedPayload{
		Payload: payload, Secret: secret, Timestamp: time.Now(),
	})
	req := httptest.NewRequest(http.MethodPost, "/api/stripe/webhook", strings.NewReader(string(signed.Payload)))
	req.Header.Set("Stripe-Signature", signed.Header)
	return m.HandleWebhook(req)
}

func processedCount(t *testing.T, database *sql.DB) int {
	t.Helper()
	var n int
	if err := database.QueryRow("SELECT COUNT(*) FROM stripe_processed_events").Scan(&n); err != nil {
		t.Fatalf("count processed events: %v", err)
	}
	return n
}

// TestHandleSubscriptionChange_CustomerLookupFailureIsRetryable is the one the
// plan flags as the open defect, and it is real.
//
// getCustomerDID returns "" for a FAILED customer.Get exactly as it does for a
// customer with no user_did metadata. handleSubscriptionChange reads that as
// "not our customer" and returns nil, so HandleWebhook records the event as
// processed and answers 200. Stripe never redelivers, and a paid upgrade is
// dropped permanently by a transient API blip — the precise "paid but never
// received tier" hole 12c55ed was written to close.
//
// The assertion that matters is the empty stripe_processed_events table: an
// error alone would still be wrong if the event had been recorded, because the
// retry would then be skipped as already-seen.
func TestHandleSubscriptionChange_CustomerLookupFailureIsRetryable(t *testing.T) {
	const secret = "whsec_test_secret"
	m, database := newTestManager(t, secret)

	// Stripe is down, not answering "no such customer".
	_ = stripeAPIReturning(t, http.StatusInternalServerError,
		`{"error":{"type":"api_error","message":"Stripe is temporarily unavailable"}}`)

	err := postWebhook(t, m, secret,
		signedSubscriptionEvent(t, secret, "evt_lookup_fail", "cus_test123", time.Now().Unix()))

	if err == nil {
		t.Error("HandleWebhook returned nil after the customer lookup failed; Stripe will not redeliver " +
			"and the subscription change is lost permanently")
	} else if !errors.Is(err, ErrWebhookProcessing) {
		t.Errorf("error = %v, want ErrWebhookProcessing so the webhook 500s and Stripe retries", err)
	}

	if n := processedCount(t, database); n != 0 {
		t.Errorf("stripe_processed_events has %d row(s); the event must not be recorded when handling "+
			"failed, or the redelivery is skipped as already-seen", n)
	}
}

// A subscription event whose customer is genuinely absent must NOT be retried:
// redelivery cannot make a customer appear, and 500ing to exhaustion buries the
// real failures. This is the case getCustomerDID's "" is legitimately for, and
// it is why the fix has to distinguish the two rather than error on both.
func TestHandleSubscriptionChange_NoCustomerMetadataIsNotRetried(t *testing.T) {
	const secret = "whsec_test_secret"
	m, database := newTestManager(t, secret)

	// Stripe answers fine; the customer simply carries no user_did.
	_ = stripeAPIReturning(t, http.StatusOK, `{"id":"cus_test123","object":"customer","metadata":{}}`)

	if err := postWebhook(t, m, secret,
		signedSubscriptionEvent(t, secret, "evt_no_meta", "cus_test123", time.Now().Unix())); err != nil {
		t.Errorf("HandleWebhook error = %v, want nil: retrying cannot conjure a user_did", err)
	}
	if n := processedCount(t, database); n != 1 {
		t.Errorf("stripe_processed_events has %d row(s), want 1: the event was handled to a conclusion", n)
	}
}

const custWithDID = `{"id":"cus_test123","object":"customer","metadata":{"user_did":"did:plc:testuser"}}`

// TestHandleWebhook_RedeliveryIsSkipped is the idempotency 12c55ed added, and
// nothing exercised it: test/stripe-integration passes a nil database, so the
// whole `m.db != nil` path — this check included — is dead there.
func TestHandleWebhook_RedeliveryIsSkipped(t *testing.T) {
	const secret = "whsec_test_secret"
	m, database := newTestManager(t, secret)
	calls := stripeAPIReturning(t, http.StatusOK, custWithDID)

	payload := signedSubscriptionEvent(t, secret, "evt_dedupe", "cus_test123", time.Now().Unix())

	if err := postWebhook(t, m, secret, payload); err != nil {
		t.Fatalf("first delivery: %v", err)
	}
	if n := processedCount(t, database); n != 1 {
		t.Fatalf("after first delivery: %d rows, want 1", n)
	}
	first := calls.Load()

	// Stripe redelivers on any non-2xx, and the dashboard can resend by hand.
	if err := postWebhook(t, m, secret, payload); err != nil {
		t.Errorf("redelivery: %v, want nil — a repeat must be a no-op, not an error", err)
	}
	if n := processedCount(t, database); n != 1 {
		t.Errorf("after redelivery: %d rows, want 1 — the event was applied twice", n)
	}
	if got := calls.Load(); got != first {
		t.Errorf("redelivery made %d Stripe call(s), want %d — the seen-check did not short-circuit "+
			"and the event was re-applied", got-first, 0)
	}
}

// Stripe does not guarantee delivery order. A stale `active` arriving after a
// cancellation must not re-grant the tier, which is what the per-customer
// event_created watermark is for.
func TestHandleSubscriptionChange_StaleEventIsIgnored(t *testing.T) {
	const secret = "whsec_test_secret"
	m, database := newTestManager(t, secret)
	calls := stripeAPIReturning(t, http.StatusOK, custWithDID)

	now := time.Now().Unix()
	if err := postWebhook(t, m, secret,
		signedSubscriptionEvent(t, secret, "evt_newer", "cus_test123", now)); err != nil {
		t.Fatalf("newer event: %v", err)
	}

	afterNewer := calls.Load()

	// An older event for the same customer, delivered second.
	if err := postWebhook(t, m, secret,
		signedSubscriptionEvent(t, secret, "evt_older", "cus_test123", now-3600)); err != nil {
		t.Errorf("stale event: %v, want nil — dropping it is a conclusion, not a failure", err)
	}

	// Both are recorded (the stale one WAS handled, by being ignored), but the
	// watermark must not have moved backwards.
	var latest int64
	if err := database.QueryRow(
		`SELECT MAX(event_created) FROM stripe_processed_events WHERE customer_id = ?`, "cus_test123",
	).Scan(&latest); err != nil {
		t.Fatalf("read watermark: %v", err)
	}
	if latest != now {
		t.Errorf("watermark = %d, want %d — a stale event moved it backwards", latest, now)
	}
	if n := processedCount(t, database); n != 2 {
		t.Errorf("processed rows = %d, want 2", n)
	}
	if got := calls.Load(); got != afterNewer {
		t.Errorf("the stale event made %d Stripe call(s) — the ordering guard did not short-circuit, "+
			"so a stale active re-granted a tier", got-afterNewer)
	}
}

// The ordering guard above line 609 checks sub.Customer != nil; the DID lookup
// right below it used to dereference sub.Customer.ID unconditionally.
func TestHandleSubscriptionChange_NilCustomerDoesNotPanic(t *testing.T) {
	const secret = "whsec_test_secret"
	m, _ := newTestManager(t, secret)
	_ = stripeAPIReturning(t, http.StatusOK, custWithDID)

	payload := fmt.Sprintf(`{
		"id": "evt_nocust",
		"object": "event",
		"api_version": %q,
		"type": "customer.subscription.updated",
		"created": %d,
		"data": {"object": {
			"id": "sub_test",
			"object": "subscription",
			"status": "active",
			"items": {"object":"list","data":[{"price":{"id":"price_monthly_test"}}]}
		}}
	}`, stripe.APIVersion, time.Now().Unix())

	// A panic here would surface as a 500 and an endless Stripe retry loop, or
	// take the appview down outright depending on the recovery middleware.
	if err := postWebhook(t, m, secret, []byte(payload)); err != nil {
		t.Errorf("HandleWebhook error = %v, want nil for an event with no customer", err)
	}
}

// TestHandleSubscriptionChange_HoldFanoutFailureIsRetryable closes the loop the
// other tests in this file only cover one half of.
//
// The tier is resolved, the customer is known, and the only thing that fails is
// the push to the managed hold. That has to reach Stripe as a 5xx and leave
// stripe_processed_events empty: a hold that is briefly down otherwise costs
// the customer their tier permanently, which is the same shape of loss as the
// customer-lookup hole above, one layer further out.
func TestHandleSubscriptionChange_HoldFanoutFailureIsRetryable(t *testing.T) {
	atproto.SetTestMode(true)
	t.Cleanup(func() { atproto.SetTestMode(false) })

	const secret = "whsec_fanout_test"
	m, database := newTestManager(t, secret)

	// The customer resolves cleanly — this test is about what happens after.
	stripeAPIReturning(t, http.StatusOK,
		`{"id":"cus_fanout","object":"customer","metadata":{"user_did":"did:plc:fanoutuser"}}`)

	_, holdDID := managedHoldServer(t, func(w http.ResponseWriter, r *http.Request) {
		http.Error(w, "hold is down", http.StatusServiceUnavailable)
	})
	m.managedHolds = []string{holdDID}

	priv, err := atcrypto.GeneratePrivateKeyP256()
	if err != nil {
		t.Fatalf("generate key: %v", err)
	}
	m.privateKey = priv

	err = postWebhook(t, m, secret,
		signedSubscriptionEvent(t, secret, "evt_fanout_fail", "cus_fanout", time.Now().Unix()))
	if err == nil {
		t.Fatal("a hold that cannot be updated must fail the webhook so Stripe redelivers")
	}
	if !strings.Contains(err.Error(), "push tier to managed holds") {
		t.Errorf("error does not identify the fan-out as the cause: %v", err)
	}
	if n := processedCount(t, database); n != 0 {
		t.Errorf("stripe_processed_events holds %d rows; a failed event must stay redeliverable", n)
	}
}