""" Team / RBAC Router — PRD-008 RBAC Multi-User Merchant Accounts. Mounted at /api/v1/team in v1.py. """ from __future__ import annotations from typing import Any, List from fastapi import APIRouter, Depends, status from sqlalchemy.orm import Session from src.core.database import get_db from src.apps.auth.utils.auth import get_current_active_user, get_current_merchant from src.apps.role_permissions import crud, services from src.apps.role_permissions.dependencies import require_permission from src.apps.role_permissions.schemas.team_schemas import ( AcceptInviteRequest, AcceptInviteResponse, ChangeMemberRoleRequest, CreateRoleRequest, InviteInfoResponse, InviteSchema, InviteUserRequest, MyPermissionsResponse, PermissionSchema, RoleMinimalSchema, RoleSchema, TeamMemberSchema, TransferOwnershipRequest, UpdateRoleRequest, ) from src.core.exceptions import APIException, NotFoundError router = APIRouter() # ─── Helpers ───────────────────────────────────────────────────────────────── def _role_to_schema(role) -> RoleSchema: return RoleSchema( id=role.id, label=role.label, slug=role.slug, is_system=role.is_system, is_default=role.is_default, role_rank=role.role_rank, description=role.description, permissions=[ PermissionSchema( id=p.id, module=p.module, submodule=p.submodule, operation=p.operation, slug=p.slug, operation_label=p.operation_label, display_order=p.display_order, ) for p in (role.permissions or []) ], member_count=None, ) def _caller_rank(current_user, merchant, db: Session) -> int: """Return the caller's role_rank for this merchant context. Superusers get max rank.""" if getattr(current_user, "is_superuser", False): return 99 ur = crud.get_user_role(db, current_user.id, merchant.id) if ur and ur.role: return ur.role.role_rank return 0 def _require_membership(current_user, merchant, db: Session): """Raise 403 if the caller is not a member of this merchant.""" if getattr(current_user, "is_superuser", False): return ur = crud.get_user_role(db, current_user.id, merchant.id) if not ur: raise APIException(status_code=403, message="You are not a member of this merchant.") # ─── Team members ───────────────────────────────────────────────────────────── @router.get( "/members", response_model=List[TeamMemberSchema], summary="List team members", dependencies=[Depends(require_permission("users:view"))], ) async def get_team_members( current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> List[TeamMemberSchema]: caller_rank = _caller_rank(current_user, merchant, db) members = services.get_members(db, merchant_id=merchant.id, caller_rank=caller_rank) result = [] for ur in members: user = ur.user if not user: continue result.append( TeamMemberSchema( user_id=user.id, email=user.email, first_name=user.first_name, last_name=user.last_name, full_name=user.full_name or f"{user.first_name or ''} {user.last_name or ''}".strip(), role=_role_to_schema(ur.role), is_owner=(ur.role.role_rank == 3), joined_at=ur.created_at, ) ) return result @router.put( "/members/{user_id}/role", summary="Change a member's role", dependencies=[Depends(require_permission("users:manage_admins"))], ) async def change_member_role( user_id: int, body: ChangeMemberRoleRequest, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: caller_rank = _caller_rank(current_user, merchant, db) # Verify target is a member target_ur = crud.get_user_role(db, user_id, merchant.id) if not target_ur: raise NotFoundError(message="User is not a member of this merchant.") # Hierarchy check services.assert_hierarchy(caller_rank, target_ur.role.role_rank, action="change role of") # Check target role rank new_role = crud.get_role_by_id(db, body.role_id) if not new_role: raise NotFoundError(message="Role not found.") services.assert_hierarchy(caller_rank, new_role.role_rank, action="assign role") user_role = services.assign_role_to_user( db, user_id=user_id, role_id=body.role_id, merchant_id=merchant.id, ) crud.write_audit_log( db, merchant_id=merchant.id, actor_id=current_user.id, action="member.role_changed", target_user_id=user_id, target_role_id=body.role_id, ) db.commit() return { "data": {"user_id": user_id, "role_id": body.role_id}, "success": True, "message": "Role updated successfully.", } @router.delete( "/members/{user_id}", status_code=status.HTTP_200_OK, summary="Remove a team member", dependencies=[Depends(require_permission("users:remove"))], ) async def remove_member( user_id: int, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: caller_rank = _caller_rank(current_user, merchant, db) target_ur = crud.get_user_role(db, user_id, merchant.id) if not target_ur: raise NotFoundError(message="User is not a member of this merchant.") services.assert_hierarchy(caller_rank, target_ur.role.role_rank, action="remove") services.remove_member(db, merchant_id=merchant.id, user_id=user_id, actor_id=current_user.id) return {"data": None, "success": True, "message": "Member removed."} @router.post( "/transfer-ownership", summary="Transfer merchant ownership", dependencies=[Depends(require_permission("users:transfer_ownership"))], ) async def transfer_ownership( body: TransferOwnershipRequest, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: services.transfer_ownership( db, merchant_id=merchant.id, from_user_id=current_user.id, to_user_id=body.to_user_id, actor_id=current_user.id, ) return {"data": None, "success": True, "message": "Ownership transferred."} # ─── Invites ────────────────────────────────────────────────────────────────── @router.get( "/invites", response_model=List[InviteSchema], summary="List pending invites", dependencies=[Depends(require_permission("users:view"))], ) async def get_invites( current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> List[InviteSchema]: invites = services.get_pending_invites(db, merchant_id=merchant.id) result = [] for inv in invites: if not inv.is_pending: continue inviter = inv.inviter invited_by_name = ( f"{inviter.first_name or ''} {inviter.last_name or ''}".strip() if inviter else "Unknown" ) result.append( InviteSchema( id=inv.id, email=inv.email, role=_role_to_schema(inv.role), invited_by_name=invited_by_name, expires_at=inv.expires_at, created_at=inv.created_at, ) ) return result @router.post( "/invites", status_code=status.HTTP_201_CREATED, summary="Send a team invite", dependencies=[Depends(require_permission("users:invite"))], ) async def send_invite( body: InviteUserRequest, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: caller_rank = _caller_rank(current_user, merchant, db) invite, raw_token = services.create_invite( db, merchant_id=merchant.id, email=str(body.email), role_id=body.role_id, invited_by_id=current_user.id, caller_rank=caller_rank, ) # Dispatch invite email task (non-blocking; errors are swallowed) if raw_token: try: from src.apps.role_permissions.tasks import send_invite_email send_invite_email.delay(invite.id, raw_token) except Exception: pass return { "data": {"invite_id": invite.id, "email": invite.email}, "success": True, "message": "Invite sent.", } @router.delete( "/invites/{invite_id}", status_code=status.HTTP_200_OK, summary="Cancel a pending invite", dependencies=[Depends(require_permission("users:invite"))], ) async def cancel_invite( invite_id: int, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: services.cancel_invite( db, merchant_id=merchant.id, invite_id=invite_id, actor_id=current_user.id, ) return {"data": None, "success": True, "message": "Invite cancelled."} @router.post( "/invites/{invite_id}/resend", status_code=status.HTTP_200_OK, summary="Re-send a pending invite", dependencies=[Depends(require_permission("users:invite"))], ) async def resend_invite( invite_id: int, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: invite, raw_token = services.resend_invite( db, merchant_id=merchant.id, invite_id=invite_id, actor_id=current_user.id, ) if raw_token: try: from src.apps.role_permissions.tasks import send_invite_email send_invite_email.delay(invite.id, raw_token) except Exception: pass return {"data": {"invite_id": invite.id, "email": invite.email}, "success": True, "message": "Invite re-sent."} # ─── Roles ──────────────────────────────────────────────────────────────────── @router.get( "/roles", response_model=List[RoleSchema], summary="List roles", dependencies=[Depends(require_permission("users:view"))], ) async def get_roles( current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> List[RoleSchema]: caller_rank = _caller_rank(current_user, merchant, db) roles = services.get_roles_for_merchant(db, merchant_id=merchant.id, caller_rank=caller_rank) return [_role_to_schema(r) for r in roles] @router.post( "/roles", status_code=status.HTTP_201_CREATED, summary="Create a custom role", dependencies=[Depends(require_permission("users:manage_admins"))], ) async def create_role( body: CreateRoleRequest, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: role = services.create_role( db, merchant_id=merchant.id, label=body.label, permission_ids=body.permission_ids, created_by_id=current_user.id, ) crud.write_audit_log( db, merchant_id=merchant.id, actor_id=current_user.id, action="role.created", target_role_id=role.id, metadata={"label": body.label}, ) db.commit() return { "data": _role_to_schema(role).model_dump(), "success": True, "message": "Role created.", } @router.put( "/roles/{role_id}", summary="Update a custom role", dependencies=[Depends(require_permission("users:manage_admins"))], ) async def update_role( role_id: int, body: UpdateRoleRequest, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: role = services.update_role( db, merchant_id=merchant.id, role_id=role_id, label=body.label, permission_ids=body.permission_ids, ) crud.write_audit_log( db, merchant_id=merchant.id, actor_id=current_user.id, action="role.updated", target_role_id=role_id, metadata={"label": body.label}, ) db.commit() return { "data": _role_to_schema(role).model_dump(), "success": True, "message": "Role updated.", } @router.delete( "/roles/{role_id}", status_code=status.HTTP_200_OK, summary="Delete a custom role", dependencies=[Depends(require_permission("users:manage_admins"))], ) async def delete_role( role_id: int, current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> dict: services.delete_role(db, merchant_id=merchant.id, role_id=role_id) # services.delete_role commits the deletion, so the role row no longer exists. # target_role_id must be None here — the FK constraint would reject a reference # to a deleted role. The ID is preserved in metadata for audit traceability. crud.write_audit_log( db, merchant_id=merchant.id, actor_id=current_user.id, action="role.deleted", target_role_id=None, metadata={"deleted_role_id": role_id}, ) db.commit() return {"data": None, "success": True, "message": "Role deleted successfully."} # ─── Permissions ────────────────────────────────────────────────────────────── @router.get( "/permissions", response_model=List[PermissionSchema], summary="List all available permissions", dependencies=[Depends(require_permission("users:manage_admins"))], ) async def get_permissions( current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> List[PermissionSchema]: permissions = services.get_permissions_list(db) return [ PermissionSchema( id=p.id, module=p.module, submodule=p.submodule, operation=p.operation, slug=p.slug, operation_label=p.operation_label, display_order=p.display_order, ) for p in permissions ] @router.get( "/my-permissions", response_model=MyPermissionsResponse, summary="Get current user's permissions for this merchant", ) async def get_my_permissions( current_user: Any = Depends(get_current_active_user), merchant: Any = Depends(get_current_merchant), db: Session = Depends(get_db), ) -> MyPermissionsResponse: if getattr(current_user, "is_superuser", False): all_perms = crud.get_all_permissions(db) return MyPermissionsResponse( permissions=[p.slug for p in all_perms], role=RoleMinimalSchema(id=0, label="Superuser", slug="superuser", role_rank=99), ) user_role = crud.get_user_role(db, current_user.id, merchant.id) if not user_role: return MyPermissionsResponse(permissions=[], role=None) perms = services.get_user_permissions(db, current_user.id, merchant.id) role_minimal = RoleMinimalSchema( id=user_role.role.id, label=user_role.role.label, slug=user_role.role.slug, role_rank=user_role.role.role_rank, ) if user_role.role else None return MyPermissionsResponse(permissions=list(perms), role=role_minimal)