permission grandted

backend/LLM_CONTEXT_PROMPT.md

@@ -0,0 +1,130 @@
+# HR-ATS Backend LLM Context Prompt
+
+Copy everything below the line into any LLM session before asking it to write or edit backend code.
+
+---
+
+You are coding inside **HR-ATS-Portal** (`backend/`). You must follow this house style **exactly**. Mirror neighboring files. Do not invent alternate patterns, layers, or response shapes. Prefer matching existing code over “cleaner” industry defaults.
+
+## Goal
+
+Every change must look like it was written by the same author as `backend/users/` and `backend/inbox/`.
+
+## Package layout (every domain)
+
+```
+backend/<domain>/
+  app.py          # routes only — HTTP in/out
+  views.py        # service class — business logic
+  models.py       # SQLModel table + classmethod DB accessors
+  serializers.py  # hand-rolled dict builders (no Pydantic response models)
+  plugins.py      # pure helpers (hash, JWT, clean payload) — NO FastAPI imports
+  permissions.py  # OAuth2 scheme + Depends aliases (auth domains only)
+```
+
+- Bare `router = APIRouter()`; mount in `main.py` with `app.include_router(...)`.
+- No package `__init__.py`. Run from `backend/` so imports are top-level (`users.app`, `db_setup`).
+- Non-DB config: module-level `load_dotenv()` + `os.getenv(...)`. Do **not** extend `db_setup.Settings` for app secrets.
+
+## Layer duties (non-negotiable)
+
+| Layer | Owns | Must NOT do |
+|---|---|---|
+| `app.py` | Routes, inline request Pydantic models, `JSONResponse`, HTTP token envelope via serializers, inject `session` / `CurrentUser` | Business rules, SQL, JWT crypto beyond calling plugin functions |
+| `views.py` | Business checks, call models, raise `HTTPException`, return ORM user (auth) or serialized dict (CRUD) | Call `serialize_token` or build login HTTP payloads |
+| `models.py` | Fields, queries, inserts/updates/soft-delete, `selectinload` when needed | HTTPException, FastAPI, serializers |
+| `serializers.py` | `serialize_*` → plain `dict` | DB, Depends |
+| `plugins.py` | Pure helpers; raise library errors (`jwt.*`) | Import FastAPI / raise HTTPException |
+| `permissions.py` | `OAuth2PasswordBearer`, `get_current_user`, `CurrentUser` alias, `require_permission` | Route handlers |
+
+## Exact route pattern (`app.py`)
+
+- Paths are verb-in-path: `/users/create`, `/users/fetch`, `/users/login` — **not** `/auth/token`, not REST-resource-only.
+- Standard wrapper on every handler:
+
+```python
+try:
+    service=User(session=session)
+    data=await service.some_method(...)
+    return JSONResponse(content={"data":data,"status_code":200})
+except HTTPException:
+    raise
+except Exception as e:
+    raise HTTPException(status_code=500,detail=str(e))
+```
+
+- List: `{"data":items,"total":total,"status_code":200}`. By id: include `"total":1`.
+- Login / refresh — **service returns ORM user**; route mints tokens and serializes:
+
+```python
+user=await service.authenticate_user(form_data.username,form_data.password)
+tokens=serialize_token(create_access_token(user),create_refresh_token(user),user)
+return JSONResponse(content={**tokens,"status_code":200})
+```
+
+- Request body models stay **inline in `app.py`** (`UserCreate`, `UserUpdate`, `TokenRefresh`). Never move them into `serializers.py`.
+- Use `session: AsyncSession = Depends(get_session)` by default. Use `Annotated` only when required (`OAuth2PasswordRequestForm`, `CurrentUser` before other defaulted params).
+- Preserve tight local spacing: `service=User(session=session)`, `detail=str(e)`. Do not pretty-reformat unrelated code.
+
+## Exact service pattern (`views.py`)
+
+```python
+class User:
+    def __init__(self,session:AsyncSession):
+        self.session=session
+
+    async def create_user(self,payload):
+        ...
+        return serialize_user(user)
+```
+
+- Leave service method parameters untyped (match existing).
+- Raise `HTTPException(status_code=...,detail="...")` for domain errors.
+- `authenticate_user` / `refresh_access_token` return the **Users ORM instance only**.
+- If you loaded via an accessor that does not `selectinload(role)`, re-fetch with `get_user_by_id` before serialization that touches `user.role`. (`get_user_by_email` and `get_user_by_id` both eager-load `role` today.)
+
+## Exact model pattern (`models.py`)
+
+- SQLModel `table=True`; accessors as `@classmethod async def`.
+- Soft delete sets `is_deleted=True` and `is_active=False`.
+- `selectinload` relations that serializers read.
+- Commits happen inside write accessors (existing convention).
+
+## Serializers
+
+- Hand-built dicts only. `str(uuid)`, `.isoformat()` for datetimes. Never include `password`.
+- Token response: OAuth2 fields at **root** (`access_token`, `refresh_token`, `token_type`, `expires_in`); user record under `data`.
+
+## Auth (when touching users auth)
+
+- PyJWT access + refresh with a `type` claim; `decode_token(..., expected_type=...)` rejects mismatches.
+- Protect `/users/*` with `current_user: CurrentUser` except `/users/login` and `/users/refresh`. Prefer `Depends(require_permission(...))` on mutating/list routes that need a specific tag; keep `/users/me` on plain `CurrentUser` so users can discover a missing-role state.
+- `get_current_user`: decode access → DB by `sub` → reject missing/deleted/inactive → return `serialize_user(user, with_permissions=True)`.
+- Login: `OAuth2PasswordRequestForm` (username = email). `tokenUrl="users/login"` (no leading slash).
+- JWT `iat`/`exp` use `datetime.now(timezone.utc)` only — never naive `datetime.now()`.
+
+## Dependencies / env
+
+- Add pins to `backend/requirements.txt` under banner comments with a trailing `# why` comment.
+- Put secrets in `backend/.env`; keep key names in `backend/.env.example`.
+
+## Hard bans
+
+1. No repository / use-case / DTO layers beyond inline request models.
+2. No Pydantic response models; no alternate envelopes; no `/api/v1` prefix.
+3. No `serialize_token` inside `views.py`.
+4. No FastAPI imports in `plugins.py`.
+5. No drive-by refactors, renames, or whole-file reformats.
+6. RBAC exists in `users/permissions.py`; do not invent a second scheme.
+7. Do not edit unrelated domains (`inbox/` vs `users/`) unless asked.
+8. Do not add `__init__.py` to make packages.
+
+## Workflow when adding an endpoint
+
+1. Model accessor (if DB).
+2. Service method in `views.py`.
+3. `serialize_*` if new shape.
+4. Route in `app.py` with the standard try/except + `JSONResponse`.
+5. Add `current_user: CurrentUser` or `Depends(require_permission(...))` if the route is protected.
+
+Before finishing, re-read the touched files and confirm they still match a sibling file’s structure, naming, spacing, and response shape.

