diff --git a/crates/billing-integrations/src/publish.rs b/crates/billing-integrations/src/publish.rs index 5cc6de3f453..2bddbeb49bc 100644 --- a/crates/billing-integrations/src/publish.rs +++ b/crates/billing-integrations/src/publish.rs @@ -10,7 +10,6 @@ use serde::{Deserialize, Serialize}; use sqlx::Postgres; use sqlx::{Pool, postgres::PgPoolOptions, types::chrono::NaiveDate}; use std::collections::HashMap; -use stripe::InvoiceStatus; const CREATED_BY_BILLING_AUTOMATION: &str = "estuary.dev/created_by_automation"; @@ -58,6 +57,15 @@ pub struct PublishInvoice { /// Clean up dangling invoices that are not in the database #[clap(long, default_value_t = false)] pub clean_up: bool, + /// Run in read-only mode: classify all invoices and report what would + /// happen, without creating or modifying anything in Stripe. + /// + /// The preview checks that each invoice could be issued (a billing email is + /// resolvable), but it cannot reconcile the final invoice total against + /// Stripe, since that requires actually creating the invoice and its line + /// items. + #[clap(long, default_value_t = false)] + pub dry_run: bool, } fn parse_date(arg: &str) -> Result { @@ -90,20 +98,32 @@ enum InvoiceResult { FutureTrialStart, NoDataMoved, NoFullPipeline, + AlreadyProcessed, Error, } impl InvoiceResult { - pub fn message(&self) -> String { + pub fn message(&self, dry_run: bool) -> String { match self { InvoiceResult::Created(provider) => { + let verb = if dry_run { + "Would publish" + } else { + "Published" + }; if provider == &PaymentProvider::Stripe { - "Published new invoice".to_string() + format!("{verb} new invoice") } else { - format!("Published new invoice for tenant using {provider:?} provider") + format!("{verb} new invoice for tenant using {provider:?} provider") + } + } + InvoiceResult::Updated => { + if dry_run { + "Would update existing invoice".to_string() + } else { + "Updated existing invoice".to_string() } } - InvoiceResult::Updated => "Updated existing invoice".to_string(), InvoiceResult::LessThanMinimum => { "Skipping invoice for less than the minimum chargable amount ($0.50)".to_string() } @@ -117,11 +137,38 @@ impl InvoiceResult { InvoiceResult::NoFullPipeline => { "Skipping invoice for tenant without an active pipeline".to_string() } + InvoiceResult::AlreadyProcessed => { + "Skipping invoice already processed in a previous billing run".to_string() + } InvoiceResult::Error => "Error publishing invoices".to_string(), } } } +/// The outcome of the classify phase: what action should be taken for this invoice. +enum InvoiceAction { + /// Invoice should not be created. Carries the skip reason and the + /// customer (if found) for potential clean-up of stale drafts. + Skip { + result: InvoiceResult, + customer: Option, + }, + /// Create a new invoice. `replace` is set when --recreate-finalized + /// requires deleting an existing invoice first. `customer` is the customer + /// found during classification, or None when none exists yet (execute then + /// creates one). + Create { + replace: Option, + customer: Option, + }, + /// Update an existing draft invoice's line items. `customer` is the invoice's + /// owner, already found during classification. + Update { + existing_invoice_id: stripe::InvoiceId, + customer: stripe::Customer, + }, +} + #[derive( Serialize, Deserialize, Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash, Copy, sqlx::Type, )] @@ -176,93 +223,229 @@ impl Invoice { Ok(invoice_search.into_iter().next()) } - #[tracing::instrument(skip(self, client, db_client), fields(tenant=self.billed_prefix, invoice_type=format!("{:?}",self.invoice_type), subtotal=format!("${:.2}", self.subtotal as f64 / 100.0)))] - async fn upsert_invoice( + /// Read-only classification: determines what action should be taken for this + /// invoice without making any writes to Stripe. + #[tracing::instrument(skip(self, client), fields(tenant=self.billed_prefix, invoice_type=format!("{:?}",self.invoice_type), subtotal=format!("${:.2}", self.subtotal as f64 / 100.0)))] + async fn classify( &self, client: &stripe::Client, - db_client: &Pool, recreate_finalized: bool, - mode: ChargeType, - ) -> anyhow::Result { + ) -> anyhow::Result { + // --- Phase 1: Cheap local checks (no Stripe calls) --- + match (&self.invoice_type, &self.extra) { (InvoiceType::Preview, _) => { bail!("Should not create Stripe invoices for preview invoices") } - (InvoiceType::Final, Some(extra)) => { - // If we have a payment method, don't skip the invoice - // If `has_payment_method` is Some, then there is a stripe customer to check - let validated_has_payment_method = - if let Some(has_payment_method) = self.has_payment_method { - // The Stripe capture in the database has been known to be unreliable. - // Let's double-check with Stripe to make sure it agrees that we really - // do not have a payment method set. - let real_default_payment_method = get_or_create_customer_for_tenant( - client, - db_client, - self.billed_prefix.to_owned(), - false, // If there's no customer, there's no way there can be a payment method - ) - .await? - .and_then(|customer| customer.invoice_settings) - .and_then(|i| i.default_payment_method); - - if has_payment_method != real_default_payment_method.is_some() { - tracing::warn!( - ?has_payment_method, - stripe_payment_method = real_default_payment_method.is_some(), - "Inconsistent payment method state" - ); - } - - real_default_payment_method.is_some() - } else { - false - }; - - let unwrapped_extra = extra.clone().0.expect( - "This is just a sqlx quirk, if the outer Option is Some then this will be Some", - ); - - if !validated_has_payment_method { - if unwrapped_extra.processed_data_gb.unwrap_or_default() == 0.0 - && !matches!(&self.invoice_type, InvoiceType::Manual) - { - return Ok(InvoiceResult::NoDataMoved); - } - - if !self.has_full_pipeline && !matches!(&self.invoice_type, InvoiceType::Manual) - { - return Ok(InvoiceResult::NoFullPipeline); - } - } - } (InvoiceType::Final, None) => { bail!("Invoice should have extra") } _ => {} }; - // An invoice should be generated in Stripe if the tenant is on a paid plan, which means: - // * The tenant has a free trial start date - // * The tenant's free trial start date is before the invoice period's end date if let InvoiceType::Final = self.invoice_type { match self.tenant_trial_start { Some(trial_start) if self.date_end < trial_start => { - return Ok(InvoiceResult::FutureTrialStart); + return Ok(InvoiceAction::Skip { + result: InvoiceResult::FutureTrialStart, + customer: None, + }); } None => { - return Ok(InvoiceResult::FreeTier); + return Ok(InvoiceAction::Skip { + result: InvoiceResult::FreeTier, + customer: None, + }); } _ => {} } } - // The minimum chargable amount of USD in Stripe is $0.50. - // https://stripe.com/docs/currencies#minimum-and-maximum-charge-amounts if self.subtotal < 50 { - return Ok(InvoiceResult::LessThanMinimum); + return Ok(InvoiceAction::Skip { + result: InvoiceResult::LessThanMinimum, + customer: None, + }); + } + + // --- Phase 2: Stripe calls (only for invoices that survived Phase 1) --- + + // Fetch the customer once, up front: it is reused for the payment-method + // validation below, the existing-invoice search, and threaded into the + // returned action so execute() need not search for it again. + let customer = find_customer(client, &self.billed_prefix).await?; + + // For Final invoices, verify the payment method state with Stripe. + // The DB capture has been known to be unreliable, so Stripe is the + // source of truth. If the tenant has no payment method, skip on + // NoDataMoved / NoFullPipeline. + if let (InvoiceType::Final, Some(extra)) = (&self.invoice_type, &self.extra) { + let validated_has_payment_method = + if let Some(has_payment_method) = self.has_payment_method { + let real_has_pm = customer + .as_ref() + .and_then(|c| c.invoice_settings.as_ref()) + .and_then(|i| i.default_payment_method.as_ref()) + .is_some(); + + if has_payment_method != real_has_pm { + tracing::warn!( + ?has_payment_method, + stripe_payment_method = real_has_pm, + "Inconsistent payment method state" + ); + } + + real_has_pm + } else { + false + }; + + if !validated_has_payment_method { + // If the outer Option is Some the inner should be too; treat a + // null inner as a data error rather than panicking and aborting + // the whole run. + let Some(unwrapped_extra) = extra.0.as_ref() else { + bail!( + "Final invoice for {tenant} has a null `extra` payload", + tenant = self.billed_prefix + ); + }; + + if unwrapped_extra.processed_data_gb.unwrap_or_default() == 0.0 { + return Ok(InvoiceAction::Skip { + result: InvoiceResult::NoDataMoved, + customer, + }); + } + + if !self.has_full_pipeline { + return Ok(InvoiceAction::Skip { + result: InvoiceResult::NoFullPipeline, + customer, + }); + } + } } + let customer = match customer { + Some(c) => c, + // No customer in Stripe means no existing invoice is possible + None => { + return Ok(InvoiceAction::Create { + replace: None, + customer: None, + }); + } + }; + + let customer_id = customer.id.to_string(); + + if let Some(invoice) = self + .get_stripe_invoice(client, customer_id.as_str()) + .await? + { + match invoice.status { + // Manual invoices are excluded from --recreate-finalized: an + // already-sent (open) manual invoice must not be voided and + // reissued, and a manual draft is refreshed via the Update arm below. + Some(stripe::InvoiceStatus::Open | stripe::InvoiceStatus::Draft) + if recreate_finalized && !matches!(self.invoice_type, InvoiceType::Manual) => + { + Ok(InvoiceAction::Create { + replace: Some(invoice.id), + customer: Some(customer), + }) + } + Some(stripe::InvoiceStatus::Draft) => { + tracing::debug!( + "Found existing draft invoice {id}", + id = invoice.id.to_string() + ); + Ok(InvoiceAction::Update { + existing_invoice_id: invoice.id, + customer, + }) + } + Some(stripe::InvoiceStatus::Open) + if matches!(self.invoice_type, InvoiceType::Manual) => + { + tracing::debug!( + "Manual invoice {id} already open, skipping", + id = invoice.id.to_string() + ); + Ok(InvoiceAction::Skip { + result: InvoiceResult::AlreadyProcessed, + customer: Some(customer), + }) + } + Some(stripe::InvoiceStatus::Open) => { + bail!( + "Found open invoice {id}. Pass --recreate-finalized to delete and recreate this invoice.", + id = invoice.id.to_string() + ) + } + Some( + status @ (stripe::InvoiceStatus::Paid + | stripe::InvoiceStatus::Void + | stripe::InvoiceStatus::Uncollectible), + ) if matches!(self.invoice_type, InvoiceType::Manual) => { + tracing::debug!( + "Manual invoice {id} already in state {status}, skipping", + id = invoice.id.to_string(), + status = status + ); + Ok(InvoiceAction::Skip { + result: InvoiceResult::AlreadyProcessed, + customer: Some(customer), + }) + } + Some(status) => { + bail!( + "Found invoice {id} in unsupported state {status}, skipping.", + id = invoice.id.to_string(), + status = status + ); + } + None => { + bail!( + "Unexpected missing status from invoice {id}", + id = invoice.id.to_string() + ); + } + } + } else { + Ok(InvoiceAction::Create { + replace: None, + customer: Some(customer), + }) + } + } + + /// Execute the classified action: performs all Stripe writes (customer creation, + /// invoice creation/update, line item management, verification). + #[tracing::instrument(skip(self, client, db_client, action), fields(tenant=self.billed_prefix, invoice_type=format!("{:?}",self.invoice_type), subtotal=format!("${:.2}", self.subtotal as f64 / 100.0)))] + async fn execute( + &self, + client: &stripe::Client, + db_client: &Pool, + action: InvoiceAction, + mode: ChargeType, + ) -> anyhow::Result { + let (replace, existing_invoice_id, found_customer) = match action { + InvoiceAction::Skip { result, .. } => return Ok(result), + InvoiceAction::Create { replace, customer } => (replace, None, customer), + InvoiceAction::Update { + existing_invoice_id, + customer, + } => (None, Some(existing_invoice_id), Some(customer)), + }; + let is_update = existing_invoice_id.is_some(); + + let customer = + ensure_customer_for_invoicing(client, db_client, &self.billed_prefix, found_customer) + .await?; + // Anything before 12:00:00 renders as the previous day in Stripe let date_start_secs = self .date_start @@ -287,114 +470,148 @@ impl Invoice { let date_start_repr = self.date_start.format("%F").to_string(); let date_end_repr = self.date_end.format("%F").to_string(); - let customer = get_or_create_customer_for_tenant( - client, - db_client, - self.billed_prefix.to_owned(), - true, - ) - .await? - .expect("Should never return None"); - let customer_id = customer.id.to_string(); - - let maybe_invoice = if let Some(invoice) = self - .get_stripe_invoice(&client, customer_id.as_str()) - .await? - { - match invoice.status { - Some(state @ (stripe::InvoiceStatus::Open | stripe::InvoiceStatus::Draft)) - if recreate_finalized => - { + // Delete existing invoice if --recreate-finalized was used + if let Some(ref replace_id) = replace { + // Re-verify the invoice status before deleting (guard against race conditions) + let existing = stripe::Invoice::retrieve(client, replace_id, &[]).await?; + match existing.status { + Some(state @ (stripe::InvoiceStatus::Open | stripe::InvoiceStatus::Draft)) => { tracing::warn!( - "Found invoice {id} in state {state} deleting and recreating", - id = invoice.id.to_string(), + "Found invoice {id} in state {state}, deleting and recreating", + id = replace_id.to_string(), state = state ); - stripe::Invoice::delete(client, &invoice.id).await?; - None + stripe::Invoice::delete(client, replace_id).await?; } - Some(stripe::InvoiceStatus::Draft) => { - tracing::debug!( - "Updating existing invoice {id}", - id = invoice.id.to_string() + Some(status) => { + bail!( + "Invoice {id} changed to state {status} since classification, cannot delete.", + id = replace_id.to_string(), + status = status ); - Some(invoice) } - Some(stripe::InvoiceStatus::Open) => { + None => { bail!( - "Found open invoice {id}. Pass --recreate-finalized to delete and recreate this invoice.", - id = invoice.id.to_string() - ) + "Unexpected missing status from invoice {id}", + id = replace_id.to_string() + ); } + } + } + + // Manual invoices should always be sent as invoices rather than + // charged to the customer's payment method. + let mode = if self.invoice_type == InvoiceType::Manual { + ChargeType::SendInvoice + } else { + mode + }; + let collection_method = match mode { + ChargeType::AutoCharge => stripe::CollectionMethod::ChargeAutomatically, + ChargeType::SendInvoice => stripe::CollectionMethod::SendInvoice, + }; + // `send_invoice` requires a due date; `charge_automatically` must not carry one. + let due_date = match mode { + ChargeType::SendInvoice => Some((Utc::now() + Duration::days(30)).timestamp()), + ChargeType::AutoCharge => None, + }; + + let invoice = if let Some(existing_id) = existing_invoice_id { + tracing::debug!( + "Updating existing invoice {id}", + id = existing_id.to_string() + ); + let existing = stripe::Invoice::retrieve(client, &existing_id, &[]).await?; + + // Re-verify the invoice is still an updatable draft. It was a draft at + // classification time, but a concurrent run or a human finalizing / + // paying / voiding it in Stripe could have changed that; mutating a + // finalized invoice's line items would otherwise fail with an opaque + // Stripe error. Mirrors the guard on the delete/recreate path above. + match existing.status { + Some(stripe::InvoiceStatus::Draft) => {} Some(status) => { bail!( - "Found invoice {id} in unsupported state {status}, skipping.", - id = invoice.id.to_string(), + "Invoice {id} changed to state {status} since classification, cannot update.", + id = existing_id.to_string(), status = status ); } None => { bail!( "Unexpected missing status from invoice {id}", - id = invoice.id.to_string() + id = existing_id.to_string() ); } } - } else { - None - }; - let invoice = match maybe_invoice.clone() { - Some(inv) => inv, - None => { - let invoice = stripe::Invoice::create( - client, - stripe::CreateInvoice { - customer: Some(customer.id.to_owned()), - // Stripe timestamps are measured in _seconds_ since epoch - // Due date must be in the future. Bill net-30, so 30 days from today - due_date: match mode { - ChargeType::SendInvoice => Some((Utc::now() + Duration::days(30)).timestamp()), - ChargeType::AutoCharge => None + // The create path sets the collection method from `mode`, but the update + // path reuses whatever the invoice was originally created with. Reconcile + // it here so a manual invoice previously stored as `charge_automatically` + // is moved to `send_invoice` instead of silently auto-charging the card. + if existing.collection_method != Some(collection_method) { + #[derive(serde::Serialize)] + struct UpdateInvoice { + collection_method: stripe::CollectionMethod, + due_date: Option, + } + tracing::debug!( + "Reconciling collection method of invoice {id} to {collection_method:?}", + id = existing_id.to_string() + ); + let updated: stripe::Invoice = client + .post_form( + &format!("/invoices/{existing_id}"), + UpdateInvoice { + collection_method, + due_date, }, - description: Some( - format!( - "Your Flow bill for the billing period between {date_start_human} - {date_end_human}. Tenant: {tenant}", - tenant=self.billed_prefix.to_owned() - ) - .as_str(), - ), - collection_method: Some(match mode { - ChargeType::AutoCharge => stripe::CollectionMethod::ChargeAutomatically, - ChargeType::SendInvoice => stripe::CollectionMethod::SendInvoice, - }), - auto_advance: Some(false), - custom_fields: Some(vec![ - stripe::CreateInvoiceCustomFields { - name: "Billing Period Start".to_string(), - value: date_start_human.to_owned(), - }, - stripe::CreateInvoiceCustomFields { - name: "Billing Period End".to_string(), - value: date_end_human.to_owned(), - }, - ]), - metadata: Some( - InvoiceMetadata { - tenant: self.billed_prefix.to_owned(), - invoice_type: self.invoice_type, - period_start: date_start_repr, - period_end: date_end_repr, - } - .to_metadata_map(), - ), - ..Default::default() - }, - ) - .await.context("Creating a new invoice")?; - tracing::debug!("Created a new invoice {id}", id = invoice.id); - invoice + ) + .await + .context("Reconciling collection method of existing invoice")?; + updated + } else { + existing } + } else { + let description_text = format!( + "Your Flow bill for the billing period between {date_start_human} - {date_end_human}. Tenant: {tenant}", + tenant = self.billed_prefix + ); + let invoice = stripe::Invoice::create( + client, + stripe::CreateInvoice { + customer: Some(customer.id.to_owned()), + due_date, + description: Some(description_text.as_str()), + collection_method: Some(collection_method), + auto_advance: Some(false), + custom_fields: Some(vec![ + stripe::CreateInvoiceCustomFields { + name: "Billing Period Start".to_string(), + value: date_start_human.to_owned(), + }, + stripe::CreateInvoiceCustomFields { + name: "Billing Period End".to_string(), + value: date_end_human.to_owned(), + }, + ]), + metadata: Some( + InvoiceMetadata { + tenant: self.billed_prefix.to_owned(), + invoice_type: self.invoice_type, + period_start: date_start_repr, + period_end: date_end_repr, + } + .to_metadata_map(), + ), + ..Default::default() + }, + ) + .await + .context("Creating a new invoice")?; + tracing::debug!("Created a new invoice {id}", id = invoice.id); + invoice }; // Clear out line items from invoice, if there are any @@ -447,11 +664,10 @@ impl Invoice { ); } - // Let's double-check that the invoice total matches the desired total + // Re-fetch invoice and customer for fresh data (balance may have changed) let check_invoice = stripe::Invoice::retrieve(client, &invoice.id, &[]).await?; - - // Customers can have an invoice credit balance, so let's make sure we take that into account. - let credit_balance = customer.balance.unwrap_or(0); + let fresh_customer = stripe::Customer::retrieve(client, &customer.id, &[]).await?; + let credit_balance = fresh_customer.balance.unwrap_or(0); let expected = (self.subtotal + (diff.ceil() as i64) + credit_balance).max(0); @@ -463,10 +679,10 @@ impl Invoice { ) } - if maybe_invoice.is_some() { - return Ok(InvoiceResult::Updated); + if is_update { + Ok(InvoiceResult::Updated) } else { - return Ok(InvoiceResult::Created(self.payment_provider)); + Ok(InvoiceResult::Created(self.payment_provider)) } } } @@ -601,76 +817,88 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { .or_default() += 1; }); - tracing::info!( - "Processing {usage} usage-based invoices, and {manual} manually-entered invoices.", - usage = invoice_type_counter - .remove(&InvoiceType::Final) - .unwrap_or_default(), - manual = invoice_type_counter - .remove(&InvoiceType::Manual) - .unwrap_or_default(), - ); + let usage = invoice_type_counter + .remove(&InvoiceType::Final) + .unwrap_or_default(); + let manual = invoice_type_counter + .remove(&InvoiceType::Manual) + .unwrap_or_default(); + if cmd.dry_run { + tracing::info!( + "[DRY RUN] Classifying {usage} usage-based invoices and {manual} manually-entered invoices without making any changes to Stripe." + ); + } else { + tracing::info!( + "Processing {usage} usage-based invoices, and {manual} manually-entered invoices." + ); + } let invoice_futures: Vec<_> = invoices .iter() .map(|response| { let client = stripe_client.clone(); let db_pool = db_pool.clone(); + + let annotation = manual_annotation(response); + async move { - let res = response - .upsert_invoice( - &client, - &db_pool, - cmd.recreate_finalized, - cmd.charge_type, - ) + let action = response + .classify(&client, cmd.recreate_finalized) .await; - match res { + + // Shared per-tenant debug line for a classified/executed result. + let log_result = |result: &InvoiceResult| { + tracing::debug!( + tenant = response.billed_prefix, + invoice_type = format!("{:?}", response.invoice_type), + subtotal = format!("${:.2}", response.subtotal as f64 / 100.0), + "{}", + result.message(cmd.dry_run) + ); + }; + + match action { Err(err) => { let formatted = format!( - "Error publishing {invoice_type:?} invoice for {tenant}", + "Error classifying {invoice_type:?} invoice for {tenant}", tenant = response.billed_prefix, invoice_type = response.invoice_type ); - Err(anyhow::anyhow!(format!( - "{}: {err:#}", - formatted, - err = err - ))) + Err(anyhow::anyhow!("{formatted}: {err:#}")) } - Ok(res) => { - tracing::debug!( - tenant = response.billed_prefix, - invoice_type = format!("{:?}", response.invoice_type), - subtotal = format!("${:.2}", response.subtotal as f64 / 100.0), - "{}", - res.message() - ); - match res { - InvoiceResult::Created(_) - | InvoiceResult::Updated - | InvoiceResult::Error => {} - // Remove any incorrectly created invoices that are now skipped for whatever reason - _ if cmd.clean_up => { - let task_res: Result<(), anyhow::Error> = async move { - let customer = match get_or_create_customer_for_tenant( - &client, - &db_pool, - response.billed_prefix.to_owned(), - false, - ) - .await? - { - Some(c) => c, - None => return Ok(()), - }; - - let customer_id = customer.id.to_string(); - - if let Some(invoice) = - response.get_stripe_invoice(&client, &customer_id).await? - { - if let Some(InvoiceStatus::Draft) = invoice.status { + Ok(InvoiceAction::Skip { result, customer }) => { + log_result(&result); + + if cmd.clean_up { + let task_res: Result<(), anyhow::Error> = async { + // FreeTier / FutureTrialStart / LessThanMinimum skip in + // classify's Phase 1 without a Stripe lookup, so their Skip + // action carries no customer. Look one up here so --clean-up + // can still remove a stale draft for those tenants. + let customer = match customer { + Some(c) => c, + None => { + match find_customer(&client, &response.billed_prefix) + .await? + { + Some(c) => c, + None => return Ok(()), + } + } + }; + let customer_id = customer.id.to_string(); + + if let Some(invoice) = + response.get_stripe_invoice(&client, &customer_id).await? + { + if let Some(stripe::InvoiceStatus::Draft) = invoice.status { + if cmd.dry_run { + tracing::warn!( + tenant = response.billed_prefix.to_string(), + "[dry-run] Would delete stale draft invoice {}", + invoice.id + ); + } else { tracing::warn!( tenant = response.billed_prefix.to_string(), "Deleting draft invoice!" @@ -678,18 +906,76 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { stripe::Invoice::delete(&client, &invoice.id).await?; } } - - Ok(()) } - .await; + Ok(()) + } + .await; - if let Err(e) = task_res { - tracing::warn!("Failed to check for or clear potential leaked draft invoices for {}, this is probably not a problem: {e:#}", response.billed_prefix.to_owned()); - } - }, - _ => {} + if let Err(e) = task_res { + tracing::warn!("Failed to check for or clear potential leaked draft invoices for {}, this is probably not a problem: {e:#}", response.billed_prefix.to_owned()); + } + } + + Ok((result, response.subtotal, response.billed_prefix.to_owned(), annotation)) + } + Ok(action) if cmd.dry_run => { + // A real run calls ensure_customer_for_invoicing, which bails + // when no email can be resolved for the tenant. Mirror that + // read-only so the preview reports such tenants as errors + // instead of claiming the invoice would be published. (The + // invoice total cannot be reconciled without creating the + // invoice in Stripe, so that check is necessarily skipped.) + if !billing_email_available(&client, &db_pool, &response.billed_prefix) + .await? + { + tracing::warn!( + tenant = response.billed_prefix, + "[dry-run] Would fail: no customer email, tenants.billing_email, or admin user to invoice" + ); + return Ok(( + InvoiceResult::Error, + response.subtotal, + response.billed_prefix.to_owned(), + annotation, + )); + } + + let result = match &action { + InvoiceAction::Create { replace: Some(id), .. } => { + tracing::info!( + tenant = response.billed_prefix, + "[dry-run] Would delete existing invoice {} and recreate", + id + ); + InvoiceResult::Created(response.payment_provider) + } + InvoiceAction::Create { .. } => { + InvoiceResult::Created(response.payment_provider) + } + InvoiceAction::Update { .. } => InvoiceResult::Updated, + InvoiceAction::Skip { .. } => unreachable!(), + }; + log_result(&result); + Ok((result, response.subtotal, response.billed_prefix.to_owned(), annotation)) + } + Ok(action) => { + let res = response + .execute(&client, &db_pool, action, cmd.charge_type) + .await; + match res { + Err(err) => { + let formatted = format!( + "Error publishing {invoice_type:?} invoice for {tenant}", + tenant = response.billed_prefix, + invoice_type = response.invoice_type + ); + Err(anyhow::anyhow!("{formatted}: {err:#}")) + } + Ok(res) => { + log_result(&res); + Ok((res, response.subtotal, response.billed_prefix.to_owned(), annotation)) + } } - Ok((res, response.subtotal, response.billed_prefix.to_owned())) } } } @@ -700,22 +986,23 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { let total = invoice_futures.len(); - let collected: HashMap)> = + let collected: HashMap)>)> = futures::stream::iter(invoice_futures) .buffer_unordered(cmd.concurrency) .or_else(|(err, invoice)| async move { if !cmd.fail_fast { tracing::error!("[{}]: {err:#}", invoice.billed_prefix); - Ok((InvoiceResult::Error, 0, invoice.billed_prefix)) + let annotation = manual_annotation(&invoice); + Ok((InvoiceResult::Error, 0, invoice.billed_prefix, annotation)) } else { Err(err) } }) .try_fold( HashMap::new(), - |mut map, (res, subtotal, tenant)| async move { + |mut map, (res, subtotal, tenant, annotation)| async move { let overall_count = map.values().map(|(_, count, _)| *count).sum::() + 1; - let msg = res.message(); + let msg = res.message(cmd.dry_run); let (subtotal_sum, count_for_result_type, tenants) = map.entry(res).or_insert((0, 0, vec![])); @@ -723,7 +1010,7 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { *count_for_result_type += 1; tracing::info!("[{overall_count}/{total}, {tenant}]: {msg}"); - tenants.push((tenant, subtotal)); + tenants.push((tenant, subtotal, annotation)); Ok(map) }, ) @@ -733,7 +1020,7 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { tracing::info!( "[{:4} invoices]: {:70}${:.2}", count, - status.message(), + status.message(cmd.dry_run), *subtotal_agg as f64 / 100.0 ); let limit = match status { @@ -741,18 +1028,24 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { InvoiceResult::NoDataMoved | InvoiceResult::NoFullPipeline | InvoiceResult::LessThanMinimum - | InvoiceResult::FreeTier => 0, + | InvoiceResult::FreeTier + | InvoiceResult::AlreadyProcessed => 0, _ => 10, }; let sorted_tenants = tenants .iter() - .sorted_by(|(_, a), (_, b)| b.cmp(a)) + .sorted_by(|(_, a, _), (_, b, _)| b.cmp(a)) .collect_vec(); let (displayed_tenants, remainder_tenants) = sorted_tenants.split_at(limit.min(tenants.len())); - for (tenant, subtotal) in displayed_tenants { - tracing::info!(" - {:} ${:.2}", tenant, *subtotal as f64 / 100.0); + for (tenant, subtotal, annotation) in displayed_tenants { + match annotation { + Some(note) => { + tracing::info!(" - {} ${:.2} {}", tenant, *subtotal as f64 / 100.0, note) + } + None => tracing::info!(" - {} ${:.2}", tenant, *subtotal as f64 / 100.0), + } } if limit > 0 && remainder_tenants.len() > 0 { tracing::info!(" - ... {} Others", remainder_tenants.len(),); @@ -762,35 +1055,72 @@ pub async fn do_publish_invoices(cmd: &PublishInvoice) -> anyhow::Result<()> { Ok(()) } -#[tracing::instrument(skip(client, db_client))] -async fn get_or_create_customer_for_tenant( +/// Display tag distinguishing a manually-entered invoice (and its period) in the +/// run summary; usage invoices get no tag. +fn manual_annotation(invoice: &Invoice) -> Option { + match invoice.invoice_type { + InvoiceType::Manual => Some(format!( + "[manual: {} - {}]", + invoice.date_start.format("%Y-%m-%d"), + invoice.date_end.format("%Y-%m-%d") + )), + _ => None, + } +} + +/// Read-only: search Stripe for an existing customer by tenant metadata. +#[tracing::instrument(skip(client))] +async fn find_customer( client: &stripe::Client, - db_client: &Pool, - tenant: String, - create: bool, + tenant: &str, ) -> anyhow::Result> { - let billing_row = sqlx::query!( - r#"SELECT billing_email, billing_address FROM tenants WHERE tenant = $1"#, - tenant, - ) - .fetch_optional(db_client) - .await?; - - let customers: Vec = stripe_search( + let mut customers: Vec = stripe_search( client, "customers", SearchParams { - query: customer_search_query(&tenant), + query: customer_search_query(tenant), ..Default::default() }, ) .await .context(format!("Searching for tenant {tenant}"))?; - let customer = if let Some(customer) = customers.into_iter().next() { + if let Some(customer) = customers.drain(..).next() { tracing::debug!("Found existing customer {id}", id = customer.id.to_string()); + Ok(Some(customer)) + } else { + Ok(None) + } +} + +/// Ensures a Stripe customer exists for this tenant and is ready for invoicing. +/// Uses `found` (the customer located during classification) when present, +/// otherwise finds an existing customer or creates a new one (carrying the +/// tenant's billing email and address from the control-plane DB), then ensures +/// the customer has an email set (falling back to the earliest admin on the +/// tenant if needed). +#[tracing::instrument(skip(client, db_client, found))] +async fn ensure_customer_for_invoicing( + client: &stripe::Client, + db_client: &Pool, + tenant: &str, + found: Option, +) -> anyhow::Result { + let billing_row = sqlx::query!( + r#"SELECT billing_email, billing_address FROM tenants WHERE tenant = $1"#, + tenant, + ) + .fetch_optional(db_client) + .await?; + + let existing = match found { + Some(customer) => Some(customer), + None => find_customer(client, tenant).await?, + }; + + let customer = if let Some(customer) = existing { customer - } else if create { + } else { tracing::debug!("Creating new customer"); // Match the deterministic Idempotency-Key used by the GraphQL path so // a setup-intent flow racing against billing automation can't produce @@ -798,7 +1128,7 @@ async fn get_or_create_customer_for_tenant( let create_client = client .clone() .with_strategy(stripe::RequestStrategy::Idempotent( - customer_create_idempotency_key(&tenant), + customer_create_idempotency_key(tenant), )); let billing_email = billing_row @@ -811,15 +1141,14 @@ async fn get_or_create_customer_for_tenant( .transpose() .context("deserializing billing_address")?; + let description = format!("Represents the billing entity for Flow tenant '{tenant}'"); let new_customer = stripe::Customer::create( &create_client, stripe::CreateCustomer { - name: Some(tenant.as_str()), + name: Some(tenant), email: billing_email, address: billing_address, - description: Some( - format!("Represents the billing entity for Flow tenant '{tenant}'").as_str(), - ), + description: Some(description.as_str()), metadata: Some(HashMap::from([ (TENANT_METADATA_KEY.to_string(), tenant.to_string()), ( @@ -838,27 +1167,52 @@ async fn get_or_create_customer_for_tenant( // Waking it during an invoicing run would add a Stripe lookup per new // customer for no contact change. new_customer - } else { - return Ok(None); }; if customer.email.is_none() { let db_email = billing_row.as_ref().and_then(|r| r.billing_email.clone()); - if let Some(email) = db_email { - tracing::info!("Using billing_email from tenants table: {email}"); - stripe::Customer::update( - client, - &customer.id, - stripe::UpdateCustomer { - email: Some(&email), - ..Default::default() - }, - ) - .await?; - } else { - let responses = sqlx::query!( - r#" + let email = match db_email { + Some(email) => { + tracing::info!("Using billing_email from tenants table: {email}"); + email + } + None => match earliest_admin_email(db_client, tenant).await? { + Some(email) => { + tracing::warn!( + "Stripe customer object is missing an email. Going with {email}, an admin on that tenant." + ); + email + } + None => bail!( + "Stripe customer object is missing an email. No admins found for tenant {tenant}, unable to create invoice without email. Skipping" + ), + }, + }; + + stripe::Customer::update( + client, + &customer.id, + stripe::UpdateCustomer { + email: Some(&email), + ..Default::default() + }, + ) + .await?; + } + Ok(customer) +} + +/// Read-only: the email of the earliest-created admin user on the tenant, if any. +async fn earliest_admin_email( + db_client: &Pool, + tenant: &str, +) -> anyhow::Result> { + // NOTE: the SQL text below is intentionally indented to stay byte-identical + // to the query this was extracted from, so it keeps hitting the same + // offline sqlx cache entry. + let responses = sqlx::query!( + r#" select users.email as email from user_grants join auth.users as users on user_grants.user_id = users.id @@ -866,35 +1220,42 @@ async fn get_or_create_customer_for_tenant( and user_grants.capability = 'admin' order by users.created_at asc "#, - tenant - ) - .fetch_all(db_client) - .await?; + tenant + ) + .fetch_all(db_client) + .await?; - if let Some(email) = responses - .iter() - .find_map(|response| response.email.to_owned()) - { - tracing::warn!( - "Stripe customer object is missing an email. Going with {email}, an admin on that tenant." - ); - stripe::Customer::update( - client, - &customer.id, - stripe::UpdateCustomer { - email: Some(&email), - ..Default::default() - }, - ) - .await?; - } else { - bail!( - "Stripe customer object is missing an email. No admins found for tenant {tenant}, unable to create invoice without email. Found users: {found:?} Skipping", - found = responses, - tenant = tenant - ); - } - } + Ok(responses.into_iter().find_map(|r| r.email)) +} + +/// Read-only mirror of the email requirement enforced by +/// `ensure_customer_for_invoicing`: an existing customer email, else +/// `tenants.billing_email`, else the earliest tenant admin. Returns whether an +/// email could be resolved without performing any writes, so --dry-run can +/// report tenants that would fail to invoice rather than over-reporting success. +async fn billing_email_available( + client: &stripe::Client, + db_client: &Pool, + tenant: &str, +) -> anyhow::Result { + if find_customer(client, tenant) + .await? + .is_some_and(|c| c.email.is_some()) + { + return Ok(true); } - Ok(Some(customer)) + + // Fetches billing_address too (unused here) to reuse the cached query that + // ensure_customer_for_invoicing already issues. + let billing_row = sqlx::query!( + r#"SELECT billing_email, billing_address FROM tenants WHERE tenant = $1"#, + tenant, + ) + .fetch_optional(db_client) + .await?; + if billing_row.and_then(|r| r.billing_email).is_some() { + return Ok(true); + } + + Ok(earliest_admin_email(db_client, tenant).await?.is_some()) } diff --git a/crates/billing-integrations/src/send.rs b/crates/billing-integrations/src/send.rs index bd626870726..7fb1f58221c 100644 --- a/crates/billing-integrations/src/send.rs +++ b/crates/billing-integrations/src/send.rs @@ -183,13 +183,15 @@ async fn update_draft_collection_methods( stripe_client: &Client, mut to_update: Vec, ) -> anyhow::Result> { - // Identify invoices that are `charge_automatically` but don't have a default payment method + // Identify invoices that need to be switched to `send_invoice`: + // - Manual invoices should always be sent as invoices, never auto-charged + // - Auto-charge invoices without a payment method on file must be sent as invoices let needs_update: HashSet = to_update .iter() .filter(|inv| { inv.collection_method().map_or(false, |cm| { cm == stripe::CollectionMethod::ChargeAutomatically - }) && !inv.has_cc() + }) && (inv.is_manual() || !inv.has_cc()) }) .map(|inv| inv.id().clone()) .collect::>(); diff --git a/crates/billing-integrations/src/stripe_utils.rs b/crates/billing-integrations/src/stripe_utils.rs index fa35f97f562..37bb20ea0ca 100644 --- a/crates/billing-integrations/src/stripe_utils.rs +++ b/crates/billing-integrations/src/stripe_utils.rs @@ -1,4 +1,4 @@ -use billing_types::{InvoiceMetadata, SearchParams, stripe_search}; +use billing_types::{InvoiceMetadata, InvoiceType, SearchParams, stripe_search}; use num_format::{Locale, ToFormattedString}; use std::ops::{Deref, DerefMut}; @@ -84,6 +84,14 @@ impl Invoice { self.0.status.clone() } + pub fn is_manual(&self) -> bool { + self.0 + .metadata + .as_ref() + .and_then(InvoiceMetadata::from_metadata_map) + .is_some_and(|m| m.invoice_type == InvoiceType::Manual) + } + pub fn period_start(&self) -> Option { self.0 .metadata