intent/intent/orgs/services.py

"""Organization membership rules. Views call these; they enforce permissions and invariants.

Rule violations raise ValidationError with a user-facing message.
Missing permissions raise PermissionDenied.
"""

from allauth.account.models import EmailAddress
from django.core.exceptions import PermissionDenied, ValidationError
from django.db import transaction
from django.db.models import Q
from django.utils import timezone

from intent.access import org_access

from .models import InviteStatus, Organization, OrgInvite, OrgMembership, Role

LAST_OWNER_MESSAGE = "An organization needs at least one owner. Make someone else an owner first."


def _require_owner(by, org):
    if not org_access(by, org).can_manage_members:
        raise PermissionDenied


def _is_last_owner(membership) -> bool:
    return (
        membership.role == Role.OWNER
        and membership.org.memberships.filter(role=Role.OWNER).count() == 1
    )


def _verified_emails(user) -> list[str]:
    return [
        e.lower()
        for e in EmailAddress.objects.filter(user=user, verified=True).values_list(
            "email", flat=True
        )
    ]


def create_org(*, slug, name, owner) -> Organization:
    if owner.is_staff:
        raise ValidationError("Administrator accounts cannot own organizations.")
    with transaction.atomic():
        org = Organization.objects.create(slug=slug, name=name)
        OrgMembership.objects.create(org=org, user=owner, role=Role.OWNER)
    return org


def invite(org, by, *, user=None, email="", role=Role.COLLABORATOR) -> tuple[OrgInvite, str]:
    """Invite an existing user or an email address. Returns the invite and its raw token.

    A new invite to the same recipient replaces any pending one.
    """
    _require_owner(by, org)
    email = email.strip().lower()
    if bool(user) == bool(email):
        raise ValueError("Invite exactly one of user or email")

    with transaction.atomic():
        if user:
            if user.is_staff:
                raise ValidationError("Administrator accounts cannot be invited.")
            if org.memberships.filter(user=user).exists():
                raise ValidationError(f"{user.username} is already a member.")
            pending = org.invites.filter(status=InviteStatus.PENDING, user=user)
        else:
            if org.memberships.filter(
                user__emailaddress__email__iexact=email, user__emailaddress__verified=True
            ).exists():
                raise ValidationError(f"Someone with {email} is already a member.")
            pending = org.invites.filter(status=InviteStatus.PENDING, email=email)

        pending.update(status=InviteStatus.REVOKED, responded_at=timezone.now())
        new_invite = OrgInvite(org=org, user=user, email=email, role=role, invited_by=by)
        token = new_invite.set_new_token()
        new_invite.save()
    return new_invite, token


def open_invites_for(user):
    """Pending, unexpired invites addressed to this user by username or verified email."""
    return (
        OrgInvite.objects.filter(status=InviteStatus.PENDING, expires_at__gt=timezone.now())
        .filter(Q(user=user) | Q(email__in=_verified_emails(user), email__gt=""))
        .select_related("org", "invited_by")
    )


def is_addressed_to(invite, user) -> bool:
    if invite.user_id:
        return invite.user_id == user.pk
    return invite.email in _verified_emails(user)


def accept_invite(invite, user) -> OrgMembership:
    with transaction.atomic():
        invite.refresh_from_db()
        if not invite.is_open:
            raise ValidationError("This invitation is no longer valid.")
        if user.is_staff:
            raise ValidationError("Administrator accounts cannot join organizations.")
        if invite.org.memberships.filter(user=user).exists():
            raise ValidationError(f"You are already a member of {invite.org}.")
        membership = OrgMembership.objects.create(org=invite.org, user=user, role=invite.role)
        invite.status = InviteStatus.ACCEPTED
        invite.accepted_by = user
        invite.responded_at = timezone.now()
        invite.save(update_fields=["status", "accepted_by", "responded_at", "updated_at"])
    return membership


def decline_invite(invite) -> None:
    with transaction.atomic():
        invite.refresh_from_db()
        if not invite.is_open:
            raise ValidationError("This invitation is no longer valid.")
        invite.status = InviteStatus.DECLINED
        invite.responded_at = timezone.now()
        invite.save(update_fields=["status", "responded_at", "updated_at"])


def revoke_invite(invite, by) -> None:
    _require_owner(by, invite.org)
    if invite.status == InviteStatus.PENDING:
        invite.status = InviteStatus.REVOKED
        invite.responded_at = timezone.now()
        invite.save(update_fields=["status", "responded_at", "updated_at"])


def update_membership(membership, by, *, role, can_create_projects) -> None:
    _require_owner(by, membership.org)
    with transaction.atomic():
        membership.refresh_from_db()
        if role != Role.OWNER and _is_last_owner(membership):
            raise ValidationError(LAST_OWNER_MESSAGE)
        membership.role = role
        membership.can_create_projects = can_create_projects
        membership.save(update_fields=["role", "can_create_projects", "updated_at"])


def remove_member(membership, by) -> None:
    """Owners can remove anyone; members can remove themselves (leave)."""
    if membership.user_id != by.pk:
        _require_owner(by, membership.org)
    with transaction.atomic():
        membership.refresh_from_db()
        if _is_last_owner(membership):
            raise ValidationError(LAST_OWNER_MESSAGE)
        # Project roles only make sense for team members.
        membership.user.project_memberships.filter(project__org=membership.org).delete()
        membership.delete()