backend/main.py

@@ -7,6 +7,7 @@ from fastapi import FastAPI,APIRouter
 from db_setup import lifespan
 from inbox.app import router as inbox_router
 from users.app import router as users_router
+from role.app import router as role_router
 # Without this the db/migration logs have no handler and are swallowed under uvicorn.
 logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s")
 
@@ -22,4 +23,5 @@ app.add_middleware(
 )
 
 app.include_router(inbox_router)
-app.include_router(users_router)
+app.include_router(users_router)
+app.include_router(role_router)

backend/role/app.py

@@ -0,0 +1,191 @@
+from fastapi import APIRouter, Depends, Query
+from fastapi.responses import JSONResponse
+from fastapi import HTTPException
+from db_setup import get_session
+from sqlalchemy.ext.asyncio import AsyncSession
+from pydantic import BaseModel
+from role.views import Role
+from users.permissions import PermissionTag, require_permission
+from dotenv import load_dotenv
+load_dotenv()
+
+router = APIRouter()
+
+
+class RoleCreate(BaseModel):
+    role_name: str
+    description: str | None = None
+    permissions: list[int] | None = None
+    is_active: bool = True
+
+
+class RoleUpdate(BaseModel):
+    role_name: str | None = None
+    description: str | None = None
+    permissions: list[int] | None = None
+    is_active: bool | None = None
+
+
+class PermissionCreate(BaseModel):
+    name: str
+    description: str | None = None
+    permission_tags: list[int] | None = None
+    is_active: bool = True
+
+
+class PermissionUpdate(BaseModel):
+    name: str | None = None
+    description: str | None = None
+    permission_tags: list[int] | None = None
+    is_active: bool | None = None
+
+
+@router.get("/roles/fetch")
+async def fetch_roles(
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
+    record_id: int | None = Query(None),
+    search: str | None = Query(None),
+    top: int | None = Query(None),
+    skip: int = Query(0, ge=0),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        if record_id is not None:
+            item=await service.get_role_by_id(record_id)
+            return JSONResponse(content={"data":item,"total":1,"status_code":200})
+        items=await service.get_roles(top,skip,search)
+        total=await service.count_roles(search)
+        return JSONResponse(content={"data":items,"total":total,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.post("/roles/create")
+async def create_role(
+    payload: RoleCreate,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_CREATE)),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        data=await service.create_role(payload.model_dump())
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.put("/roles/update")
+async def update_role(
+    payload: RoleUpdate,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_EDIT)),
+    record_id: int = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        data=await service.update_role(record_id,payload.model_dump())
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.delete("/roles/delete")
+async def delete_role(
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_DELETE)),
+    record_id: int = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        data=await service.delete_role(record_id)
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.get("/permissions/fetch")
+async def fetch_permissions(
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
+    record_id: int | None = Query(None),
+    search: str | None = Query(None),
+    top: int | None = Query(None),
+    skip: int = Query(0, ge=0),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        if record_id is not None:
+            item=await service.get_permission_by_id(record_id)
+            return JSONResponse(content={"data":item,"total":1,"status_code":200})
+        items=await service.get_permissions(top,skip,search)
+        total=await service.count_permissions(search)
+        return JSONResponse(content={"data":items,"total":total,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.post("/permissions/create")
+async def create_permission(
+    payload: PermissionCreate,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_MANAGE)),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        data=await service.create_permission(payload.model_dump())
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.put("/permissions/update")
+async def update_permission(
+    payload: PermissionUpdate,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_MANAGE)),
+    record_id: int = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        data=await service.update_permission(record_id,payload.model_dump())
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.get("/permission-tags/fetch")
+async def fetch_permission_tags(
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
+    record_id: int | None = Query(None),
+    search: str | None = Query(None),
+    top: int | None = Query(None),
+    skip: int = Query(0, ge=0),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=Role(session=session)
+        if record_id is not None:
+            item=await service.get_permission_tag_by_id(record_id)
+            return JSONResponse(content={"data":item,"total":1,"status_code":200})
+        items=await service.get_permission_tags(top,skip,search)
+        total=await service.count_permission_tags(search)
+        return JSONResponse(content={"data":items,"total":total,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))

backend/role/models.py

@@ -1,13 +1,14 @@
 from datetime import datetime
 from enum import Enum
-from typing import Any
-
-from sqlalchemy import Column, Enum as SAEnum
+from sqlalchemy import Column, UniqueConstraint, func, or_
 from sqlalchemy.dialects.postgresql import JSONB
-from sqlmodel import Field, Relationship, SQLModel
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel import Field, Relationship, SQLModel, select
 
 
 class EnumRoles(str, Enum):
+    """Canonical keys for the eight seeded system roles. `Roles.role_name` is a varchar."""
+
     SYSTEM_ADMINISTRATOR = "system_administrator"
     HR_ADMINISTRATOR = "hr_administrator"
     RECRUITER = "recruiter"
@@ -18,19 +19,317 @@ class EnumRoles(str, Enum):
     CANDIDATE = "candidate"
 
 
+class PermissionTags(SQLModel, table=True):
+    __tablename__ = "permission_tags"
+    __table_args__ = (
+        UniqueConstraint("module", "action", name="uq_permission_tags_module_action"),
+    )
+
+    id: int | None = Field(default=None, primary_key=True)
+    tag_name: str = Field(max_length=64, unique=True, nullable=False, index=True)
+    module: str = Field(max_length=32, nullable=False, index=True)
+    action: str = Field(max_length=32, nullable=False)
+    description: str | None = Field(default=None)
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+    is_active: bool = Field(default=True)
+    is_deleted: bool = Field(default=False)
+
+    @classmethod
+    def _search_filter(cls, search: str):
+        pattern = f"%{search}%"
+        return or_(
+            cls.tag_name.ilike(pattern),
+            cls.module.ilike(pattern),
+            cls.action.ilike(pattern),
+            cls.description.ilike(pattern),
+        )
+
+    @classmethod
+    async def get_permission_tags(
+        cls, session: AsyncSession, top: int | None, skip: int, search: str | None
+    ):
+        statement = (
+            select(cls)
+            .where(cls.is_deleted == False)  # noqa: E712
+            .order_by(cls.module.asc(), cls.action.asc())
+        )
+        if search:
+            statement = statement.where(cls._search_filter(search))
+        if skip:
+            statement = statement.offset(skip)
+        if top is not None:
+            statement = statement.limit(top)
+        result = await session.execute(statement)
+        return result.scalars().all()
+
+    @classmethod
+    async def get_permission_tag_by_id(cls, session: AsyncSession, record_id: int):
+        statement = select(cls).where(cls.id == record_id)
+        result = await session.execute(statement)
+        return result.scalars().first()
+
+    @classmethod
+    async def get_permission_tags_by_ids(cls, session: AsyncSession, ids: list[int]):
+        if not ids:
+            return []
+        statement = select(cls).where(
+            cls.id.in_(ids),
+            cls.is_active == True,  # noqa: E712
+            cls.is_deleted == False,  # noqa: E712
+        )
+        result = await session.execute(statement)
+        return result.scalars().all()
+
+    @classmethod
+    async def count_permission_tags(cls, session: AsyncSession, search: str | None):
+        statement = (
+            select(func.count())
+            .select_from(cls)
+            .where(cls.is_deleted == False)  # noqa: E712
+        )
+        if search:
+            statement = statement.where(cls._search_filter(search))
+        result = await session.execute(statement)
+        return result.scalar_one()
+
+
+class Permissions(SQLModel, table=True):
+    """Named permission bundles — each row holds a JSONB array of permission_tags.id."""
+
+    __tablename__ = "permissions"
+
+    id: int | None = Field(default=None, primary_key=True)
+    name: str = Field(max_length=64, unique=True, nullable=False)
+    description: str | None = Field(default=None)
+    permission_tags: list | None = Field(default=None, sa_column=Column(JSONB))
+    is_system: bool = Field(default=False)
+    created_at: datetime = Field(default_factory=datetime.now)
+    updated_at: datetime = Field(default_factory=datetime.now)
+    is_active: bool = Field(default=True)
+    is_deleted: bool = Field(default=False)
+
+    @classmethod
+    def _search_filter(cls, search: str):
+        pattern = f"%{search}%"
+        return or_(cls.name.ilike(pattern), cls.description.ilike(pattern))
+
+    @classmethod
+    async def get_permissions(
+        cls, session: AsyncSession, top: int | None, skip: int, search: str | None
+    ):
+        statement = (
+            select(cls)
+            .where(cls.is_deleted == False)  # noqa: E712
+            .order_by(cls.name.asc())
+        )
+        if search:
+            statement = statement.where(cls._search_filter(search))
+        if skip:
+            statement = statement.offset(skip)
+        if top is not None:
+            statement = statement.limit(top)
+        result = await session.execute(statement)
+        return result.scalars().all()
+
+    @classmethod
+    async def get_permission_by_id(cls, session: AsyncSession, record_id: int):
+        statement = select(cls).where(cls.id == record_id)
+        result = await session.execute(statement)
+        return result.scalars().first()
+
+    @classmethod
+    async def get_permission_by_name(cls, session: AsyncSession, name: str):
+        statement = select(cls).where(cls.name == name, cls.is_deleted == False)  # noqa: E712
+        result = await session.execute(statement)
+        return result.scalars().first()
+
+    @classmethod
+    async def get_permissions_by_ids(cls, session: AsyncSession, ids: list[int]):
+        if not ids:
+            return []
+        statement = select(cls).where(
+            cls.id.in_(ids),
+            cls.is_active == True,  # noqa: E712
+            cls.is_deleted == False,  # noqa: E712
+        )
+        result = await session.execute(statement)
+        return result.scalars().all()
+
+    @classmethod
+    async def count_permissions(cls, session: AsyncSession, search: str | None):
+        statement = (
+            select(func.count())
+            .select_from(cls)
+            .where(cls.is_deleted == False)  # noqa: E712
+        )
+        if search:
+            statement = statement.where(cls._search_filter(search))
+        result = await session.execute(statement)
+        return result.scalar_one()
+
+    @classmethod
+    async def insert_permission(cls, session: AsyncSession, fields: dict):
+        row = cls(**fields)
+        session.add(row)
+        await session.commit()
+        return await cls.get_permission_by_id(session, row.id)
+
+    @classmethod
+    async def update_permission(cls, session: AsyncSession, record_id: int, fields: dict):
+        row = await cls.get_permission_by_id(session, record_id)
+        if not row:
+            return None
+        for key, value in fields.items():
+            setattr(row, key, value)
+        row.updated_at = datetime.now()
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+    @classmethod
+    async def soft_delete_permission(cls, session: AsyncSession, record_id: int):
+        row = await cls.get_permission_by_id(session, record_id)
+        if not row:
+            return None
+        row.is_deleted = True
+        row.is_active = False
+        row.updated_at = datetime.now()
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+
 class Roles(SQLModel, table=True):
     __tablename__ = "roles"
 
     id: int | None = Field(default=None, primary_key=True)
-    role_name: EnumRoles = Field(
-        default=EnumRoles.SYSTEM_ADMINISTRATOR,
-        sa_column=Column(SAEnum(EnumRoles), nullable=False, unique=True),
-    )
-    description: str | None = Field(default="System Administrator")
-    permissions: dict[str, Any] | None = Field(default=None, sa_column=Column(JSONB))
+    role_name: str = Field(max_length=64, unique=True, nullable=False)
+    description: str | None = Field(default=None)
+    permissions: list | None = Field(default=None, sa_column=Column(JSONB))
+    is_system: bool = Field(default=False)
     created_at: datetime = Field(default_factory=datetime.now)
     updated_at: datetime = Field(default_factory=datetime.now)
     is_active: bool = Field(default=True)
     is_deleted: bool = Field(default=False)
 
     users: list["Users"] = Relationship(back_populates="role")
+
+    @classmethod
+    def _search_filter(cls, search: str):
+        pattern = f"%{search}%"
+        return or_(cls.role_name.ilike(pattern), cls.description.ilike(pattern))
+
+    @classmethod
+    async def get_roles(
+        cls, session: AsyncSession, top: int | None, skip: int, search: str | None
+    ):
+        statement = (
+            select(cls)
+            .where(cls.is_deleted == False)  # noqa: E712
+            .order_by(cls.role_name.asc())
+        )
+        if search:
+            statement = statement.where(cls._search_filter(search))
+        if skip:
+            statement = statement.offset(skip)
+        if top is not None:
+            statement = statement.limit(top)
+        result = await session.execute(statement)
+        return result.scalars().all()
+
+    @classmethod
+    async def get_role_by_id(cls, session: AsyncSession, record_id: int):
+        statement = select(cls).where(cls.id == record_id)
+        result = await session.execute(statement)
+        return result.scalars().first()
+
+    @classmethod
+    async def get_role_by_name(cls, session: AsyncSession, role_name: str):
+        statement = select(cls).where(
+            cls.role_name == role_name, cls.is_deleted == False  # noqa: E712
+        )
+        result = await session.execute(statement)
+        return result.scalars().first()
+
+    @classmethod
+    async def count_roles(cls, session: AsyncSession, search: str | None):
+        statement = (
+            select(func.count())
+            .select_from(cls)
+            .where(cls.is_deleted == False)  # noqa: E712
+        )
+        if search:
+            statement = statement.where(cls._search_filter(search))
+        result = await session.execute(statement)
+        return result.scalar_one()
+
+    @classmethod
+    async def insert_role(cls, session: AsyncSession, fields: dict):
+        row = cls(**fields)
+        session.add(row)
+        await session.commit()
+        return await cls.get_role_by_id(session, row.id)
+
+    @classmethod
+    async def update_role(cls, session: AsyncSession, record_id: int, fields: dict):
+        row = await cls.get_role_by_id(session, record_id)
+        if not row:
+            return None
+        for key, value in fields.items():
+            setattr(row, key, value)
+        row.updated_at = datetime.now()
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+    @classmethod
+    async def soft_delete_role(cls, session: AsyncSession, record_id: int):
+        row = await cls.get_role_by_id(session, record_id)
+        if not row:
+            return None
+        row.is_deleted = True
+        row.is_active = False
+        row.updated_at = datetime.now()
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+    @classmethod
+    async def resolve_tags(cls, session: AsyncSession, role: "Roles | None") -> tuple[str, ...]:
+        """roles.permissions[] → permissions.permission_tags[] → permission_tags.tag_name.
+
+        Dangling / inactive ids contribute nothing (deny, never error). NULL or [] denies all.
+        """
+        if role is None or not role.is_active or role.is_deleted:
+            return ()
+        perm_ids = role.permissions
+        if not perm_ids or not isinstance(perm_ids, list):
+            return ()
+        bundles = await Permissions.get_permissions_by_ids(session, [int(i) for i in perm_ids])
+        tag_ids: list[int] = []
+        for bundle in bundles:
+            raw = bundle.permission_tags
+            if not raw or not isinstance(raw, list):
+                continue
+            tag_ids.extend(int(i) for i in raw)
+        if not tag_ids:
+            return ()
+        tags = await PermissionTags.get_permission_tags_by_ids(session, tag_ids)
+        # Stable unique order by tag id (seed order), then name as tiebreaker.
+        ordered = sorted(tags, key=lambda t: (t.id or 0, t.tag_name))
+        seen: set[str] = set()
+        names: list[str] = []
+        for tag in ordered:
+            if tag.tag_name not in seen:
+                seen.add(tag.tag_name)
+                names.append(tag.tag_name)
+        return tuple(names)
+
+
+# Register Users so Roles.users Relationship can resolve (safe under circular import).
+import users.models as _users_models  # noqa: E402, F401

backend/role/serializers.py

@@ -0,0 +1,55 @@
+from role.models import PermissionTags, Permissions, Roles
+
+
+def serialize_permission_tag(tag: PermissionTags) -> dict:
+    return {
+        "id": tag.id,
+        "tag_name": tag.tag_name,
+        "module": tag.module,
+        "action": tag.action,
+        "description": tag.description,
+        "is_active": tag.is_active,
+        "is_deleted": tag.is_deleted,
+        "created_at": tag.created_at.isoformat() if tag.created_at else None,
+        "updated_at": tag.updated_at.isoformat() if tag.updated_at else None,
+    }
+
+
+def serialize_permission(
+    permission: Permissions,
+    *,
+    tag_names: list[str] | None = None,
+) -> dict:
+    return {
+        "id": permission.id,
+        "name": permission.name,
+        "description": permission.description,
+        "permission_tags": list(permission.permission_tags or []),
+        "tag_names": list(tag_names or []),
+        "is_system": permission.is_system,
+        "is_active": permission.is_active,
+        "is_deleted": permission.is_deleted,
+        "created_at": permission.created_at.isoformat() if permission.created_at else None,
+        "updated_at": permission.updated_at.isoformat() if permission.updated_at else None,
+    }
+
+
+def serialize_role(
+    role: Roles,
+    *,
+    bundles: list[dict] | None = None,
+    permissions: list[str] | None = None,
+) -> dict:
+    return {
+        "id": role.id,
+        "role_name": role.role_name,
+        "description": role.description,
+        "permissions": list(role.permissions or []),
+        "bundles": list(bundles or []),
+        "effective_permissions": list(permissions or []),
+        "is_system": role.is_system,
+        "is_active": role.is_active,
+        "is_deleted": role.is_deleted,
+        "created_at": role.created_at.isoformat() if role.created_at else None,
+        "updated_at": role.updated_at.isoformat() if role.updated_at else None,
+    }

backend/role/views.py

@@ -0,0 +1,161 @@
+from fastapi import HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from role.models import PermissionTags, Permissions, Roles
+from role.serializers import serialize_permission, serialize_permission_tag, serialize_role
+
+
+class Role:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def _bundle_payload(self, permission: Permissions) -> dict:
+        tag_ids = [int(i) for i in (permission.permission_tags or [])]
+        tags = await PermissionTags.get_permission_tags_by_ids(self.session, tag_ids)
+        by_id = {t.id: t.tag_name for t in tags}
+        tag_names = [by_id[i] for i in tag_ids if i in by_id]
+        return serialize_permission(permission, tag_names=tag_names)
+
+    async def _role_payload(self, role: Roles) -> dict:
+        perm_ids = [int(i) for i in (role.permissions or [])]
+        bundles_orm = await Permissions.get_permissions_by_ids(self.session, perm_ids)
+        by_id = {b.id: b for b in bundles_orm}
+        bundles = []
+        for pid in perm_ids:
+            bundle = by_id.get(pid)
+            if bundle is not None:
+                bundles.append(await self._bundle_payload(bundle))
+        tags = await Roles.resolve_tags(self.session, role)
+        return serialize_role(role, bundles=bundles, permissions=list(tags))
+
+    async def get_roles(self, top, skip, search=None):
+        rows = await Roles.get_roles(self.session, top, skip, search)
+        return [await self._role_payload(r) for r in rows]
+
+    async def get_role_by_id(self, record_id):
+        role = await Roles.get_role_by_id(self.session, int(record_id))
+        if not role or role.is_deleted:
+            raise HTTPException(status_code=404, detail="Role not found")
+        return await self._role_payload(role)
+
+    async def count_roles(self, search=None):
+        return await Roles.count_roles(self.session, search)
+
+    async def create_role(self, payload):
+        name = (payload.get("role_name") or "").strip()
+        if not name:
+            raise HTTPException(status_code=400, detail="role_name is required")
+        if await Roles.get_role_by_name(self.session, name):
+            raise HTTPException(status_code=409, detail="Role name already exists")
+        fields = {
+            "role_name": name,
+            "description": payload.get("description"),
+            "permissions": list(payload.get("permissions") or []),
+            "is_system": False,
+            "is_active": payload.get("is_active", True),
+            "is_deleted": False,
+        }
+        role = await Roles.insert_role(self.session, fields)
+        return await self._role_payload(role)
+
+    async def update_role(self, record_id, payload):
+        role = await Roles.get_role_by_id(self.session, int(record_id))
+        if not role or role.is_deleted:
+            raise HTTPException(status_code=404, detail="Role not found")
+        fields = {}
+        if "role_name" in payload and payload["role_name"] is not None:
+            new_name = payload["role_name"].strip()
+            if role.is_system and new_name != role.role_name:
+                raise HTTPException(status_code=409, detail="System roles cannot be renamed")
+            if new_name != role.role_name:
+                clash = await Roles.get_role_by_name(self.session, new_name)
+                if clash:
+                    raise HTTPException(status_code=409, detail="Role name already exists")
+            fields["role_name"] = new_name
+        if "description" in payload and payload["description"] is not None:
+            fields["description"] = payload["description"]
+        if "permissions" in payload and payload["permissions"] is not None:
+            fields["permissions"] = list(payload["permissions"])
+        if "is_active" in payload and payload["is_active"] is not None:
+            fields["is_active"] = payload["is_active"]
+        updated = await Roles.update_role(self.session, int(record_id), fields)
+        return await self._role_payload(updated)
+
+    async def delete_role(self, record_id):
+        role = await Roles.get_role_by_id(self.session, int(record_id))
+        if not role or role.is_deleted:
+            raise HTTPException(status_code=404, detail="Role not found")
+        if role.is_system:
+            raise HTTPException(status_code=409, detail="System roles cannot be deleted")
+        deleted = await Roles.soft_delete_role(self.session, int(record_id))
+        return await self._role_payload(deleted)
+
+    async def get_permissions(self, top, skip, search=None):
+        rows = await Permissions.get_permissions(self.session, top, skip, search)
+        return [await self._bundle_payload(r) for r in rows]
+
+    async def get_permission_by_id(self, record_id):
+        row = await Permissions.get_permission_by_id(self.session, int(record_id))
+        if not row or row.is_deleted:
+            raise HTTPException(status_code=404, detail="Permission bundle not found")
+        return await self._bundle_payload(row)
+
+    async def count_permissions(self, search=None):
+        return await Permissions.count_permissions(self.session, search)
+
+    async def create_permission(self, payload):
+        name = (payload.get("name") or "").strip()
+        if not name:
+            raise HTTPException(status_code=400, detail="name is required")
+        if await Permissions.get_permission_by_name(self.session, name):
+            raise HTTPException(status_code=409, detail="Permission bundle name already exists")
+        fields = {
+            "name": name,
+            "description": payload.get("description"),
+            "permission_tags": list(payload.get("permission_tags") or []),
+            "is_system": False,
+            "is_active": payload.get("is_active", True),
+            "is_deleted": False,
+        }
+        row = await Permissions.insert_permission(self.session, fields)
+        return await self._bundle_payload(row)
+
+    async def update_permission(self, record_id, payload):
+        row = await Permissions.get_permission_by_id(self.session, int(record_id))
+        if not row or row.is_deleted:
+            raise HTTPException(status_code=404, detail="Permission bundle not found")
+        fields = {}
+        if "name" in payload and payload["name"] is not None:
+            new_name = payload["name"].strip()
+            if row.is_system and new_name != row.name:
+                raise HTTPException(
+                    status_code=409, detail="System permission bundles cannot be renamed"
+                )
+            if new_name != row.name:
+                clash = await Permissions.get_permission_by_name(self.session, new_name)
+                if clash:
+                    raise HTTPException(
+                        status_code=409, detail="Permission bundle name already exists"
+                    )
+            fields["name"] = new_name
+        if "description" in payload and payload["description"] is not None:
+            fields["description"] = payload["description"]
+        if "permission_tags" in payload and payload["permission_tags"] is not None:
+            fields["permission_tags"] = list(payload["permission_tags"])
+        if "is_active" in payload and payload["is_active"] is not None:
+            fields["is_active"] = payload["is_active"]
+        updated = await Permissions.update_permission(self.session, int(record_id), fields)
+        return await self._bundle_payload(updated)
+
+    async def get_permission_tags(self, top, skip, search=None):
+        rows = await PermissionTags.get_permission_tags(self.session, top, skip, search)
+        return [serialize_permission_tag(r) for r in rows]
+
+    async def get_permission_tag_by_id(self, record_id):
+        row = await PermissionTags.get_permission_tag_by_id(self.session, int(record_id))
+        if not row or row.is_deleted:
+            raise HTTPException(status_code=404, detail="Permission tag not found")
+        return serialize_permission_tag(row)
+
+    async def count_permission_tags(self, search=None):
+        return await PermissionTags.count_permission_tags(self.session, search)

backend/users/app.py

@@ -5,7 +5,7 @@ from db_setup import get_session
 from sqlalchemy.ext.asyncio import AsyncSession
 from pydantic import BaseModel, EmailStr, model_validator
 from users.views import User
-from users.permissions import CurrentUser
+from users.permissions import CurrentUser, PermissionTag, require_permission
 from users.serializers import serialize_token
 from users.plugins import create_access_token,create_refresh_token
 from dotenv import load_dotenv
@@ -83,7 +83,11 @@ async def me(current_user: CurrentUser):
 
 
 @router.post("/users/create")
-async def create_user(payload: UserCreate,session: AsyncSession = Depends(get_session)):
+async def create_user(
+    payload: UserCreate,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_CREATE)),
+    session: AsyncSession = Depends(get_session),
+):
     try:
         service=User(session=session)
         user=await service.create_user(payload.model_dump())
