leocrm/alembic/versions/0015_rls_policies.py

"""Enable Row Level Security with tenant isolation policies on core tables.

Idempotent: drops existing policies before creating them.

Revision ID: 0015_rls_policies
Revises: 0014_currency_unique_fix
Create Date: 2026-07-04
"""

from __future__ import annotations

from typing import Sequence, Union

from alembic import op

revision: str = "0015_rls_policies"
down_revision: Union[str, None] = "0014_currency_unique_fix"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

# Tables with tenant_id column that get RLS
RLS_TABLES = [
    "companies",
    "contacts",
    "company_contacts",
    "currencies",
    "tax_rates",
    "sequences",
    "system_settings",
    "attachments",
    "addresses",
    "users",
    "roles",
    "sessions",
    "audit_log",
    "deletion_log",
    "notifications",
    "ai_conversations",
    "ai_messages",
    "workflows",
    "workflow_instances",
    "workflow_step_history",
    "password_reset_tokens",
]


def upgrade() -> None:
    for table_name in RLS_TABLES:
        # Enable RLS on the table (idempotent — ENABLE is safe to repeat)
        op.execute(f"ALTER TABLE {table_name} ENABLE ROW LEVEL SECURITY")

        # Drop existing policy if it exists (idempotent — prevents DuplicateObjectError on restart)
        op.execute(f"DROP POLICY IF EXISTS tenant_isolation ON {table_name}")

        # Create tenant isolation policy
        # USING clause: tenant_id must match the session variable set by the app
        op.execute(
            f"CREATE POLICY tenant_isolation ON {table_name} "
            f"USING (tenant_id = current_setting('app.current_tenant_id')::uuid)"
        )


def downgrade() -> None:
    for table_name in reversed(RLS_TABLES):
        op.execute(f"DROP POLICY IF EXISTS tenant_isolation ON {table_name}")
        op.execute(f"ALTER TABLE {table_name} DISABLE ROW LEVEL SECURITY")