membersystem/src/services/models.py

import typing

from django.contrib.sites.models import Site
from django.db import models
from django.db.models import TextChoices
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from utils.matrix import notify_admins
from utils.mixins import CreatedModifiedAbstract

from .registry import ServiceRegistry
from .registry import ServiceRequests

if typing.TYPE_CHECKING:
    from membership.models import Membership


class ServiceRequestStatus(TextChoices):
    NEW = "NEW", _("New")
    RESOLVED = "RESOLVED", _("Resolved")


class ServiceAccess(CreatedModifiedAbstract):
    """Access to a service for a user."""

    member = models.ForeignKey("membership.Member", on_delete=models.PROTECT)

    service = ServiceRegistry.choices_field(verbose_name=_("service"))

    subscription_data = models.JSONField(
        verbose_name=_("subscription data"),
        null=True,
        blank=True,
    )

    class Meta:
        verbose_name = _("service access")
        verbose_name_plural = _("service accesses")
        constraints = (
            models.UniqueConstraint(
                fields=["member", "service"],
                name="unique_user_service",
            ),
        )

    def __str__(self) -> str:
        return f"{self.member} - {self.service}"

    def save(self, *args, **kwargs) -> None:
        """Ensure that existing ServiceRequest objects are automatically resolved."""
        is_new = not self.pk
        super().save(*args, **kwargs)
        # When a new Service Access is created for a user, we should ensure that all Service Requests for this service
        # are set as RESOLVED.
        if is_new:
            self.member.service_requests.filter(
                service=self.service, request=ServiceRequests.CREATION, status=ServiceRequestStatus.NEW
            ).update(status=ServiceRequestStatus.RESOLVED)


class ServiceRequest(CreatedModifiedAbstract):
    member = models.ForeignKey("membership.Member", on_delete=models.CASCADE, related_name="service_requests")
    service = ServiceRegistry.choices_field()
    request = models.CharField(max_length=24, choices=ServiceRequests.choices)

    is_auto_created = models.BooleanField(default=False)

    status = models.CharField(max_length=24, choices=ServiceRequestStatus.choices, default=ServiceRequestStatus.NEW)

    member_notes = models.TextField(
        blank=True, help_text=_("Notes from the member, intended to guide the resolution of the request.")
    )
    admin_notes = models.TextField(
        blank=True, help_text=_("Readable by member: Notes from the admin / status updates, resolutions etc.")
    )

    class Meta:
        verbose_name = _("service request")
        verbose_name_plural = _("service requests")
        constraints = (
            models.CheckConstraint(
                name="%(app_label)s_%(class)s_status_valid",
                condition=models.Q(status__in=ServiceRequestStatus.values),
            ),
        )

    @classmethod
    def create_defaults(cls, membership: "Membership") -> None:
        """Ensure that a membership has service requests for all 'default' (auto_create=True) services."""
        services_with_access = [
            sa.service for sa in ServiceAccess.objects.filter(member=membership.user).values("service")
        ]
        for __, service in ServiceRegistry.get_items():
            if service.auto_create and service not in services_with_access:
                # Use get_or_create so we don't end up with multiple requests for the same service+user
                cls.objects.get_or_create(
                    member=membership.user, service=service.slug, request=ServiceRequests.CREATION
                )

    def save(self, *args, **kwargs) -> None:
        """Create notifications when new service requests are added."""
        is_new = not self.pk

        super().save(*args, **kwargs)

        # When a new Service Request is saved, we should send a notification to admins
        if is_new:
            base_domain = Site.objects.get_current()
            change_url = reverse("admin:services_servicerequest_change", kwargs={"object_id": self.pk})
            notify_admins(f"New service request: https://{base_domain}{change_url}")