@@ -97,7 +101,7 @@ async def create_user(payload: UserCreate,session: AsyncSession = Depends(get_se
 
 @router.get("/users/fetch")
 async def fetch_users(
-    current_user: CurrentUser,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
     record_id: str | None = Query(None),
     search: str | None = Query(None),
     top: int | None = Query(None),
@@ -120,7 +124,12 @@ async def fetch_users(
 
 
 @router.put("/users/update")
-async def update_user(payload: UserUpdate,current_user: CurrentUser,record_id: str = Query(...),session: AsyncSession = Depends(get_session)):
+async def update_user(
+    payload: UserUpdate,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_EDIT)),
+    record_id: str = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
     try:
         service=User(session=session)
         data=await service.update_user(record_id,payload.model_dump())
@@ -132,7 +141,11 @@ async def update_user(payload: UserUpdate,current_user: CurrentUser,record_id: s
 
 
 @router.delete("/users/delete")
-async def delete_user(current_user: CurrentUser,record_id: str = Query(...),session: AsyncSession = Depends(get_session)):
+async def delete_user(
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_DELETE)),
+    record_id: str = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
     try:
         service=User(session=session)
         data=await service.delete_user(record_id)

backend/users/models.py

@@ -68,7 +68,8 @@ class Users(SQLModel, table=True):
 
     @classmethod
     async def get_user_by_email(cls, session: AsyncSession, email: str):
-        result = await session.execute(select(cls).where(cls.email == email))
+        statement = select(cls).options(selectinload(cls.role)).where(cls.email == email)
+        result = await session.execute(statement)
         return result.scalars().first()
 
     @classmethod

backend/users/permissions.py

@@ -1,7 +1,13 @@
-"""HTTP Bearer scheme and the current-user dependency for `/users/*` routes."""
+"""HTTP Bearer scheme, current-user dependency, and RBAC enforcement.
+
+PermissionTag is a str Enum of every "module.action" tag. With class PermissionTag(str, Enum),
+f"{PermissionTag.JOBS_VIEW}" renders "PermissionTag.JOBS_VIEW" — always use .value in JSON
+and HTTPException details.
+"""
 
 from __future__ import annotations
 
+from enum import Enum
 from typing import Annotated
 
 import jwt
@@ -10,6 +16,7 @@ from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from db_setup import get_session
+from role.models import Roles
 from users.models import Users
 from users.plugins import decode_token
 from users.serializers import serialize_user
@@ -17,6 +24,196 @@ from users.serializers import serialize_user
 bearer_scheme = HTTPBearer()
 
 
+class PermissionModule(str, Enum):
+    DASHBOARD = "dashboard"
+    INBOX = "inbox"
+    JOBS = "jobs"
+    CANDIDATES = "candidates"
+    PIPELINE = "pipeline"
+    INTERVIEWS = "interviews"
+    ASSESSMENTS = "assessments"
+    OFFERS = "offers"
+    REPORTS = "reports"
+    ANALYTICS = "analytics"
+    JOB_BOARD = "job_board"
+    SETTINGS = "settings"
+    RBAC_USERS = "rbac_users"
+
+
+class PermissionAction(str, Enum):
+    VIEW = "view"
+    CREATE = "create"
+    EDIT = "edit"
+    DELETE = "delete"
+    APPROVE = "approve"
+    EXPORT = "export"
+    MANAGE = "manage"
+    CONFIGURE = "configure"
+
+
+class PermissionTag(str, Enum):
+    DASHBOARD_VIEW = "dashboard.view"
+    DASHBOARD_CREATE = "dashboard.create"
+    DASHBOARD_EDIT = "dashboard.edit"
+    DASHBOARD_DELETE = "dashboard.delete"
+    DASHBOARD_APPROVE = "dashboard.approve"
+    DASHBOARD_EXPORT = "dashboard.export"
+    DASHBOARD_MANAGE = "dashboard.manage"
+    DASHBOARD_CONFIGURE = "dashboard.configure"
+    INBOX_VIEW = "inbox.view"
+    INBOX_CREATE = "inbox.create"
+    INBOX_EDIT = "inbox.edit"
+    INBOX_DELETE = "inbox.delete"
+    INBOX_APPROVE = "inbox.approve"
+    INBOX_EXPORT = "inbox.export"
+    INBOX_MANAGE = "inbox.manage"
+    INBOX_CONFIGURE = "inbox.configure"
+    JOBS_VIEW = "jobs.view"
+    JOBS_CREATE = "jobs.create"
+    JOBS_EDIT = "jobs.edit"
+    JOBS_DELETE = "jobs.delete"
+    JOBS_APPROVE = "jobs.approve"
+    JOBS_EXPORT = "jobs.export"
+    JOBS_MANAGE = "jobs.manage"
+    JOBS_CONFIGURE = "jobs.configure"
+    CANDIDATES_VIEW = "candidates.view"
+    CANDIDATES_CREATE = "candidates.create"
+    CANDIDATES_EDIT = "candidates.edit"
+    CANDIDATES_DELETE = "candidates.delete"
+    CANDIDATES_APPROVE = "candidates.approve"
+    CANDIDATES_EXPORT = "candidates.export"
+    CANDIDATES_MANAGE = "candidates.manage"
+    CANDIDATES_CONFIGURE = "candidates.configure"
+    PIPELINE_VIEW = "pipeline.view"
+    PIPELINE_CREATE = "pipeline.create"
+    PIPELINE_EDIT = "pipeline.edit"
+    PIPELINE_DELETE = "pipeline.delete"
+    PIPELINE_APPROVE = "pipeline.approve"
+    PIPELINE_EXPORT = "pipeline.export"
+    PIPELINE_MANAGE = "pipeline.manage"
+    PIPELINE_CONFIGURE = "pipeline.configure"
+    INTERVIEWS_VIEW = "interviews.view"
+    INTERVIEWS_CREATE = "interviews.create"
+    INTERVIEWS_EDIT = "interviews.edit"
+    INTERVIEWS_DELETE = "interviews.delete"
+    INTERVIEWS_APPROVE = "interviews.approve"
+    INTERVIEWS_EXPORT = "interviews.export"
+    INTERVIEWS_MANAGE = "interviews.manage"
+    INTERVIEWS_CONFIGURE = "interviews.configure"
+    ASSESSMENTS_VIEW = "assessments.view"
+    ASSESSMENTS_CREATE = "assessments.create"
+    ASSESSMENTS_EDIT = "assessments.edit"
+    ASSESSMENTS_DELETE = "assessments.delete"
+    ASSESSMENTS_APPROVE = "assessments.approve"
+    ASSESSMENTS_EXPORT = "assessments.export"
+    ASSESSMENTS_MANAGE = "assessments.manage"
+    ASSESSMENTS_CONFIGURE = "assessments.configure"
+    OFFERS_VIEW = "offers.view"
+    OFFERS_CREATE = "offers.create"
+    OFFERS_EDIT = "offers.edit"
+    OFFERS_DELETE = "offers.delete"
+    OFFERS_APPROVE = "offers.approve"
+    OFFERS_EXPORT = "offers.export"
+    OFFERS_MANAGE = "offers.manage"
+    OFFERS_CONFIGURE = "offers.configure"
+    REPORTS_VIEW = "reports.view"
+    REPORTS_CREATE = "reports.create"
+    REPORTS_EDIT = "reports.edit"
+    REPORTS_DELETE = "reports.delete"
+    REPORTS_APPROVE = "reports.approve"
+    REPORTS_EXPORT = "reports.export"
+    REPORTS_MANAGE = "reports.manage"
+    REPORTS_CONFIGURE = "reports.configure"
+    ANALYTICS_VIEW = "analytics.view"
+    ANALYTICS_CREATE = "analytics.create"
+    ANALYTICS_EDIT = "analytics.edit"
+    ANALYTICS_DELETE = "analytics.delete"
+    ANALYTICS_APPROVE = "analytics.approve"
+    ANALYTICS_EXPORT = "analytics.export"
+    ANALYTICS_MANAGE = "analytics.manage"
+    ANALYTICS_CONFIGURE = "analytics.configure"
+    JOB_BOARD_VIEW = "job_board.view"
+    JOB_BOARD_CREATE = "job_board.create"
+    JOB_BOARD_EDIT = "job_board.edit"
+    JOB_BOARD_DELETE = "job_board.delete"
+    JOB_BOARD_APPROVE = "job_board.approve"
+    JOB_BOARD_EXPORT = "job_board.export"
+    JOB_BOARD_MANAGE = "job_board.manage"
+    JOB_BOARD_CONFIGURE = "job_board.configure"
+    SETTINGS_VIEW = "settings.view"
+    SETTINGS_CREATE = "settings.create"
+    SETTINGS_EDIT = "settings.edit"
+    SETTINGS_DELETE = "settings.delete"
+    SETTINGS_APPROVE = "settings.approve"
+    SETTINGS_EXPORT = "settings.export"
+    SETTINGS_MANAGE = "settings.manage"
+    SETTINGS_CONFIGURE = "settings.configure"
+    RBAC_USERS_VIEW = "rbac_users.view"
+    RBAC_USERS_CREATE = "rbac_users.create"
+    RBAC_USERS_EDIT = "rbac_users.edit"
+    RBAC_USERS_DELETE = "rbac_users.delete"
+    RBAC_USERS_APPROVE = "rbac_users.approve"
+    RBAC_USERS_EXPORT = "rbac_users.export"
+    RBAC_USERS_MANAGE = "rbac_users.manage"
+    RBAC_USERS_CONFIGURE = "rbac_users.configure"
+
+
+def _assert_vocabulary_complete() -> None:
+    expected = {
+        f"{m.value}.{a.value}"
+        for m in PermissionModule
+        for a in PermissionAction
+    }
+    actual = {t.value for t in PermissionTag}
+    if expected != actual:
+        missing = sorted(expected - actual)
+        extra = sorted(actual - expected)
+        raise RuntimeError(
+            f"PermissionTag vocabulary drift: missing={missing!r} extra={extra!r}"
+        )
+
+
+_assert_vocabulary_complete()
+
+
+def has_permission(
+    granted: set[str] | list[str] | tuple[str, ...],
+    *required: PermissionTag,
+    require_all: bool = True,
+) -> bool:
+    needed = {t.value for t in required}
+    have = set(granted or ())
+    if require_all:
+        return needed.issubset(have)
+    return bool(needed & have)
+
+
+def require_permission(*required: PermissionTag, require_all: bool = True):
+    """FastAPI dependency: enforce one or more PermissionTag values (AND by default)."""
+
+    async def dependency(current_user: CurrentUser) -> dict:
+        if current_user.get("role_id") is None:
+            raise HTTPException(status_code=403, detail="User has no role assigned")
+        granted = current_user.get("permissions") or []
+        if not has_permission(granted, *required, require_all=require_all):
+            if require_all and len(required) == 1:
+                detail = f"Missing required permission: {required[0].value}"
+            elif require_all:
+                detail = (
+                    "Missing required permissions: "
+                    + ", ".join(t.value for t in required)
+                )
+            else:
+                detail = (
+                    "Missing any of required permissions: "
+                    + ", ".join(t.value for t in required)
+                )
+            raise HTTPException(status_code=403, detail=detail)
+        return current_user
+
+    return dependency
+
+
 async def get_current_user(
     credentials: Annotated[HTTPAuthorizationCredentials, Depends(bearer_scheme)],
     session: Annotated[AsyncSession, Depends(get_session)],
@@ -38,8 +235,8 @@ async def get_current_user(
             detail="User is inactive or does not exist",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    return serialize_user(user)
+    permissions = await Roles.resolve_tags(session, user.role)
+    return serialize_user(user, with_permissions=True, permissions=permissions)
 
 
 CurrentUser = Annotated[dict, Depends(get_current_user)]
-

backend/users/serializers.py

@@ -2,14 +2,19 @@ from users.models import Users
 from users.plugins import ACCESS_TOKEN_EXPIRE_SECONDS
 
 
-def serialize_user(user: Users) -> dict:
+def serialize_user(
+    user: Users,
+    *,
+    with_permissions: bool = False,
+    permissions: tuple[str, ...] | list[str] | None = None,
+) -> dict:
     """users row -> the shape the #rbac Users tab renders. Never includes password."""
     role = getattr(user, "role", None)
     role_name = None
     if role is not None:
         role_name = getattr(role.role_name, "value", role.role_name)
 
-    return {
+    data = {
         "id": str(user.id),
         "name": user.name,
         "email": user.email,
@@ -21,6 +26,9 @@ def serialize_user(user: Users) -> dict:
         "created_at": user.created_at.isoformat() if user.created_at else None,
         "updated_at": user.updated_at.isoformat() if user.updated_at else None,
     }
+    if with_permissions:
+        data["permissions"] = list(permissions or ())
+    return data
 
 
 def serialize_token(access_token: str, refresh_token: str, user: Users) -> dict: