smart-org/apps/api/app/domain/identity/models.py

"""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)