leocrm/alembic/versions/0019_rbac_groups.py

"""RBAC: groups, user_groups, system_admin, tenant-scoped role_id, denied_permissions.

Revision ID: 0019_rbac_groups
Revises: 0018_fix_notif
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID

revision = "0019_rbac_groups"
down_revision = "0018_fix_notif"
branch_labels = None
depends_on = None


def upgrade() -> None:
    # ── groups table ──
    op.create_table(
        "groups",
        sa.Column("id", PGUUID(as_uuid=True), primary_key=True),
        sa.Column("tenant_id", PGUUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True),
        sa.Column("name", sa.String(100), nullable=False),
        sa.Column("description", sa.String(500), nullable=True),
        sa.Column("permissions", JSONB, nullable=False, server_default="{}"),
        sa.Column("denied_permissions", JSONB, nullable=False, server_default="[]"),
        sa.Column("field_permissions", JSONB, nullable=False, server_default="{}"),
        sa.Column("permission_version", sa.Integer, nullable=False, server_default="1"),
        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()),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.UniqueConstraint("tenant_id", "name", name="uq_groups_tenant_name"),
    )

    # ── user_groups table ──
    op.create_table(
        "user_groups",
        sa.Column("user_id", PGUUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("group_id", PGUUID(as_uuid=True), sa.ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True),
        sa.Column("tenant_id", PGUUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.UniqueConstraint("user_id", "group_id", "tenant_id", name="uq_user_groups_user_group_tenant"),
    )

    # ── users: add is_system_admin ──
    op.add_column(
        "users",
        sa.Column("is_system_admin", sa.Boolean, nullable=False, server_default="false"),
    )

    # ── user_tenants: add role_id ──
    op.add_column(
        "user_tenants",
        sa.Column("role_id", PGUUID(as_uuid=True), sa.ForeignKey("roles.id", ondelete="SET NULL"), nullable=True),
    )
    op.create_index("ix_user_tenants_role_id", "user_tenants", ["role_id"])

    # ── roles: add denied_permissions + permission_version + missing mixin columns ──
    op.add_column(
        "roles",
        sa.Column("denied_permissions", JSONB, nullable=False, server_default="[]"),
    )
    op.add_column(
        "roles",
        sa.Column("permission_version", sa.Integer, nullable=False, server_default="1"),
    )
    # Add missing TimestampMixin + SoftDeleteMixin columns
    op.add_column(
        "roles",
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
    )
    op.add_column(
        "roles",
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
    )

    # ── Seed default roles per tenant ──
    # For each tenant, create admin/editor/viewer role records if they don't exist
    # Use sa.text() with bindparams to avoid SQLAlchemy interpreting :read/:write as bind params
    for role_name, perms_json in [
        ("admin", '{"*:*": true}'),
        ("editor", '{"core:*:read": true, "core:*:write": true, "core:*:create": true}'),
        ("viewer", '{"core:*:read": true}'),
    ]:
        op.execute(
            sa.text("""
                INSERT INTO roles (id, tenant_id, name, permissions, denied_permissions, field_permissions, permission_version, created_at)
                SELECT
                    gen_random_uuid(),
                    t.id,
                    :role_name,
                    CAST(:perms AS jsonb),
                    '[]'::jsonb,
                    '{}'::jsonb,
                    1,
                    now()
                FROM tenants t
                WHERE NOT EXISTS (
                    SELECT 1 FROM roles ro
                    WHERE ro.tenant_id = t.id AND ro.name = :role_name
                )
            """).bindparams(
                sa.bindparam("role_name", value=role_name),
                sa.bindparam("perms", value=perms_json),
            )
        )

    # ── Migrate existing user.role_id to user_tenants.role_id ──
    # For each user_tenants row, set role_id from users table if the user has one
    op.execute("""
        UPDATE user_tenants ut
        SET role_id = u.role_id
        FROM users u
        WHERE ut.user_id = u.id
          AND u.role_id IS NOT NULL
          AND ut.role_id IS NULL
    """)

    # ── Migrate legacy role strings to role records ──
    # For users with role='admin'/'editor'/'viewer' but no role_id on user_tenants,
    # link to the seeded default roles
    op.execute("""
        UPDATE user_tenants ut
        SET role_id = ro.id
        FROM users u, roles ro
        WHERE ut.user_id = u.id
          AND ro.tenant_id = ut.tenant_id
          AND ro.name = u.role
          AND ut.role_id IS NULL
          AND u.role IN ('admin', 'editor', 'viewer')
    """)


def downgrade() -> None:
    op.drop_column("roles", "deleted_at")
    op.drop_column("roles", "updated_at")
    op.drop_column("roles", "permission_version")
    op.drop_column("roles", "denied_permissions")
    op.drop_index("ix_user_tenants_role_id", table_name="user_tenants")
    op.drop_column("user_tenants", "role_id")
    op.drop_column("users", "is_system_admin")
    op.drop_table("user_groups")
    op.drop_table("groups")