"""Workspaces domain - business logic.""" from __future__ import annotations import re from datetime import datetime, timedelta, timezone from typing import Optional from urllib.parse import urlencode from uuid import UUID from src.core.protocols import EmailProviderProtocol from src.core.security.hashing import digest_token_for_lookup, generate_secure_token from src.core.exceptions import ForbiddenError, NotFoundError, PlanLimitError, ValidationError from src.domain.auth.repository import UserRepository from src.domain.auth.email_templates import build_workspace_invite_email from src.domain.workspaces.models import Workspace from src.domain.workspaces.repository import ( MembershipRepository, RoleRepository, WorkspaceInviteRepository, WorkspaceRepository, WorkspaceSettingsRepository, ) def build_workspace_invite_landing_url(app_base_url: str, workspace_id: UUID, token: str) -> str: """SPA URL for accepting an invite (frontend reads workspace_id + token from query).""" base = app_base_url.rstrip("/") q = urlencode({"workspace_id": str(workspace_id), "token": token}) return f"{base}/invite?{q}" def slugify(name: str) -> str: """Convert name to URL-safe slug.""" s = name.lower().strip() s = re.sub(r"[^\w\s-]", "", s) s = re.sub(r"[-\s]+", "-", s) return s[:100] or "workspace" class WorkspaceService: """Workspace business logic.""" def __init__( self, workspace_repo: WorkspaceRepository, role_repo: RoleRepository, membership_repo: MembershipRepository, invite_repo: WorkspaceInviteRepository, settings_repo: WorkspaceSettingsRepository, user_repo: UserRepository, email_provider: Optional[EmailProviderProtocol] = None, subscription_repo=None, plan_repo=None, allowance_repo=None, ): self._workspace_repo = workspace_repo self._role_repo = role_repo self._membership_repo = membership_repo self._invite_repo = invite_repo self._settings_repo = settings_repo self._user_repo = user_repo self._email_provider = email_provider self._subscription_repo = subscription_repo self._plan_repo = plan_repo self._allowance_repo = allowance_repo async def create_workspace(self, name: str, owner_id: UUID, plan_slug: str) -> Workspace: """Create workspace with plan selection. Plan required - without subscription, app usage is blocked.""" from src.config import get_settings from src.domain.billing.constants import STAFF_PLAN_SLUG from src.domain.billing.credit_policy import email_is_staff owner = await self._user_repo.get_by_id(owner_id) skip_plan_quota_limits = email_is_staff( getattr(owner, "email", None), get_settings().staff_email_domain_set ) if skip_plan_quota_limits: plan_slug = STAFF_PLAN_SLUG plan = None if self._plan_repo: plan = await self._plan_repo.get_by_slug(plan_slug) if not plan or not plan.is_active: raise ValidationError(f"Invalid or inactive plan: {plan_slug}") base_slug = slugify(name) slug = base_slug counter = 0 while await self._workspace_repo.slug_exists(slug): counter += 1 slug = f"{base_slug}-{counter}" owner_role = await self._role_repo.get_or_create_system_role("owner") ws = await self._workspace_repo.create( slug=slug, name=name, owner_id=owner_id, selected_plan_slug=plan_slug, onboarding_status="plan_selected", skip_plan_quota_limits=skip_plan_quota_limits, ) await self._membership_repo.create( workspace_id=ws.id, user_id=owner_id, role_id=owner_role.id, invited_by=None, ) await self._settings_repo.get_or_create(ws.id) # Free plan: provision allowance without Stripe (no subscription) if plan_slug == "free" and not skip_plan_quota_limits and self._allowance_repo and plan: included = 0 if plan.limits: included = plan.limits.get("included_ai_units") or plan.limits.get("ai_minutes") or 0 if included >= 0: period_start = datetime.now(timezone.utc).replace( hour=0, minute=0, second=0, microsecond=0 ) period_end = period_start + timedelta(days=31) await self._allowance_repo.create( workspace_id=ws.id, period_start=period_start, period_end=period_end, included_units=included, subscription_id=None, ) return ws async def get_workspace(self, workspace_id: UUID, user_id: UUID) -> Optional[Workspace]: """Get workspace if user has access.""" has_access = await self._membership_repo.get(workspace_id, user_id) if not has_access: return None return await self._workspace_repo.get_by_id(workspace_id) async def list_workspaces(self, user_id: UUID) -> list[Workspace]: return await self._workspace_repo.list_for_user(user_id) async def update_workspace( self, workspace_id: UUID, user_id: UUID, name: Optional[str] = None, selected_plan_slug: Optional[str] = None, ) -> Optional[Workspace]: if not await self._membership_repo.has_permission(workspace_id, user_id, "workspaces.manage"): raise ForbiddenError() if selected_plan_slug and self._plan_repo: plan = await self._plan_repo.get_by_slug(selected_plan_slug) if not plan or not plan.is_active: raise ValidationError(f"Invalid or inactive plan: {selected_plan_slug}") return await self._workspace_repo.update(workspace_id, name=name, selected_plan_slug=selected_plan_slug) async def delete_workspace(self, workspace_id: UUID, user_id: UUID) -> bool: if not await self._membership_repo.has_permission(workspace_id, user_id, "workspaces.manage"): raise ForbiddenError() await self._workspace_repo.soft_delete(workspace_id) return True async def invite( self, workspace_id: UUID, inviter_id: UUID, email: str, role_name: str = "editor", *, invite_app_base_url: Optional[str] = None, inviter_display: Optional[str] = None, ) -> tuple[str, bool]: """Create invite; return (token, email_sent). Sends email when provider and base URL are set.""" if not await self._membership_repo.has_permission(workspace_id, inviter_id, "members.manage"): raise ForbiddenError() if self._subscription_repo and self._plan_repo: await self._check_invite_limit(workspace_id) role = await self._role_repo.get_or_create_system_role(role_name) token = generate_secure_token(32) token_hash = digest_token_for_lookup(token) expires_days = 7 expires_at = datetime.now(timezone.utc) + timedelta(days=expires_days) await self._invite_repo.create( workspace_id=workspace_id, email=email.lower(), role_id=role.id, token_hash=token_hash, expires_at=expires_at, invited_by=inviter_id, ) email_sent = False if self._email_provider and invite_app_base_url: ws_obj = await self._workspace_repo.get_by_id(workspace_id) ws_name = (ws_obj.name if ws_obj else None) or "a workspace" landing = build_workspace_invite_landing_url(invite_app_base_url, workspace_id, token) subject, html_body, text_plain = build_workspace_invite_email( accept_landing_url=landing, workspace_name=ws_name, inviter_display=inviter_display or "A teammate", role_name=role_name, expires_days=expires_days, ) result = await self._email_provider.send(email.lower(), subject, html_body, text_plain) email_sent = bool(result and result.success) return token, email_sent async def list_pending_invites( self, workspace_id: UUID, user_id: UUID ) -> list[tuple]: """List pending invites. Same permission as creating invites.""" if not await self._membership_repo.has_permission(workspace_id, user_id, "members.manage"): raise ForbiddenError() return await self._invite_repo.list_pending_for_workspace(workspace_id) async def inviter_display(self, user_id: UUID) -> Optional[str]: """Return a safe display value for invite-management responses.""" inviter = await self._user_repo.get_by_id(user_id) if not inviter: return None return ( (inviter.display_name or "").strip() or (f"{inviter.first_name or ''} {inviter.last_name or ''}".strip()) or inviter.email ) async def resend_invite( self, workspace_id: UUID, inviter_id: UUID, invite_id: UUID, *, invite_app_base_url: Optional[str] = None, inviter_display: Optional[str] = None, ) -> tuple[str, bool, str]: """Replace an active invite with a fresh seven-day invite and email it.""" if not await self._membership_repo.has_permission(workspace_id, inviter_id, "members.manage"): raise ForbiddenError() invite = await self._invite_repo.get_pending_for_workspace(invite_id, workspace_id) if not invite: raise NotFoundError("Invite not found or no longer pending") await self._invite_repo.revoke(invite.id, workspace_id) role = await self._role_repo.get_by_id(invite.role_id) if invite.role_id else None role_name = role.name if role else "member" token = generate_secure_token(32) await self._invite_repo.create( workspace_id=workspace_id, email=invite.email, role_id=invite.role_id, token_hash=digest_token_for_lookup(token), expires_at=datetime.now(timezone.utc) + timedelta(days=7), invited_by=inviter_id, ) email_sent = False if self._email_provider and invite_app_base_url: ws_obj = await self._workspace_repo.get_by_id(workspace_id) landing = build_workspace_invite_landing_url(invite_app_base_url, workspace_id, token) subject, html_body, text_plain = build_workspace_invite_email( accept_landing_url=landing, workspace_name=(ws_obj.name if ws_obj else None) or "a workspace", inviter_display=inviter_display or "A teammate", role_name=role_name, expires_days=7, ) result = await self._email_provider.send(invite.email, subject, html_body, text_plain) email_sent = bool(result and result.success) return token, email_sent, invite.email async def revoke_invite(self, workspace_id: UUID, inviter_id: UUID, invite_id: UUID) -> bool: """Revoke one pending invite within the caller's workspace.""" if not await self._membership_repo.has_permission(workspace_id, inviter_id, "members.manage"): raise ForbiddenError() if not await self._invite_repo.revoke(invite_id, workspace_id): raise NotFoundError("Invite not found or no longer pending") return True async def accept_invite(self, workspace_id: UUID, token: str, user_id: UUID) -> Optional[Workspace]: """Accept invite with token. Validates invite belongs to workspace_id in path.""" inv = await self._invite_repo.resolve_pending_invite_token(workspace_id, token) if not inv: raise NotFoundError("Invalid or expired invite") if inv.workspace_id != workspace_id: from src.core.exceptions import InvitePathMismatchError raise InvitePathMismatchError( message="Invite token does not belong to this workspace.", workspace_id=str(workspace_id), ) user = await self._user_repo.get_by_id(user_id) if not user: raise NotFoundError("User not found") if (user.email or "").strip().lower() != (inv.email or "").strip().lower(): raise ForbiddenError("This invitation was sent to a different email address.") existing = await self._membership_repo.get(inv.workspace_id, user_id) if not existing: await self._membership_repo.create( workspace_id=inv.workspace_id, user_id=user_id, role_id=inv.role_id, invited_by=inv.invited_by, ) await self._invite_repo.mark_accepted(inv.id) await _schedule_membership_projection(inv.workspace_id) return await self._workspace_repo.get_by_id(inv.workspace_id) async def preview_invite(self, workspace_id: UUID, token: str): """Return non-sensitive invite metadata before authentication.""" inv = await self._invite_repo.resolve_pending_invite_token(workspace_id, token) if not inv: raise NotFoundError("Invalid or expired invite") role = await self._role_repo.get_by_id(inv.role_id) if inv.role_id else None return inv, role async def update_member_role(self, workspace_id: UUID, actor_id: UUID, member_id: UUID, role_name: str) -> bool: if not await self._membership_repo.has_permission(workspace_id, actor_id, "members.manage"): raise ForbiddenError() role = await self._role_repo.get_or_create_system_role(role_name) changed = await self._membership_repo.update_role(workspace_id, member_id, role.id) if changed: await _schedule_membership_projection(workspace_id) return changed async def remove_member(self, workspace_id: UUID, actor_id: UUID, member_id: UUID) -> bool: if not await self._membership_repo.has_permission(workspace_id, actor_id, "members.manage"): raise ForbiddenError() removed = await self._membership_repo.delete(workspace_id, member_id) if removed: await _schedule_membership_projection(workspace_id) return removed async def list_members(self, workspace_id: UUID, user_id: UUID) -> list[tuple]: """List members - requires any membership to view.""" if not await self._membership_repo.get(workspace_id, user_id): raise ForbiddenError() return await self._membership_repo.list_members(workspace_id) async def _check_invite_limit(self, workspace_id: UUID) -> None: """Raise PlanLimitError if workspace would exceed plan max_members.""" members_count = len(await self._membership_repo.list_members(workspace_id)) pending_count = await self._invite_repo.count_pending_for_workspace(workspace_id) current_count = members_count + pending_count max_allowed = await self._get_plan_max_members(workspace_id) if max_allowed is not None and current_count >= max_allowed: raise PlanLimitError( message=f"This plan allows up to {max_allowed} members.", limit_code="max_members", current_count=current_count, max_allowed=max_allowed, ) async def _get_plan_max_members(self, workspace_id: UUID) -> Optional[int]: """Get max_members from workspace's plan. Returns None if unlimited.""" plan_slug = None ws = await self._workspace_repo.get_by_id(workspace_id) if ws and getattr(ws, "selected_plan_slug", None): plan_slug = ws.selected_plan_slug sub = await self._subscription_repo.get_by_workspace(workspace_id) if sub and sub.plan_id: plan = await self._plan_repo.get_by_stripe_price_id(sub.plan_id) if plan: plan_slug = plan.slug if not plan_slug: # No plan: only owner allowed (safe default; seed a free plan for team capacity) return 1 plan = await self._plan_repo.get_by_slug(plan_slug) if not plan or not plan.limits: return 1 return plan.limits.get("max_members", 1) async def _schedule_membership_projection(workspace_id: UUID) -> None: """Ask for a convergence pass over the workspace's projects. Best-effort: the hourly sweep converges any pass that is lost, so a failed enqueue must not fail the request.""" from src.config import get_settings if not get_settings().project_membership_projection_enabled: return try: from src.workers.enqueue import enqueue_reconcile_workspace_projection await enqueue_reconcile_workspace_projection(str(workspace_id)) except Exception: import structlog structlog.get_logger().warning( "membership_projection_enqueue_failed", workspace_id=str(workspace_id) )