stripe: Send invoice to customer at the start of free trial.

We send customer an invoice at the start of free trial, if customer
pays we upgrade them to the active plan at the end of free trial,
else we downgrade them and show a custom message on the upgrade
page regarding the current status.

corporate/lib/stripe.py

@@ -685,6 +685,8 @@ class UpgradePageContext(TypedDict):
     is_sponsorship_pending: bool
     sponsorship_plan_name: str
     scheduled_upgrade_invoice_amount_due: Optional[str]
+    is_free_trial_invoice_expired_notice: bool
+    free_trial_invoice_expired_notice_page_plan_name: Optional[str]
 
 
 class SponsorshipRequestForm(forms.Form):
@@ -868,6 +870,9 @@ class BillingSession(ABC):
         charge_automatically: bool,
         invoice_period: Dict[str, int],
         license_management: Optional[str] = None,
+        days_until_due: Optional[int] = None,
+        on_free_trial: bool = False,
+        current_plan_id: Optional[int] = None,
     ) -> stripe.Invoice:
         plan_name = CustomerPlan.name_from_tier(plan_tier)
         assert price_per_license is None or fixed_price is None
@@ -910,19 +915,21 @@ class BillingSession(ABC):
 
         if charge_automatically:
             collection_method = "charge_automatically"
-            days_until_due = None
         else:
             collection_method = "send_invoice"
             # days_until_due is required for `send_invoice` collection method. Since this is an invoice
             # for upgrade, the due date is irrelevant since customer will upgrade once they pay the invoice
             # regardless of the due date. Using `1` shows `Due today / tomorrow` which seems nice.
-            days_until_due = 1
+            if days_until_due is None:
+                days_until_due = 1
 
         metadata = {
             "plan_tier": plan_tier,
             "billing_schedule": billing_schedule,
             "licenses": licenses,
             "license_management": license_management,
+            "on_free_trial": on_free_trial,
+            "current_plan_id": current_plan_id,
         }
 
         if hasattr(self, "user"):
@@ -1156,6 +1163,8 @@ class BillingSession(ABC):
             assert stripe_customer.invoice_settings.default_payment_method is not None
         stripe_invoice = None
         try:
+            current_plan_id = metadata.get("current_plan_id")
+            on_free_trial = bool(metadata.get("on_free_trial"))
             stripe_invoice = self.generate_invoice_for_upgrade(
                 customer,
                 metadata["price_per_license"],
@@ -1166,13 +1175,20 @@ class BillingSession(ABC):
                 charge_automatically=charge_automatically,
                 license_management=metadata["license_management"],
                 invoice_period=metadata["invoice_period"],
+                days_until_due=metadata.get("days_until_due"),
+                on_free_trial=on_free_trial,
+                current_plan_id=current_plan_id,
             )
             assert stripe_invoice.id is not None
+
             invoice = Invoice.objects.create(
                 stripe_invoice_id=stripe_invoice.id,
                 customer=customer,
                 status=Invoice.SENT,
+                plan_id=current_plan_id,
+                is_created_for_free_trial_upgrade=current_plan_id is not None and on_free_trial,
             )
+
             if charge_automatically:
                 # Stripe takes its sweet hour to charge customers after creating an invoice.
                 # Since we want to charge customers immediately, we charge them manually.
@@ -1583,6 +1599,9 @@ class BillingSession(ABC):
         license_management: str,
         billing_schedule: int,
         billing_modality: str,
+        on_free_trial: bool = False,
+        days_until_due: Optional[int] = None,
+        current_plan_id: Optional[int] = None,
     ) -> str:
         customer = self.update_or_create_stripe_customer()
         assert customer is not None  # for mypy
@@ -1596,6 +1615,9 @@ class BillingSession(ABC):
             "fixed_price": None,
             "type": "upgrade",
             "plan_tier": plan_tier,
+            "on_free_trial": on_free_trial,
+            "days_until_due": days_until_due,
+            "current_plan_id": current_plan_id,
         }
         discount_for_plan = customer.get_discount_for_plan_tier(plan_tier)
         (
@@ -1607,9 +1629,7 @@ class BillingSession(ABC):
             plan_tier,
             billing_schedule,
             discount_for_plan,
-            # TODO: Use the correct value for free_trial when we switch behaviour to send invoice
-            # at the start of free trial.
-            False,
+            on_free_trial,
             None,
             not isinstance(self, RealmBillingSession),
         )
