mnemosyne/docs/Pattern_Notification_V1-00.md

Notification Trigger Pattern v1.0.0

Standard pattern for triggering notifications from domain-specific events in Django applications that use Themis for notification infrastructure.

🐾 Red Panda Approval™

This pattern follows Red Panda Approval standards.

---

Overview

Themis provides the notification mailbox — the model, UI (bell + dropdown + list page), polling, browser notifications, user preferences, and cleanup. What Themis does not provide is the trigger logic — the rules that decide when a notification should be created.

Trigger logic is inherently domain-specific:

• A task tracker sends "Task overdue" notifications
• A calendar sends "Event starting in 15 minutes" reminders
• A finance app sends "Invoice payment received" alerts
• A monitoring system sends "Server CPU above 90%" warnings

This pattern documents how consuming apps should create notifications using Themis infrastructure.

---

The Standard Interface

All notification creation goes through one function:

from themis.notifications import notify_user

notify_user(
    user=user,              # Django User instance
    title="Task overdue",   # Short headline (max 200 chars)
    message="Task 'Deploy v2' was due yesterday.",  # Optional body
    level="warning",        # info | success | warning | danger
    url="/tasks/42/",       # Optional: where to navigate on click
    source_app="tasks",     # Your app label (for tracking/cleanup)
    source_model="Task",    # Model that triggered this
    source_id="42",         # PK of the source object (as string)
    deduplicate=True,       # Skip if unread duplicate exists
    expires_at=None,        # Optional: auto-expire datetime
)

Never create UserNotification objects directly. The notify_user() function handles:

• Checking if the user has notifications enabled
• Filtering by the user's minimum notification level
• Deduplication (when deduplicate=True)
• Returning None when skipped (so callers can check)

---

Trigger Patterns

1. Signal-Based Triggers

The most common pattern — listen to Django signals and create notifications:

# myapp/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver

from themis.notifications import notify_user

from .models import Task


@receiver(post_save, sender=Task)
def notify_task_assigned(sender, instance, created, **kwargs):
    """Notify user when a task is assigned to them."""
    if not created and instance.assignee and instance.tracker.has_changed("assignee"):
        notify_user(
            user=instance.assignee,
            title=f"Task assigned: {instance.title}",
            message=f"You've been assigned to '{instance.title}'",
            level="info",
            url=instance.get_absolute_url(),
            source_app="tasks",
            source_model="Task",
            source_id=str(instance.pk),
            deduplicate=True,
        )

2. View-Based Triggers

Create notifications during request processing:

# myapp/views.py
from themis.notifications import notify_user


@login_required
def approve_request(request, pk):
    req = get_object_or_404(Request, pk=pk)
    req.status = "approved"
    req.save()

    # Notify the requester
    notify_user(
        user=req.requester,
        title="Request approved",
        message=f"Your request '{req.title}' has been approved.",
        level="success",
        url=req.get_absolute_url(),
        source_app="requests",
        source_model="Request",
        source_id=str(req.pk),
    )

    messages.success(request, "Request approved.")
    return redirect("request-list")

3. Management Command Triggers

For scheduled checks (e.g., daily overdue detection):

# myapp/management/commands/check_overdue.py
from django.core.management.base import BaseCommand
from django.utils import timezone

from themis.notifications import notify_user

from myapp.models import Task


class Command(BaseCommand):
    help = "Send notifications for overdue tasks"

    def handle(self, *args, **options):
        overdue = Task.objects.filter(
            due_date__lt=timezone.now().date(),
            status__in=["open", "in_progress"],
        )
        count = 0
        for task in overdue:
            result = notify_user(
                user=task.assignee,
                title=f"Overdue: {task.title}",
                message=f"Task was due {task.due_date}",
                level="danger",
                url=task.get_absolute_url(),
                source_app="tasks",
                source_model="Task",
                source_id=str(task.pk),
                deduplicate=True,  # Don't send again if unread
            )
            if result:
                count += 1
        self.stdout.write(f"Sent {count} overdue notification(s)")

Schedule with cron or Kubernetes CronJob:

# Kubernetes CronJob
apiVersion: batch/v1
kind: CronJob
metadata:
  name: check-overdue-tasks
spec:
  schedule: "0 8 * * *"  # Daily at 8 AM
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: check-overdue
            command: ["python", "manage.py", "check_overdue"]

4. Celery Task Triggers

For apps with background workers:

# myapp/tasks.py
from celery import shared_task
from django.contrib.auth import get_user_model

from themis.notifications import notify_user

User = get_user_model()


@shared_task
def notify_report_ready(user_id, report_id):
    """Notify user when their report has been generated."""
    from myapp.models import Report

    user = User.objects.get(pk=user_id)
    report = Report.objects.get(pk=report_id)

    notify_user(
        user=user,
        title="Report ready",
        message=f"Your {report.report_type} report is ready to download.",
        level="success",
        url=report.get_absolute_url(),
        source_app="reports",
        source_model="Report",
        source_id=str(report.pk),
    )

