feat: add role-based admin support for multiple admin users

- Add Role model and user_roles association table (many-to-many)
- Add password_hash column to User model (nullable, for future bcrypt)
- Keep is_admin boolean for backward compatibility
- Add is_admin_effective property: true if legacy is_admin OR 'admin' role
- Update auth dependency (get_current_admin_user) to use is_admin_effective
- Update bootstrap login to auto-create 'admin' role and assign on login
- Update Google OAuth login to use is_admin_effective for token creation
- Add migrate_roles.py: idempotent migration script to backfill roles
- Add unit tests for Role, User, association table, and is_admin_effective

backend/app/api/auth.py

@@ -57,8 +57,18 @@ async def login(payload: LoginRequest):
 
     # This is handled in the auth router to keep it self-contained
     from app.database import async_session
+    from app.user_models import Role
 
     async with async_session() as session:
+        # Ensure the 'admin' role exists
+        result = await session.execute(select(Role).where(Role.name == "admin"))
+        admin_role = result.scalar_one_or_none()
+        if admin_role is None:
+            admin_role = Role(name="admin", description="Full administrative access")
+            session.add(admin_role)
+            await session.commit()
+            await session.refresh(admin_role)
+
         # Find or create the bootstrap admin user
         result = await session.execute(
             select(User).where(
@@ -74,11 +84,16 @@ async def login(payload: LoginRequest):
                 auth_provider="local",
                 is_admin=True,
             )
+            user.roles.append(admin_role)
             session.add(user)
+        else:
+            # Ensure existing bootstrap user also has the role
+            if admin_role not in user.roles:
+                user.roles.append(admin_role)
         await session.commit()
         await session.refresh(user)
 
-    token = create_access_token(user.id, user.is_admin)
+    token = create_access_token(user.id, user.is_admin_effective)
     return TokenResponse(access_token=token)
 
 
@@ -102,7 +117,7 @@ async def login_with_google(payload: GoogleLoginRequest, session: AsyncSession =
         await session.commit()
         await session.refresh(user)
 
-    token = create_access_token(user.id, user.is_admin)
+    token = create_access_token(user.id, user.is_admin_effective)
     return TokenResponse(access_token=token)
 
 

backend/app/auth.py

@@ -98,8 +98,11 @@ async def get_current_user(
 async def get_current_admin_user(
     current_user: User = Depends(get_current_user),
 ) -> User:
-    """Raise 403 if the current user is not an admin."""
-    if not current_user.is_admin:
+    """Raise 403 if the current user is not an admin.
+
+    Checks both the legacy is_admin flag and the new role-based admin role.
+    """
+    if not current_user.is_admin_effective:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
     return current_user
 

backend/app/user_models.py

@@ -1,11 +1,47 @@
+"""User, Role, and UserRole models — supports multiple admin roles."""
+
 from datetime import datetime
 
-from sqlalchemy import Column, Integer, String, Boolean, DateTime
-from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy import (
+    Boolean,
+    Column,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Table,
+    Text,
+    UniqueConstraint,
+)
+from sqlalchemy.orm import DeclarativeBase, relationship
 
 from app.models import Base
 
 
+# ── Many-to-many: users ↔ roles ──────────────────────────────────────────────
+
+user_roles = Table(
+    "user_roles",
+    Base.metadata,
+    Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
+    Column("role_id", Integer, ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
+)
+
+
+# ── Role ─────────────────────────────────────────────────────────────────────
+
+class Role(Base):
+    __tablename__ = "roles"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    name = Column(String(50), unique=True, nullable=False)
+    description = Column(Text, nullable=True)
+
+    users = relationship("User", secondary=user_roles, back_populates="roles")
+
+
+# ── User ─────────────────────────────────────────────────────────────────────
+
 class User(Base):
     __tablename__ = "users"
 
@@ -14,5 +50,16 @@ class User(Base):
     display_name = Column(String(100), nullable=False)
     avatar_url = Column(String(500), nullable=True)
     auth_provider = Column(String(20), nullable=False, default="google")
+    password_hash = Column(String(255), nullable=True)
     is_admin = Column(Boolean, default=False)
     created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
+
+    # Role-based admin (many-to-many)
+    roles = relationship("Role", secondary=user_roles, back_populates="users")
+
+    @property
+    def is_admin_effective(self) -> bool:
+        """True when is_admin=True (legacy) OR the user has an 'admin' role."""
+        if self.is_admin:
+            return True
+        return any(role.name == "admin" for role in self.roles)

backend/migrate_roles.py

@@ -0,0 +1,60 @@
+"""
+Migration seed: create the 'admin' role and assign it to existing is_admin=True users.
+
+Run once after deploying the updated schema. Idempotent — safe to run multiple times.
+
+Usage:
+    KMTN_DATABASE_URL=... python migrate_roles.py
+    (run from the backend/ directory)
+"""
+
+import asyncio
+from sqlalchemy import select, text
+
+from app.database import async_session
+from app.user_models import User, Role, user_roles
+
+
+async def migrate() -> None:
+    """Ensure 'admin' role exists and assign to existing is_admin=True users."""
+    async with async_session() as session:
+        # 1) Create 'admin' role if it doesn't exist
+        result = await session.execute(select(Role).where(Role.name == "admin"))
+        admin_role = result.scalar_one_or_none()
+        if admin_role is None:
+            admin_role = Role(name="admin", description="Full administrative access")
+            session.add(admin_role)
+            await session.commit()
+            await session.refresh(admin_role)
+            print(f"  ✓ Created 'admin' role (id={admin_role.id})")
+        else:
+            print(f"  ✓ 'admin' role already exists (id={admin_role.id})")
+
+        # 2) Assign 'admin' role to all existing is_admin=True users
+        result = await session.execute(
+            select(User).where(User.is_admin == True)  # noqa: E712
+        )
+        admin_users = result.scalars().all()
+        migrated = 0
+        for user in admin_users:
+            if admin_role not in user.roles:
+                user.roles.append(admin_role)
+                migrated += 1
+        if migrated:
+            await session.commit()
+            print(f"  ✓ Assigned 'admin' role to {migrated} existing admin user(s)")
+        else:
+            print("  ✓ No new role assignments needed")
+
+        # 3) Summary
+        result = await session.execute(
+            select(User).where(user_roles.c.role_id == admin_role.id)
+        )
+        role_admins = result.scalars().all()
+        print(f"  → {len(role_admins)} user(s) now have the 'admin' role")
+
+
+if __name__ == "__main__":
+    print("Running role migration...")
+    asyncio.run(migrate())
+    print("Done.")

backend/tests/test_user_models.py

@@ -0,0 +1,87 @@
+"""Tests for app.user_models — User, Role, and UserRole relationships."""
+
+import pytest
+
+from app.user_models import User, Role, user_roles
+
+
+class TestRoleModel:
+    """Verify Role model definition."""
+
+    def test_role_table_name(self):
+        assert Role.__tablename__ == "roles"
+
+    def test_role_has_name_column(self):
+        col = Role.__table__.columns["name"]
+        assert col.type.__visit_name__ == "VARCHAR"
+
+    def test_role_name_is_unique(self):
+        col = Role.__table__.columns["name"]
+        assert col.unique is True
+
+    def test_role_name_not_nullable(self):
+        col = Role.__table__.columns["name"]
+        assert col.nullable is False
+
+
+class TestUserModel:
+    """Verify User model still works with new role fields."""
+
+    def test_user_table_name(self):
+        assert User.__tablename__ == "users"
+
+    def test_user_has_is_admin_column(self):
+        col = User.__table__.columns["is_admin"]
+        assert col.default.arg is False
+
+    def test_user_has_password_hash_column(self):
+        col = User.__table__.columns["password_hash"]
+        assert col.nullable is True
+
+    def test_user_roles_relationship(self):
+        assert hasattr(User, "roles")
+
+
+class TestUserRoleAssociation:
+    """Verify the user_roles association table."""
+
+    def test_user_roles_table_name(self):
+        assert user_roles.name == "user_roles"
+
+    def test_user_roles_has_user_id(self):
+        assert "user_id" in user_roles.columns
+
+    def test_user_roles_has_role_id(self):
+        assert "role_id" in user_roles.columns
+
+    def test_user_roles_composite_primary_key(self):
+        pk = [col for col in user_roles.columns if col.primary_key]
+        assert len(pk) == 2
+
+
+class TestIsAdminEffective:
+    """Test the is_admin_effective property."""
+
+    def test_legacy_is_admin_true(self):
+        user = User(email="a@test.com", display_name="A", is_admin=True)
+        assert user.is_admin_effective is True
+
+    def test_legacy_is_admin_false_no_roles(self):
+        user = User(email="b@test.com", display_name="B", is_admin=False)
+        assert user.is_admin_effective is False
+
+    def test_role_based_admin(self):
+        user = User(email="c@test.com", display_name="C", is_admin=False)
+        role = Role(name="admin")
+        user.roles.append(role)
+        assert user.is_admin_effective is True
+
+    def test_non_admin_role_does_not_grant_admin(self):
+        user = User(email="d@test.com", display_name="D", is_admin=False)
+        role = Role(name="editor")
+        user.roles.append(role)
+        assert user.is_admin_effective is False
+
+    def test_legacy_is_admin_true_overrides_empty_roles(self):
+        user = User(email="e@test.com", display_name="E", is_admin=True)
+        assert user.is_admin_effective is True