"""add subscriptions and subscription_activities tables Revision ID: a9b0c1d2e3f4 Revises: 641a39544141 Create Date: 2026-03-17 14:00:00.000000 Changes: 1. Create ``subscriptions`` table — core recurring-billing contract record 2. Create ``subscription_activities`` table — append-only audit trail 3. Add ``subscriptions.subscription_id`` indexed FK to ``invoices`` 4. Add ``is_reauth`` boolean column to ``payment_requests_links`` """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision: str = 'a9b0c1d2e3f4' down_revision: Union[str, Sequence[str], None] = '641a39544141' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: # ── 1. subscriptions ────────────────────────────────────────────────────── # Core record for a recurring billing contract. One per recurring # PaymentRequest (unique FK enforces the 1-to-1 relationship). # # Status is stored as an integer using the SubscriptionStatus enum: # 100 INITIALIZING | 200 ACTIVE | 201 PAST_DUE | 202 PAUSED # 300 CANCELLED | 301 COMPLETED | 400 DUNNING_EXHAUSTED # # Dunning columns track automated retry state; the Celery task writes them. op.create_table( 'subscriptions', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), # Public identifiers sa.Column('subscription_id', sa.String(length=64), nullable=False), sa.Column('subscription_literal', sa.String(length=20), nullable=False), sa.Column('name', sa.String(length=255), nullable=True), # Status — integer enum (SubscriptionStatus) sa.Column( 'status', sa.Integer(), nullable=False, server_default=sa.text('100'), # INITIALIZING ), # Multi-tenancy / relations sa.Column('merchant_id', sa.Integer(), nullable=False), sa.Column('customer_id', sa.Integer(), nullable=True), sa.Column('payment_request_id', sa.Integer(), nullable=False), # Running totals — maintained by invoice listener sa.Column('total_billed', sa.Float(), nullable=False, server_default=sa.text('0.0')), sa.Column('total_paid', sa.Float(), nullable=False, server_default=sa.text('0.0')), sa.Column('invoices_generated', sa.Integer(), nullable=False, server_default=sa.text('0')), sa.Column('invoices_paid', sa.Integer(), nullable=False, server_default=sa.text('0')), # Scheduling sa.Column('next_billing_date', sa.DateTime(timezone=True), nullable=True), sa.Column('past_due_since', sa.DateTime(timezone=True), nullable=True), # Dunning state sa.Column('dunning_retry_count', sa.Integer(), nullable=False, server_default=sa.text('0')), sa.Column('dunning_last_retry_at', sa.DateTime(timezone=True), nullable=True), sa.Column('dunning_next_retry_at', sa.DateTime(timezone=True), nullable=True), sa.Column('dunning_exhausted_at', sa.DateTime(timezone=True), nullable=True), # Timestamps sa.Column( 'created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True), sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True), # Foreign key constraints sa.ForeignKeyConstraint( ['merchant_id'], ['merchants.id'], name='fk_subscriptions_merchant_id', ), sa.ForeignKeyConstraint( ['customer_id'], ['customers.id'], name='fk_subscriptions_customer_id', ), sa.ForeignKeyConstraint( ['payment_request_id'], ['payment_requests.id'], name='fk_subscriptions_payment_request_id', ), # Primary key and unique constraints sa.PrimaryKeyConstraint('id'), sa.UniqueConstraint('subscription_id', name='uq_subscriptions_subscription_id'), sa.UniqueConstraint('subscription_literal', name='uq_subscriptions_subscription_literal'), # One subscription per payment_request (1-to-1) sa.UniqueConstraint('payment_request_id', name='uq_subscriptions_payment_request_id'), ) # Indexes on subscriptions op.create_index( op.f('ix_subscriptions_id'), 'subscriptions', ['id'], unique=False, ) op.create_index( 'ix_subscriptions_subscription_id', 'subscriptions', ['subscription_id'], unique=True, ) # Scheduler query: SELECT * FROM subscriptions WHERE status = 200 AND next_billing_date <= NOW() op.create_index( 'ix_subscriptions_status', 'subscriptions', ['status'], unique=False, ) # Multi-tenancy query: SELECT * FROM subscriptions WHERE merchant_id = ? AND deleted_at IS NULL op.create_index( 'ix_subscriptions_merchant_id', 'subscriptions', ['merchant_id'], unique=False, ) # Customer portal query: SELECT * FROM subscriptions WHERE customer_id = ? op.create_index( 'ix_subscriptions_customer_id', 'subscriptions', ['customer_id'], unique=False, ) # Compound index to efficiently find active subscriptions due for billing # by merchant (scheduler task scoped to a merchant batch). op.create_index( 'ix_subscriptions_merchant_status_next_billing', 'subscriptions', ['merchant_id', 'status', 'next_billing_date'], unique=False, ) # Dunning scheduler: find PAST_DUE subscriptions with a retry due op.create_index( 'ix_subscriptions_status_dunning_next_retry', 'subscriptions', ['status', 'dunning_next_retry_at'], unique=False, ) # ── 2. subscription_activities ──────────────────────────────────────────── # Append-only audit trail. Follows the same structure as invoice_activities. # No soft-delete: rows are never deleted. op.create_table( 'subscription_activities', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('subscription_id', sa.Integer(), nullable=False), # Event metadata sa.Column('activity_type', sa.String(length=64), nullable=False), sa.Column('description', sa.Text(), nullable=True), sa.Column('actor_type', sa.String(length=20), nullable=True), sa.Column('actor_id', sa.Integer(), nullable=True), sa.Column('metadata', sa.JSON(), nullable=True), # Timestamp (server-set, never updated — this is an event log) sa.Column( 'created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.ForeignKeyConstraint( ['subscription_id'], ['subscriptions.id'], name='fk_subscription_activities_subscription_id', ), sa.PrimaryKeyConstraint('id'), ) op.create_index( op.f('ix_subscription_activities_id'), 'subscription_activities', ['id'], unique=False, ) # Fast lookup of all activities for a given subscription (detail view) op.create_index( 'ix_subscription_activities_subscription_id', 'subscription_activities', ['subscription_id'], unique=False, ) # Filter/sort activities by type (e.g. show only failed charges) op.create_index( 'ix_subscription_activities_activity_type', 'subscription_activities', ['activity_type'], unique=False, ) # Chronological ordering index — created_at is queried in every list call op.create_index( 'ix_subscription_activities_created_at', 'subscription_activities', ['created_at'], unique=False, ) # ── 3. invoices: add subscription_id FK ─────────────────────────────────── # NULL for one-time invoices; non-NULL for invoices generated by the # recurring billing Celery task. op.add_column( 'invoices', sa.Column('subscription_id', sa.Integer(), nullable=True), ) op.create_foreign_key( 'fk_invoices_subscription_id', 'invoices', 'subscriptions', ['subscription_id'], ['id'], ) op.create_index( 'ix_invoices_subscription_id', 'invoices', ['subscription_id'], unique=False, ) # ── 4. payment_requests_links: add is_reauth flag ───────────────────────── # TRUE when this HPP link was issued during the dunning flow to allow the # customer to update their payment method before retrying the charge. # Defaults to FALSE for all existing rows (safe, backward-compatible). op.add_column( 'payment_requests_links', sa.Column( 'is_reauth', sa.Boolean(), nullable=False, server_default=sa.text('false'), ), ) def downgrade() -> None: # ── 4. Remove is_reauth from payment_requests_links ─────────────────────── op.drop_column('payment_requests_links', 'is_reauth') # ── 3. Remove subscription_id from invoices ─────────────────────────────── op.drop_index('ix_invoices_subscription_id', table_name='invoices') op.drop_constraint('fk_invoices_subscription_id', 'invoices', type_='foreignkey') op.drop_column('invoices', 'subscription_id') # ── 2. Drop subscription_activities ────────────────────────────────────── op.drop_index('ix_subscription_activities_created_at', table_name='subscription_activities') op.drop_index('ix_subscription_activities_activity_type', table_name='subscription_activities') op.drop_index('ix_subscription_activities_subscription_id', table_name='subscription_activities') op.drop_index(op.f('ix_subscription_activities_id'), table_name='subscription_activities') op.drop_table('subscription_activities') # ── 1. Drop subscriptions ───────────────────────────────────────────────── op.drop_index('ix_subscriptions_status_dunning_next_retry', table_name='subscriptions') op.drop_index('ix_subscriptions_merchant_status_next_billing', table_name='subscriptions') op.drop_index('ix_subscriptions_customer_id', table_name='subscriptions') op.drop_index('ix_subscriptions_merchant_id', table_name='subscriptions') op.drop_index('ix_subscriptions_status', table_name='subscriptions') op.drop_index('ix_subscriptions_subscription_id', table_name='subscriptions') op.drop_index(op.f('ix_subscriptions_id'), table_name='subscriptions') op.drop_table('subscriptions')