# app/modules/tenancy/routes/api/store_team.py """ Store team member management endpoints. Implements complete team management with: - Team member listing - Invitation system - Role management - Permission checking - RBAC integration """ import logging from fastapi import APIRouter, Depends, Request from sqlalchemy.orm import Session from app.api.deps import ( get_current_store_api, get_user_permissions, require_store_owner, require_store_permission, ) from app.core.database import get_db from app.modules.tenancy.schemas.auth import UserContext from app.modules.tenancy.schemas.team import ( BulkRemoveRequest, BulkRemoveResponse, InvitationAccept, InvitationAcceptResponse, InvitationResponse, PermissionCatalogResponse, RoleCreate, RoleListResponse, RoleResponse, RoleUpdate, TeamMemberInvite, TeamMemberListResponse, TeamMemberResponse, TeamMemberUpdate, TeamStatistics, UserPermissionsResponse, ) # Permission IDs are now defined in module definition.py files # and discovered by PermissionDiscoveryService from app.modules.tenancy.services.permission_discovery_service import ( permission_discovery_service, ) from app.modules.tenancy.services.store_team_service import store_team_service from app.utils.i18n import translate store_team_router = APIRouter(prefix="/team") logger = logging.getLogger(__name__) # ============================================================================ # Team Member Routes # ============================================================================ @store_team_router.get("/members", response_model=TeamMemberListResponse) def list_team_members( request: Request, include_inactive: bool = False, db: Session = Depends(get_db), current_user: UserContext = Depends( require_store_permission("team.view") ), ): """ Get all team members for current store. **Required Permission:** `team.view` **Query Parameters:** - `include_inactive`: Include inactive team members (default: False) **Returns:** - List of team members with their roles and permissions - Statistics (total, active, pending) """ store = request.state.store members = store_team_service.get_team_members( db=db, store=store, include_inactive=include_inactive ) # Calculate statistics total = len(members) active = sum(1 for m in members if m["is_active"]) pending = sum(1 for m in members if m["invitation_pending"]) logger.info( f"Listed {total} team members for store {store.store_code} " f"(active: {active}, pending: {pending})" ) return TeamMemberListResponse( members=members, total=total, active_count=active, pending_invitations=pending ) @store_team_router.post("/invite", response_model=InvitationResponse) def invite_team_member( invitation: TeamMemberInvite, request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), # Owner only ): """ Invite a new team member to the store. **Required:** Store owner role **Process:** 1. Create user account (if doesn't exist) 2. Create StoreUser with invitation token 3. Send invitation email **Request Body:** - `email`: Email address of invitee - `first_name`, `last_name`: Optional names - `role_name`: Preset role (manager, staff, support, viewer, marketing) - `role_id`: Use existing role (alternative to role_name) - `custom_permissions`: Override role permissions (requires role_name) **Returns:** - Invitation details - Confirmation of email sent """ store = request.state.store # Determine role approach if invitation.role_id: # Use existing role by ID result = store_team_service.invite_team_member( db=db, store=store, inviter=current_user, email=invitation.email, role_id=invitation.role_id, first_name=invitation.first_name, last_name=invitation.last_name, ) elif invitation.role_name: # Use role name with optional custom permissions result = store_team_service.invite_team_member( db=db, store=store, inviter=current_user, email=invitation.email, role_name=invitation.role_name, first_name=invitation.first_name, last_name=invitation.last_name, custom_permissions=invitation.custom_permissions, ) else: # Default to Staff role result = store_team_service.invite_team_member( db=db, store=store, inviter=current_user, email=invitation.email, role_name="staff", ) db.commit() logger.info( f"Invitation sent: {invitation.email} to {store.store_code} " f"by {current_user.username}" ) return InvitationResponse( message="Invitation sent successfully", email=result["email"], role=result["role"], invitation_sent=True, ) @store_team_router.post("/accept-invitation", response_model=InvitationAcceptResponse) # public def accept_invitation(acceptance: InvitationAccept, db: Session = Depends(get_db)): """ Accept a team invitation and activate account. **No authentication required** - uses invitation token. **Request Body:** - `invitation_token`: Token from invitation email - `password`: New password (min 8 chars, must have upper, lower, digit) - `first_name`, `last_name`: User's name **Returns:** - Confirmation message - Store information - User information - Assigned role """ result = store_team_service.accept_invitation( db=db, invitation_token=acceptance.invitation_token, password=acceptance.password, first_name=acceptance.first_name, last_name=acceptance.last_name, ) db.commit() logger.info( f"Invitation accepted: {result['user'].email} " f"for store {result['store'].store_code}" ) return InvitationAcceptResponse( message="Invitation accepted successfully. You can now login.", store={ "id": result["store"].id, "store_code": result["store"].store_code, "name": result["store"].name, "subdomain": result["store"].subdomain, }, user={ "id": result["user"].id, "email": result["user"].email, "username": result["user"].username, "full_name": result["user"].full_name, }, role=result["role"], ) @store_team_router.get("/members/{user_id}", response_model=TeamMemberResponse) def get_team_member( user_id: int, request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends( require_store_permission("team.view") ), ): """ Get details of a specific team member. **Required Permission:** `team.view` """ store = request.state.store members = store_team_service.get_team_members( db=db, store=store, include_inactive=True ) member = next((m for m in members if m["id"] == user_id), None) if not member: from app.modules.tenancy.exceptions import UserNotFoundException raise UserNotFoundException(str(user_id)) return TeamMemberResponse(**member) @store_team_router.put("/members/{user_id}", response_model=TeamMemberResponse) def update_team_member( user_id: int, update_data: TeamMemberUpdate, request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), # Owner only ): """ Update a team member's role or status. **Required:** Store owner role **Cannot:** - Change owner's role - Remove owner **Request Body:** - `role_id`: New role ID (optional) - `is_active`: Active status (optional) """ store = request.state.store store_team_service.update_member( db=db, store=store, user_id=user_id, role_id=update_data.role_id, is_active=update_data.is_active, actor_user_id=current_user.id, ) db.commit() logger.info( f"Team member updated: {user_id} in {store.store_code} " f"by {current_user.username}" ) # Return updated member details members = store_team_service.get_team_members(db, store, include_inactive=True) member = next((m for m in members if m["id"] == user_id), None) return TeamMemberResponse(**member) @store_team_router.post("/members/{user_id}/resend") def resend_team_invitation( user_id: int, request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), ): """ Resend invitation to a pending team member. **Required:** Store owner role """ store = request.state.store result = store_team_service.resend_invitation( db, store=store, user_id=user_id, inviter=current_user, ) db.commit() return result @store_team_router.delete("/members/{user_id}") def remove_team_member( user_id: int, request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), # Owner only ): """ Remove a team member from the store. **Required:** Store owner role **Cannot remove:** - Store owner **Action:** - Soft delete (sets is_active = False) - Member can be re-invited later """ store = request.state.store store_team_service.remove_team_member(db=db, store=store, user_id=user_id) db.commit() logger.info( f"Team member removed: {user_id} from {store.store_code} " f"by {current_user.username}" ) return {"message": "Team member removed successfully", "user_id": user_id} @store_team_router.post("/members/bulk-remove", response_model=BulkRemoveResponse) def bulk_remove_team_members( bulk_remove: BulkRemoveRequest, request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), ): """ Remove multiple team members at once. **Required:** Store owner role """ store = request.state.store success_count = 0 failed_count = 0 errors = [] for user_id in bulk_remove.user_ids: try: store_team_service.remove_team_member( db=db, store=store, user_id=user_id ) success_count += 1 except Exception as e: failed_count += 1 errors.append({"user_id": user_id, "error": str(e)}) db.commit() logger.info( f"Bulk remove completed: {success_count} removed, {failed_count} failed " f"in {store.store_code}" ) return BulkRemoveResponse( success_count=success_count, failed_count=failed_count, errors=errors ) # ============================================================================ # Role Management Routes # ============================================================================ @store_team_router.get("/roles", response_model=RoleListResponse) def list_roles( request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends( require_store_permission("team.view") ), ): """ Get all available roles for the store. **Required Permission:** `team.view` **Returns:** - List of roles with permissions - Includes both preset and custom roles """ store = request.state.store roles = store_team_service.get_store_roles(db=db, store_id=store.id) db.commit() # Commit in case default roles were created return RoleListResponse(roles=roles, total=len(roles)) @store_team_router.post("/roles", response_model=RoleResponse, status_code=201) def create_role( request: Request, role_data: RoleCreate, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), ): """ Create a custom role for the store. **Required:** Store owner only. Preset role names (manager, staff, support, viewer, marketing) cannot be used. """ store = request.state.store role = store_team_service.create_custom_role( db=db, store_id=store.id, name=role_data.name, permissions=role_data.permissions, actor_user_id=current_user.id, ) db.commit() return role @store_team_router.put("/roles/{role_id}", response_model=RoleResponse) def update_role( request: Request, role_id: int, role_data: RoleUpdate, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), ): """ Update a role's name and/or permissions. **Required:** Store owner only. """ store = request.state.store role = store_team_service.update_role( db=db, store_id=store.id, role_id=role_id, name=role_data.name, permissions=role_data.permissions, actor_user_id=current_user.id, ) db.commit() return role @store_team_router.delete("/roles/{role_id}", status_code=204) def delete_role( request: Request, role_id: int, db: Session = Depends(get_db), current_user: UserContext = Depends(require_store_owner), ): """ Delete a custom role. **Required:** Store owner only. Preset roles cannot be deleted. Roles with assigned team members cannot be deleted. """ store = request.state.store store_team_service.delete_role( db=db, store_id=store.id, role_id=role_id, actor_user_id=current_user.id, ) db.commit() # ============================================================================ # Permission Routes # ============================================================================ @store_team_router.get( "/permissions/catalog", response_model=PermissionCatalogResponse ) def get_permission_catalog( request: Request, current_user: UserContext = Depends( require_store_permission("team.view") ), ): """ Get the full permission catalog grouped by category. **Required Permission:** `team.view` Returns all available permissions with labels and descriptions, grouped by category. Used by the role editor UI for displaying permission checkboxes with human-readable names and tooltips. """ categories = permission_discovery_service.get_permissions_by_category() lang = current_user.preferred_language or getattr( request.state, "language", "en" ) def _t(key: str) -> str: """Translate key, falling back to readable version.""" translated = translate(key, language=lang) if translated == key: parts = key.split(".") return parts[-1].replace("_", " ").title() return translated return PermissionCatalogResponse( categories=[ { "id": cat.id, "label": _t(cat.label_key), "permissions": [ { "id": p.id, "label": _t(p.label_key), "description": _t(p.description_key), "is_owner_only": p.is_owner_only, } for p in cat.permissions ], } for cat in categories ] ) @store_team_router.get("/me/permissions", response_model=UserPermissionsResponse) def get_my_permissions( request: Request, permissions: list[str] = Depends(get_user_permissions), current_user: UserContext = Depends(get_current_store_api), ): """ Get current user's permissions in this store. **Use this endpoint to:** - Determine what UI elements to show/hide - Check permissions before making API calls - Display user's role and capabilities **Returns:** - Complete list of permissions - Whether user is owner - Role name (if team member) Requires Authorization header (API endpoint). """ store = request.state.store is_owner = current_user.is_owner_of(store.id) role_name = current_user.get_store_role(store.id) return UserPermissionsResponse( permissions=permissions, permission_count=len(permissions), is_owner=is_owner, role_name=role_name, ) # ============================================================================ # Statistics Routes # ============================================================================ @store_team_router.get("/statistics", response_model=TeamStatistics) def get_team_statistics( request: Request, db: Session = Depends(get_db), current_user: UserContext = Depends( require_store_permission("team.view") ), ): """ Get team statistics for the store. **Required Permission:** `team.view` **Returns:** - Total members - Active/inactive breakdown - Pending invitations - Owner count - Role distribution """ store = request.state.store members = store_team_service.get_team_members( db=db, store=store, include_inactive=True ) # Calculate statistics total = len(members) active = sum(1 for m in members if m["is_active"]) inactive = total - active pending = sum(1 for m in members if m["invitation_pending"]) owners = sum(1 for m in members if m["is_owner"]) team_members = total - owners # Role breakdown roles_breakdown = {} for member in members: role = member["role_name"] roles_breakdown[role] = roles_breakdown.get(role, 0) + 1 return TeamStatistics( total_members=total, active_members=active, inactive_members=inactive, pending_invitations=pending, owners=owners, team_members=team_members, roles_breakdown=roles_breakdown, )