"""Identity & Access domain: users, organizations, memberships, invites.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import UTC, datetime from enum import StrEnum from app.domain.base import new_id, utcnow class Role(StrEnum): OWNER = "owner" ADMIN = "admin" MANAGER = "manager" MEMBER = "member" VIEWER = "viewer" class PlatformRole(StrEnum): """Platform-wide role — a different axis from the in-organization ``Role``. The platform has exactly three levels, and exactly one ``OWNER`` account exists (enforced by a partial unique index on ``users.platform_role``): - ``OWNER`` — the single owner of the whole system. The only account that may enter the admin console, create or demote managers, and transfer ownership. - ``MANAGER`` — an account created by the owner. A manager may create their own organizations (becoming their organization owner) and join other managers' organizations by invitation. - ``USER`` — a plain account. Has no platform privileges: cannot create organizations, only works inside organizations it was invited to. """ OWNER = "owner" MANAGER = "manager" USER = "user" #: Roles that may create a new organization. Everyone else must be invited. PLATFORM_ROLES_CAN_CREATE_ORG = frozenset({PlatformRole.OWNER, PlatformRole.MANAGER}) #: Higher rank implies every permission of the lower ranks. ROLE_RANK: dict[Role, int] = { Role.VIEWER: 10, Role.MEMBER: 20, Role.MANAGER: 30, Role.ADMIN: 40, Role.OWNER: 50, } #: Capabilities used across the application layer. CAN_MANAGE_ORG = "org.manage" CAN_MANAGE_MEMBERS = "org.members" CAN_MANAGE_PROJECTS = "projects.manage" CAN_MANAGE_EMPLOYEES = "employees.manage" CAN_RUN_TASKS = "tasks.run" CAN_VIEW_BILLING = "billing.view" CAN_MANAGE_BILLING = "billing.manage" CAN_VIEW_AUDIT = "audit.view" CAN_USE_SANDBOX = "sandbox.use" _ROLE_CAPABILITIES: dict[Role, set[str]] = { Role.VIEWER: set(), Role.MEMBER: {CAN_RUN_TASKS, CAN_VIEW_BILLING}, Role.MANAGER: { CAN_RUN_TASKS, CAN_VIEW_BILLING, CAN_MANAGE_PROJECTS, CAN_MANAGE_EMPLOYEES, CAN_USE_SANDBOX, }, Role.ADMIN: { CAN_RUN_TASKS, CAN_VIEW_BILLING, CAN_MANAGE_BILLING, CAN_MANAGE_PROJECTS, CAN_MANAGE_EMPLOYEES, CAN_USE_SANDBOX, CAN_MANAGE_MEMBERS, CAN_VIEW_AUDIT, }, Role.OWNER: { CAN_RUN_TASKS, CAN_VIEW_BILLING, CAN_MANAGE_BILLING, CAN_MANAGE_PROJECTS, CAN_MANAGE_EMPLOYEES, CAN_USE_SANDBOX, CAN_MANAGE_MEMBERS, CAN_VIEW_AUDIT, CAN_MANAGE_ORG, }, } def role_has_capability(role: Role, capability: str) -> bool: return capability in _ROLE_CAPABILITIES.get(role, set()) class MembershipStatus(StrEnum): ACTIVE = "active" INVITED = "invited" SUSPENDED = "suspended" @dataclass(slots=True, kw_only=True) class User: email: str password_hash: str name: str = "" id: str = field(default_factory=new_id) avatar_url: str | None = None #: Platform-wide role. ``is_platform_owner`` / ``is_manager`` are derived #: from it; only one ``OWNER`` may exist (see the partial unique index on #: ``users.platform_role``). platform_role: PlatformRole = PlatformRole.USER is_active: bool = True created_at: datetime = field(default_factory=utcnow) updated_at: datetime = field(default_factory=utcnow) @property def is_platform_owner(self) -> bool: return self.platform_role == PlatformRole.OWNER @property def is_manager(self) -> bool: return self.platform_role == PlatformRole.MANAGER @property def can_create_organization(self) -> bool: return self.platform_role in PLATFORM_ROLES_CAN_CREATE_ORG @property def display_name(self) -> str: return self.name.strip() or self.email.split("@")[0] def normalized_email(self) -> str: return self.email.strip().lower() @dataclass(slots=True, kw_only=True) class Organization: name: str slug: str owner_user_id: str id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class Membership: organization_id: str user_id: str role: Role = Role.MEMBER status: MembershipStatus = MembershipStatus.ACTIVE id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @property def is_active(self) -> bool: return self.status == MembershipStatus.ACTIVE def has(self, capability: str) -> bool: return self.is_active and role_has_capability(self.role, capability) def outranks(self, other_role: Role) -> bool: return ROLE_RANK[self.role] > ROLE_RANK[other_role] @dataclass(slots=True, kw_only=True) class Invite: organization_id: str email: str role: Role token_hash: str expires_at: datetime id: str = field(default_factory=new_id) invited_by_user_id: str | None = None accepted_at: datetime | None = None created_at: datetime = field(default_factory=utcnow) def is_expired(self, now: datetime | None = None) -> bool: return (now or utcnow()) >= self.expires_at def is_pending(self, now: datetime | None = None) -> bool: return self.accepted_at is None and not self.is_expired(now) @dataclass(slots=True, kw_only=True) class PlatformInvite: """The platform owner's invitation to create a new manager account (D12). Different in kind from an ``Invite``: an org invite joins someone to an organization that already exists, while this one mints the platform account itself. The token lives only as a hash. """ token_hash: str created_by_user_id: str expires_at: datetime email: str | None = None platform_role: PlatformRole = PlatformRole.MANAGER used_at: datetime | None = None used_by_user_id: str | None = None revoked_at: datetime | None = None revoked_by_user_id: str | None = None note: str = "" id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @property def is_redeemed(self) -> bool: return self.used_at is not None @property def is_revoked(self) -> bool: return self.revoked_at is not None def is_expired(self, now: datetime | None = None) -> bool: """Redeeming a used or revoked link is also "expired" — one helper, one rule. Both sides are normalised to aware before comparing: the row comes back from SQLite with the offset stripped, while ``utcnow()`` carries one, and comparing the two raises rather than answering. """ if self.used_at is not None or self.revoked_at is not None: return True return _aware(self.expires_at) <= _aware(now or utcnow()) def _aware(value: datetime) -> datetime: """A datetime guaranteed to carry a UTC offset.""" return value if value.tzinfo is not None else value.replace(tzinfo=UTC)