from django.conf import settings from django.db import models from django.utils import timezone from apps.common.models import TimeStampedModel class Notification(TimeStampedModel): class Type(models.TextChoices): TASK = "task", "Task" TEAM = "team", "Team" BILLING = "billing", "Billing" SYSTEM = "system", "System" class Priority(models.TextChoices): OK = "ok", "OK" WARN = "warn", "Warn" ERR = "err", "Error" INFO = "info", "Info" team = models.ForeignKey("accounts.Team", on_delete=models.CASCADE, related_name="notifications") recipient = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, null=True, blank=True, related_name="notifications", ) project = models.ForeignKey( "projects.Project", on_delete=models.SET_NULL, null=True, blank=True, related_name="notifications", ) notification_type = models.CharField(max_length=24, choices=Type.choices, default=Type.SYSTEM) priority = models.CharField(max_length=24, choices=Priority.choices, default=Priority.INFO) title = models.CharField(max_length=200) brief = models.CharField(max_length=300, blank=True) body = models.TextField(blank=True) source = models.CharField(max_length=120, blank=True) stage = models.CharField(max_length=120, blank=True) owner_label = models.CharField(max_length=120, blank=True) cost_label = models.CharField(max_length=64, blank=True) related_url = models.CharField(max_length=300, blank=True) dedupe_key = models.CharField(max_length=160, blank=True, null=True) is_read = models.BooleanField(default=False) read_at = models.DateTimeField(null=True, blank=True) archived_at = models.DateTimeField(null=True, blank=True) metadata = models.JSONField(default=dict, blank=True) class Meta: ordering = ["-created_at"] indexes = [ models.Index(fields=["team", "recipient", "is_read", "-created_at"]), models.Index(fields=["team", "archived_at", "-created_at"]), models.Index(fields=["team", "dedupe_key"]), ] constraints = [ models.UniqueConstraint( fields=["team", "dedupe_key"], name="ops_notification_team_dedupe_key_unique", ) ] def mark_read(self): if not self.is_read: self.is_read = True self.read_at = timezone.now() self.save(update_fields=["is_read", "read_at", "updated_at"]) def mark_unread(self): if self.is_read or self.read_at: self.is_read = False self.read_at = None self.save(update_fields=["is_read", "read_at", "updated_at"]) def archive(self): if self.archived_at is None: self.archived_at = timezone.now() self.save(update_fields=["archived_at", "updated_at"]) def __str__(self) -> str: return self.title