""" Notifications models. Provides models for: - Subscriber: User notification profile (legacy, kept for compatibility) - NotificationPreference: User notification preferences - NotificationLog: Audit trail of sent notifications - SystemAnnouncement: System-wide announcements Note: Now using django-notifications-hq for the core notification system. Subscriber model is kept for backward compatibility but is optional. """ from django.conf import settings from django.db import models class Subscriber(models.Model): """ User notification profile. Note: This model is kept for backward compatibility. The new django-notifications-hq system uses User directly for notifications. This can be used for storing additional notification-related user data. """ user = models.OneToOneField( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="notification_subscriber", ) # Legacy field - kept for migration compatibility novu_subscriber_id = models.CharField( max_length=255, unique=True, db_index=True, help_text="Legacy Novu subscriber ID (deprecated)" ) first_name = models.CharField(max_length=100, blank=True) last_name = models.CharField(max_length=100, blank=True) email = models.EmailField(blank=True) phone = models.CharField(max_length=20, blank=True) avatar = models.URLField(blank=True) locale = models.CharField(max_length=10, default="en") data = models.JSONField(default=dict, blank=True, help_text="Custom subscriber data") created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: verbose_name = "Notification Subscriber" verbose_name_plural = "Notification Subscribers" def __str__(self): return f"Subscriber({self.user.username})" class NotificationPreference(models.Model): """ User notification preferences across channels and workflows. """ user = models.OneToOneField( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="novu_notification_prefs", # Renamed to avoid conflict with User.notification_preferences JSONField ) # Channel preferences channel_preferences = models.JSONField( default=dict, blank=True, help_text="Preferences per channel (email, push, in_app, sms)", ) # Workflow-specific preferences workflow_preferences = models.JSONField( default=dict, blank=True, help_text="Preferences per notification workflow", ) # Frequency settings frequency_settings = models.JSONField( default=dict, blank=True, help_text="Digest and frequency settings", ) # Global opt-out is_opted_out = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: verbose_name = "Notification Preference" verbose_name_plural = "Notification Preferences" def __str__(self): return f"Preferences({self.user.username})" class NotificationLog(models.Model): """ Audit log of sent notifications. """ class Status(models.TextChoices): PENDING = "pending", "Pending" SENT = "sent", "Sent" DELIVERED = "delivered", "Delivered" FAILED = "failed", "Failed" user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, related_name="notification_logs", ) workflow_id = models.CharField(max_length=100, db_index=True) notification_type = models.CharField(max_length=50) channel = models.CharField(max_length=20) # email, push, in_app, sms status = models.CharField( max_length=20, choices=Status.choices, default=Status.PENDING, ) payload = models.JSONField(default=dict, blank=True) error_message = models.TextField(blank=True) novu_transaction_id = models.CharField(max_length=255, blank=True, db_index=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: verbose_name = "Notification Log" verbose_name_plural = "Notification Logs" ordering = ["-created_at"] indexes = [ models.Index(fields=["user", "-created_at"]), models.Index(fields=["workflow_id", "-created_at"]), ] def __str__(self): return f"Log({self.workflow_id}, {self.status})" class SystemAnnouncement(models.Model): """ System-wide announcements. """ class Severity(models.TextChoices): INFO = "info", "Information" WARNING = "warning", "Warning" CRITICAL = "critical", "Critical" title = models.CharField(max_length=255) message = models.TextField() severity = models.CharField( max_length=20, choices=Severity.choices, default=Severity.INFO, ) action_url = models.URLField(blank=True) is_active = models.BooleanField(default=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, related_name="announcements_created", ) created_at = models.DateTimeField(auto_now_add=True) expires_at = models.DateTimeField(null=True, blank=True) class Meta: verbose_name = "System Announcement" verbose_name_plural = "System Announcements" ordering = ["-created_at"] def __str__(self): return f"{self.title} ({self.severity})" class Notification(models.Model): """ In-app notification model. This is a Django-native implementation for storing user notifications, supporting both in-app and email notification channels. """ class Level(models.TextChoices): INFO = "info", "Info" SUCCESS = "success", "Success" WARNING = "warning", "Warning" ERROR = "error", "Error" # Who receives the notification recipient = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="in_app_notifications", # Renamed to avoid clash with accounts.UserNotification ) # Who triggered the notification (can be null for system notifications) actor = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="notifications_sent", ) # What happened verb = models.CharField(max_length=255) description = models.TextField(blank=True) level = models.CharField( max_length=20, choices=Level.choices, default=Level.INFO, ) # The object that was acted upon (generic foreign key) action_object_content_type = models.ForeignKey( "contenttypes.ContentType", on_delete=models.CASCADE, blank=True, null=True, related_name="notification_action_objects", ) action_object_id = models.PositiveIntegerField(blank=True, null=True) # The target of the action (generic foreign key) target_content_type = models.ForeignKey( "contenttypes.ContentType", on_delete=models.CASCADE, blank=True, null=True, related_name="notification_targets", ) target_id = models.PositiveIntegerField(blank=True, null=True) # Additional data data = models.JSONField(default=dict, blank=True) # Status unread = models.BooleanField(default=True, db_index=True) # Timestamps timestamp = models.DateTimeField(auto_now_add=True) read_at = models.DateTimeField(null=True, blank=True) class Meta: verbose_name = "Notification" verbose_name_plural = "Notifications" ordering = ["-timestamp"] indexes = [ models.Index(fields=["recipient", "-timestamp"]), models.Index(fields=["recipient", "unread"]), ] def __str__(self): return f"{self.verb} -> {self.recipient}" def mark_as_read(self): """Mark this notification as read.""" if self.unread: from django.utils import timezone self.unread = False self.read_at = timezone.now() self.save(update_fields=["unread", "read_at"]) @property def action_object(self): """Get the action object instance.""" if self.action_object_content_type and self.action_object_id: return self.action_object_content_type.get_object_for_this_type( pk=self.action_object_id ) return None @property def target(self): """Get the target instance.""" if self.target_content_type and self.target_id: return self.target_content_type.get_object_for_this_type(pk=self.target_id) return None class NotificationManager(models.Manager): """Custom manager for Notification model.""" def unread(self): """Return only unread notifications.""" return self.filter(unread=True) def read(self): """Return only read notifications.""" return self.filter(unread=False) def mark_all_as_read(self): """Mark all notifications as read.""" from django.utils import timezone return self.filter(unread=True).update(unread=False, read_at=timezone.now()) # Add custom manager to Notification model Notification.objects = NotificationManager() Notification.objects.model = Notification