"""Workspace service: creation, membership, invitation token flow. Invitation security contract: - The service generates a random 32-byte token. - Only SHA-256(token) is persisted (token_hash). - The raw token is returned to the caller exactly once and never stored. - Redemption: hash(presented_token) must match stored token_hash. """ from __future__ import annotations import hashlib import secrets from datetime import datetime, timedelta from uuid import UUID, uuid4 from app.entities.personal_data import PersonalData from app.entities.workspace import Workspace from app.entities.workspace_invitation import WorkspaceInvitation from app.repositories.leave_wish_repository import LeaveWishRepository from app.repositories.parental_leave_repository import ParentalLeaveRepository from app.repositories.personal_data_repository import PersonalDataRepository from app.repositories.planned_leave_repository import PlannedLeaveRepository from app.repositories.user_repository import UserRepository from app.repositories.vacation_days_repository import VacationDaysRepository from app.repositories.workspace_repository import ( WorkspaceInvitationRepository, WorkspaceRepository, ) from app.utils.errors import InvitationError, NotFoundError, ValidationError _INVITE_TTL_HOURS = 72 class WorkspaceService: def __init__( self, ws_repo: WorkspaceRepository, inv_repo: WorkspaceInvitationRepository, pd_repo: PersonalDataRepository, user_repo: UserRepository, leave_wish_repo: LeaveWishRepository, planned_leave_repo: PlannedLeaveRepository, parental_leave_repo: ParentalLeaveRepository, vacation_days_repo: VacationDaysRepository, ) -> None: self._ws = ws_repo self._inv = inv_repo self._pd = pd_repo self._user = user_repo self._leave_wish = leave_wish_repo self._planned_leave = planned_leave_repo self._parental_leave = parental_leave_repo self._vacation_days = vacation_days_repo # ------------------------------------------------------------------ # Workspace CRUD # ------------------------------------------------------------------ async def create_workspace(self, name: str) -> Workspace: ws = Workspace(id=uuid4(), name=name, created_at=datetime.now()) return await self._ws.create(ws) async def get_workspace(self, workspace_id: UUID) -> Workspace: ws = await self._ws.get_by_id(workspace_id) if ws is None: raise NotFoundError( f"Workspace {workspace_id} not found", message_key="workspace.notFound", data={"workspace_id": str(workspace_id)}, ) return ws async def update_workspace( self, workspace_id: UUID, name: str | None = None, planning_goal: str | None = None, planning_mode: str | None = None, ) -> Workspace: ws = await self.get_workspace(workspace_id) if name is not None: ws.name = name if planning_goal is not None: valid = {"max_time", "max_money", "balanced"} if planning_goal not in valid: raise ValidationError(f"planning_goal must be one of {valid}") ws.planning_goal = planning_goal if planning_mode is not None: valid_modes = {"solo", "couple"} if planning_mode not in valid_modes: raise ValidationError(f"planning_mode must be one of {valid_modes}") ws.planning_mode = planning_mode return await self._ws.update(workspace_id, ws) async def list_members(self, workspace_id: UUID) -> list[PersonalData]: """Return all (non-deleted) parent profiles in the workspace.""" return await self._pd.list_by_workspace(workspace_id) async def add_placeholder_parent(self, workspace_id: UUID, display_name: str) -> PersonalData: """Create a parent profile without a login (solo planning for a partner).""" _validate_display_name(display_name) await self.get_workspace(workspace_id) # verify workspace exists pd = PersonalData( uuid=uuid4(), workspace_id=workspace_id, name=display_name, created_at=datetime.now(), ) return await self._pd.create(pd) async def rename_member( self, workspace_id: UUID, personal_data_uuid: UUID, display_name: str ) -> PersonalData: """Rename a workspace member's display name. Allowed for the placeholder partner and for the logged-in user's own profile alike — there is no other endpoint that can correct a typo'd name (the settings profile-update path only covers salary_type / employment_status / min_vacation_days_per_year), so this is the one place a name gets fixed. """ _validate_display_name(display_name) pd = await self._get_member(workspace_id, personal_data_uuid) pd.name = display_name return await self._pd.update(personal_data_uuid, pd) async def remove_member( self, workspace_id: UUID, personal_data_uuid: UUID, requesting_personal_data_uuid: UUID, ) -> None: """Soft-delete a placeholder partner from a workspace. Refuses (ValidationError, 422 at the router) when: - the target is the caller's own profile, - the target has its own login (it is not actually a "placeholder" — e.g. a real partner who redeemed an invite into this workspace), - the target still has dependent plan data (leave wishes, planned leave days, parental-leave rows, or vacation-days rows) that would be orphaned. """ await self._get_member(workspace_id, personal_data_uuid) # 404 if not a member if personal_data_uuid == requesting_personal_data_uuid: raise ValidationError( "Cannot remove your own profile from the workspace", message_key="workspace.parents.remove.self", data={}, ) if await self._has_login(personal_data_uuid): raise ValidationError( "Cannot remove a member that has their own login", message_key="workspace.parents.remove.hasLogin", data={}, ) blockers = await self._dependent_data_blockers(personal_data_uuid) if blockers: raise ValidationError( f"Cannot remove member with existing {', '.join(blockers)}", message_key="workspace.parents.remove.hasDependentData", data={"blockers": blockers}, ) await self._pd.soft_delete(personal_data_uuid) async def _get_member(self, workspace_id: UUID, personal_data_uuid: UUID) -> PersonalData: """Return the member, raising NotFoundError if the workspace or membership is missing. Deliberately checks membership via `list_by_workspace` rather than a bare `pd_repo.get_by_id` — a `personal_data_uuid` that exists but belongs to a *different* workspace must 404 here, not succeed. """ ws = await self._ws.get_by_id(workspace_id) if ws is None: raise NotFoundError( f"Workspace {workspace_id} not found", message_key="workspace.notFound", data={"workspace_id": str(workspace_id)}, ) members = await self._pd.list_by_workspace(workspace_id) for member in members: if member.uuid == personal_data_uuid: return member raise NotFoundError( f"PersonalData {personal_data_uuid} is not a member of workspace {workspace_id}", message_key="workspace.parents.notFound", data={"personal_data_uuid": str(personal_data_uuid)}, ) async def _has_login(self, personal_data_uuid: UUID) -> bool: """True if a User account is linked to this personal_data_uuid. A genuine placeholder (created via `add_placeholder_parent`) never has one; a partner who registered and redeemed an invite into the same workspace does — removing them here would silently break their login. """ users = await self._user.list() return any(u.personal_data_uuid == personal_data_uuid for u in users) async def _dependent_data_blockers(self, personal_data_uuid: UUID) -> list[str]: """Return the kinds of plan data that would be orphaned by removing this member.""" blockers: list[str] = [] # LeaveWishRepository.list() only recognizes workspace_id/child_id # filters today — passing parent_personal_data_uuid would be silently # ignored rather than applied (returning every wish in every # workspace unfiltered). Fetch unfiltered and filter in Python # instead, so an unsupported filter can never be mistaken for "no # wishes for this parent". all_wishes = await self._leave_wish.list() if any(w.parent_personal_data_uuid == personal_data_uuid for w in all_wishes): blockers.append("leave_wishes") if await self._planned_leave.list(parent_personal_data_uuid=personal_data_uuid): blockers.append("planned_leave") if await self._parental_leave.list(personal_data_uuid=personal_data_uuid): blockers.append("parental_leave") if await self._vacation_days.list(personal_data_uuid=personal_data_uuid): blockers.append("vacation_days") return blockers # ------------------------------------------------------------------ # Invitation flow # ------------------------------------------------------------------ async def invite_partner(self, workspace_id: UUID, invited_by_user_id: UUID) -> str: """Generate a single-use invite token. Returns the RAW token (share out-of-band). Only the SHA-256 hash is stored. The raw token is lost after this call. """ await self.get_workspace(workspace_id) raw_token = secrets.token_urlsafe(32) token_hash = _sha256(raw_token) invitation = WorkspaceInvitation( id=uuid4(), workspace_id=workspace_id, token_hash=token_hash, invited_by_user_id=invited_by_user_id, status="pending", expires_at=datetime.now() + timedelta(hours=_INVITE_TTL_HOURS), created_at=datetime.now(), ) await self._inv.create(invitation) return raw_token async def redeem_invitation( self, raw_token: str, personal_data_uuid: UUID ) -> WorkspaceInvitation: """Link personal_data to the workspace and mark invitation accepted.""" token_hash = _sha256(raw_token) inv = await self._inv.get_by_token_hash(token_hash) if inv is None: raise InvitationError("Invitation not found") if inv.status != "pending": raise InvitationError(f"Invitation is {inv.status}") if datetime.now() > inv.expires_at: inv.status = "expired" await self._inv.update(inv.id, inv) raise InvitationError("Invitation has expired") # Link the personal_data to the workspace. pd = await self._pd.get_by_id(personal_data_uuid) if pd is None: raise NotFoundError(f"PersonalData {personal_data_uuid} not found") pd.workspace_id = inv.workspace_id await self._pd.update(pd.uuid, pd) inv.status = "accepted" inv.accepted_at = datetime.now() return await self._inv.update(inv.id, inv) async def revoke_invitation(self, invitation_id: UUID) -> None: inv = await self._inv.get_by_id(invitation_id) if inv is None: raise NotFoundError(f"Invitation {invitation_id} not found") if inv.status != "pending": raise ValidationError(f"Cannot revoke a {inv.status} invitation") inv.status = "revoked" await self._inv.update(inv.id, inv) def _sha256(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() def _validate_display_name(display_name: str) -> None: """Shared validation for add_placeholder_parent and rename_member so the two paths agree on what counts as a valid name (Pydantic's `min_length=1` on `display_name` does not reject a whitespace-only string). """ if not display_name.strip(): raise ValidationError( "display_name must not be empty or whitespace-only", message_key="workspace.parents.invalidName", data={}, )