@@ -1621,6 +1641,14 @@ class BillingSession(ABC):
                 invoice_period_start, CustomerPlan.FIXED_PRICE_PLAN_DURATION_MONTHS
             )
 
+        if on_free_trial and billing_modality == "send_invoice":
+            # Paid plan starts at the end of free trial.
+            invoice_period_start = invoice_period_end
+            purchased_months = 1
+            if billing_schedule == CustomerPlan.BILLING_SCHEDULE_ANNUAL:
+                purchased_months = 12
+            invoice_period_end = add_months(invoice_period_end, purchased_months)
+
         general_metadata["invoice_period"] = {
             "start": datetime_to_timestamp(invoice_period_start),
             "end": datetime_to_timestamp(invoice_period_end),
@@ -1855,6 +1883,8 @@ class BillingSession(ABC):
         if not stripe_invoice_paid and not (
             free_trial or should_schedule_upgrade_for_legacy_remote_server
         ):
+            # We don't actually expect to ever reach here but this is just a safety net
+            # in case any future changes make this possible.
             assert plan is not None
             self.generate_invoice_for_upgrade(
                 customer,
@@ -1869,6 +1899,22 @@ class BillingSession(ABC):
                     "end": datetime_to_timestamp(period_end),
                 },
             )
+        elif free_trial and not charge_automatically:
+            assert stripe_invoice_paid is False
+            assert plan is not None
+            assert plan.next_invoice_date is not None
+            # Send an invoice to the customer which expires at the end of free trial. If the customer
+            # fails to pay the invoice before expiration, we downgrade the customer.
+            self.generate_stripe_invoice(
+                plan_tier,
+                licenses=billed_licenses,
+                license_management="automatic" if automanage_licenses else "manual",
+                billing_schedule=billing_schedule,
+                billing_modality="send_invoice",
+                on_free_trial=True,
+                days_until_due=(plan.next_invoice_date - event_time).days,
+                current_plan_id=plan.id,
+            )
 
     def do_upgrade(self, upgrade_request: UpgradeRequest) -> Dict[str, Any]:
         customer = self.get_customer()
@@ -1950,6 +1996,9 @@ class BillingSession(ABC):
         return data
 
     def do_change_schedule_after_free_trial(self, plan: CustomerPlan, schedule: int) -> None:
+        # NOTE: Schedule change for free trial with invoice payments is not supported due to complication
+        # involving sending another invoice and handling payment difference if customer already paid.
+        assert plan.charge_automatically
         # Change the billing frequency of the plan after the free trial ends.
         assert schedule in (
             CustomerPlan.BILLING_SCHEDULE_MONTHLY,
@@ -2074,13 +2123,28 @@ class BillingSession(ABC):
                     licenses_at_next_renewal=licenses_at_next_renewal,
                 )
             if plan.is_free_trial():
+                is_renewal = True
+                # Check if user has already paid for the plan by invoice.
+                if not plan.charge_automatically:
+                    last_sent_invoice = Invoice.objects.filter(plan=plan).order_by("-id").first()
+                    if last_sent_invoice and last_sent_invoice.status == Invoice.PAID:
+                        # This will create invoice for any additional licenses that user has at the time of
+                        # switching from free trial to paid plan since they already paid for the plan's this billing cycle.
+                        is_renewal = False
+                    else:
+                        # We end the free trial since customer hasn't paid.
+                        plan.status = CustomerPlan.DOWNGRADE_AT_END_OF_FREE_TRIAL
+                        plan.save(update_fields=["status"])
+                        self.make_end_of_cycle_updates_if_needed(plan, event_time)
+                        return None, None
+
                 plan.invoiced_through = last_ledger_entry
                 plan.billing_cycle_anchor = next_billing_cycle.replace(microsecond=0)
                 plan.status = CustomerPlan.ACTIVE
                 plan.save(update_fields=["invoiced_through", "billing_cycle_anchor", "status"])
                 return None, LicenseLedger.objects.create(
                     plan=plan,
-                    is_renewal=True,
+                    is_renewal=is_renewal,
                     event_time=next_billing_cycle,
                     licenses=licenses_at_next_renewal,
                     licenses_at_next_renewal=licenses_at_next_renewal,
@@ -2245,6 +2309,7 @@ class BillingSession(ABC):
         last_ledger_entry: LicenseLedger,
         now: datetime,
     ) -> Dict[str, Any]:
+        is_self_hosted_billing = not isinstance(self, RealmBillingSession)
         downgrade_at_end_of_cycle = plan.status == CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE
         downgrade_at_end_of_free_trial = plan.status == CustomerPlan.DOWNGRADE_AT_END_OF_FREE_TRIAL
         switch_to_annual_at_end_of_cycle = (
@@ -2272,6 +2337,27 @@ class BillingSession(ABC):
                 dt=start_of_next_billing_cycle(plan, now)
             )
 
+        has_paid_invoice_for_free_trial = False
+        free_trial_next_renewal_date_after_invoice_paid = None
+        if plan.is_free_trial() and not plan.charge_automatically:
+            last_sent_invoice = Invoice.objects.filter(plan=plan).order_by("-id").first()
+            # If the customer doesn't have any invoice, this likely means a bug and customer needs to be handled manually.
+            assert last_sent_invoice is not None
+            has_paid_invoice_for_free_trial = last_sent_invoice.status == Invoice.PAID
+
+            if has_paid_invoice_for_free_trial:
+                assert plan.next_invoice_date is not None
+                free_trial_days = get_free_trial_days(is_self_hosted_billing, plan.tier)
+                assert free_trial_days is not None
+                free_trial_next_renewal_date_after_invoice_paid = (
+                    "{dt:%B} {dt.day}, {dt.year}".format(
+                        dt=(
+                            start_of_next_billing_cycle(plan, plan.next_invoice_date)
+                            + timedelta(days=free_trial_days)
+                        )
+                    )
+                )
+
         billing_frequency = CustomerPlan.BILLING_SCHEDULES[plan.billing_schedule]
         discount_for_current_plan = plan.discount
 
@@ -2328,7 +2414,6 @@ class BillingSession(ABC):
             customer, status=CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END
         )
         legacy_remote_server_next_plan_name = self.get_legacy_remote_server_next_plan_name(customer)
-        is_self_hosted_billing = not isinstance(self, RealmBillingSession)
         context = {
             "plan_name": plan.name,
             "has_active_plan": True,
@@ -2364,6 +2449,8 @@ class BillingSession(ABC):
             "pre_discount_renewal_cents": cents_to_dollar_string(pre_discount_renewal_cents),
             "flat_discount": format_money(customer.flat_discount),
             "discounted_months_left": customer.flat_discounted_months,
+            "has_paid_invoice_for_free_trial": has_paid_invoice_for_free_trial,
+            "free_trial_next_renewal_date_after_invoice_paid": free_trial_next_renewal_date_after_invoice_paid,
         }
         return context
 
@@ -2461,6 +2548,8 @@ class BillingSession(ABC):
         fixed_price = None
         pay_by_invoice_payments_page = None
         scheduled_upgrade_invoice_amount_due = None
+        is_free_trial_invoice_expired_notice = False
+        free_trial_invoice_expired_notice_page_plan_name = None
         if customer is not None:
             fixed_price_plan_offer = get_configured_fixed_price_plan_offer(customer, tier)
             if fixed_price_plan_offer:
@@ -2485,6 +2574,17 @@ class BillingSession(ABC):
                         scheduled_upgrade_invoice_amount_due = format_money(invoice.amount_due)
                         pay_by_invoice_payments_page = f"{self.billing_base_url}/invoices"
 
+                        if (
+                            last_send_invoice.plan is not None
+                            and last_send_invoice.is_created_for_free_trial_upgrade
+                        ):
+                            # Automatic payment invoice would have been marked void already.
+                            assert not last_send_invoice.plan.charge_automatically
+                            is_free_trial_invoice_expired_notice = True
+                            free_trial_invoice_expired_notice_page_plan_name = (
+                                last_send_invoice.plan.name
+                            )
+
         percent_off = Decimal(0)
         if customer is not None:
             discount_for_plan_tier = customer.get_discount_for_plan_tier(tier)
@@ -2571,19 +2671,23 @@ class BillingSession(ABC):
                 customer, is_self_hosted_billing
             ),
             "scheduled_upgrade_invoice_amount_due": scheduled_upgrade_invoice_amount_due,
+            "is_free_trial_invoice_expired_notice": is_free_trial_invoice_expired_notice,
+            "free_trial_invoice_expired_notice_page_plan_name": free_trial_invoice_expired_notice_page_plan_name,
         }
 
         return None, context
 
     def min_licenses_for_flat_discount_to_self_hosted_basic_plan(
-        self, customer: Optional[Customer]
+        self,
+        customer: Optional[Customer],
+        is_plan_free_trial_with_invoice_payment: bool = False,
     ) -> int:
         # Since monthly and annual TIER_SELF_HOSTED_BASIC plans have same per user price we only need to do this calculation once.
         # If we decided to apply this for other tiers, then we will have to do this calculation based on billing schedule selected by the user.
         price_per_license = get_price_per_license(
             CustomerPlan.TIER_SELF_HOSTED_BASIC, CustomerPlan.BILLING_SCHEDULE_MONTHLY
         )
-        if customer is None:
+        if customer is None or is_plan_free_trial_with_invoice_payment:
             return (
                 Customer._meta.get_field("flat_discount").get_default() // price_per_license
             ) + 1
@@ -2592,14 +2696,22 @@ class BillingSession(ABC):
         # If flat discount is not applied.
         return 1
 
-    def min_licenses_for_plan(self, tier: int) -> int:
+    def min_licenses_for_plan(
+        self, tier: int, is_plan_free_trial_with_invoice_payment: bool = False
+    ) -> int:
         customer = self.get_customer()
         if customer is not None and customer.minimum_licenses:
             assert customer.default_discount is not None
             return customer.minimum_licenses
 
         if tier == CustomerPlan.TIER_SELF_HOSTED_BASIC:
-            return min(self.min_licenses_for_flat_discount_to_self_hosted_basic_plan(customer), 10)
+            return min(
+                self.min_licenses_for_flat_discount_to_self_hosted_basic_plan(
+                    customer,
+                    is_plan_free_trial_with_invoice_payment,
+                ),
+                10,
+            )
         if tier == CustomerPlan.TIER_SELF_HOSTED_BUSINESS:
             return 25
         return 1
@@ -2705,8 +2817,11 @@ class BillingSession(ABC):
                 self.downgrade_now_without_creating_additional_invoices(plan=plan)
             elif status == CustomerPlan.DOWNGRADE_AT_END_OF_FREE_TRIAL:
                 assert plan.is_free_trial()
+                # For payment by invoice, we don't allow changing plan schedule and status.
+                assert plan.charge_automatically
                 do_change_plan_status(plan, status)
             elif status == CustomerPlan.FREE_TRIAL:
+                assert plan.charge_automatically
                 if update_plan_request.schedule is not None:
                     self.do_change_schedule_after_free_trial(plan, update_plan_request.schedule)
                 else:
@@ -2769,16 +2884,47 @@ class BillingSession(ABC):
                         "Your plan is already scheduled to renew with {licenses_at_next_renewal} licenses."
                     ).format(licenses_at_next_renewal=licenses_at_next_renewal)
                 )
+            is_plan_free_trial_with_invoice_payment = (
+                plan.is_free_trial() and not plan.charge_automatically
+            )
             validate_licenses(
                 plan.charge_automatically,
                 licenses_at_next_renewal,
                 self.current_count_for_billed_licenses(),
                 plan.customer.exempt_from_license_number_check,
-                self.min_licenses_for_plan(plan.tier),
-            )
-            self.update_license_ledger_for_manual_plan(
-                plan, timezone_now(), licenses_at_next_renewal=licenses_at_next_renewal
+                self.min_licenses_for_plan(plan.tier, is_plan_free_trial_with_invoice_payment),
             )
+
+            # User is trying to change licenses while in free trial.
+            if is_plan_free_trial_with_invoice_payment:  # nocoverage
+                invoice = Invoice.objects.filter(plan=plan).order_by("-id").first()
+                assert invoice is not None
+                # Don't allow customer to reduce licenses for next billing cycle if they have paid invoice.
+                if invoice.status == Invoice.PAID:
+                    assert last_ledger_entry.licenses_at_next_renewal is not None
+                    if last_ledger_entry.licenses_at_next_renewal > licenses_at_next_renewal:
+                        raise JsonableError(
+                            _(
+                                "You’ve already purchased {licenses_at_next_renewal} licenses for the next billing period."
+                            ).format(
+                                licenses_at_next_renewal=last_ledger_entry.licenses_at_next_renewal
+                            )
+                        )
+                    else:
+                        # If customer has paid already, we will send them an invoice for additional
+                        # licenses at the end of free trial.
+                        self.update_license_ledger_for_manual_plan(
+                            plan, timezone_now(), licenses_at_next_renewal=licenses_at_next_renewal
+                        )
+                else:
+                    # Discard the old invoice and create a new one with updated licenses.
+                    self.update_free_trial_invoice_with_licenses(
+                        plan, timezone_now(), licenses_at_next_renewal
+                    )
+            else:
+                self.update_license_ledger_for_manual_plan(
+                    plan, timezone_now(), licenses_at_next_renewal=licenses_at_next_renewal
+                )
             return
 
         raise JsonableError(_("Nothing to change."))
