leocrm/alembic/versions/0144_personal_dashboards.py

"""Personal dashboards table (Phase M2) + RLS policy-role convergence.

Revision ID: 0144
Revises: 0143
Create Date: 2026-08-30

Part 1 — dashboards: personal per-user dashboard layouts (JSONB tabs /
widgets). RLS follows the 0090 fail-closed pattern scoped to BOTH runtime
roles (crm_api, crm_worker).

Part 2 — convergence fix (measured live on production 2026-08-30):
migration 0143 created the letterheads/print_templates/document_assets
tenant-isolation policies with ``TO crm_api`` only, while the established
pattern (0090, verified by tests/test_rls_coverage.py) requires both
crm_api AND crm_worker. This migration recreates those policies with both
roles so both install paths (plugin-SQL 0003 / alembic 0143) converge.
"""

import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import UUID as PGUUID

from alembic import op

revision = "0144"
down_revision = "0143"
branch_labels = None
depends_on = None

_TENANT_USING = (
    "tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid"
)


def _table_exists(conn, table_name: str) -> bool:
    row = conn.execute(
        sa.text("SELECT to_regclass(:tname) IS NOT NULL"),
        {"tname": f"public.{table_name}"},
    ).scalar()
    return bool(row)


def _create_policy(table: str) -> None:
    op.execute(f"DROP POLICY IF EXISTS {table}_tenant_isolation ON {table}")
    op.execute(
        f"CREATE POLICY {table}_tenant_isolation ON {table} AS PERMISSIVE "
        f"FOR ALL TO crm_api, crm_worker "
        f"USING ({_TENANT_USING}) "
        f"WITH CHECK ({_TENANT_USING})"
    )


def _rls(table: str) -> None:
    op.execute(f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY")
    op.execute(f"ALTER TABLE {table} FORCE ROW LEVEL SECURITY")
    _create_policy(table)


def upgrade() -> None:
    conn = op.get_bind()

    # ── Part 1: dashboards table ──
    if not _table_exists(conn, "dashboards"):
        op.create_table(
            "dashboards",
            sa.Column("id", PGUUID(as_uuid=True), primary_key=True),
            sa.Column("name", sa.String(100), nullable=False),
            sa.Column("layout", JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb")),
            sa.Column("is_default", sa.Boolean(), nullable=False, server_default=sa.false()),
            sa.Column("user_id", PGUUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
            sa.Column("tenant_id", PGUUID(as_uuid=True), nullable=False),
            sa.Column("deleted_at", sa.DateTime(timezone=True)),
            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        )
        op.create_index(
            "uq_dashboards_tenant_user_name",
            "dashboards",
            ["tenant_id", "user_id", "name"],
            unique=True,
            postgresql_where=sa.text("deleted_at IS NULL"),
        )
        op.create_index("ix_dashboards_tenant_user", "dashboards", ["tenant_id", "user_id"])
        _rls("dashboards")
    else:
        # Dual-path convergence: table exists (plugin SQL), ensure policy roles
        _create_policy("dashboards")

    # ── Part 2: converge Phase L policies to crm_api + crm_worker ──
    for table in ("letterheads", "print_templates", "document_assets"):
        if _table_exists(conn, table):
            _create_policy(table)


def downgrade() -> None:
    conn = op.get_bind()
    # Revert the convergence fix to the (buggy) Phase L state first…
    for table in ("letterheads", "print_templates", "document_assets"):
        if _table_exists(conn, table):
            op.execute(f"DROP POLICY IF EXISTS {table}_tenant_isolation ON {table}")
            op.execute(
                f"CREATE POLICY {table}_tenant_isolation ON {table} AS PERMISSIVE "
                f"FOR ALL TO crm_api "
                f"USING ({_TENANT_USING}) "
                f"WITH CHECK ({_TENANT_USING})"
            )
    if _table_exists(conn, "dashboards"):
        op.execute("DROP POLICY IF EXISTS dashboards_tenant_isolation ON dashboards")
        op.drop_index("ix_dashboards_tenant_user", table_name="dashboards")
        op.drop_index("uq_dashboards_tenant_user_name", table_name="dashboards")
        op.drop_table("dashboards")