arrr-erp/app/modules/core/rbac/api.py

from __future__ import annotations

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.core.db.deps import get_common_db
from app.core.security.session_auth import require_login
from app.modules.core.audit.service import write_audit_log
from app.modules.core.iam.models import User
from app.modules.core.iam.scope import build_scope, scope_to_http, assert_can_manage_role_object
from app.modules.core.rbac.models import Permission, Role, RolePermission

router = APIRouter(prefix="/rbac", tags=["rbac"])


class RoleCreateRequest(BaseModel):
    name: str
    is_active: bool = True


class PermissionCreateRequest(BaseModel):
    code: str
    name: str
    is_active: bool = True


class RolePermissionUpdateRequest(BaseModel):
    permission_ids: list[int] = []


def _require_system_admin_scope(db: Session, current_user: User):
    scope = build_scope(db, current_user)
    if not scope.is_system_admin:
        raise HTTPException(status_code=403, detail="Only System Admin can access RBAC.")
    return scope


@router.get("/roles")
def list_roles(
    current_user: User = Depends(require_login),
    db: Session = Depends(get_common_db),
):
    _require_system_admin_scope(db, current_user)
    roles = db.execute(select(Role).order_by(Role.name)).scalars().all()
    return [{"id": r.id, "name": r.name, "is_active": r.is_active} for r in roles]


@router.post("/roles")
def create_role(
    payload: RoleCreateRequest,
    current_user: User = Depends(require_login),
    db: Session = Depends(get_common_db),
):
    _require_system_admin_scope(db, current_user)

    name = payload.name.strip()
    if not name:
        raise HTTPException(status_code=400, detail="Role name is required")

    exists = db.execute(select(Role).where(Role.name == name)).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=400, detail="Role already exists")

    role = Role(name=name, is_active=payload.is_active)
    db.add(role)
    db.commit()
    db.refresh(role)

    write_audit_log(
        db,
        action="role.create.api",
        entity_type="role",
        actor=current_user,
        entity_id=role.id,
        entity_name=role.name,
        target_tenant_id=current_user.tenant_id,
        target_branch_id=current_user.branch_id,
        details={"name": role.name, "is_active": role.is_active},
    )
    return {"status": "ok", "id": role.id}


@router.get("/permissions")
def list_permissions(
    current_user: User = Depends(require_login),
    db: Session = Depends(get_common_db),
):
    _require_system_admin_scope(db, current_user)
    permissions = db.execute(select(Permission).order_by(Permission.code)).scalars().all()
    return [{"id": p.id, "code": p.code, "name": p.name, "is_active": p.is_active} for p in permissions]


@router.post("/permissions")
def create_permission(
    payload: PermissionCreateRequest,
    current_user: User = Depends(require_login),
    db: Session = Depends(get_common_db),
):
    _require_system_admin_scope(db, current_user)

    code = payload.code.strip()
    if not code:
        raise HTTPException(status_code=400, detail="Permission code is required")

    exists = db.execute(select(Permission).where(Permission.code == code)).scalar_one_or_none()
    if exists:
        raise HTTPException(status_code=400, detail="Permission already exists")

    permission = Permission(code=code, name=payload.name.strip(), is_active=payload.is_active)
    db.add(permission)
    db.commit()
    db.refresh(permission)

    write_audit_log(
        db,
        action="permission.create.api",
        entity_type="permission",
        actor=current_user,
        entity_id=permission.id,
        entity_name=permission.code,
        target_tenant_id=current_user.tenant_id,
        target_branch_id=current_user.branch_id,
        details={"code": permission.code, "name": permission.name, "is_active": permission.is_active},
    )
    return {"status": "ok", "id": permission.id}


@router.get("/roles/{role_id}")
def role_detail(
    role_id: int,
    current_user: User = Depends(require_login),
    db: Session = Depends(get_common_db),
):
    scope = _require_system_admin_scope(db, current_user)

    role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
    if not role:
        raise HTTPException(status_code=404, detail="Role not found")

    try:
        assert_can_manage_role_object(scope, role)
    except Exception as exc:
        raise scope_to_http(exc)

    permission_ids = db.execute(
        select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
    ).scalars().all()

    return {
        "id": role.id,
        "name": role.name,
        "is_active": role.is_active,
        "permission_ids": list(permission_ids),
    }


@router.put("/roles/{role_id}/permissions")
def update_role_permissions(
    role_id: int,
    payload: RolePermissionUpdateRequest,
    current_user: User = Depends(require_login),
    db: Session = Depends(get_common_db),
):
    _require_system_admin_scope(db, current_user)

    role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
    if not role:
        raise HTTPException(status_code=404, detail="Role not found")

    old_permission_ids = db.execute(
        select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
    ).scalars().all()

    db.execute(RolePermission.__table__.delete().where(RolePermission.role_id == role.id))
    for permission_id in payload.permission_ids:
        db.add(RolePermission(role_id=role.id, permission_id=permission_id))
    db.commit()

    write_audit_log(
        db,
        action="role.permissions.update.api",
        entity_type="role",
        actor=current_user,
        entity_id=role.id,
        entity_name=role.name,
        target_tenant_id=current_user.tenant_id,
        target_branch_id=current_user.branch_id,
        details={
            "old_permission_ids": list(old_permission_ids),
            "new_permission_ids": list(payload.permission_ids),
        },
    )
    return {"status": "ok"}