Pricing Rehearsal Report

Plan:

All invoices matched the plan. Rehearsed on Stripe Test Clocks in test mode. Applying this to live customers requires human approval.

Revenue impact over the rehearsal window

Invoice timeline: expected vs Stripe actual

monthly_pro

annual_pro

monthly_pro_coupon

Verified migration script

"""Pro $49 -> $59 migration for existing subscriptions.

New customers get the new price immediately at signup (handled at
checkout, outside this script's scope). Existing subscriptions keep their
current price for every billing cycle that starts within
GRANDFATHER_MONTHS of this migration running, then move to the new price
indefinitely. Any active discount (e.g. a forever coupon) is carried over
to the new price so it isn't dropped.

This is the script that would run against the live book. Here it runs
against the fixtures created by fixtures.py.

Run once, after fixtures.py and before advance.py.
"""
import json

from common import attr, grandfather_iterations, init_stripe, obj_id

stripe = init_stripe()


def migrate(sub_id: str, new_price_by_old: dict, interval_months: int) -> str:
    # expand=["discounts"] turns sub.discounts from bare "di_..." id strings
    # into Discount objects -- the coupon id lives at discount.source.coupon.
    sub = stripe.Subscription.retrieve(sub_id, expand=["discounts"])
    item = attr(attr(sub, "items"), "data")[0]
    old_price_id = obj_id(attr(item, "price"))
    new_price_id = new_price_by_old[old_price_id]
    # Anchor phase 1 to the sub's current cycle; a bare phases[] update needs
    # an explicit start_date on at least one phase.
    current_period_start = attr(item, "current_period_start", attr(sub, "current_period_start"))

    discounts = [
        {"coupon": c}
        for c in (obj_id(attr(attr(d, "source"), "coupon")) for d in (attr(sub, "discounts") or []))
        if c
    ]
    # "iterations" is deprecated in favor of a calendar "duration"; express the
    # grandfather window in months regardless of the sub's own billing interval.
    grandfather_months = grandfather_iterations(interval_months) * interval_months

    # Idempotent: a retry after a partial failure reuses the schedule already
    # attached to this subscription instead of erroring on a second create.
    schedule_id = obj_id(attr(sub, "schedule"))
    if not schedule_id:
        schedule_id = stripe.SubscriptionSchedule.create(from_subscription=sub_id).id

    stripe.SubscriptionSchedule.modify(
        schedule_id,
        end_behavior="release",
        phases=[
            {
                "items": [{"price": old_price_id, "discounts": discounts}],
                "start_date": current_period_start,
                "duration": {"interval": "month", "interval_count": grandfather_months},
            },
            {"items": [{"price": new_price_id, "discounts": discounts}]},
        ],
    )
    return schedule_id


def main():
    catalog = json.load(open("catalog.json"))
    fixtures = json.load(open("fixtures.json"))
    new_price_by_old = {
        catalog["pro_monthly_old"]: catalog["pro_monthly_new"],
        catalog["pro_annual_old"]: catalog["pro_annual_new"],
    }

    for fx in fixtures:
        schedule_id = migrate(fx["subscription_id"], new_price_by_old, fx["interval_months"])
        print(f"{fx['type']:20} schedule={schedule_id}")


if __name__ == "__main__":
    main()