stripe: Fix $0 invoices being generated on upgrade.

Instead of charging the customer using the attached payment
method and then creating the invoice, we create an invoice and
force an immediate payment for the invoice via the attached
payment method.

corporate/lib/stripe.py

@@ -32,7 +32,6 @@ from corporate.models import (
     CustomerPlanOffer,
     Invoice,
     LicenseLedger,
-    PaymentIntent,
     Session,
     SponsoredPlanTypes,
     ZulipSponsorshipRequest,
@@ -529,14 +528,6 @@ class StripeCustomerData:
     metadata: Dict[str, Any]
 
 
-@dataclass
-class StripePaymentIntentData:
-    amount: int
-    description: str
-    plan_name: str
-    email: str
-
-
 @dataclass
 class UpgradeRequest:
     billing_modality: str
@@ -567,7 +558,7 @@ class UpdatePlanRequest:
 @dataclass
 class EventStatusRequest:
     stripe_session_id: Optional[str]
-    stripe_payment_intent_id: Optional[str]
+    stripe_invoice_id: Optional[str]
 
 
 class SupportType(Enum):
@@ -747,7 +738,7 @@ class BillingSession(ABC):
         pass
 
     @abstractmethod
-    def update_data_for_checkout_session_and_payment_intent(
+    def update_data_for_checkout_session_and_invoice_payment(
         self, metadata: Dict[str, Any]
     ) -> Dict[str, Any]:
         pass
@@ -787,7 +778,7 @@ class BillingSession(ABC):
             return_url=f"{self.billing_session_url}/billing/",
         ).url
 
-    def get_data_for_stripe_payment_intent(
+    def generate_invoice_for_upgrade(
         self,
         customer: Customer,
         price_per_license: Optional[int],
@@ -795,36 +786,77 @@ class BillingSession(ABC):
         licenses: int,
         plan_tier: int,
         billing_schedule: int,
-        email: str,
+        charge_automatically: bool,
-    ) -> StripePaymentIntentData:
+        license_management: Optional[str] = None,
-        if hasattr(self, "support_session") and self.support_session:  # nocoverage
+    ) -> stripe.Invoice:
-            raise BillingError(
-                "invalid support session",
-                "Support requests do not set any stripe billing information.",
-            )
-
         plan_name = CustomerPlan.name_from_tier(plan_tier)
         assert price_per_license is None or fixed_price is None
-        if price_per_license is not None:
+        price_args: PriceArgs = {}
-            amount = price_per_license * licenses
+        if fixed_price is None:
-            description = f"Upgrade to {plan_name}, ${price_per_license/100} x {licenses}"
+            assert price_per_license is not None
+            price_args = {
+                "quantity": licenses,
+                "unit_amount": price_per_license,
+            }
         else:
             assert fixed_price is not None
-            amount = get_amount_due_fixed_price_plan(fixed_price, billing_schedule)
+            amount_due = get_amount_due_fixed_price_plan(fixed_price, billing_schedule)
-            description = plan_name
+            price_args = {"amount": amount_due}
+
+        stripe.InvoiceItem.create(
+            currency="usd",
+            customer=customer.stripe_customer_id,
+            description=plan_name,
+            discountable=False,
+            **price_args,
+        )
 
         if fixed_price is None and customer.flat_discounted_months > 0:
             num_months = 12 if billing_schedule == CustomerPlan.BILLING_SCHEDULE_ANNUAL else 1
             flat_discounted_months = min(customer.flat_discounted_months, num_months)
-            amount -= customer.flat_discount * flat_discounted_months
+            discount = customer.flat_discount * flat_discounted_months
-            description += f" - ${customer.flat_discount/100} x {flat_discounted_months}"
+            customer.flat_discounted_months -= flat_discounted_months
+            customer.save(update_fields=["flat_discounted_months"])
 
-        return StripePaymentIntentData(
+            stripe.InvoiceItem.create(
-            amount=amount,
+                currency="usd",
-            description=description,
+                customer=customer.stripe_customer_id,
-            plan_name=plan_name,
+                description=f"${cents_to_dollar_string(customer.flat_discount)}/month new customer discount",
-            email=email,
+                # Negative value to apply discount.
+                amount=(-1 * discount),
+            )
+
+        if charge_automatically:
+            collection_method = "charge_automatically"
+            days_until_due = None
+        else:
+            collection_method = "send_invoice"
+            days_until_due = DEFAULT_INVOICE_DAYS_UNTIL_DUE
+
+        metadata = {
+            "plan_tier": plan_tier,
+            "billing_schedule": billing_schedule,
+            "licenses": licenses,
+            "license_management": license_management,
+        }
+
+        if hasattr(self, "user"):
+            metadata["user_id"] = self.user.id
+
+        # We only need to email customer about open invoice for manual billing.
+        # If automatic charge fails, we simply void the invoice.
+        # https://stripe.com/docs/invoicing/integration/automatic-advancement-collection
+        auto_advance = not charge_automatically
+        stripe_invoice = stripe.Invoice.create(
+            auto_advance=auto_advance,
+            collection_method=collection_method,
+            customer=customer.stripe_customer_id,
+            days_until_due=days_until_due,
+            statement_descriptor=plan_name,
+            metadata=metadata,
         )
+        stripe.Invoice.finalize_invoice(stripe_invoice)
+        return stripe_invoice
 
     @abstractmethod
     def update_or_create_customer(
@@ -1000,9 +1032,8 @@ class BillingSession(ABC):
     def update_or_create_stripe_customer(self, payment_method: Optional[str] = None) -> Customer:
         customer = self.get_customer()
         if customer is None or customer.stripe_customer_id is None:
-            # A stripe.PaymentMethod should be attached to a stripe.Customer via
+            # A stripe.PaymentMethod should be attached to a stripe.Customer via replace_payment_method.
-            # a stripe.SetupIntent or stripe.PaymentIntent. Here we just want to
+            # Here we just want to create a new stripe.Customer.
-            # create a new stripe.Customer.
             assert payment_method is None
             # We could do a better job of handling race conditions here, but if two
             # people try to upgrade at exactly the same time, the main bad thing that
@@ -1013,22 +1044,13 @@ class BillingSession(ABC):
             self.replace_payment_method(customer.stripe_customer_id, payment_method, True)
         return customer
 
-    def create_stripe_payment_intent(
+    def create_stripe_invoice_and_charge(
         self,
         metadata: Dict[str, Any],
     ) -> str:
         # NOTE: This charges users immediately.
         customer = self.get_customer()
         assert customer is not None and customer.stripe_customer_id is not None
-        payment_intent_data = self.get_data_for_stripe_payment_intent(
-            customer,
-            metadata["price_per_license"],
-            metadata["fixed_price"],
-            metadata["licenses"],
-            metadata["plan_tier"],
-            metadata["billing_schedule"],
-            self.get_email(),
-        )
         # Ensure customers have a default payment method set.
         stripe_customer = stripe_get_customer(customer.stripe_customer_id)
         if not stripe_customer_has_credit_card_as_default_payment_method(stripe_customer):
@@ -1039,29 +1061,42 @@ class BillingSession(ABC):
 
         assert stripe_customer.invoice_settings is not None
         assert stripe_customer.invoice_settings.default_payment_method is not None
+        stripe_invoice = None
         try:
-            # Try to charge user immediately, and if that fails, we inform the user about the failure.
+            stripe_invoice = self.generate_invoice_for_upgrade(
-            stripe_payment_intent = stripe.PaymentIntent.create(
+                customer,
-                amount=payment_intent_data.amount,
+                metadata["price_per_license"],
-                currency="usd",
+                metadata["fixed_price"],
-                customer=customer.stripe_customer_id,
+                metadata["licenses"],
-                description=payment_intent_data.description,
+                metadata["plan_tier"],
-                receipt_email=payment_intent_data.email,
+                metadata["billing_schedule"],
-                confirm=True,
+                charge_automatically=True,
-                statement_descriptor=payment_intent_data.plan_name,
+                license_management=metadata["license_management"],
-                metadata=metadata,
-                off_session=True,
-                payment_method=stripe_customer.invoice_settings.default_payment_method,
             )
-        except stripe.CardError as e:
+            assert stripe_invoice.id is not None
-            raise StripeCardError("card error", e.user_message)
+            invoice = Invoice.objects.create(
+                stripe_invoice_id=stripe_invoice.id,
+                customer=customer,
+                status=Invoice.SENT,
+            )
+            # Stripe takes its sweet hour to charge customers after creating an invoice.
+            # Since we want to charge customers immediately, we charge them manually.
+            # Then poll for the status of the invoice to see if the payment succeeded.
+            stripe_invoice = stripe.Invoice.pay(stripe_invoice.id)
+        except Exception as e:
+            if stripe_invoice is not None:
+                assert stripe_invoice.id is not None
+                # Void invoice to avoid double charging if customer tries to upgrade again.
+                stripe.Invoice.void_invoice(stripe_invoice.id)
+                invoice.status = Invoice.VOID
+                invoice.save(update_fields=["status"])
+            if isinstance(e, stripe.CardError):
+                raise StripeCardError("card error", e.user_message)
+            else:  # nocoverage
+                raise e
 
-        PaymentIntent.objects.create(
+        assert stripe_invoice.id is not None
-            customer=customer,
+        return stripe_invoice.id
-            stripe_payment_intent_id=stripe_payment_intent.id,
-            status=PaymentIntent.get_status_integer_from_status_text(stripe_payment_intent.status),
-        )
-        return stripe_payment_intent.id
 
     def create_card_update_session_for_upgrade(
         self,
@@ -1392,7 +1427,7 @@ class BillingSession(ABC):
             f"No current plan for {self.billing_entity_display_name}."
         )  # nocoverage
 
-    def setup_upgrade_payment_intent_and_charge(
+    def generate_stripe_invoice_and_charge_immediately(
         self,
         plan_tier: int,
         seat_count: int,
@@ -1423,11 +1458,10 @@ class BillingSession(ABC):
             general_metadata["price_per_license"] = price_per_license
         else:
             general_metadata["fixed_price"] = fixed_price_plan_offer.fixed_price
-        updated_metadata = self.update_data_for_checkout_session_and_payment_intent(
+        updated_metadata = self.update_data_for_checkout_session_and_invoice_payment(
             general_metadata
         )
-        stripe_payment_intent_id = self.create_stripe_payment_intent(updated_metadata)
+        return self.create_stripe_invoice_and_charge(updated_metadata)
-        return stripe_payment_intent_id
 
     def ensure_current_plan_is_upgradable(self, customer: Customer, new_plan_tier: int) -> None:
         # Upgrade for customers with an existing plan is only supported for remote realm / server right now.
@@ -1614,12 +1648,31 @@ class BillingSession(ABC):
                     plan=plan,
                     is_renewal=True,
                     event_time=event_time,
-                    licenses=billed_licenses,
+                    licenses=licenses,
-                    licenses_at_next_renewal=billed_licenses,
+                    licenses_at_next_renewal=licenses,
                 )
                 plan.invoiced_through = ledger_entry
                 plan.save(update_fields=["invoiced_through"])
 
+                # TODO: Do a check for max licenses for fixed price plans here after we add that.
+                if (
+                    stripe_invoice_paid
+                    and billed_licenses != licenses
+                    and not customer.exempt_from_license_number_check
+                    and not fixed_price_plan_offer
+                ):
+                    # Customer paid for less licenses than they have.
+                    # We need to create a new ledger entry to track the additional licenses.
+                    LicenseLedger.objects.create(
+                        plan=plan,
+                        is_renewal=False,
+                        event_time=event_time,
+                        licenses=billed_licenses,
+                        licenses_at_next_renewal=billed_licenses,
+                    )
+                    # Creates due today invoice for additional licenses.
+                    self.invoice_plan(plan, event_time)
+
             self.write_to_audit_log(
                 event_type=AuditLogEventType.CUSTOMER_PLAN_CREATED,
                 event_time=event_time,
@@ -1630,60 +1683,15 @@ class BillingSession(ABC):
             free_trial or should_schedule_upgrade_for_legacy_remote_server
         ):
             assert plan is not None
-            price_args: PriceArgs = {}
+            self.generate_invoice_for_upgrade(
-            if plan.fixed_price is None:
+                customer,
-                price_args = {
+                price_per_license=price_per_license,
-                    "quantity": billed_licenses,
+                fixed_price=plan.fixed_price,
-                    "unit_amount": price_per_license,
+                licenses=billed_licenses,
-                }
+                plan_tier=plan.tier,
-            else:
+                billing_schedule=billing_schedule,
-                assert plan.fixed_price is not None
+                charge_automatically=False,
-                amount_due = get_amount_due_fixed_price_plan(plan.fixed_price, billing_schedule)
-                price_args = {"amount": amount_due}
-            assert customer.stripe_customer_id is not None
-            stripe.InvoiceItem.create(
-                currency="usd",
-                customer=customer.stripe_customer_id,
-                description=plan.name,
-                discountable=False,
-                period={
-                    "start": datetime_to_timestamp(billing_cycle_anchor),
-                    "end": datetime_to_timestamp(period_end),
-                },
-                **price_args,
             )
-
-            if plan.fixed_price is None and customer.flat_discounted_months > 0:
-                num_months = 12 if billing_schedule == CustomerPlan.BILLING_SCHEDULE_ANNUAL else 1
-                flat_discounted_months = min(customer.flat_discounted_months, num_months)
-                discount = customer.flat_discount * flat_discounted_months
-                customer.flat_discounted_months -= flat_discounted_months
-                customer.save(update_fields=["flat_discounted_months"])
-
-                stripe.InvoiceItem.create(
-                    currency="usd",
-                    customer=customer.stripe_customer_id,
-                    description=f"${cents_to_dollar_string(customer.flat_discount)}/month new customer discount",
-                    # Negative value to apply discount.
-                    amount=(-1 * discount),
-                )
-
-            if charge_automatically:
-                collection_method = "charge_automatically"
-                days_until_due = None
-            else:
-                collection_method = "send_invoice"
-                days_until_due = DEFAULT_INVOICE_DAYS_UNTIL_DUE
-
-            stripe_invoice = stripe.Invoice.create(
-                auto_advance=True,
-                collection_method=collection_method,
-                customer=customer.stripe_customer_id,
-                days_until_due=days_until_due,
-                statement_descriptor=plan.name,
-            )
-            stripe.Invoice.finalize_invoice(stripe_invoice)
-
         if plan.status < CustomerPlan.LIVE_STATUS_THRESHOLD:
             # Tier and usage limit change will happen when plan becomes live.
             self.do_change_plan_type(tier=plan_tier)
@@ -1761,7 +1769,7 @@ class BillingSession(ABC):
             )
             data["organization_upgrade_successful"] = True
         else:
-            stripe_payment_intent_id = self.setup_upgrade_payment_intent_and_charge(
+            stripe_invoice_id = self.generate_stripe_invoice_and_charge_immediately(
                 upgrade_request.tier,
                 seat_count,
                 licenses,
@@ -1769,7 +1777,7 @@ class BillingSession(ABC):
                 billing_schedule,
                 billing_modality,
             )
-            data["stripe_payment_intent_id"] = stripe_payment_intent_id
+            data["stripe_invoice_id"] = stripe_invoice_id
         return data
 
     def do_change_schedule_after_free_trial(self, plan: CustomerPlan, schedule: int) -> None:
@@ -2827,18 +2835,18 @@ class BillingSession(ABC):
                 raise JsonableError(_("Must be a billing administrator or an organization owner"))
             return {"session": session.to_dict()}
 
-        stripe_payment_intent_id = event_status_request.stripe_payment_intent_id
+        stripe_invoice_id = event_status_request.stripe_invoice_id
-        if stripe_payment_intent_id is not None:
+        if stripe_invoice_id is not None:
-            payment_intent = PaymentIntent.objects.filter(
+            stripe_invoice = Invoice.objects.filter(
-                stripe_payment_intent_id=stripe_payment_intent_id,
+                stripe_invoice_id=stripe_invoice_id,
                 customer=customer,
             ).last()
 
-            if payment_intent is None:
+            if stripe_invoice is None:
                 raise JsonableError(_("Payment intent not found"))
-            return {"payment_intent": payment_intent.to_dict()}
+            return {"stripe_invoice": stripe_invoice.to_dict()}
 
-        raise JsonableError(_("Pass stripe_session_id or stripe_payment_intent_id"))
+        raise JsonableError(_("Pass stripe_session_id or stripe_invoice_id"))
 
     def get_sponsorship_plan_name(
         self, customer: Optional[Customer], is_remotely_hosted: bool
@@ -3378,7 +3386,7 @@ class RealmBillingSession(BillingSession):
         return realm_stripe_customer_data
 
     @override
-    def update_data_for_checkout_session_and_payment_intent(
+    def update_data_for_checkout_session_and_invoice_payment(
         self, metadata: Dict[str, Any]
     ) -> Dict[str, Any]:
         assert self.user is not None
@@ -3771,7 +3779,7 @@ class RemoteRealmBillingSession(BillingSession):
         return realm_stripe_customer_data
 
     @override
-    def update_data_for_checkout_session_and_payment_intent(
+    def update_data_for_checkout_session_and_invoice_payment(
         self, metadata: Dict[str, Any]
     ) -> Dict[str, Any]:
         assert self.remote_billing_user is not None
@@ -4194,7 +4202,7 @@ class RemoteServerBillingSession(BillingSession):
         return realm_stripe_customer_data
 
     @override
-    def update_data_for_checkout_session_and_payment_intent(
+    def update_data_for_checkout_session_and_invoice_payment(
         self, metadata: Dict[str, Any]
     ) -> Dict[str, Any]:
         assert self.remote_billing_user is not None

corporate/lib/stripe_event_handler.py

@@ -1,19 +1,19 @@
 import logging
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Optional, Union
 
 import stripe
 from django.conf import settings
 
 from corporate.lib.stripe import (
+    BILLING_SUPPORT_EMAIL,
     BillingError,
-    InvalidPlanUpgradeError,
     RealmBillingSession,
     RemoteRealmBillingSession,
     RemoteServerBillingSession,
-    UpgradeWithExistingPlanError,
     get_configured_fixed_price_plan_offer,
 )
-from corporate.models import Customer, CustomerPlan, Event, Invoice, PaymentIntent, Session
+from corporate.models import Customer, CustomerPlan, Event, Invoice, Session
+from zerver.lib.send_email import FromAddress, send_email
 from zerver.models.users import get_active_user_profile_by_id_in_realm
 
 billing_logger = logging.getLogger("corporate.stripe")
@@ -21,9 +21,9 @@ billing_logger = logging.getLogger("corporate.stripe")
 
 def error_handler(
     func: Callable[[Any, Any], None],
-) -> Callable[[Union[stripe.checkout.Session, stripe.PaymentIntent, stripe.Invoice], Event], None]:
+) -> Callable[[Union[stripe.checkout.Session, stripe.Invoice], Event], None]:
     def wrapper(
-        stripe_object: Union[stripe.checkout.Session, stripe.PaymentIntent, stripe.Invoice],
+        stripe_object: Union[stripe.checkout.Session, stripe.Invoice],
         event: Event,
     ) -> None:
         event.status = Event.EVENT_HANDLER_STARTED
@@ -32,7 +32,7 @@ def error_handler(
         try:
             func(stripe_object, event.content_object)
         except BillingError as e:
-            billing_logger.warning(
+            message = (
                 "BillingError in %s event handler: %s. stripe_object_id=%s, customer_id=%s metadata=%s",
                 event.type,
                 e.error_description,
@@ -40,12 +40,23 @@ def error_handler(
                 stripe_object.customer,
                 stripe_object.metadata,
             )
+            billing_logger.warning(message)
             event.status = Event.EVENT_HANDLER_FAILED
             event.handler_error = {
                 "message": e.msg,
                 "description": e.error_description,
             }
             event.save(update_fields=["status", "handler_error"])
+            if type(stripe_object) == stripe.Invoice:
+                # For Invoice processing errors, send email to billing support.
+                send_email(
+                    "zerver/emails/error_processing_invoice",
+                    to_emails=[BILLING_SUPPORT_EMAIL],
+                    from_address=FromAddress.tokenized_no_reply_address(),
+                    context={
+                        "message": message,
+                    },
+                )
         except Exception:
             billing_logger.exception(
                 "Uncaught exception in %s event handler:",
@@ -102,58 +113,6 @@ def handle_checkout_session_completed_event(
         billing_session.update_or_create_stripe_customer(payment_method)
 
 
-@error_handler
-def handle_payment_intent_succeeded_event(
-    stripe_payment_intent: stripe.PaymentIntent, payment_intent: PaymentIntent
-) -> None:
-    payment_intent.status = PaymentIntent.SUCCEEDED
-    payment_intent.save()
-    metadata: Dict[str, Any] = stripe_payment_intent.metadata
-
-    description = ""
-    charge: stripe.Charge
-    for charge in stripe_payment_intent.charges:  # type: ignore[attr-defined] # https://stripe.com/docs/upgrades#2022-11-15
-        assert charge.payment_method_details is not None
-        assert charge.payment_method_details.card is not None
-        description = f"Payment (Card ending in {charge.payment_method_details.card.last4})"
-        break
-
-    stripe.InvoiceItem.create(
-        amount=stripe_payment_intent.amount * -1,
-        currency="usd",
-        customer=stripe_payment_intent.customer,
-        description=description,
-        discountable=False,
-    )
-    billing_session = get_billing_session_for_stripe_webhook(
-        payment_intent.customer, metadata.get("user_id")
-    )
-    plan_tier = int(metadata["plan_tier"])
-    try:
-        billing_session.ensure_current_plan_is_upgradable(payment_intent.customer, plan_tier)
-    except (UpgradeWithExistingPlanError, InvalidPlanUpgradeError) as e:
-        stripe_invoice = stripe.Invoice.create(
-            auto_advance=True,
-            collection_method="charge_automatically",
-            customer=stripe_payment_intent.customer,
-            days_until_due=None,
-            statement_descriptor=CustomerPlan.name_from_tier(plan_tier).replace("Zulip ", "")
-            + " Credit",
-        )
-        stripe.Invoice.finalize_invoice(stripe_invoice)
-        raise e
-
-    billing_session.process_initial_upgrade(
-        plan_tier,
-        int(metadata["licenses"]),
-        metadata["license_management"] == "automatic",
-        int(metadata["billing_schedule"]),
-        True,
-        False,
-        billing_session.get_remote_server_legacy_plan(payment_intent.customer),
-    )
-
-
 @error_handler
 def handle_invoice_paid_event(stripe_invoice: stripe.Invoice, invoice: Invoice) -> None:
     invoice.status = Invoice.PAID
@@ -183,3 +142,35 @@ def handle_invoice_paid_event(stripe_invoice: stripe.Invoice, invoice: Invoice)
             remote_server_legacy_plan=remote_server_legacy_plan,
             stripe_invoice_paid=True,
         )
+    elif stripe_invoice.collection_method == "charge_automatically":
+        metadata = stripe_invoice.metadata
+        assert metadata is not None
+        billing_session = get_billing_session_for_stripe_webhook(customer, metadata.get("user_id"))
+        remote_server_legacy_plan = billing_session.get_remote_server_legacy_plan(customer)
+        billing_schedule = int(metadata["billing_schedule"])
+        plan_tier = int(metadata["plan_tier"])
+        if configured_fixed_price_plan and customer.required_plan_tier == plan_tier:
+            assert customer.required_plan_tier is not None
+            billing_session.process_initial_upgrade(
+                plan_tier=customer.required_plan_tier,
+                # TODO: Currently licenses don't play any role for fixed price plan.
+                # We plan to introduce max_licenses allowed soon.
+                licenses=0,
+                automanage_licenses=True,
+                billing_schedule=billing_schedule,
+                charge_automatically=True,
+                free_trial=False,
+                remote_server_legacy_plan=remote_server_legacy_plan,
+                stripe_invoice_paid=True,
+            )
+        else:
+            billing_session.process_initial_upgrade(
+                plan_tier,
+                int(metadata["licenses"]),
+                metadata["license_management"] == "automatic",
+                billing_schedule=billing_schedule,
+                charge_automatically=True,
+                free_trial=False,
+                remote_server_legacy_plan=remote_server_legacy_plan,
+                stripe_invoice_paid=True,
+            )

corporate/models.py

@@ -113,7 +113,7 @@ class Event(models.Model):
 
 
 def get_last_associated_event_by_type(
-    content_object: Union["PaymentIntent", "Session"], event_type: str
+    content_object: Union["Invoice", "PaymentIntent", "Session"], event_type: str
 ) -> Optional[Event]:
     content_type = ContentType.objects.get_for_model(type(content_object))
     return Event.objects.filter(
@@ -168,7 +168,7 @@ class Session(models.Model):
         return get_last_associated_event_by_type(self, "checkout.session.completed")
 
 
-class PaymentIntent(models.Model):
+class PaymentIntent(models.Model):  # nocoverage
     customer = models.ForeignKey(Customer, on_delete=CASCADE)
     stripe_payment_intent_id = models.CharField(max_length=255, unique=True)
 
@@ -221,8 +221,32 @@ class Invoice(models.Model):
 
     SENT = 1
     PAID = 2
+    VOID = 3
     status = models.SmallIntegerField()
 
+    def get_status_as_string(self) -> str:
+        return {
+            Invoice.SENT: "sent",
+            Invoice.PAID: "paid",
+            Invoice.VOID: "void",
+        }[self.status]
+
+    def get_last_associated_event(self) -> Optional[Event]:
+        if self.status == Invoice.PAID:
+            event_type = "invoice.paid"
+        # TODO: Add test for this case. Not sure how to trigger naturally.
+        else:  # nocoverage
+            return None  # nocoverage
+        return get_last_associated_event_by_type(self, event_type)
+
+    def to_dict(self) -> Dict[str, Any]:
+        stripe_invoice_dict: Dict[str, Any] = {}
+        stripe_invoice_dict["status"] = self.get_status_as_string()
+        event = self.get_last_associated_event()
+        if event is not None:
+            stripe_invoice_dict["event_handler"] = event.get_event_handler_details_as_dict()
+        return stripe_invoice_dict
+
 
 class AbstractCustomerPlan(models.Model):
     # A customer can only have one ACTIVE / CONFIGURED plan,