@@ -2897,11 +3043,14 @@ class BillingSession(ABC):
                     plan_renewal_or_end_date = get_plan_renewal_or_end_date(
                         plan, ledger_entry.event_time
                     )
-                    proration_fraction = (plan_renewal_or_end_date - ledger_entry.event_time) / (
-                        billing_period_end - last_renewal
-                    )
+                    unit_amount = plan.price_per_license
+                    if not plan.is_free_trial():
+                        proration_fraction = (
+                            plan_renewal_or_end_date - ledger_entry.event_time
+                        ) / (billing_period_end - last_renewal)
+                        unit_amount = int(plan.price_per_license * proration_fraction + 0.5)
                     price_args = {
-                        "unit_amount": int(plan.price_per_license * proration_fraction + 0.5),
+                        "unit_amount": unit_amount,
                         "quantity": ledger_entry.licenses - licenses_base,
                     }
                     description = "Additional license ({} - {})".format(
@@ -3249,6 +3398,82 @@ class BillingSession(ABC):
 
         return success_message
 
+    def update_free_trial_invoice_with_licenses(
+        self,
+        plan: CustomerPlan,
+        event_time: datetime,
+        licenses: int,
+    ) -> None:  # nocoverage
+        assert (
+            self.get_billable_licenses_for_customer(plan.customer, plan.tier, licenses) <= licenses
+        )
+        last_sent_invoice = Invoice.objects.filter(plan=plan).order_by("-id").first()
+        assert last_sent_invoice is not None
+        assert last_sent_invoice.status == Invoice.SENT
+
+        assert plan.automanage_licenses is False
+        assert plan.charge_automatically is False
+        assert plan.fixed_price is None
+        assert plan.is_free_trial()
+
+        # Create a new renewal invoice with updated licenses so that this becomes the last
+        # renewal invoice for customer which will be used for any future comparisons.
+        LicenseLedger.objects.create(
+            plan=plan,
+            is_renewal=True,
+            event_time=event_time,
+            licenses=licenses,
+            licenses_at_next_renewal=licenses,
+        )
+
+        # Update the last sent invoice with the new licenses. We just need to update `quantity` in
+        # the first invoice item. So, we void the current invoice and create a new copy of it with
+        # the updated quantity.
+        stripe_invoice = stripe.Invoice.retrieve(last_sent_invoice.stripe_invoice_id)
+        assert stripe_invoice.status == "open"
+        invoice_items = stripe_invoice.lines.data
+        # Stripe does something weird and puts the discount item first, so we need to reverse the order here.
+        invoice_items.reverse()
+        for invoice_item in invoice_items:
+            price_args = {}
+            # If amount is positive, this must be non-discount item we need to update.
+            if invoice_item.amount > 0:
+                assert invoice_item.price is not None
+                price_args = {
+                    "quantity": licenses,
+                    "unit_amount": invoice_item.price.unit_amount,
+                }
+            else:
+                price_args = {
+                    "amount": invoice_item.amount,
+                }
+            stripe.InvoiceItem.create(
+                currency=invoice_item.currency,
+                customer=stripe_invoice.customer,
+                description=invoice_item.description,
+                period=invoice_item.period,
+                **price_args,
+            )
+
+        assert plan.next_invoice_date is not None
+        # Difference between end of free trial and event time
+        days_until_due = (plan.next_invoice_date - event_time).days
+
+        new_stripe_invoice = stripe.Invoice.create(
+            auto_advance=False,
+            collection_method="send_invoice",
+            customer=stripe_invoice.customer,
+            days_until_due=days_until_due,
+            statement_descriptor=stripe_invoice.statement_descriptor,
+            metadata=stripe_invoice.metadata,
+        )
+        new_stripe_invoice = stripe.Invoice.finalize_invoice(new_stripe_invoice)
+        last_sent_invoice.stripe_invoice_id = str(new_stripe_invoice.id)
+        last_sent_invoice.save(update_fields=["stripe_invoice_id"])
+
+        assert stripe_invoice.id is not None
+        stripe.Invoice.void_invoice(stripe_invoice.id)
+
     def update_license_ledger_for_manual_plan(
         self,
         plan: CustomerPlan,
@@ -4962,6 +5187,7 @@ def invoice_plans_as_needed(event_time: Optional[datetime] = None) -> None:
                 plan.save(update_fields=["reminder_to_review_plan_email_sent"])
 
             free_plan_with_no_next_plan = not plan.is_paid() and plan.status == CustomerPlan.ACTIVE
+            free_trial_pay_by_invoice_plan = plan.is_free_trial() and not plan.charge_automatically
             last_audit_log_update = remote_server.last_audit_log_update
             if not free_plan_with_no_next_plan and (
                 last_audit_log_update is None or plan.next_invoice_date > last_audit_log_update
@@ -4987,7 +5213,11 @@ def invoice_plans_as_needed(event_time: Optional[datetime] = None) -> None:
                     )
                     plan.invoice_overdue_email_sent = True
                     plan.save(update_fields=["invoice_overdue_email_sent"])
-                continue
+
+                # We still process free trial plans so that we can directly downgrade them.
+                # Above emails can serve as a reminder to followup for additional feedback.
+                if not free_trial_pay_by_invoice_plan:
+                    continue
 
         while (
             plan.next_invoice_date is not None  # type: ignore[redundant-expr] # plan.next_invoice_date can be None after calling invoice_plan.

corporate/lib/stripe_event_handler.py

@@ -12,7 +12,14 @@ from corporate.lib.stripe import (
     RemoteServerBillingSession,
     get_configured_fixed_price_plan_offer,
 )
-from corporate.models import Customer, CustomerPlan, Event, Invoice, Session
+from corporate.models import (
+    Customer,
+    CustomerPlan,
+    Event,
+    Invoice,
+    Session,
+    get_current_plan_by_customer,
+)
 from zerver.lib.send_email import FromAddress, send_email
 from zerver.models.users import get_active_user_profile_by_id_in_realm
 
@@ -172,14 +179,27 @@ def handle_invoice_paid_event(stripe_invoice: stripe.Invoice, invoice: Invoice)
                 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=charge_automatically,
-                free_trial=False,
-                remote_server_legacy_plan=remote_server_legacy_plan,
-                stripe_invoice_paid=True,
-            )
+            return
+        elif metadata.get("on_free_trial") and invoice.is_created_for_free_trial_upgrade:
+            free_trial_plan = invoice.plan
+            assert free_trial_plan is not None
+            if free_trial_plan.is_free_trial():
+                # We don't need to do anything here. When the free trial ends we will
+                # check if user has paid the invoice, if not we downgrade the user.
+                return
+
+            # If customer paid after end of free trial, we just upgrade via default method below.
+            assert free_trial_plan.status == CustomerPlan.ENDED
+            # Also check if customer is not on any other active plan.
+            assert get_current_plan_by_customer(customer) is None
+
+        billing_session.process_initial_upgrade(
+            plan_tier,
+            int(metadata["licenses"]),
+            metadata["license_management"] == "automatic",
+            billing_schedule=billing_schedule,
+            charge_automatically=charge_automatically,
+            free_trial=False,
+            remote_server_legacy_plan=remote_server_legacy_plan,
+            stripe_invoice_paid=True,
+        )

corporate/migrations/0042_invoice_is_created_for_free_trial_upgrade_and_more.py

@@ -0,0 +1,28 @@
+# Generated by Django 4.2.11 on 2024-04-10 03:17
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("corporate", "0041_fix_plans_on_free_trial_with_changes_in_schedule"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="invoice",
+            name="is_created_for_free_trial_upgrade",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name="invoice",
+            name="plan",
+            field=models.ForeignKey(
+                default=None,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                to="corporate.customerplan",
+            ),
+        ),
+    ]

corporate/models.py

@@ -5,7 +5,7 @@ from typing import Any, Dict, Optional, Union
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.contrib.contenttypes.models import ContentType
 from django.db import models
-from django.db.models import CASCADE, Q
+from django.db.models import CASCADE, SET_NULL, Q
 from typing_extensions import override
 
 from zerver.models import Realm, UserProfile
@@ -218,6 +218,8 @@ class PaymentIntent(models.Model):  # nocoverage
 class Invoice(models.Model):
     customer = models.ForeignKey(Customer, on_delete=CASCADE)
     stripe_invoice_id = models.CharField(max_length=255, unique=True)
+    plan = models.ForeignKey("CustomerPlan", null=True, default=None, on_delete=SET_NULL)
+    is_created_for_free_trial_upgrade = models.BooleanField(default=False)
 
     SENT = 1
     PAID = 2