from __future__ import annotations import secrets import hashlib from typing import Any from app.projects.repositories.collaboration_repository import CollaborationRepository from app.projects.schemas.collaboration import TeamResponse, InvitationResponse, MemberResponse from app.security.models import WorkspaceMembership from app.projects.errors import ( CollaborationUnauthorizedError, TeamNotFoundError, TeamMemberNotFoundError, ) class CollaborationService: def __init__(self, repository: CollaborationRepository) -> None: self.repository = repository async def list_teams(self, workspace_id: str) -> list[TeamResponse]: teams = await self.repository.list_teams(workspace_id) return [TeamResponse(id=t.id, workspace_id=t.workspace_id, name=t.name, created_at=t.created_at.isoformat()) for t in teams] async def create_team(self, workspace_id: str, name: str) -> TeamResponse: team = await self.repository.create_team(workspace_id, name) return TeamResponse(id=team.id, workspace_id=team.workspace_id, name=team.name, created_at=team.created_at.isoformat()) async def update_team(self, workspace_id: str, team_id: str, name: str) -> TeamResponse: try: team = await self.repository.update_team(workspace_id, team_id, name) except Exception: raise TeamNotFoundError(f"Team {team_id} not found.") return TeamResponse(id=team.id, workspace_id=team.workspace_id, name=team.name, created_at=team.created_at.isoformat()) async def archive_team(self, workspace_id: str, team_id: str) -> None: await self.repository.archive_team(workspace_id, team_id) async def invite_member(self, workspace_id: str, email: str, role: str) -> InvitationResponse: if role not in {"admin", "member"}: raise ValueError("Invitation role must be admin or member.") token = secrets.token_urlsafe(32) token_hash = hashlib.sha256(token.encode()).hexdigest() invitation = await self.repository.invite_member_with_hash(workspace_id, email, role, token_hash) response = InvitationResponse( id=invitation.id, workspace_id=invitation.workspace_id, email=invitation.email, role=invitation.role, status=invitation.status, expires_at=invitation.expires_at.isoformat(), created_at=invitation.created_at.isoformat() ) setattr(response, "token", token) return response async def list_members(self, workspace_id: str) -> list[MemberResponse]: members = await self.repository.list_members(workspace_id) return [MemberResponse(id=m.id, workspace_id=m.workspace_id, user_id=m.user_id, role=m.role, created_at=m.created_at.isoformat()) for m in members] async def list_invitations(self, workspace_id: str) -> list[InvitationResponse]: invitations = await self.repository.list_invitations(workspace_id) return [InvitationResponse( id=i.id, workspace_id=i.workspace_id, email=i.email, role=i.role, status=i.status, expires_at=i.expires_at.isoformat(), created_at=i.created_at.isoformat() ) for i in invitations] async def remove_member(self, workspace_id: str, actor_user_id: str, target_user_id: str) -> None: # Check permissions: actor must be admin actor_membership = await self.repository.get_membership(workspace_id, actor_user_id) if not actor_membership or actor_membership.role != 'admin': raise CollaborationUnauthorizedError("Unauthorized: Only admins can remove members.") # Prevent removing last admin if actor_membership.user_id == target_user_id: raise CollaborationUnauthorizedError("Cannot remove yourself.") target_membership = await self.repository.get_membership(workspace_id, target_user_id) if not target_membership: raise TeamMemberNotFoundError("Member not found.") if target_membership.role == 'admin': admin_count = await self.repository.count_workspace_admins(workspace_id) if admin_count <= 1: raise CollaborationUnauthorizedError("Cannot remove the last administrator.") # Perform removal await self.repository.remove_member(workspace_id, target_user_id) async def update_member_role(self, workspace_id: str, actor_user_id: str, target_user_id: str, new_role: str) -> None: # Check permissions: actor must be admin actor_membership = await self.repository.get_membership(workspace_id, actor_user_id) if not actor_membership or actor_membership.role != 'admin': raise CollaborationUnauthorizedError("Unauthorized: Only admins can update member roles.") # Prevent self-elevation if actor_user_id == target_user_id and new_role == 'admin': raise CollaborationUnauthorizedError("Cannot elevate your own privileges.") # Perform update await self.repository.update_member_role(workspace_id, target_user_id, new_role) async def get_membership(self, workspace_id: str, user_id: str) -> WorkspaceMembership | None: return await self.repository.get_membership(workspace_id, user_id) async def list_project_collaborators(self, project_id: str) -> list[MemberResponse]: collaborators = await self.repository.list_project_collaborators(project_id) return [MemberResponse(id=c.id, workspace_id=c.workspace_id, user_id=c.user_id, role=c.role, created_at=c.created_at.isoformat()) for c in collaborators] async def add_project_collaborator(self, workspace_id: str, project_id: str, user_id: str, role: str) -> MemberResponse: collaborator = await self.repository.add_project_collaborator(workspace_id, project_id, user_id, role) return MemberResponse(id=collaborator.id, workspace_id=collaborator.workspace_id, user_id=collaborator.user_id, role=collaborator.role, created_at=collaborator.created_at.isoformat()) async def record_activity(self, workspace_id: str, user_id: str, action: str, entity_id: str, entity_type: str, metadata: dict[str, Any]) -> None: await self.repository.record_activity(workspace_id, user_id, action, entity_id, entity_type, metadata) async def list_activity(self, workspace_id: str) -> list[dict[str, Any]]: activities = await self.repository.list_activity(workspace_id) return [ { "id": a.id, "user_id": a.user_id, "action": a.action, "entity_id": a.entity_id, "entity_type": a.entity_type, "metadata": a.metadata, "created_at": a.created_at.isoformat(), } for a in activities ]