billing: list invoices per customer instead of per-invoice Stripe search

crates/control-plane-api/src/billing/memory.rs

@@ -200,12 +200,15 @@ impl BillingProvider for InMemoryBillingProvider {
         Ok(method)
     }
 
-    async fn search_invoices(&self, query: &str) -> anyhow::Result<Vec<stripe::Invoice>> {
+    async fn list_invoices(
+        &self,
+        customer_id: &stripe::CustomerId,
+    ) -> anyhow::Result<Vec<stripe::Invoice>> {
         let state = self.state.lock().unwrap();
         Ok(state
             .invoices
             .iter()
-            .filter(|(customer_id, _)| query.contains(customer_id.as_str()))
+            .filter(|(id, _)| id == customer_id)
             .map(|(_, invoice)| invoice.clone())
             .collect())
     }

crates/control-plane-api/src/billing/provider.rs

@@ -43,7 +43,13 @@ pub trait BillingProvider: Send + Sync + std::fmt::Debug {
         payment_method_id: &stripe::PaymentMethodId,
     ) -> anyhow::Result<stripe::PaymentMethod>;
 
-    async fn search_invoices(&self, query: &str) -> anyhow::Result<Vec<stripe::Invoice>>;
+    /// List a customer's invoices, newest first. The resolver matches these to
+    /// catalog invoice rows locally by metadata, so this uses the standard list
+    /// endpoint rather than the more aggressively rate-limited Search API.
+    async fn list_invoices(
+        &self,
+        customer_id: &stripe::CustomerId,
+    ) -> anyhow::Result<Vec<stripe::Invoice>>;
 
     async fn retrieve_payment_intent(
         &self,

crates/control-plane-api/src/billing/stripe_impl.rs

@@ -151,16 +151,32 @@ impl BillingProvider for StripeBillingProvider {
         Ok(pm)
     }
 
-    async fn search_invoices(&self, query: &str) -> anyhow::Result<Vec<stripe::Invoice>> {
-        stripe_search(
-            &self.client,
-            "invoices",
-            SearchParams {
-                query: query.to_string(),
-                ..Default::default()
-            },
-        )
-        .await
+    async fn list_invoices(
+        &self,
+        customer_id: &stripe::CustomerId,
+    ) -> anyhow::Result<Vec<stripe::Invoice>> {
+        // Page through all of the customer's invoices. Stripe caps `limit` at
+        // 100 and paginates with a `starting_after` object-ID cursor.
+        let mut all = Vec::new();
+        let mut starting_after: Option<stripe::InvoiceId> = None;
+        loop {
+            let mut params = stripe::ListInvoices::new();
+            params.customer = Some(customer_id.clone());
+            params.limit = Some(100);
+            params.starting_after = starting_after.clone();
+
+            let page = stripe::Invoice::list(&self.client, &params).await?;
+            let next_cursor = page.data.last().map(|invoice| invoice.id.clone());
+            all.extend(page.data);
+
+            if !page.has_more {
+                break;
+            }
+            // `has_more` is true, so the page was non-empty and `next_cursor`
+            // is set; advance past the last invoice we just collected.
+            starting_after = next_cursor;
+        }
+        Ok(all)
     }
 
     async fn retrieve_payment_intent(

crates/control-plane-api/src/server/public/graphql/billing/invoices.rs

@@ -317,6 +317,18 @@ mod tests {
                 invoice_pdf: Some("https://example.test/invoice.pdf".to_string()),
                 hosted_invoice_url: Some("https://example.test/hosted".to_string()),
                 payment_intent: Some(stripe::Expandable::Id("pi_test_123".parse().unwrap())),
+                // The loader matches list results to rows by this metadata, so
+                // it must mirror the row the billing_historical above produces:
+                // a Final invoice for the February 2024 period.
+                metadata: Some(
+                    billing::InvoiceMetadata {
+                        tenant: "invoicefields/".to_string(),
+                        invoice_type: billing::InvoiceType::Final,
+                        period_start: "2024-02-01".to_string(),
+                        period_end: "2024-02-29".to_string(),
+                    }
+                    .to_metadata_map(),
+                ),
                 ..Default::default()
             },
         );

crates/control-plane-api/src/server/public/graphql/billing/loaders.rs

@@ -3,12 +3,17 @@ use std::future::Future;
 use std::sync::Arc;
 
 use async_graphql::{Result, dataloader::Loader};
-use billing_types::{InvoiceSearch, StatusFilter};
+use billing_types::InvoiceMetadata;
 
 use crate::billing::{BillingProvider, InvoiceType};
 
-/// Compound key used by `StripeInvoiceLoader` to dedup search calls within a
-/// request: one Stripe search per (customer, period, type).
+/// Metadata identity that links a Stripe invoice back to a catalog invoice row:
+/// `(invoice_type, period_start, period_end)`, scoped to a single customer.
+type InvoiceIdentity = (InvoiceType, String, String);
+
+/// Compound key that identifies the Stripe invoice for one catalog invoice row.
+/// Batched keys are grouped by `customer_id`, so the loader resolves them with a
+/// single `list_invoices` call per customer.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub(super) struct StripeInvoiceKey {
     pub(super) customer_id: stripe::CustomerId,
@@ -38,7 +43,13 @@ impl Loader<String> for CustomerDataLoader {
     }
 }
 
-/// Request-scoped loader that resolves a Stripe invoice by its metadata key.
+/// Request-scoped loader that resolves Stripe invoices for catalog invoice rows.
+///
+/// Every row in an `invoices` connection belongs to one tenant, hence one Stripe
+/// customer. The loader groups the batched keys by customer, issues a single
+/// `list_invoices` per customer, and matches each row locally by its metadata
+/// identity. Searching Stripe once per row instead would fan out one Search API
+/// call per invoice and burst past Stripe's rate limit on large pages.
 pub(crate) struct StripeInvoiceLoader(pub Arc<dyn BillingProvider>);
 
 impl Loader<StripeInvoiceKey> for StripeInvoiceLoader {
@@ -49,25 +60,63 @@ impl Loader<StripeInvoiceKey> for StripeInvoiceLoader {
         &self,
         keys: &[StripeInvoiceKey],
     ) -> Result<HashMap<StripeInvoiceKey, Self::Value>> {
-        fan_out_optional(keys, |key| {
-            let provider = self.0.clone();
-            let query = InvoiceSearch {
-                customer_id: Some(key.customer_id.as_str()),
-                invoice_type: Some(key.invoice_type),
-                period_start: Some(&key.date_start),
-                period_end: Some(&key.date_end),
-                status: StatusFilter::Exclude(stripe::InvoiceStatus::Draft),
+        let mut keys_by_customer: HashMap<stripe::CustomerId, Vec<&StripeInvoiceKey>> =
+            HashMap::new();
+        for key in keys {
+            keys_by_customer
+                .entry(key.customer_id.clone())
+                .or_default()
+                .push(key);
+        }
+
+        let mut resolved = HashMap::new();
+        for (customer_id, keys) in keys_by_customer {
+            let invoices = self
+                .0
+                .list_invoices(&customer_id)
+                .await
+                .map_err(|err| async_graphql::Error::new(err.to_string()))?;
+
+            // Index the customer's invoices by metadata identity so each
+            // requested key resolves with no further Stripe calls. Drafts are
+            // skipped: they're in-progress and shouldn't surface amounts or
+            // PDFs. Invoices without our metadata (e.g. created outside the
+            // billing pipeline) can't be matched and are ignored. `or_insert`
+            // keeps the first match, which is the newest since Stripe lists
+            // invoices newest-first.
+            let mut by_identity: HashMap<InvoiceIdentity, stripe::Invoice> = HashMap::new();
+            for invoice in invoices {
+                if matches!(invoice.status, Some(stripe::InvoiceStatus::Draft)) {
+                    continue;
+                }
+                let Some(metadata) = invoice
+                    .metadata
+                    .as_ref()
+                    .and_then(InvoiceMetadata::from_metadata_map)
+                else {
+                    continue;
+                };
+                by_identity
+                    .entry((
+                        metadata.invoice_type,
+                        metadata.period_start,
+                        metadata.period_end,
+                    ))
+                    .or_insert(invoice);
             }
-            .to_query();
-            async move {
-                provider
-                    .search_invoices(&query)
-                    .await
-                    .map(|invoices| invoices.into_iter().next())
-                    .map_err(|err| async_graphql::Error::new(err.to_string()))
+
+            for key in keys {
+                let identity = (
+                    key.invoice_type,
+                    key.date_start.clone(),
+                    key.date_end.clone(),
+                );
+                if let Some(invoice) = by_identity.get(&identity) {
+                    resolved.insert(key.clone(), invoice.clone());
+                }
             }
-        })
-        .await
+        }
+        Ok(resolved)
     }
 }