Source code for swh.coarnotify.server.models

# Copyright (C) 2025 - 2026  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information

"""Models."""

from __future__ import annotations

from typing import cast
import uuid

from django.contrib.auth.models import AbstractUser, BaseUserManager
from django.db import models
from django.urls import reverse
from rdflib import Node

from swh.coarnotify.namespaces import AS, CN




class Organization(models.Model):
    """An organization."""

    name = models.CharField(max_length=150, blank=True)

    url = models.URLField(null=False, unique=True)
    """Organization URL (for discovery)."""
    inbox = models.URLField(null=False, unique=True)
    """LDN inbox URL."""

    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

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





class ActorManager(BaseUserManager):


    def create_user(
        self, email: str, password: str | None = None, **extra_fields
    ) -> "Actor":
        if not email:
            raise ValueError("Email is required")
        instance = self.model(email=self.normalize_email(email), **extra_fields)
        user = cast(Actor, instance)
        user.set_password(password)
        user.save()
        return user




    def create_superuser(self, email: str, password: str, **extra_fields) -> "Actor":
        extra_fields.setdefault("is_staff", True)
        extra_fields.setdefault("is_superuser", True)

        return self.create_user(email, password, **extra_fields)






class Actor(AbstractUser):
    """A member of an organization."""

    email = models.EmailField(unique=True)
    organization = models.ForeignKey(Organization, on_delete=models.PROTECT)
    name = models.CharField(max_length=150, blank=True)

    # mypy is not happy with this but it follows the recommended way of overriding
    # django's user model
    username = None  # type: ignore[assignment]
    first_name = None  # type: ignore[assignment]
    last_name = None  # type: ignore[assignment]
    objects = ActorManager()  # type: ignore[misc,assignment]

    USERNAME_FIELD = "email"
    EMAIL_FIELD = "email"
    REQUIRED_FIELDS = ["organization"]

    def __str__(self) -> str:
        return f"{self.name} <{self.email}>"





class Statuses(models.TextChoices):
    PENDING = "pending", "Pending"
    REJECTED = "rejected", "Rejected"
    ACCEPTED = "accepted", "Accepted"
    PROCESSED = "processed", "Processed"
    UNPROCESSABLE = "unprocessable", "Unprocessable"
    RETRY = "retry", "Retry"





class Notification(models.Model):
    """An abstract model for COAR Notification."""

    id = models.UUIDField(primary_key=True, default=uuid.uuid4)
    """UUID."""
    status = models.CharField(max_length=20, choices=Statuses, default=Statuses.PENDING)
    """Processing status"""
    created_at = models.DateTimeField(auto_now_add=True)
    """Receipt date."""
    updated_at = models.DateTimeField(auto_now=True)
    """Last update date."""
    payload = models.JSONField()
    """JSON-LD payload, could be compacted or expanded."""
    error_message = models.TextField()
    """Optional error message."""



    class Meta:
        abstract = True


    def __str__(self) -> str:
        return str(self.id)





class Handlers(models.TextChoices):
    MENTION = "mention", "Mention"



HANDLER_BY_TYPES: dict[frozenset[Node], Handlers] = {
    frozenset([AS.Announce, CN.RelationshipAction]): Handlers.MENTION
}




class InboundNotification(Notification):
    """Our inbox"""

    raw_payload = models.JSONField()
    """The notification verbatim. ``payload`` contains an expanded version of it."""
    in_reply_to = models.ForeignKey(
        "OutboundNotification",
        null=True,
        on_delete=models.SET_NULL,
        related_name="replied_by",
    )
    """Set if this notification is an answer to an :class:`OutboundNotification`."""
    sender = models.ForeignKey(Organization, on_delete=models.SET_NULL, null=True)
    """Which partner sent this notification."""
    process_task_id = models.BigIntegerField(null=True)
    """Task ID associated to the handling of this notification."""
    handler = models.CharField(max_length=20, choices=Handlers, null=False, blank=True)
    """Which handler will be used to process the task."""



    def get_absolute_url(self) -> str:
        return reverse("read", kwargs={"pk": self.pk})






class OutboundNotification(Notification):
    """Our outbox."""

    in_reply_to = models.ForeignKey(
        "InboundNotification",
        null=True,
        on_delete=models.SET_NULL,
        related_name="replied_by",
    )
    """Set if this notification is an answer to an :class:`InboundNotification`."""
    inbox = models.URLField(null=False)
    """The URL of the inbox it should be sent to."""
    send_task_id = models.BigIntegerField(null=True)
    """Task ID associated to the sending of this notification."""