"""Phase 1 billing hardening: plan_price_versions, workspace_billing_discounts, grandfathering. Revision ID: 007 Revises: 006 Create Date: 2025-03-18 Invariants: - Prices are append-only (plan_price_versions) - Plans reference current sellable version; historical pricing reconstructible - Discounts are first-class (workspace_billing_discounts) - Grandfathered subscriptions explicitly tracked (subscriptions.is_grandfathered) """ from typing import Sequence, Union import sqlalchemy as sa from alembic import op from sqlalchemy.dialects import postgresql revision: str = "007" down_revision: Union[str, None] = "006" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: # ---- plan_price_versions (append-only price history) ---- op.create_table( "plan_price_versions", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")), sa.Column("subscription_plan_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("subscription_plans.id", ondelete="CASCADE"), nullable=False), sa.Column("interval", sa.String(20), nullable=False), # month, year, metered sa.Column("stripe_price_id", sa.String(100), nullable=False, unique=True), sa.Column("price_minor", sa.BigInteger(), nullable=False), sa.Column("currency", sa.String(10), nullable=False, server_default="usd"), sa.Column("stripe_product_id", sa.String(100)), sa.Column("is_current", sa.Boolean(), nullable=False, server_default="true"), sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()), ) op.create_index("idx_plan_price_versions_plan_interval", "plan_price_versions", ["subscription_plan_id", "interval"]) op.create_index("idx_plan_price_versions_stripe_price_id", "plan_price_versions", ["stripe_price_id"], unique=True) # Backfill from existing subscription_plans (one current version per plan+interval) op.execute(""" INSERT INTO plan_price_versions (subscription_plan_id, interval, stripe_price_id, price_minor, currency, stripe_product_id, is_current) SELECT id, 'month', stripe_monthly_price_id, price_monthly, currency, stripe_product_id, true FROM subscription_plans WHERE stripe_monthly_price_id IS NOT NULL AND stripe_monthly_price_id != '' """) op.execute(""" INSERT INTO plan_price_versions (subscription_plan_id, interval, stripe_price_id, price_minor, currency, stripe_product_id, is_current) SELECT id, 'year', stripe_yearly_price_id, COALESCE(price_yearly, 0), currency, stripe_product_id, true FROM subscription_plans WHERE stripe_yearly_price_id IS NOT NULL AND stripe_yearly_price_id != '' """) # Ensure only one is_current per (plan, interval) - take latest by created_at if dupes op.execute(""" WITH ranked AS ( SELECT id, subscription_plan_id, interval, ROW_NUMBER() OVER (PARTITION BY subscription_plan_id, interval ORDER BY created_at DESC) AS rn FROM plan_price_versions WHERE is_current = true ) UPDATE plan_price_versions ppv SET is_current = false FROM ranked r WHERE ppv.id = r.id AND r.rn > 1 """) # ---- subscription_plans: reference current version (nullable for backward compat) ---- op.add_column( "subscription_plans", sa.Column("current_monthly_price_version_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("plan_price_versions.id", ondelete="SET NULL")), ) op.add_column( "subscription_plans", sa.Column("current_yearly_price_version_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("plan_price_versions.id", ondelete="SET NULL")), ) op.add_column( "subscription_plans", sa.Column("current_metered_price_version_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("plan_price_versions.id", ondelete="SET NULL")), ) # Backfill: link to the current versions we just created op.execute(""" UPDATE subscription_plans sp SET current_monthly_price_version_id = ppv.id FROM plan_price_versions ppv WHERE ppv.subscription_plan_id = sp.id AND ppv.interval = 'month' AND ppv.is_current = true """) op.execute(""" UPDATE subscription_plans sp SET current_yearly_price_version_id = ppv.id FROM plan_price_versions ppv WHERE ppv.subscription_plan_id = sp.id AND ppv.interval = 'year' AND ppv.is_current = true """) # ---- subscriptions: grandfathering ---- op.add_column("subscriptions", sa.Column("is_grandfathered", sa.Boolean(), server_default="false")) op.add_column("subscriptions", sa.Column("grandfather_metadata", postgresql.JSONB())) # ---- workspace_billing_discounts (first-class discount records) ---- op.create_table( "workspace_billing_discounts", sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")), sa.Column("workspace_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False), sa.Column("source", sa.String(50), nullable=False), # manual, campaign, partner, user_segment sa.Column("scope", sa.String(50), nullable=False), # subscription, credit_purchase, both sa.Column("duration_type", sa.String(30), nullable=False), # one_time, repeating_n_months, lifetime sa.Column("duration_months", sa.Integer()), sa.Column("starts_at", sa.TIMESTAMP(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("ends_at", sa.TIMESTAMP(timezone=True)), sa.Column("stripe_coupon_id", sa.String(100)), # execution detail; DB is source of truth sa.Column("stripe_promotion_code_id", sa.String(100)), sa.Column("discount_percent", sa.Integer()), sa.Column("discount_amount_minor", sa.BigInteger()), sa.Column("currency", sa.String(10), server_default="usd"), sa.Column("reason", sa.Text()), sa.Column("stacking_rule", sa.String(30), server_default="replace"), # replace, stack sa.Column("metadata_json", postgresql.JSONB()), sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()), sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()), sa.Column("created_by", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL")), ) op.create_index("idx_workspace_billing_discounts_workspace", "workspace_billing_discounts", ["workspace_id"]) op.create_index("idx_workspace_billing_discounts_active", "workspace_billing_discounts", ["workspace_id", "ends_at"]) def downgrade() -> None: op.drop_index("idx_workspace_billing_discounts_active", table_name="workspace_billing_discounts") op.drop_index("idx_workspace_billing_discounts_workspace", table_name="workspace_billing_discounts") op.drop_table("workspace_billing_discounts") op.drop_column("subscriptions", "grandfather_metadata") op.drop_column("subscriptions", "is_grandfathered") op.drop_column("subscription_plans", "current_metered_price_version_id") op.drop_column("subscription_plans", "current_yearly_price_version_id") op.drop_column("subscription_plans", "current_monthly_price_version_id") op.drop_index("idx_plan_price_versions_stripe_price_id", table_name="plan_price_versions") op.drop_index("idx_plan_price_versions_plan_interval", table_name="plan_price_versions") op.drop_table("plan_price_versions")