feat(organizations): create endpoints to handle organization invitations TASK-969

kobo/apps/organizations/constants.py

@@ -1,4 +1,26 @@
+INVITE_OWNER_ERROR = (
+    'This account is already the owner of ##organization_name##. '
+    'You cannot join multiple organizations with the same account. '
+    'To accept this invitation, you must either transfer ownership of '
+    '##organization_name## to a different account or sign in using a different '
+    'account with the same email address. If you do not already have another '
+    'account, you can create one.'
+)
+
+INVITE_MEMBER_ERROR = (
+    'This account is already a member in ##organization_name##. '
+    'You cannot join multiple organizations with the same account. '
+    'To accept this invitation, sign in using a different account with the '
+    'same email address. If you do not already have another account, you can '
+    'create one.'
+)
+
+INVITE_ALREADY_ACCEPTED_ERROR = 'Invite has already been accepted.'
+INVITE_NOT_FOUND_ERROR = 'Invite not found.'
 ORG_ADMIN_ROLE = 'admin'
 ORG_EXTERNAL_ROLE = 'external'
 ORG_MEMBER_ROLE = 'member'
 ORG_OWNER_ROLE = 'owner'
+USER_DOES_NOT_EXIST_ERROR = (
+    'User with username or email {invitee} does not exist or is not active.'
+)

kobo/apps/organizations/migrations/0010_add_status_and_invitee_role_to_organization_invitation.py

@@ -0,0 +1,38 @@
+# Generated by Django 4.2.15 on 2025-01-02 12:25
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('organizations', '0009_update_db_state_with_auth_user'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='organizationinvitation',
+            name='invitee_role',
+            field=models.CharField(
+                choices=[('admin', 'Admin'), ('member', 'Member')],
+                default='member',
+                max_length=10,
+            ),
+        ),
+        migrations.AddField(
+            model_name='organizationinvitation',
+            name='status',
+            field=models.CharField(
+                choices=[
+                    ('accepted', 'Accepted'),
+                    ('cancelled', 'Cancelled'),
+                    ('declined', 'Declined'),
+                    ('expired', 'Expired'),
+                    ('pending', 'Pending'),
+                    ('resent', 'Resent'),
+                ],
+                default='pending',
+                max_length=11,
+            ),
+        ),
+    ]

kobo/apps/organizations/models.py

@@ -1,12 +1,13 @@
 from functools import partial
 from typing import Literal
 
+from django.apps import apps
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
 from django.db.models import F
 from django_request_cache import cache_for_request
-from django.utils.translation import gettext_lazy as t
+from django.utils.translation import gettext_lazy as t, override
 
 if settings.STRIPE_ENABLED:
     from djstripe.models import Customer, Subscription
@@ -24,6 +25,8 @@
 from organizations.utils import create_organization as create_organization_base
 
 from kpi.fields import KpiUidField
