"""add user profile schema PRD-006 — User Profile Management Creates: user_password_history, password_policies, user_notification_preferences, email_change_requests. Alters: users (9 new columns), auth_sessions (11 new geo/device columns). Inserts: global default password_policies seed row (merchant_id = NULL). Revision ID: 20260316001a Revises: 20260309001a Create Date: 2026-03-16 00:00:00.000000 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy import inspect as sa_inspect # --------------------------------------------------------------------------- # Revision metadata # --------------------------------------------------------------------------- revision: str = '20260316001a' down_revision: Union[str, Sequence[str], None] = '20260309001a' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None # --------------------------------------------------------------------------- # upgrade # --------------------------------------------------------------------------- def upgrade() -> None: conn = op.get_bind() inspector = sa_inspect(conn) # ------------------------------------------------------------------ # 1. ALTER TABLE users — add profile, address, security columns # ------------------------------------------------------------------ users_cols = {c['name'] for c in inspector.get_columns('users')} users_fks = {fk['name'] for fk in inspector.get_foreign_keys('users')} if 'avatar_file_id' not in users_cols: op.add_column( 'users', sa.Column( 'avatar_file_id', sa.Integer(), sa.ForeignKey('files.id', name='fk_users_avatar_file_id', ondelete='SET NULL'), nullable=True, comment='FK to files table; preferred source for avatar image', ), ) elif 'fk_users_avatar_file_id' not in users_fks: # Column exists but FK may be missing op.create_foreign_key( 'fk_users_avatar_file_id', 'users', 'files', ['avatar_file_id'], ['id'], ) if 'address_line_1' not in users_cols: op.add_column('users', sa.Column('address_line_1', sa.String(255), nullable=True, comment='Primary street address')) if 'address_line_2' not in users_cols: op.add_column('users', sa.Column('address_line_2', sa.String(255), nullable=True, comment='Apartment, suite, unit, etc.')) if 'city' not in users_cols: op.add_column('users', sa.Column('city', sa.String(100), nullable=True)) if 'state' not in users_cols: op.add_column('users', sa.Column('state', sa.String(100), nullable=True)) if 'zip_code' not in users_cols: op.add_column('users', sa.Column('zip_code', sa.String(20), nullable=True)) if 'country' not in users_cols: op.add_column( 'users', sa.Column('country', sa.String(100), nullable=True, server_default='US', comment='ISO country code; defaults to US'), ) if 'twofa_enabled' not in users_cols: op.add_column( 'users', sa.Column('twofa_enabled', sa.Boolean(), nullable=False, server_default='false', comment='Whether TOTP/2FA is currently enabled for this user'), ) if 'last_password_changed_at' not in users_cols: op.add_column( 'users', sa.Column('last_password_changed_at', sa.DateTime(timezone=True), nullable=True, comment='Timestamp of the most recent password change; NULL until first explicit change'), ) # ------------------------------------------------------------------ # 2. ALTER TABLE auth_sessions — add geo + device enrichment columns # ------------------------------------------------------------------ sessions_cols = {c['name'] for c in inspector.get_columns('auth_sessions')} if 'geo_country' not in sessions_cols: op.add_column('auth_sessions', sa.Column('geo_country', sa.String(100), nullable=True)) if 'geo_region' not in sessions_cols: op.add_column('auth_sessions', sa.Column('geo_region', sa.String(100), nullable=True)) if 'geo_city' not in sessions_cols: op.add_column('auth_sessions', sa.Column('geo_city', sa.String(100), nullable=True)) if 'geo_latitude' not in sessions_cols: op.add_column('auth_sessions', sa.Column('geo_latitude', sa.Float(), nullable=True, comment='DOUBLE PRECISION; stored as Float in SA')) if 'geo_longitude' not in sessions_cols: op.add_column('auth_sessions', sa.Column('geo_longitude', sa.Float(), nullable=True)) if 'device_browser' not in sessions_cols: op.add_column('auth_sessions', sa.Column('device_browser', sa.String(100), nullable=True)) if 'device_browser_version' not in sessions_cols: op.add_column('auth_sessions', sa.Column('device_browser_version', sa.String(50), nullable=True)) if 'device_os' not in sessions_cols: op.add_column('auth_sessions', sa.Column('device_os', sa.String(100), nullable=True)) if 'device_os_version' not in sessions_cols: op.add_column('auth_sessions', sa.Column('device_os_version', sa.String(50), nullable=True)) if 'device_type' not in sessions_cols: op.add_column('auth_sessions', sa.Column('device_type', sa.String(50), nullable=True, comment='mobile | tablet | desktop | bot')) if 'device_brand' not in sessions_cols: op.add_column('auth_sessions', sa.Column('device_brand', sa.String(100), nullable=True, comment='e.g. Apple, Samsung, Dell')) # ------------------------------------------------------------------ # 3. CREATE TABLE user_password_history # ------------------------------------------------------------------ existing_tables = sa_inspect(conn).get_table_names() if 'user_password_history' not in existing_tables: op.create_table( 'user_password_history', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False, comment='Surrogate primary key'), sa.Column('user_id', sa.Integer(), nullable=False, comment='FK → users.id; CASCADE DELETE'), sa.Column('hashed_password', sa.String(255), nullable=False, comment='bcrypt hash — never plaintext'), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), sa.PrimaryKeyConstraint('id', name='pk_user_password_history'), sa.ForeignKeyConstraint( ['user_id'], ['users.id'], name='fk_user_password_history_user_id', ondelete='CASCADE', ), comment='Stores hashed copies of previous passwords to enforce non-reuse policy', ) op.create_index( 'idx_user_password_history_user_created', 'user_password_history', ['user_id', sa.text('created_at DESC')], unique=False, ) # ------------------------------------------------------------------ # 4. CREATE TABLE password_policies # ------------------------------------------------------------------ if 'password_policies' not in existing_tables: op.create_table( 'password_policies', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False, comment='Surrogate primary key'), sa.Column('merchant_id', sa.Integer(), nullable=True, comment='NULL = global default policy; UNIQUE NULLS DISTINCT enforced below'), sa.Column('min_length', sa.Integer(), nullable=False, server_default='8'), sa.Column('require_uppercase', sa.Boolean(), nullable=False, server_default='true'), sa.Column('require_lowercase', sa.Boolean(), nullable=False, server_default='true'), sa.Column('require_digit', sa.Boolean(), nullable=False, server_default='true'), sa.Column('require_special', sa.Boolean(), nullable=False, server_default='false'), sa.Column('max_age_days', sa.Integer(), nullable=True, comment='NULL = password never expires'), sa.Column('grace_period_days', sa.Integer(), nullable=False, server_default='7', comment='Days the user may still log in after password expiry before forced reset'), sa.Column('history_count', sa.Integer(), nullable=False, server_default='5', comment='Number of previous password hashes retained to prevent reuse'), sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True, comment='Soft delete timestamp'), sa.PrimaryKeyConstraint('id', name='pk_password_policies'), sa.ForeignKeyConstraint( ['merchant_id'], ['merchants.id'], name='fk_password_policies_merchant_id', ondelete='CASCADE', ), comment=( 'Per-merchant password complexity and rotation policy. ' 'The row with merchant_id IS NULL is the global default applied to any ' 'merchant that has not defined a custom policy.' ), ) op.create_index( 'uq_password_policies_merchant_id', 'password_policies', ['merchant_id'], unique=True, postgresql_where=sa.text('merchant_id IS NOT NULL AND deleted_at IS NULL'), ) op.create_index( 'uq_password_policies_global_default', 'password_policies', [sa.text('(merchant_id IS NULL)')], unique=True, postgresql_where=sa.text('merchant_id IS NULL AND deleted_at IS NULL'), ) # ------------------------------------------------------------------ # 5. CREATE TABLE user_notification_preferences # ------------------------------------------------------------------ if 'user_notification_preferences' not in existing_tables: op.create_table( 'user_notification_preferences', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('user_id', sa.Integer(), nullable=False, comment='FK → users.id; one-to-one via UNIQUE constraint'), sa.Column('notify_on_new_login', sa.Boolean(), nullable=False, server_default='true'), sa.Column('notify_on_password_change', sa.Boolean(), nullable=False, server_default='true'), sa.Column('notify_on_new_device', sa.Boolean(), nullable=False, server_default='true'), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), sa.PrimaryKeyConstraint('id', name='pk_user_notification_preferences'), sa.ForeignKeyConstraint( ['user_id'], ['users.id'], name='fk_user_notification_prefs_user_id', ondelete='CASCADE', ), sa.UniqueConstraint('user_id', name='uq_user_notification_preferences_user_id'), comment='One-to-one with users; stores per-user notification opt-in flags', ) # ------------------------------------------------------------------ # 6. CREATE TABLE email_change_requests # ------------------------------------------------------------------ if 'email_change_requests' not in existing_tables: op.create_table( 'email_change_requests', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('user_id', sa.Integer(), nullable=False, comment='FK → users.id; multiple in-flight requests per user are allowed'), sa.Column('new_email', sa.String(255), nullable=False, comment='The email address the user wants to change to'), sa.Column('token_hash', sa.String(64), nullable=False, comment='SHA-256 hex digest of the one-time confirmation token'), sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False, comment='Token expiry; typically now() + 24h'), sa.Column('confirmed_at', sa.DateTime(timezone=True), nullable=True, comment='Populated when the user clicks the confirmation link; NULL = pending'), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), sa.PrimaryKeyConstraint('id', name='pk_email_change_requests'), sa.ForeignKeyConstraint( ['user_id'], ['users.id'], name='fk_email_change_requests_user_id', ondelete='CASCADE', ), sa.UniqueConstraint('token_hash', name='uq_email_change_requests_token_hash'), comment=( 'Tracks pending email-change confirmation flows. ' 'A token is generated server-side, hashed (SHA-256) before storage, ' 'and emailed in plaintext to new_email. Confirmed by matching hash.' ), ) op.create_index( 'idx_email_change_requests_token_hash', 'email_change_requests', ['token_hash'], unique=False, ) op.create_index( 'idx_email_change_requests_expires_pending', 'email_change_requests', ['expires_at'], unique=False, postgresql_where=sa.text('confirmed_at IS NULL'), ) # ------------------------------------------------------------------ # 7. Seed: global default password policy (merchant_id = NULL) # ------------------------------------------------------------------ op.execute( """ INSERT INTO password_policies ( merchant_id, min_length, require_uppercase, require_lowercase, require_digit, require_special, max_age_days, grace_period_days, history_count, is_active, created_at, updated_at ) VALUES ( NULL, 8, TRUE, TRUE, TRUE, FALSE, NULL, 7, 5, TRUE, now(), now() ) ON CONFLICT DO NOTHING """ ) # --------------------------------------------------------------------------- # downgrade # --------------------------------------------------------------------------- def downgrade() -> None: # 6. Drop email_change_requests op.drop_index('idx_email_change_requests_expires_pending', table_name='email_change_requests') op.drop_index('idx_email_change_requests_token_hash', table_name='email_change_requests') op.drop_table('email_change_requests') # 5. Drop user_notification_preferences op.drop_table('user_notification_preferences') # 4. Drop password_policies op.drop_index('uq_password_policies_global_default', table_name='password_policies') op.drop_index('uq_password_policies_merchant_id', table_name='password_policies') op.drop_table('password_policies') # 3. Drop user_password_history op.drop_index('idx_user_password_history_user_created', table_name='user_password_history') op.drop_table('user_password_history') # 2. Remove auth_sessions geo/device columns (reverse order) op.drop_column('auth_sessions', 'device_brand') op.drop_column('auth_sessions', 'device_type') op.drop_column('auth_sessions', 'device_os_version') op.drop_column('auth_sessions', 'device_os') op.drop_column('auth_sessions', 'device_browser_version') op.drop_column('auth_sessions', 'device_browser') op.drop_column('auth_sessions', 'geo_longitude') op.drop_column('auth_sessions', 'geo_latitude') op.drop_column('auth_sessions', 'geo_city') op.drop_column('auth_sessions', 'geo_region') op.drop_column('auth_sessions', 'geo_country') # 1. Remove users profile/address/security columns (reverse order) op.drop_column('users', 'last_password_changed_at') op.drop_column('users', 'twofa_enabled') op.drop_column('users', 'country') op.drop_column('users', 'zip_code') op.drop_column('users', 'state') op.drop_column('users', 'city') op.drop_column('users', 'address_line_2') op.drop_column('users', 'address_line_1') op.drop_column('users', 'avatar_file_id')