DEP-262: add SUPER_ADMIN role and associated functionality to db; remove staff_user tenant_id column

Author: NatSquared

CHANGELOG.MD

@@ -1,3 +1,11 @@
+## May 4, 2026
+
+- **Feature** Sync SUPER_ADMIN tenant membership and protect super admins in user management [🎟️ DEP-262](https://citz-gdx.atlassian.net/browse/DEP-262)
+  - Added a migration to create the `SUPER_ADMIN` group, the `super_admin` role, and their mapping so super admins can be represented in tenant memberships.
+  - Updated the API to add or remove `SUPER_ADMIN` group membership based on the user's Keycloak roles during login/request processing, including unit tests for assignment and removal.
+  - Updated the user management UI to recognize Super Admin as a composite role and prevent administrators and super admins from being deactivated, while showing the correct permission messaging.
+  - Removed the tenant_id column from staff_user table, as tenant membership is determined by group memberships rather than a direct column on the user model, making the column redundant.
+
 ## April 30, 2026
 
 - **Feature** Updated all S3 uploaded objects to dynamically generate document URLs from the bucket and object key, rather than storing the full URL in the database. [🎟️ DEP-261](https://citz-gdx.atlassian.net/browse/DEP-261)

api/migrations/versions/9d2d16f5f3aa_add_super_admin_role_group_mapping.py

@@ -0,0 +1,80 @@
+"""Add super admin role/group mapping.
+
+Revision ID: 9d2d16f5f3aa
+Revises: 8e50e47b7d32
+Create Date: 2026-04-30 12:00:00.000000
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '9d2d16f5f3aa'
+down_revision = '8e50e47b7d32'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute(
+        """
+        INSERT INTO user_group (created_date, updated_date, id, name, created_by, updated_by)
+        SELECT NOW(), NOW(), COALESCE((SELECT MAX(id) + 1 FROM user_group), 1), 'SUPER_ADMIN', NULL, NULL
+        WHERE NOT EXISTS (
+            SELECT 1 FROM user_group WHERE name = 'SUPER_ADMIN'
+        )
+        """
+    )
+
+    op.execute(
+        """
+        INSERT INTO user_role (created_date, updated_date, id, name, description, created_by, updated_by)
+        SELECT NOW(), NOW(), COALESCE((SELECT MAX(id) + 1 FROM user_role), 1),
+               'super_admin', 'Role for system-wide super administrator access', NULL, NULL
+        WHERE NOT EXISTS (
+            SELECT 1 FROM user_role WHERE name = 'super_admin'
+        )
+        """
+    )
+
+    op.execute(
+        """
+        INSERT INTO group_role_mapping (created_date, updated_date, id, role_id, group_id, created_by, updated_by)
+        SELECT NOW(), NOW(), COALESCE((SELECT MAX(id) + 1 FROM group_role_mapping), 1),
+               r.id, g.id, NULL, NULL
+        FROM user_group g
+        JOIN user_role r ON r.name = 'super_admin'
+        WHERE g.name = 'SUPER_ADMIN'
+          AND NOT EXISTS (
+              SELECT 1
+              FROM group_role_mapping m
+              WHERE m.group_id = g.id
+                AND m.role_id = r.id
+          )
+        """
+    )
+
+
+def downgrade():
+    op.execute(
+        """
+        DELETE FROM user_group_membership
+        WHERE group_id IN (
+            SELECT id FROM user_group WHERE name = 'SUPER_ADMIN'
+        )
+        """
+    )
+
+    op.execute(
+        """
+        DELETE FROM group_role_mapping
+        WHERE group_id IN (
+            SELECT id FROM user_group WHERE name = 'SUPER_ADMIN'
+        )
+        OR role_id IN (
+            SELECT id FROM user_role WHERE name = 'super_admin'
+        )
+        """
+    )
+
+    op.execute("DELETE FROM user_role WHERE name = 'super_admin'")
+    op.execute("DELETE FROM user_group WHERE name = 'SUPER_ADMIN'")

api/migrations/versions/a6f15ee2c7b4_remove_staff_user_tenant_id.py

@@ -0,0 +1,48 @@
+"""Remove redundant tenant_id from staff_users.
+
+Revision ID: a6f15ee2c7b4
+Revises: 9d2d16f5f3aa
+Create Date: 2026-05-04 16:40:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'a6f15ee2c7b4'
+down_revision = '9d2d16f5f3aa'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    _drop_matching_foreign_keys('staff_users', 'tenant_id', 'tenant')
+    with op.batch_alter_table('staff_users', schema=None) as batch_op:
+        batch_op.drop_column('tenant_id')
+
+
+def downgrade():
+    with op.batch_alter_table('staff_users', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('tenant_id', sa.Integer(), nullable=True))
+
+    op.create_foreign_key(None, 'staff_users', 'tenant', ['tenant_id'], ['id'])
+
+
+def _drop_matching_foreign_keys(table_name, local_column, referred_table):
+    """Drop FK constraints for a specific local column and referenced table."""
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    foreign_keys = inspector.get_foreign_keys(table_name)
+
+    for foreign_key in foreign_keys:
+        name = foreign_key.get('name')
+        constrained_columns = foreign_key.get('constrained_columns') or []
+        current_referred_table = foreign_key.get('referred_table')
+
+        if (
+            name
+            and local_column in constrained_columns
+            and current_referred_table == referred_table
+        ):
+            op.drop_constraint(name, table_name, type_='foreignkey')

api/src/api/__init__.py

@@ -205,6 +205,12 @@ def get_roles(token_info) -> list:
         # ... so any extraneous roles are discarded
         user_roles = list(set(roles_from_token).intersection(keycloak_forwarded_roles))
 
+        StaffUserService.sync_super_admin_membership(
+            token_info=token_info,
+            token_roles=roles_from_token,
+            tenant_id=getattr(g, 'tenant_id', None),
+        )
+
         # Retrieve user by external ID from token info
         user = StaffUserService.get_user_by_external_id(token_info['sub'])
 

api/src/api/models/base_model.py

@@ -19,8 +19,6 @@
 from sqlalchemy.ext.declarative import declared_attr
 
 from .db import db
-from ..utils.token_info import TokenInfo
-
 TENANT_ID = 'tenant_id'
 
 
@@ -48,7 +46,8 @@ def _get_current_user():
 
         Used to populate the created_by and modified_by relationships on all models.
         """
-        return TokenInfo.get_id()
+        token_info = getattr(g, 'jwt_oidc_token_info', None) or {}
+        return token_info.get('sub')
 
     @classmethod
     def find_by_id(cls, identifier: int):

api/src/api/models/staff_user.py

@@ -7,7 +7,7 @@
 from typing import Optional
 
 from flask import g
-from sqlalchemy import Column, ForeignKey, String, asc, desc, func, or_
+from sqlalchemy import Column, ForeignKey, String, asc, desc, func
 from sqlalchemy.orm import column_property
 from sqlalchemy.sql import text
 from sqlalchemy.sql.operators import ilike_op
@@ -36,7 +36,6 @@ class StaffUser(BaseModel):
     contact_number = Column(db.String(50), nullable=True)
     external_id = Column(db.String(50), nullable=False, unique=True)
     status_id = db.Column(db.Integer, ForeignKey('user_status.id'), nullable=False, default=1)
-    tenant_id = db.Column(db.Integer, db.ForeignKey('tenant.id'), nullable=True)
 
     @classmethod
     def get_all_paginated(cls, pagination_options: PaginationOptions, search_text='', include_inactive=False):
@@ -67,18 +66,12 @@ def get_all_paginated(cls, pagination_options: PaginationOptions, search_text=''
     @classmethod
     def _add_tenant_filter(cls, query):
         """Add tenant filtering to the query based on user group membership."""
-        has_tenant_id = hasattr(cls, TENANT_ID)
         has_g_tenant_id = hasattr(g, TENANT_ID) and g.tenant_id
-        if has_tenant_id and has_g_tenant_id:
-            return query.outerjoin(
+        if has_g_tenant_id:
+            return query.join(
                 UserGroupMembership,
                 (UserGroupMembership.staff_user_external_id == cls.external_id) &
                 (UserGroupMembership.tenant_id == g.tenant_id)
-            ).filter(
-                or_(
-                    UserGroupMembership.id != None,  # noqa: E711  # pylint: disable=C0121
-                    cls.tenant_id == g.tenant_id,   # new user, no role yet
-                )
             )
         return query
 

api/src/api/schemas/staff_user.py

@@ -25,5 +25,4 @@ class Meta:  # pylint: disable=too-few-public-methods
     created_date = fields.Str(data_key='created_date')
     updated_date = fields.Str(data_key='updated_date')
     roles = fields.List(fields.Str(data_key='roles'))
-    tenant_id = fields.Str(data_key='tenant_id')
     status_id = fields.Int(data_key='status_id')

api/src/api/services/staff_user_service.py

@@ -15,6 +15,7 @@
 from api.utils import notification
 from api.utils.constants import CompositeRoles
 from api.utils.enums import CompositeRoleId, CompositeRoleNames, UserStatus
+from api.utils.roles import Role
 from api.utils.template import Template
 
 
@@ -48,20 +49,62 @@ def create_or_update_user(self, user: dict):
         if db_user is None:
             new_user = StaffUserModel.create_user(user)
             if len(user.get('roles', [])) == 0:
-                self._send_access_request_email(new_user)
+                self._send_access_request_email(new_user, getattr(g, 'tenant_id', None))
             return new_user
 
         return StaffUserModel.update_user(db_user.id, user)
 
+    @classmethod
+    def sync_super_admin_membership(cls, token_info: dict, token_roles, tenant_id=None):
+        """Sync SUPER_ADMIN tenant membership from keycloak roles at login/request time."""
+        if getattr(g, 'syncing_super_admin_membership', False) or getattr(db.session, '_flushing', False):
+            return
+
+        g.syncing_super_admin_membership = True
+        if not token_info:
+            g.syncing_super_admin_membership = False
+            return
+
+        try:
+            external_id = token_info.get('sub')
+            if not external_id:
+                return
+
+            has_super_admin_role = Role.SUPER_ADMIN.value in set(token_roles or [])
+
+            db_user = StaffUserModel.get_user_by_external_id(external_id, include_inactive=True)
+            if db_user is None:
+                user_data = {
+                    'external_id': external_id,
+                    'first_name': token_info.get('given_name'),
+                    'last_name': token_info.get('family_name'),
+                    'email_address': token_info.get('email'),
+                    'username': token_info.get('preferred_username'),
+                    'identity_provider': token_info.get('identity_provider', ''),
+                    'roles': set(token_roles or []),
+                }
+                required = ['external_id', 'first_name', 'last_name', 'email_address']
+                if all(user_data.get(field) for field in required):
+                    cls().create_or_update_user(user_data)
+
+            if has_super_admin_role and tenant_id:
+                UserGroupMembershipService.ensure_group_membership(external_id, tenant_id, 'SUPER_ADMIN')
+                return
+
+            if not has_super_admin_role:
+                UserGroupMembershipService.remove_group_memberships_by_group_name(external_id, 'SUPER_ADMIN')
+        finally:
+            g.syncing_super_admin_membership = False
+
     @staticmethod
-    def _send_access_request_email(user: StaffUserModel) -> None:
+    def _send_access_request_email(user: StaffUserModel, tenant_id: Optional[int] = None) -> None:
         """Send a new user email.Throws error if fails."""
         templates = current_app.config['EMAIL_TEMPLATES']
         to_email_address = templates['ACCESS_REQUEST']['DEST_EMAIL_ADDRESS']
         if to_email_address is None:
             return
         template_id = templates['ACCESS_REQUEST']['ID']
-        subject, body, args = StaffUserService._render_email_template(user)
+        subject, body, args = StaffUserService._render_email_template(user, tenant_id)
         try:
             notification.send_email(subject=subject,
                                     email=to_email_address,
@@ -75,13 +118,14 @@ def _send_access_request_email(user: StaffUserModel) -> None:
                 status_code=HTTPStatus.INTERNAL_SERVER_ERROR) from exc
 
     @staticmethod
-    def _render_email_template(user: StaffUserModel):
+    def _render_email_template(user: StaffUserModel, tenant_id: Optional[int] = None):
         template = Template.get_template('email_access_request.html')
         templates = current_app.config['EMAIL_TEMPLATES']
         paths = current_app.config['PATHS']
         subject = templates['ACCESS_REQUEST']['SUBJECT']
+        tenant_id = tenant_id if tenant_id is not None else getattr(g, 'tenant_id', None)
         grant_access_url = notification.get_tenant_site_url(
-            user.tenant_id, paths['USER_MANAGEMENT']
+            tenant_id, paths['USER_MANAGEMENT']
         ).replace('{id}', str(user.id))
         email_environment = templates['ENVIRONMENT']
         args = {

api/src/api/services/user_group_membership_service.py

@@ -1,6 +1,8 @@
 """Service for user group membership management."""
 from typing import List, Tuple
 
+from api.models.db import db
+from api.models.user_group import UserGroup
 from api.models.user_group_membership import UserGroupMembership
 from api.models.user_role import UserRole
 
@@ -52,3 +54,57 @@ def assign_composite_role_to_user(membership_data):
     def reassign_composite_role_to_user(membership_data):
         """Update user_group_membership."""
         return UserGroupMembership.update_user_group_membership(membership_data)
+
+    @staticmethod
+    def ensure_group_membership(external_id: str, tenant_id: int, group_name: str):
+        """Ensure the user has a membership for the given tenant and group name."""
+        if not external_id or not tenant_id or not group_name:
+            return None
+
+        group = UserGroup.query.filter(UserGroup.name == group_name).first()
+        if not group:
+            return None
+
+        membership = UserGroupMembership.query.filter(
+            UserGroupMembership.staff_user_external_id == external_id,
+            UserGroupMembership.tenant_id == tenant_id,
+        ).first()
+
+        if membership:
+            if membership.group_id != group.id or not membership.is_active:
+                membership.group_id = group.id
+                membership.is_active = True
+                db.session.commit()
+            return membership
+
+        membership = UserGroupMembership(
+            staff_user_external_id=external_id,
+            group_id=group.id,
+            tenant_id=tenant_id,
+            is_active=True,
+        )
+        membership.save()
+        return membership
+
+    @staticmethod
+    def remove_group_memberships_by_group_name(external_id: str, group_name: str) -> int:
+        """Remove all memberships for a user that belong to the given group name."""
+        if not external_id or not group_name:
+            return 0
+
+        memberships = UserGroupMembership.query.join(
+            UserGroup,
+            UserGroupMembership.group_id == UserGroup.id,
+        ).filter(
+            UserGroupMembership.staff_user_external_id == external_id,
+            UserGroup.name == group_name,
+        ).all()
+
+        if not memberships:
+            return 0
+
+        for membership in memberships:
+            db.session.delete(membership)
+
+        db.session.commit()
+        return len(memberships)

api/src/api/utils/constants.py

@@ -20,6 +20,7 @@ class CompositeRoles(Enum):
     """Enumeration representing user roles."""
 
     ADMIN = 'Administrator'
+    SUPER_ADMIN = 'Super Admin'
     TEAM_MEMBER = 'Team Member'
     REVIEWER = 'Reviewer'
     VIEWER = 'Viewer'