"""RBAC roles: add role column to portal_users, user_tenant_roles, portal_invitations Revision ID: 006 Revises: 005 Create Date: 2026-03-24 This migration adds: 1. `role` column to `portal_users` (TEXT + CHECK constraint): - Nullable initially, backfilled, then made NOT NULL - CHECK: role IN ('platform_admin', 'customer_admin', 'customer_operator') - Backfill: is_admin=TRUE -> 'platform_admin', else -> 'customer_admin' - Drops `is_admin` column after backfill 2. `user_tenant_roles` table: - Maps portal users to tenants with a role - UNIQUE(user_id, tenant_id) — one role per user per tenant - ON DELETE CASCADE for both FKs 3. `portal_invitations` table: - Invitation records for invite-only onboarding flow - token_hash: SHA-256 of the raw HMAC token (unique, for lookup) - status: 'pending' | 'accepted' | 'revoked' - expires_at: TIMESTAMPTZ, 48h from creation Design notes: - TEXT + CHECK constraint (not sa.Enum) per Phase 1 ADR — avoids DDL type conflicts in Alembic - portal_users: not RLS-protected (auth before tenant context) - user_tenant_roles and portal_invitations: not RLS-protected (RBAC layer handles isolation) - konstruct_app granted appropriate permissions per table """ from __future__ import annotations from typing import Sequence, Union import sqlalchemy as sa from alembic import op from sqlalchemy.dialects.postgresql import UUID # revision identifiers, used by Alembic. revision: str = "006" down_revision: Union[str, None] = "005" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: # ========================================================================= # 1. Add role column to portal_users # ========================================================================= # Step 1a: Add as nullable first (to allow backfill) op.add_column( "portal_users", sa.Column("role", sa.Text, nullable=True), ) # Step 1b: Backfill — is_admin=TRUE -> 'platform_admin', else -> 'customer_admin' op.execute(""" UPDATE portal_users SET role = CASE WHEN is_admin = TRUE THEN 'platform_admin' ELSE 'customer_admin' END """) # Step 1c: Make NOT NULL op.alter_column("portal_users", "role", nullable=False) # Step 1d: Add CHECK constraint (TEXT + CHECK pattern, not sa.Enum) op.create_check_constraint( "ck_portal_users_role", "portal_users", "role IN ('platform_admin', 'customer_admin', 'customer_operator')", ) # Step 1e: Drop is_admin column op.drop_column("portal_users", "is_admin") # ========================================================================= # 2. user_tenant_roles table # ========================================================================= op.create_table( "user_tenant_roles", sa.Column( "id", UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()"), ), sa.Column( "user_id", UUID(as_uuid=True), sa.ForeignKey("portal_users.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "tenant_id", UUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "role", sa.Text, nullable=False, comment="customer_admin | customer_operator", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), ), sa.UniqueConstraint("user_id", "tenant_id", name="uq_user_tenant_role"), ) op.create_index("ix_user_tenant_roles_user", "user_tenant_roles", ["user_id"]) op.create_index("ix_user_tenant_roles_tenant", "user_tenant_roles", ["tenant_id"]) op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON user_tenant_roles TO konstruct_app") # ========================================================================= # 3. portal_invitations table # ========================================================================= op.create_table( "portal_invitations", sa.Column( "id", UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()"), ), sa.Column( "email", sa.String(255), nullable=False, ), sa.Column( "name", sa.String(255), nullable=False, ), sa.Column( "tenant_id", UUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, ), sa.Column( "role", sa.Text, nullable=False, comment="customer_admin | customer_operator", ), sa.Column( "invited_by", UUID(as_uuid=True), sa.ForeignKey("portal_users.id"), nullable=True, comment="ID of the user who created the invitation (NULL for system-generated)", ), sa.Column( "token_hash", sa.String(255), nullable=False, unique=True, comment="SHA-256 hex digest of the raw HMAC invite token", ), sa.Column( "status", sa.String(20), nullable=False, server_default="pending", comment="pending | accepted | revoked", ), sa.Column( "expires_at", sa.DateTime(timezone=True), nullable=False, ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("NOW()"), ), ) op.create_index("ix_portal_invitations_tenant", "portal_invitations", ["tenant_id"]) op.create_index("ix_portal_invitations_email", "portal_invitations", ["email"]) op.create_index("ix_portal_invitations_token_hash", "portal_invitations", ["token_hash"], unique=True) op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON portal_invitations TO konstruct_app") def downgrade() -> None: # Remove portal_invitations op.execute("REVOKE ALL ON portal_invitations FROM konstruct_app") op.drop_index("ix_portal_invitations_token_hash", table_name="portal_invitations") op.drop_index("ix_portal_invitations_email", table_name="portal_invitations") op.drop_index("ix_portal_invitations_tenant", table_name="portal_invitations") op.drop_table("portal_invitations") # Remove user_tenant_roles op.execute("REVOKE ALL ON user_tenant_roles FROM konstruct_app") op.drop_index("ix_user_tenant_roles_tenant", table_name="user_tenant_roles") op.drop_index("ix_user_tenant_roles_user", table_name="user_tenant_roles") op.drop_table("user_tenant_roles") # Restore is_admin column on portal_users op.drop_constraint("ck_portal_users_role", "portal_users", type_="check") op.add_column( "portal_users", sa.Column("is_admin", sa.Boolean, nullable=True), ) op.execute(""" UPDATE portal_users SET is_admin = CASE WHEN role = 'platform_admin' THEN TRUE ELSE FALSE END """) op.alter_column("portal_users", "is_admin", nullable=False) op.drop_column("portal_users", "role")