feat(cloud): GET /admin/billing-users pages the users the reconcile sweep re-syncs (TASK-3564, pad half) (#2042)

The Pad Cloud sidecar's reconcile sweep (TASK-3564, lead ruling day 91) is
the backstop for a Stripe webhook whose effect was lost. It re-runs the same
entitlement sync a webhook runs, for every user pad holds billing state for.
This is pad's half: a cloud-secret gated, paged list of those users.

GET /api/v1/admin/billing-users?after=<user_id>&limit=N returns
{users: [{user_id, stripe_customer_id, plan, plan_source, plan_expires_at}],
next}, ordered by id. limit defaults to 200, max 1000, and next is '' on the
last page. A user is listed when it has a stored Stripe customer id OR a
stripe-sourced plan: each finds a case the other misses. App principals are
excluded. Same credential rule as user-by-id: the cloud secret, or a
platform admin. The sidecar's startup key check reads a small page of it.

Wired like its siblings: cloudAdminPaths, requireCloudMode, the cloud-admin
rate limiter, the BUG-3361 cloud-only route list, and the TASK-2225 compress
census (compress: customer ids are not secrets, as for user-by-id). Its
response shape is pinned in the cloud contract goldens.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

internal/server/bug3361_admin_pat_test.go

@@ -161,6 +161,7 @@ func TestBUG3361_EscapeHatchIsSelfHostOnly(t *testing.T) {
 // alternative) and the billing dashboard.
 var cloudOnlyAdminRoutes = []string{
 	"GET /api/v1/admin/billing-stats",
+	"GET /api/v1/admin/billing-users",
 	"GET /api/v1/admin/user-by-customer",
 	"GET /api/v1/admin/user-by-id",
 	"POST /api/v1/admin/payment-failed",

internal/server/cloud_contract_test.go

@@ -107,6 +107,8 @@ func TestCloudContract(t *testing.T) {
 			"user_id": u.ID, "customer_id": "cus_contract_other", "cloud_secret": cloudContractSecret,
 		}),
 		"admin_user_by_customer": call("GET", "/api/v1/admin/user-by-customer?customer_id=cus_contract", nil),
+		// The reconcile sweep pages billing users (TASK-3564).
+		"admin_billing_users": call("GET", "/api/v1/admin/billing-users?limit=10", nil),
 		"admin_plan": call("POST", "/api/v1/admin/plan", map[string]any{
 			"user_id": u.ID, "plan": "pro", "expires_at": "2099-01-01T00:00:00Z", "source": "stripe",
 			"revision": 1, "subscription_id": "sub_contract", "cloud_secret": cloudContractSecret,

internal/server/handlers_cloud.go

@@ -127,6 +127,7 @@ var cloudAdminPaths = map[string]struct{}{
 	"/api/v1/admin/stripe-customer-id":     {},
 	"/api/v1/admin/user-by-customer":       {},
 	"/api/v1/admin/user-by-id":             {},
+	"/api/v1/admin/billing-users":          {},
 	"/api/v1/admin/stripe-event-processed": {},
 	"/api/v1/admin/stripe-event-unmark":    {},
 	"/api/v1/admin/payment-failed":         {},
@@ -1051,6 +1052,45 @@ func (s *Server) handleGetUserByID(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
+// billingUsersMaxLimit bounds one page of GET /admin/billing-users.
+const billingUsersMaxLimit = 1000
+
+// handleListBillingUsers handles GET /api/v1/admin/billing-users?after=<user_id>&limit=N
+// (TASK-3564). The Pad Cloud sidecar's reconcile sweep pages through every
+// user with a stored Stripe customer or a stripe-sourced plan and re-syncs
+// each from Stripe, the backstop for a webhook whose effect was lost. Its
+// startup key check reads a small page to probe that its Stripe key can see
+// these customers. Same credential rule as user-by-id. limit defaults to 200;
+// `next` is the cursor for the following page, "" on the last one.
+func (s *Server) handleListBillingUsers(w http.ResponseWriter, r *http.Request) {
+	user := currentUser(r)
+	isAdmin := user != nil && user.Role == "admin"
+	if !isAdmin {
+		if !s.validateCloudSecret(r.Header.Get("X-Cloud-Secret"), w) {
+			return
+		}
+	}
+	limit := 200
+	if v := r.URL.Query().Get("limit"); v != "" {
+		n, err := strconv.Atoi(v)
+		if err != nil || n < 1 {
+			writeError(w, http.StatusBadRequest, "bad_request", "limit must be a positive integer")
+			return
+		}
+		limit = min(n, billingUsersMaxLimit)
+	}
+	users, err := s.store.ListBillingUsers(r.URL.Query().Get("after"), limit)
+	if err != nil {
+		writeInternalError(w, err)
+		return
+	}
+	next := ""
+	if len(users) == limit {
+		next = users[len(users)-1].UserID
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"users": users, "next": next})
+}
+
 // --- Stripe Webhook Idempotency (TASK-696) ---
 
 // handleStripeEventProcessed handles POST /api/v1/admin/stripe-event-processed.

internal/server/middleware_ratelimit.go

@@ -741,7 +741,7 @@ func (s *Server) RateLimit(next http.Handler) http.Handler {
 		// Cloud admin endpoints (sidecar → pad): plan changes, Stripe mapping, user lookup
 		if strings.HasPrefix(path, "/api/v1/admin/") {
 			switch path {
-			case "/api/v1/admin/plan", "/api/v1/admin/stripe-customer-id", "/api/v1/admin/user-by-customer", "/api/v1/admin/user-by-id", "/api/v1/admin/stripe-event-processed", "/api/v1/admin/stripe-event-unmark", "/api/v1/admin/payment-failed":
+			case "/api/v1/admin/plan", "/api/v1/admin/stripe-customer-id", "/api/v1/admin/user-by-customer", "/api/v1/admin/user-by-id", "/api/v1/admin/billing-users", "/api/v1/admin/stripe-event-processed", "/api/v1/admin/stripe-event-unmark", "/api/v1/admin/payment-failed":
 				if !s.rateLimiters.CloudAdmin.allow(addr) {
 					slog.Warn("rate limited", "ip", ip, "path", path, "limiter", "cloud_admin")
 					writeRateLimitResponse(w, s.rateLimiters.CloudAdmin.config)

internal/server/server.go

@@ -1841,6 +1841,7 @@ func (s *Server) setupRouter() {
 					r.Post("/stripe-customer-id", s.handleSetStripeCustomerID)      // Cloud: sidecar stores Stripe customer ID after checkout
 					r.Get("/user-by-customer", s.handleGetUserByCustomerID)         // Cloud: sidecar looks up user by Stripe customer ID
 					r.Get("/user-by-id", s.handleGetUserByID)                       // Cloud: sidecar reads a user's stored Stripe customer (TASK-3549)
+					r.Get("/billing-users", s.handleListBillingUsers)               // Cloud: sidecar reconcile sweep pages billing users (TASK-3564)
 					r.Post("/stripe-event-processed", s.handleStripeEventProcessed) // Cloud: sidecar webhook idempotency (TASK-696)
 					r.Post("/stripe-event-unmark", s.handleStripeEventUnmark)       // Cloud: sidecar handler-failure rollback (TASK-736)
 					r.Post("/payment-failed", s.handlePaymentFailed)                // Cloud: sidecar forwards invoice.payment_failed to trigger email (TASK-712)

internal/server/task3564_billing_users_test.go

@@ -0,0 +1,84 @@
+package server
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/PerpetualSoftware/pad/internal/models"
+)
+
+// TASK-3564: GET /admin/billing-users pages every user the reconcile sweep
+// must re-sync (a stored Stripe customer, or a stripe-sourced plan), and only
+// those, behind the same cloud-secret rule as its siblings.
+func TestBillingUsersPagesTheUsersTheSweepNeeds(t *testing.T) {
+	srv := testServer(t)
+	bootstrapFirstUser(t, srv, "admin@example.com", "Admin")
+	srv.SetCloudMode(cloudContractSecret)
+	mk := func(email string) string {
+		u, err := srv.store.CreateUser(models.UserCreate{Email: email, Name: "U", Password: "pw-billing-x"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return u.ID
+	}
+	linked := mk("linked@example.com")
+	if err := srv.store.SetUserStripeCustomerID(linked, "cus_linked"); err != nil {
+		t.Fatal(err)
+	}
+	stripeSourced := mk("sourced@example.com")
+	if _, err := srv.store.DB().Exec(`UPDATE users SET plan = 'free', plan_source = 'stripe' WHERE id = ?`, stripeSourced); err != nil {
+		t.Fatal(err)
+	}
+	mk("nobody@example.com") // no link, manual plan: not a billing user
+
+	get := func(path string, hdr map[string]string) (*httptest.ResponseRecorder, struct {
+		Users []map[string]any `json:"users"`
+		Next  string           `json:"next"`
+	}) {
+		t.Helper()
+		rr := httptest.NewRecorder()
+		srv.ServeHTTP(rr, cloudAdminReq(t, "GET", path, nil, hdr))
+		var out struct {
+			Users []map[string]any `json:"users"`
+			Next  string           `json:"next"`
+		}
+		_ = json.Unmarshal(rr.Body.Bytes(), &out)
+		return rr, out
+	}
+	hdr := map[string]string{"X-Cloud-Secret": cloudContractSecret}
+
+	// Paged one at a time, the cursor walks exactly the two billing users.
+	var seen []string
+	after := ""
+	for i := 0; i < 5; i++ {
+		rr, page := get("/api/v1/admin/billing-users?limit=1&after="+after, hdr)
+		if rr.Code != http.StatusOK {
+			t.Fatalf("status %d: %s", rr.Code, rr.Body.String())
+		}
+		for _, u := range page.Users {
+			seen = append(seen, u["user_id"].(string))
+		}
+		if page.Next == "" {
+			break
+		}
+		after = page.Next
+	}
+	want := map[string]bool{linked: true, stripeSourced: true}
+	if len(seen) != 2 || !want[seen[0]] || !want[seen[1]] {
+		t.Fatalf("pages walked %v, want exactly the linked and the stripe-sourced user", seen)
+	}
+
+	// Without the secret (and without an admin session) it is refused.
+	if rr, _ := get("/api/v1/admin/billing-users", nil); rr.Code == http.StatusOK {
+		t.Fatalf("no credential answered %d", rr.Code)
+	}
+	if rr, _ := get("/api/v1/admin/billing-users", map[string]string{"X-Cloud-Secret": "wrong"}); rr.Code == http.StatusOK {
+		t.Fatalf("a wrong secret answered %d", rr.Code)
+	}
+	// A bad limit is a 400.
+	if rr, _ := get("/api/v1/admin/billing-users?limit=0", hdr); rr.Code != http.StatusBadRequest {
+		t.Fatalf("limit=0 answered %d, want 400", rr.Code)
+	}
+}

internal/server/testdata/cloud_contract/admin_billing_users.json

@@ -0,0 +1,12 @@
+{
+  "next": "",
+  "users": [
+    {
+      "plan": "<string>",
+      "plan_expires_at": "",
+      "plan_source": "<string>",
+      "stripe_customer_id": "<string>",
+      "user_id": "<string>"
+    }
+  ]
+}

internal/server/testdata/compress_get_routes.txt

@@ -19,6 +19,7 @@ compress /api/app/v1/workspaces/{ws}/items/{itemID}/comments
 compress /api/app/v1/workspaces/{ws}/me
 compress /api/v1/admin/apps
 compress /api/v1/admin/billing-stats
+compress /api/v1/admin/billing-users
 compress /api/v1/admin/decision-provider
 compress /api/v1/admin/invitations
 compress /api/v1/admin/limits

internal/store/billing_users.go

@@ -0,0 +1,45 @@
+package store
+
+import "fmt"
+
+// BillingUser is one user the Pad Cloud reconcile sweep re-syncs (TASK-3564):
+// someone pad holds a Stripe link for, or whose plan Stripe set.
+type BillingUser struct {
+	UserID           string `json:"user_id"`
+	StripeCustomerID string `json:"stripe_customer_id"`
+	Plan             string `json:"plan"`
+	PlanSource       string `json:"plan_source"`
+	PlanExpiresAt    string `json:"plan_expires_at"`
+}
+
+// ListBillingUsers pages through human users with a stored Stripe customer id
+// or a stripe-sourced plan, ordered by id: the users after afterID, at most
+// limit of them. Both conditions, because each finds a case the other misses:
+// a link pad holds for a customer Stripe no longer bills, and a stripe plan
+// whose link was cleared. App principals never pay and are excluded.
+func (s *Store) ListBillingUsers(afterID string, limit int) ([]BillingUser, error) {
+	if limit < 1 {
+		return nil, fmt.Errorf("list billing users: limit must be positive")
+	}
+	rows, err := s.db.Query(s.q(`
+		SELECT id, COALESCE(stripe_customer_id, ''), COALESCE(plan, ''), COALESCE(plan_source, ''), COALESCE(plan_expires_at, '')
+		FROM users
+		WHERE kind = 'human'
+		  AND (COALESCE(stripe_customer_id, '') <> '' OR plan_source = ?)
+		  AND id > ?
+		ORDER BY id ASC
+		LIMIT ?`), PlanSourceStripe, afterID, limit)
+	if err != nil {
+		return nil, fmt.Errorf("list billing users: %w", err)
+	}
+	defer rows.Close()
+	out := []BillingUser{}
+	for rows.Next() {
+		var u BillingUser
+		if err := rows.Scan(&u.UserID, &u.StripeCustomerID, &u.Plan, &u.PlanSource, &u.PlanExpiresAt); err != nil {
+			return nil, fmt.Errorf("list billing users (scan): %w", err)
+		}
+		out = append(out, u)
+	}
+	return out, rows.Err()
+}