---

Notification Levels

Choose the appropriate level for each notification type:

Level	Weight	Use For
info	0	Informational updates (assigned, comment added)
success	0	Positive outcomes (approved, completed, payment received)
warning	1	Needs attention (approaching deadline, low balance)
danger	2	Urgent/error (overdue, failed, system error)

Users can set a minimum notification level in their preferences:

• info (default) — receive all notifications
• warning — only warnings and errors
• danger — only errors

Note that info and success have the same weight (0), so setting minimum to "warning" filters out both.

---

Source Tracking

The three source tracking fields enable two important features:

Deduplication

When deduplicate=True, notify_user() checks for existing unread notifications with the same source_app, source_model, and source_id. This prevents notification spam when the same event is checked multiple times (e.g., a daily cron job for overdue tasks).

Bulk Cleanup

When a source object is deleted, clean up its notifications:

# In your model's delete signal or post_delete:
from themis.models import UserNotification

@receiver(post_delete, sender=Task)
def cleanup_task_notifications(sender, instance, **kwargs):
    UserNotification.objects.filter(
        source_app="tasks",
        source_model="Task",
        source_id=str(instance.pk),
    ).delete()

---

Expiring Notifications

For time-sensitive notifications, use expires_at:

from datetime import timedelta
from django.utils import timezone

# Event reminder that expires when the event starts
notify_user(
    user=attendee,
    title=f"Starting soon: {event.title}",
    level="info",
    url=event.get_absolute_url(),
    expires_at=event.start_time,
    source_app="events",
    source_model="Event",
    source_id=str(event.pk),
    deduplicate=True,
)

Expired notifications are automatically excluded from counts and lists. The cleanup_notifications management command deletes them permanently.

---

Multi-User Notifications

For events that affect multiple users, call notify_user() in a loop:

def notify_team(team, title, message, **kwargs):
    """Send a notification to all members of a team."""
    for member in team.members.all():
        notify_user(user=member, title=title, message=message, **kwargs)

For large recipient lists, consider using a Celery task to avoid blocking the request.

---

Notification Cleanup

Themis provides automatic cleanup via the management command:

# Uses THEMIS_NOTIFICATION_MAX_AGE_DAYS (default: 90)
python manage.py cleanup_notifications

# Override max age
python manage.py cleanup_notifications --max-age-days=60

What gets deleted:

• Read notifications older than the max age
• Dismissed notifications older than the max age
• Expired notifications (past their expires_at)

What is preserved:

• Unread notifications (regardless of age)

Schedule this as a daily cron job or Kubernetes CronJob.

---

Settings

Themis recognizes these settings for notification behavior:

# Polling interval for the notification bell (seconds, 0 = disabled)
THEMIS_NOTIFICATION_POLL_INTERVAL = 60

# Hard ceiling for notification cleanup (days)
THEMIS_NOTIFICATION_MAX_AGE_DAYS = 90

Users control their own preferences in Settings:

• Enable notifications — master on/off switch
• Minimum level — filter low-priority notifications
• Browser desktop notifications — opt-in for OS-level alerts
• Retention days — how long to keep read notifications

---

Anti-Patterns

• ❌ Don't create UserNotification objects directly — use notify_user()
• ❌ Don't send notifications in tight loops without deduplicate=True
• ❌ Don't use notifications for real-time chat — use WebSocket channels
• ❌ Don't store sensitive data in notification messages (they're visible in admin)
• ❌ Don't rely on notifications as the sole delivery mechanism — they may be disabled by the user
• ❌ Don't forget source_app/source_model/source_id — they enable cleanup and dedup

---

Testing Notifications

from themis.notifications import notify_user
from themis.models import UserNotification


class MyAppNotificationTest(TestCase):
    def test_task_overdue_notification(self):
        """Overdue task creates a danger notification."""
        user = User.objects.create_user(username="test", password="pass")
        task = Task.objects.create(
            title="Deploy v2",
            assignee=user,
            due_date=date.today() - timedelta(days=1),
        )

        # Trigger your notification logic
        check_overdue_tasks()

        # Verify notification was created
        notif = UserNotification.objects.get(
            user=user,
            source_app="tasks",
            source_model="Task",
            source_id=str(task.pk),
        )
        self.assertEqual(notif.level, "danger")
        self.assertIn("Deploy v2", notif.title)

    def test_disabled_user_gets_no_notification(self):
        """Users with notifications disabled get nothing."""
        user = User.objects.create_user(username="quiet", password="pass")
        user.profile.notifications_enabled = False
        user.profile.save()

        result = notify_user(user, "Should be skipped")
        self.assertIsNone(result)
        self.assertEqual(UserNotification.objects.count(), 0)