""" The org/owner/role model — the foundation everything else (mailbox connections, contacts, messages, sends) is scoped by. See docs/blueprint/blueprint.md "Core entities" and docs/product/srs.md FR-1 through FR-3. Firestore layout (see backend/firestore_client.py for the local/real seam): orgs/{org_id} — {name, owner_uid, created_at} orgs/{org_id}/members/{uid} — {role, added_at, email?} orgs/{org_id}/roles/{key} — {name, permissions} (an edited built-in role, or a custom one; built-ins not stored use their defaults) user_orgs/{uid} — {org_ids: {org_id: role, ...}} (reverse index, kept in sync on every membership write, so "which orgs does this user belong to" doesn't require scanning every org) One person (one Firebase uid) can own or belong to multiple orgs — the reverse index is what makes that lookup cheap. """ from __future__ import annotations import uuid from datetime import datetime, timezone from backend import firestore_client as db from backend.core.constants import BUILTIN_ROLES, ORG_CREATOR_ROLE, PERMISSIONS, ROLE_ALIASES, ROLE_RANK class PermissionError_(Exception): """Raised when an actor's role doesn't meet the required rank for an action.""" class NotFoundError(Exception): """Raised when an org or membership doesn't exist.""" def _now() -> str: return datetime.now(timezone.utc).isoformat() def _member_path(org_id: str, uid: str) -> str: return f"orgs/{org_id}/members/{uid}" def _members_prefix(org_id: str) -> str: return f"orgs/{org_id}/members/" def _user_orgs_path(uid: str) -> str: return f"user_orgs/{uid}" def _set_reverse_index(uid: str, org_id: str, role: str) -> None: doc = db.get_doc(_user_orgs_path(uid)) or {"org_ids": {}} doc["org_ids"][org_id] = role db.set_doc(_user_orgs_path(uid), doc) def _clear_reverse_index(uid: str, org_id: str) -> None: doc = db.get_doc(_user_orgs_path(uid)) if doc and org_id in doc.get("org_ids", {}): del doc["org_ids"][org_id] db.set_doc(_user_orgs_path(uid), doc) def create_org(owner_uid: str, name: str, kind: str | None = None) -> str: """Creates a new org with `owner_uid` as its owner. Returns the new org_id. A single uid can own — and separately create — more than one org (SRS note under FR-2). `kind` is "personal" for an individual's own org (see backend/accounts.py).""" org_id = uuid.uuid4().hex doc = {"name": name, "owner_uid": owner_uid, "created_at": _now()} if kind: doc["kind"] = kind db.set_doc(f"orgs/{org_id}", doc) db.set_doc(_member_path(org_id, owner_uid), {"role": ORG_CREATOR_ROLE, "added_at": _now()}) _set_reverse_index(owner_uid, org_id, ORG_CREATOR_ROLE) return org_id def get_org(org_id: str) -> dict: org = db.get_doc(f"orgs/{org_id}") if org is None: raise NotFoundError(f"No org {org_id!r}") return org def get_role(org_id: str, uid: str) -> str | None: """The uid's role in this org, or None if they aren't a member.""" membership = db.get_doc(_member_path(org_id, uid)) return membership["role"] if membership else None def require_role(org_id: str, uid: str, minimum_role: str) -> str: """Raises PermissionError_ unless uid's role in org_id ranks at or above minimum_role. Returns the actor's actual role on success. Kept for rank checks between built-in roles; feature checks use require_permission. Custom roles rank lowest.""" role = get_role(org_id, uid) if role is None: raise PermissionError_(f"{uid!r} is not a member of org {org_id!r}") if ROLE_RANK.get(role, 0) < ROLE_RANK[minimum_role]: raise PermissionError_( f"{uid!r} has role {role!r} in org {org_id!r}, needs at least {minimum_role!r}" ) return role def _role_path(org_id: str, key: str) -> str: return f"orgs/{org_id}/roles/{key}" def list_roles(org_id: str) -> list[dict]: """Every role in the org: the built-ins (with any edits) first, then custom ones.""" prefix = f"orgs/{org_id}/roles/" stored = {path[len(prefix):]: data for path, data in db.query_prefix(prefix).items()} roles = [] for key, default in BUILTIN_ROLES.items(): data = default if key == ORG_CREATOR_ROLE else stored.get(key, default) roles.append({"key": key, "name": default["name"], "permissions": list(data["permissions"]), "builtin": True}) for key, data in sorted(stored.items(), key=lambda kv: kv[1].get("name", "").lower()): if key not in BUILTIN_ROLES: roles.append({"key": key, "name": data["name"], "permissions": list(data["permissions"]), "builtin": False}) return roles def permissions_for(org_id: str, role: str) -> set[str]: role = ROLE_ALIASES.get(role, role) if role == ORG_CREATOR_ROLE: return set(PERMISSIONS) stored = db.get_doc(_role_path(org_id, role)) if stored is not None: return set(stored["permissions"]) return set(BUILTIN_ROLES.get(role, {"permissions": []})["permissions"]) def require_permission(org_id: str, uid: str, permission: str | None = None) -> str: """Raises PermissionError_ unless uid is a member of org_id whose role allows `permission` (None: being a member is enough). Returns the actor's role.""" role = get_role(org_id, uid) if role is None: raise PermissionError_(f"{uid!r} is not a member of org {org_id!r}") if permission is not None and permission not in permissions_for(org_id, role): raise PermissionError_(f"Your role doesn't allow this ({PERMISSIONS[permission].lower()})") return role def _role_exists(org_id: str, role: str) -> bool: return role in BUILTIN_ROLES or role in ROLE_ALIASES or db.get_doc(_role_path(org_id, role)) is not None def _check_grant(org_id: str, actor_uid: str, role: str) -> str: """The checks shared by inviting, adding and changing someone's role. Nobody can hand out ownership, only an owner can hand out admin, and nobody can hand out more than they can do themselves.""" if role == ORG_CREATOR_ROLE: raise PermissionError_("Ownership can't be granted — only by creating the org") if not _role_exists(org_id, role): raise ValueError(f"Unknown role {role!r}") actor_role = require_permission(org_id, actor_uid, "manage_team") if ROLE_ALIASES.get(role, role) == "admin" and actor_role != ORG_CREATOR_ROLE: raise PermissionError_("Only an owner can make someone an admin") if not permissions_for(org_id, role) <= permissions_for(org_id, actor_role): raise PermissionError_("You can't give someone more access than you have") return actor_role def _check_manage(org_id: str, actor_uid: str, target_uid: str) -> str: """Checks before changing or removing an existing member. Returns their role.""" target_role = get_role(org_id, target_uid) if target_role is None: raise NotFoundError(f"{target_uid!r} is not a member of org {org_id!r}") if target_role == ORG_CREATOR_ROLE: raise PermissionError_("The owner can't be changed or removed") actor_role = require_permission(org_id, actor_uid, "manage_team") if target_role == "admin" and actor_role != ORG_CREATOR_ROLE: raise PermissionError_("Only an owner can change or remove an admin") return target_role def add_member(org_id: str, actor_uid: str, target_uid: str, role: str, email: str = "") -> None: """actor_uid adds target_uid to org_id with the given role (see _check_grant).""" _check_grant(org_id, actor_uid, role) doc = {"role": role, "added_at": _now()} if email: doc["email"] = email db.set_doc(_member_path(org_id, target_uid), doc) _set_reverse_index(target_uid, org_id, role) def join(org_id: str, uid: str, role: str, email: str = "") -> None: """Adds uid with an already-approved role (an accepted invite). Never lowers or replaces the role of someone who is already a member.""" if get_role(org_id, uid) is not None: return doc = {"role": role, "added_at": _now()} if email: doc["email"] = email db.set_doc(_member_path(org_id, uid), doc) _set_reverse_index(uid, org_id, role) def change_member_role(org_id: str, actor_uid: str, target_uid: str, role: str) -> None: _check_manage(org_id, actor_uid, target_uid) _check_grant(org_id, actor_uid, role) doc = db.get_doc(_member_path(org_id, target_uid)) doc["role"] = role db.set_doc(_member_path(org_id, target_uid), doc) _set_reverse_index(target_uid, org_id, role) def remove_member(org_id: str, actor_uid: str, target_uid: str) -> None: """Someone who can manage the team can remove anyone except the owner; only an owner can remove an admin.""" _check_manage(org_id, actor_uid, target_uid) db.delete_doc(_member_path(org_id, target_uid)) _clear_reverse_index(target_uid, org_id) def leave(org_id: str, uid: str) -> None: """A member leaves on their own. The owner can't leave their own org.""" role = require_permission(org_id, uid) if role == ORG_CREATOR_ROLE: raise PermissionError_("The owner can't leave their own organisation") db.delete_doc(_member_path(org_id, uid)) _clear_reverse_index(uid, org_id) def save_role(org_id: str, actor_uid: str, key: str | None, name: str, permissions: list[str]) -> str: """Creates a custom role (key None) or edits one, built-ins included except Owner. Returns the role's key.""" actor_role = require_permission(org_id, actor_uid, "manage_team") unknown = [p for p in permissions if p not in PERMISSIONS] if unknown: raise ValueError(f"Unknown permission {unknown[0]!r}") if not set(permissions) <= permissions_for(org_id, actor_role): raise PermissionError_("You can't give a role more access than you have") if key == ORG_CREATOR_ROLE: raise PermissionError_("The Owner role can't be changed") if key is None: name = (name or "").strip() if not name or len(name) > 40: raise ValueError("Give the role a name of up to 40 characters") if any(r["name"].lower() == name.lower() for r in list_roles(org_id)): raise ValueError(f"There is already a role called {name!r}") key = "custom-" + uuid.uuid4().hex[:10] elif not _role_exists(org_id, key): raise NotFoundError(f"No role {key!r}") if key in BUILTIN_ROLES: name = BUILTIN_ROLES[key]["name"] if key == "admin" and actor_role != ORG_CREATOR_ROLE: raise PermissionError_("Only an owner can change the Admin role") else: stored = db.get_doc(_role_path(org_id, key)) name = (name or "").strip() or (stored or {}).get("name", "") db.set_doc(_role_path(org_id, key), {"name": name, "permissions": sorted(set(permissions))}) return key def delete_role(org_id: str, actor_uid: str, key: str) -> None: """Removes a custom role nobody holds, or resets an edited built-in to its default.""" require_permission(org_id, actor_uid, "manage_team") if key == ORG_CREATOR_ROLE: raise PermissionError_("The Owner role can't be changed") if key not in BUILTIN_ROLES: if any(m["role"] == key for m in list_members(org_id).values()): raise ValueError("Move everyone off this role before deleting it") db.delete_doc(_role_path(org_id, key)) def list_members(org_id: str) -> dict[str, dict]: """{uid: {role, added_at}} for every member of org_id.""" prefix = _members_prefix(org_id) docs = db.query_prefix(prefix) return {path[len(prefix):]: data for path, data in docs.items()} def list_orgs_for_user(uid: str) -> dict[str, str]: """{org_id: role} for every org this uid belongs to, across all their orgs.""" doc = db.get_doc(_user_orgs_path(uid)) return dict(doc["org_ids"]) if doc else {}