+from kpi.utils.mailer import EmailMessage, Mailer
+from kpi.utils.placeholders import replace_placeholders
 
 from .constants import (
     ORG_ADMIN_ROLE,
@@ -46,6 +49,16 @@ class OrganizationType(models.TextChoices):
     NONE = 'none', t('I am not associated with any organization')
 
 
+class OrganizationInviteStatusChoices(models.TextChoices):
+
+    ACCEPTED = 'accepted'
+    CANCELLED = 'cancelled'
+    DECLINED = 'declined'
+    EXPIRED = 'expired'
+    PENDING = 'pending'
+    RESENT = 'resent'
+
+
 class Organization(AbstractOrganization):
     id = KpiUidField(uid_prefix='org', primary_key=True)
     mmo_override = models.BooleanField(
@@ -273,7 +286,146 @@ class OrganizationOwner(AbstractOrganizationOwner):
 
 
 class OrganizationInvitation(AbstractOrganizationInvitation):
-    pass
+    status = models.CharField(
+        max_length=11,
+        choices=OrganizationInviteStatusChoices.choices,
+        default=OrganizationInviteStatusChoices.PENDING,
+    )
+    invitee_role = models.CharField(
+        max_length=10,
+        choices=[('admin', 'Admin'), ('member', 'Member')],
+        default='member',
+    )
+
+    def send_acceptance_email(self):
+        """
+        Send an email to the sender of the invitation to notify them that the
+        invitee has accepted the invitation
+        """
+        sender_language = self.invited_by.extra_details.data.get(
+            'last_ui_language', settings.LANGUAGE_CODE
+        )
+
+        template_variables = {
+            'sender_username': self.invited_by.username,
+            'sender_email': self.invited_by.email,
+            'recipient_username': self.invitee.username,
+            'recipient_email': self.invitee.email,
+            'organization_name': self.invited_by.organization.name,
+            'base_url': settings.KOBOFORM_URL,
+        }
+
+        email_message = EmailMessage(
+            to=self.invited_by.email,
+            subject='KoboToolbox organization invitation accepted',
+            plain_text_content_or_template='emails/accepted_invite.txt',
+            template_variables=template_variables,
+            html_content_or_template='emails/accepted_invite.html',
+            language=sender_language
+        )
+
+        Mailer.send(email_message)
+
+    def send_invite_email(self):
+        is_registered_user = bool(self.invitee)
+        to_email = (
+            self.invitee.email
+            if is_registered_user
+            else self.invitee_identifier
+        )
+
+        # Get recipient role with an article
+        recipient_role = (
+            t('an admin') if self.invitee_role == 'admin' else t('a member')
+        )
+        # To avoid circular import
+        User = apps.get_model('kobo_auth', 'User')
+        has_multiple_accounts = User.objects.filter(email=to_email).count() > 1
+        organization_name = self.invited_by.organization.name
+        current_language = settings.LANGUAGE_CODE
+        invitee_language = (
+            self.invitee.extra_details.data.get(
+                'last_ui_language', current_language
+            )
+            if is_registered_user
+            else current_language
+        )
+
+        template_variables = {
+            'sender_name': self.invited_by.extra_details.data['name'],
+            'sender_username': self.invited_by.username,
+            'sender_email': self.invited_by.email,
+            'recipient_username': (
+                self.invitee.username
+                if is_registered_user
+                else self.invitee_identifier
+            ),
+            'recipient_email': to_email,
+            'recipient_role': recipient_role,
+            'organization_name': organization_name,
+            'base_url': settings.KOBOFORM_URL,
+            'invite_uid': self.guid,
+            'is_registered_user': is_registered_user,
+            'has_multiple_accounts': has_multiple_accounts,
+        }
+
+        if is_registered_user:
+            html_template = 'emails/registered_user_invite.html'
+            text_template = 'emails/registered_user_invite.txt'
+        else:
+            html_template = 'emails/unregistered_user_invite.html'
+            text_template = 'emails/unregistered_user_invite.txt'
+
+        with override(invitee_language):
+            # Because `subject` contains a placeholder, it cannot be translated
+            # by EmailMessage
+            subject = replace_placeholders(
+                t("You're invited to join ##organization_name## organization"),
+                organization_name=organization_name
+            )
+
+        email_message = EmailMessage(
+            to=to_email,
+            subject=subject,
+            plain_text_content_or_template=text_template,
+            template_variables=template_variables,
+            html_content_or_template=html_template,
+            language=invitee_language,
+        )
+
+        Mailer.send(email_message)
+
+    def send_refusal_email(self):
+        """
+        Send an email to the sender of the invitation to notify them that the
+        invitee has declined the invitation
+        """
+        sender_language = self.invited_by.extra_details.data.get(
+            'last_ui_language', settings.LANGUAGE_CODE
+        )
+
+        template_variables = {
+            'sender_username': self.invited_by.username,
+            'sender_email': self.invited_by.email,
+            'recipient': (
+                self.invitee.username
+                if self.invitee
+                else self.invitee_identifier
+            ),
+            'organization_name': self.invited_by.organization.name,
+            'base_url': settings.KOBOFORM_URL,
+        }
+
+        email_message = EmailMessage(
+            to=self.invited_by.email,
+            subject='KoboToolbox organization invitation declined',
+            plain_text_content_or_template='emails/declined_invite.txt',
+            template_variables=template_variables,
+            html_content_or_template='emails/declined_invite.html',
+            language=sender_language,
+        )
+
+        Mailer.send(email_message)
 
 
 create_organization = partial(create_organization_base, model=Organization)

kobo/apps/organizations/permissions.py

@@ -2,7 +2,11 @@
 from rest_framework import permissions
 from rest_framework.permissions import IsAuthenticated
 
-from kobo.apps.organizations.constants import ORG_EXTERNAL_ROLE
+from kobo.apps.organizations.constants import (
+    ORG_EXTERNAL_ROLE,
+    ORG_OWNER_ROLE,
+    ORG_ADMIN_ROLE
+)
 from kobo.apps.organizations.models import Organization
 from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
 from kpi.utils.object_permission import get_database_user
@@ -58,3 +62,39 @@ def has_object_permission(self, request, view, obj):
         is validated in `has_permission()`. Therefore, this method always returns True.
         """
         return True
+
+
+class OrgMembershipInvitePermission(
+    ValidationPasswordPermissionMixin, IsAuthenticated
+):
+    def has_permission(self, request, view):
+        self.validate_password(request)
+        if not super().has_permission(request=request, view=view):
+            return False
+
+        organization_id = view.kwargs.get('organization_id')
+        try:
+            organization = Organization.objects.get(id=organization_id)
+        except Organization.DoesNotExist:
+            raise Http404
+
+        # Fetch and attach the user role to the view for reuse in the viewset
+        user = get_database_user(request.user)
+        user_role = organization.get_user_role(user)
+        view.user_role = user_role
+
+        allowed_roles = [ORG_OWNER_ROLE, ORG_ADMIN_ROLE]
+        if request.method in ['POST', 'DELETE'] or (
+            request.method == 'PATCH' and
+            request.data.get('status') in ['resent', 'cancelled']
+        ):
+            if user_role in allowed_roles:
+                return True
+            elif user_role == ORG_EXTERNAL_ROLE:
+                raise Http404
+            return False
+
+        if request.method == 'GET' and user_role == ORG_EXTERNAL_ROLE:
+            raise Http404
+
+        return True