Notification System — Backend Django Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use compose:subagent (recommended) or compose:execute to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Implement the complete notification system backend — model, triggers, preferences, real-time push, and wire up the existing headless API stubs.

Architecture: A Notification model stores all notifications with a generic FK target. Notifications are created via a notify() helper function called explicitly from DRF views (not signals — following the gamification pattern). A NotificationPreference model stores per-user, per-medium, per-space settings. WebSocket broadcasting pushes new notifications in real-time. The existing headless API stubs are replaced with real queries.

Tech Stack: Django 6, Django Rest Framework, Django Channels (WebSocket), PostgreSQL, Redis

File Structure

apps/
├── communities/
│   ├── models/
│   │   └── notification.py           # Notification, NotificationPreference models
│   ├── notifications.py              # notify() helper, notification types enum
│   ├── signals.py                    # (modify) no changes needed
│   ├── views/
│   │   └── notifications.py          # (new) internal DRF views for preferences
│   ├── serializers/
│   │   └── notifications.py          # (new) serializers for preferences
│   ├── urls.py                       # (modify) add preference endpoints
│   └── migrations/
│       └── 0004_notification.notificationpreference.py
├── spaces/
│   ├── views/
│   │   ├── posts.py                  # (modify) call notify() on create/comment
│   │   ├── comments.py               # (modify) call notify() on reply
│   │   ├── reactions.py              # (modify) call notify() on react
│   │   ├── chat_messages.py          # (modify) call notify() on mention
│   │   ├── dm_messages.py            # (modify) call notify() on DM
│   │   └── lesson_progress.py        # (modify) call notify() on complete
│   ├── consumers.py                  # (modify) add notification consumer
│   └── routing.py                    # (modify) add notification WS route
├── headless/
│   └── views.py                      # (modify) replace notification stubs
└── tests/
    └── test_notifications.py         # (new) comprehensive tests

Task 1: Notification Model + Migration

Covers: Core data model for storing notifications.

Files:

Create: apps/communities/models/notification.py
Modify: apps/communities/models/__init__.py
Create: apps/communities/migrations/0004_notification_notificationpreference.py
[ ] Step 1: Create the Notification model

python

# apps/communities/models/notification.py
from django.conf import settings
from django.db import models
from django.utils import timezone

from apps.utils.models import BaseModel


class Notification(BaseModel):
    class Type(models.TextChoices):
        POST_CREATED = "post_created"
        POST_COMMENTED = "post_commented"
        COMMENT_REPLIED = "comment_replied"
        REACTION = "reaction"
        MENTION = "mention"
        DM_RECEIVED = "dm_received"
        EVENT_RSVP = "event_rsvp"
        LESSON_COMPLETED = "lesson_completed"
        MEMBER_JOINED = "member_joined"

    class Status(models.TextChoices):
        UNREAD = "unread"
        READ = "read"
        ARCHIVED = "archived"

    community = models.ForeignKey(
        "communities.Community",
        on_delete=models.CASCADE,
        related_name="notifications",
    )
    recipient = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name="notifications",
    )
    actor = models.ForeignKey(
        "communities.Membership",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
        related_name="acted_notifications",
    )
    notification_type = models.CharField(max_length=32, choices=Type.choices)
    status = models.CharField(
        max_length=16,
        choices=Status.choices,
        default=Status.UNREAD,
    )

    # Generic FK — the thing being notified about
    content_type = models.ForeignKey(
        "contenttypes.ContentType",
        on_delete=models.CASCADE,
        null=True,
        blank=True,
    )
    object_id = models.PositiveBigIntegerField(null=True, blank=True)

    # Snapshot fields for display (denormalized from target)
    title = models.CharField(max_length=255, blank=True)
    space_name = models.CharField(max_length=100, blank=True)
    space_id = models.PositiveBigIntegerField(null=True, blank=True)

    read_at = models.DateTimeField(null=True, blank=True)
    archived_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        ordering = ["-created_at"]
        indexes = [
            models.Index(fields=["recipient", "status", "-created_at"]),
            models.Index(fields=["community", "recipient", "-created_at"]),
            models.Index(fields=["content_type", "object_id"]),
        ]

    def mark_read(self):
        if self.status != self.Status.READ:
            self.status = self.Status.READ
            self.read_at = timezone.now()
            self.save(update_fields=["status", "read_at", "updated_at"])

    def archive(self):
        self.status = self.Status.ARCHIVED
        self.archived_at = timezone.now()
        self.save(update_fields=["status", "archived_at", "updated_at"])

    def __str__(self):
        return f"{self.notification_type} → {self.recipient}"


class NotificationPreference(BaseModel):
    class Medium(models.TextChoices):
        IN_APP = "in_app"
        EMAIL = "email"
        PUSH = "push"

    class Frequency(models.TextChoices):
        ALL = "all"
        MENTIONS = "mentions"
        NEVER = "never"

    community = models.ForeignKey(
        "communities.Community",
        on_delete=models.CASCADE,
        related_name="notification_preferences",
    )
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        related_name="notification_preferences",
    )
    medium = models.CharField(max_length=16, choices=Medium.choices)
    notification_type = models.CharField(max_length=32, choices=Notification.Type.choices)
    enabled = models.BooleanField(default=True)

    class Meta:
        unique_together = ["community", "user", "medium", "notification_type"]

    def __str__(self):
        return f"{self.user} — {self.medium} — {self.notification_type}: {self.enabled}"


class SpaceNotificationPreference(BaseModel):
    community = models.ForeignKey(
        "communities.Community",
        on_delete=models.CASCADE,
    )
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
    )
    space = models.ForeignKey(
        "spaces.Space",
        on_delete=models.CASCADE,
        related_name="notification_preferences",
    )
    medium = models.CharField(max_length=16, choices=NotificationPreference.Medium.choices)
    frequency = models.CharField(
        max_length=16,
        choices=NotificationPreference.Frequency.choices,
        default=NotificationPreference.Frequency.ALL,
    )

    class Meta:
        unique_together = ["community", "user", "space", "medium"]

    def __str__(self):
        return f"{self.user} — {self.space} — {self.medium}: {self.frequency}"

[ ] Step 2: Export models from __init__.py

python

# apps/communities/models/__init__.py — add these imports
from apps.communities.models.notification import (
    Notification,
    NotificationPreference,
    SpaceNotificationPreference,
)

[ ] Step 3: Create migration

Run: make manage ARGS='makemigrations communities' Expected: Creates apps/communities/migrations/0004_notification_notificationpreference.py

[ ] Step 4: Apply migration

Run: make migrate Expected: Migration applies successfully

[ ] Step 5: Commit

bash

git add apps/communities/models/notification.py apps/communities/models/__init__.py apps/communities/migrations/
git commit -m "feat: add Notification, NotificationPreference, and SpaceNotificationPreference models"

Task 2: Notification Helper + Types

Covers: The notify() function that views call to create notifications, plus the notification type constants.

Files:

Create: apps/communities/notifications.py
[ ] Step 1: Create the notifications helper module

python

# apps/communities/notifications.py
from django.contrib.contenttypes.models import ContentType
from django.utils import timezone

from apps.communities.models import (
    Membership,
    Notification,
    NotificationPreference,
    SpaceNotificationPreference,
)
from apps.spaces.broadcasting import broadcast


NOTIFICATION_TEXT = {
    Notification.Type.POST_CREATED: "{actor} posted in {space}",
    Notification.Type.POST_COMMENTED: "{actor} commented on your post in {space}",
    Notification.Type.COMMENT_REPLIED: "{actor} replied to your comment in {space}",
    Notification.Type.REACTION: "{actor} reacted to your {target_type} in {space}",
    Notification.Type.MENTION: "{actor} mentioned you in {space}",
    Notification.Type.DM_RECEIVED: "{actor} sent you a message",
    Notification.Type.EVENT_RSVP: "{actor} is going to {title}",
    Notification.Type.LESSON_COMPLETED: "You completed a lesson in {space}",
    Notification.Type.MEMBER_JOINED: "{actor} joined the community",
}


def _should_notify(recipient_user, community, notification_type, space=None):
    """Check if the recipient wants this notification via in_app medium."""
    pref = NotificationPreference.objects.filter(
        community=community,
        user=recipient_user,
        medium=NotificationPreference.Medium.IN_APP,
        notification_type=notification_type,
    ).first()

    if pref and not pref.enabled:
        return False

    if space:
        space_pref = SpaceNotificationPreference.objects.filter(
            community=community,
            user=recipient_user,
            space=space,
            medium=NotificationPreference.Medium.IN_APP,
        ).first()

        if space_pref:
            if space_pref.frequency == SpaceNotificationPreference.Frequency.NEVER:
                return False
            if space_pref.frequency == SpaceNotificationPreference.Frequency.MENTIONS:
                return notification_type != Notification.Type.MENTION

    return True


def _build_title(notification_type, actor, target, space):
    """Build the notification title from the type and context."""
    actor_name = actor.user.get_display_name() if actor else "Someone"
    space_name = space.name if space else ""
    target_type = ""
    title = ""

    if target and hasattr(target, "title"):
        title = target.title
    elif target and hasattr(target, "body"):
        target_type = target.__class__.__name__.lower()

    return NOTIFICATION_TEXT[notification_type].format(
        actor=actor_name,
        space=space_name,
        target_type=target_type,
        title=title,
    )


def notify(notification_type, recipient_user, community, actor=None, target=None, space=None):
    """
    Create a notification and broadcast it in real-time.

    This is the single entry point for all notification creation.
    Called explicitly from DRF views (not from signals).
    """
    if recipient_user == (actor.user if actor else None):
        return None

    if not _should_notify(recipient_user, community, notification_type, space):
        return None

    content_type = None
    object_id = None
    if target is not None:
        content_type = ContentType.objects.get_for_model(target)
        object_id = target.pk

    title = _build_title(notification_type, actor, target, space)
    space_name = space.name if space else ""
    space_id = space.pk if space else None

    notification = Notification.objects.create(
        community=community,
        recipient=recipient_user,
        actor=actor,
        notification_type=notification_type,
        content_type=content_type,
        object_id=object_id,
        title=title,
        space_name=space_name,
        space_id=space_id,
    )

    _broadcast_notification(notification)

    return notification


def _broadcast_notification(notification):
    """Push notification to recipient's WebSocket group."""
    payload = {
        "type": "notification.new",
        "id": notification.id,
        "notification_type": notification.notification_type,
        "title": notification.title,
        "space_name": notification.space_name,
        "actor": {
            "id": notification.actor.id,
            "display_name": notification.actor.user.get_display_name(),
        } if notification.actor else None,
        "created_at": notification.created_at.isoformat(),
    }
    group = f"user_{notification.recipient_id}_notifications"
    broadcast(group, payload)


def get_unread_count(user, community):
    """Return the count of unread notifications for a user in a community."""
    return Notification.objects.filter(
        community=community,
        recipient=user,
        status=Notification.Status.UNREAD,
    ).count()


def mark_all_read(user, community):
    """Mark all unread notifications as read."""
    now = timezone.now()
    Notification.objects.filter(
        community=community,
        recipient=user,
        status=Notification.Status.UNREAD,
    ).update(
        status=Notification.Status.READ,
        read_at=now,
        updated_at=now,
    )

[ ] Step 2: Commit

bash

git add apps/communities/notifications.py
git commit -m "feat: add notify() helper and notification type constants"

Task 3: Notification Consumer (WebSocket)

Covers: Real-time push of new notifications to the browser.

Files:

Create: apps/spaces/consumers.py — add NotificationConsumer class
Modify: apps/spaces/routing.py — add WS route
[ ] Step 1: Add NotificationConsumer

python

# Add to apps/spaces/consumers.py

class NotificationConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.user = self.scope.get("user")
        if self.user is None:
            await self.close(code=4001)
            return

        self.group_name = f"user_{self.user.id}_notifications"
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        if hasattr(self, "group_name"):
            await self.channel_layer.group_discard(self.group_name, self.channel_name)

    async def broadcast_payload(self, event):
        await self.send_json(event["payload"])

[ ] Step 2: Add WebSocket route

python

# Add to apps/spaces/routing.py
from apps.spaces.consumers import NotificationConsumer

websocket_urlpatterns = [
    path("ws/chat/spaces/<int:space_id>/", ChatConsumer.as_asgi()),
    path("ws/dm/threads/<int:thread_id>/", DMConsumer.as_asgi()),
    path("ws/notifications/", NotificationConsumer.as_asgi()),
]

[ ] Step 3: Commit

bash

git add apps/spaces/consumers.py apps/spaces/routing.py
git commit -m "feat: add NotificationConsumer for real-time notification push"

Task 4: Notification Preference Endpoints

Covers: API endpoints for users to manage their notification preferences.

Files:

Create: apps/communities/serializers/notifications.py
Create: apps/communities/views/notifications.py
Modify: apps/communities/urls.py
[ ] Step 1: Create serializers

python

# apps/communities/serializers/notifications.py
from rest_framework import serializers

from apps.communities.models import (
    NotificationPreference,
    SpaceNotificationPreference,
)


class NotificationPreferenceSerializer(serializers.ModelSerializer):
    class Meta:
        model = NotificationPreference
        fields = ["id", "medium", "notification_type", "enabled"]


class NotificationPreferenceUpdateSerializer(serializers.Serializer):
    enabled = serializers.BooleanField()


class SpaceNotificationPreferenceSerializer(serializers.ModelSerializer):
    space_name = serializers.CharField(source="space.name", read_only=True)

    class Meta:
        model = SpaceNotificationPreference
        fields = ["id", "space", "space_name", "medium", "frequency"]


class SpaceNotificationPreferenceUpdateSerializer(serializers.Serializer):
    frequency = serializers.ChoiceField(choices=NotificationPreference.Frequency.choices)


class BulkSpacePreferenceUpdateSerializer(serializers.Serializer):
    frequency = serializers.ChoiceField(choices=NotificationPreference.Frequency.choices)
    space_ids = serializers.ListField(
        child=serializers.IntegerField(),
        required=False,
        help_text="If omitted, updates all spaces.",
    )

[ ] Step 2: Create views

python

# apps/communities/views/notifications.py
from rest_framework import generics, permissions, status
from rest_framework.response import Response

from apps.communities.models import (
    NotificationPreference,
    SpaceNotificationPreference,
)
from apps.communities.permissions import RequiresActiveMembership
from apps.communities.serializers.notifications import (
    NotificationPreferenceSerializer,
    NotificationPreferenceUpdateSerializer,
    SpaceNotificationPreferenceSerializer,
    SpaceNotificationPreferenceUpdateSerializer,
    BulkSpacePreferenceUpdateSerializer,
)


class NotificationPreferenceDetailView(generics.RetrieveUpdateAPIView):
    serializer_class = NotificationPreferenceSerializer
    permission_classes = [permissions.IsAuthenticated, RequiresActiveMembership]
    lookup_field = "medium"

    def get_object(self):
        return NotificationPreference.objects.filter(
            community=self.request.community,
            user=self.request.user,
            medium=self.kwargs["medium"],
        )

    def get(self, request, *args, **kwargs):
        prefs = self.get_object()
        serializer = self.get_serializer(prefs, many=True)
        medium = self.kwargs["medium"]
        enabled = all(p["enabled"] for p in serializer.data) if serializer.data else True
        return Response({
            "medium": medium,
            "enabled": enabled,
            "channels": serializer.data,
        })

    def put(self, request, *args, **kwargs):
        ser = NotificationPreferenceUpdateSerializer(data=request.data)
        ser.is_valid(raise_exception=True)
        enabled = ser.validated_data["enabled"]

        NotificationPreference.objects.filter(
            community=request.community,
            user=request.user,
            medium=kwargs["medium"],
        ).update(enabled=enabled)

        return Response({"success": True})


class SpaceNotificationPreferenceListView(generics.ListCreateAPIView):
    serializer_class = SpaceNotificationPreferenceSerializer
    permission_classes = [permissions.IsAuthenticated, RequiresActiveMembership]

    def get_queryset(self):
        return SpaceNotificationPreference.objects.filter(
            community=self.request.community,
            user=self.request.user,
            medium=self.kwargs["medium"],
        )

    def create(self, request, *args, **kwargs):
        ser = SpaceNotificationPreferenceUpdateSerializer(data=request.data)
        ser.is_valid(raise_exception=True)

        pref, created = SpaceNotificationPreference.objects.update_or_create(
            community=request.community,
            user=request.user,
            space_id=request.data.get("space_id"),
            medium=kwargs["medium"],
            defaults={"frequency": ser.validated_data["frequency"]},
        )

        return Response(
            SpaceNotificationPreferenceSerializer(pref).data,
            status=status.HTTP_201_CREATED if created else status.HTTP_200_OK,
        )


class SpaceNotificationPreferenceBulkUpdateView(generics.GenericAPIView):
    permission_classes = [permissions.IsAuthenticated, RequiresActiveMembership]

    def put(self, request, *args, **kwargs):
        ser = BulkSpacePreferenceUpdateSerializer(data=request.data)
        ser.is_valid(raise_exception=True)

        frequency = ser.validated_data["frequency"]
        space_ids = ser.validated_data.get("space_ids")

        qs = SpaceNotificationPreference.objects.filter(
            community=request.community,
            user=request.user,
            medium=kwargs["medium"],
        )
        if space_ids:
            qs = qs.filter(space_id__in=space_ids)

        qs.update(frequency=frequency)

        return Response({"success": True})


class SpaceNotificationPreferenceDetailView(generics.UpdateAPIView):
    serializer_class = SpaceNotificationPreferenceUpdateSerializer
    permission_classes = [permissions.IsAuthenticated, RequiresActiveMembership]

    def get_object(self):
        return SpaceNotificationPreference.objects.get(
            community=self.request.community,
            user=self.request.user,
            space_id=self.kwargs["space_id"],
            medium=self.kwargs["medium"],
        )

[ ] Step 3: Add URL patterns

python

# Add to apps/communities/urls.py
from apps.communities.views.notifications import (
    NotificationPreferenceDetailView,
    SpaceNotificationPreferenceListView,
    SpaceNotificationPreferenceBulkUpdateView,
    SpaceNotificationPreferenceDetailView,
)

urlpatterns += [
    path("community/notification_preferences/<str:medium>/", NotificationPreferenceDetailView.as_view(), name="notification-preferences-detail"),
    path("community/notification_preferences/<str:medium>/spaces/", SpaceNotificationPreferenceListView.as_view(), name="notification-preferences-spaces"),
    path("community/notification_preferences/<str:medium>/spaces/bulk/", SpaceNotificationPreferenceBulkUpdateView.as_view(), name="notification-preferences-spaces-bulk"),
    path("community/notification_preferences/<str:medium>/spaces/<int:space_id>/", SpaceNotificationPreferenceDetailView.as_view(), name="notification-preferences-space-detail"),
]

[ ] Step 4: Commit

bash

git add apps/communities/serializers/notifications.py apps/communities/views/notifications.py apps/communities/urls.py
git commit -m "feat: add notification preference API endpoints"

Task 5: Wire Up Notification Triggers in Views

Covers: Calling notify() from the content creation views.

Files:

Modify: apps/spaces/views/posts.py
Modify: apps/spaces/views/comments.py
Modify: apps/spaces/views/reactions.py
Modify: apps/spaces/views/dm_messages.py
Modify: apps/spaces/views/lesson_progress.py
[ ] Step 1: Post creation → notify post author's followers

In apps/spaces/views/posts.py, add to PostListCreateView.perform_create:

python

# After award_points call, add:
from apps.communities.notifications import notify, Notification

# Notify space members (optional — could be expensive for large spaces)
# For now, skip broadcast notification on post creation to avoid noise.
# Notifications for posts are triggered when someone comments or reacts.

[ ] Step 2: Comment creation → notify post author

In apps/spaces/views/comments.py, add to CommentListCreateView.perform_create:

python

from apps.communities.notifications import notify, Notification

def perform_create(self, serializer):
    post = self._post
    membership = self.request.membership

    comment = serializer.save(
        author=membership,
        community=self.request.community,
        post=post,
    )

    # Notify post author
    if post.author.user != membership.user:
        notify(
            notification_type=Notification.Type.POST_COMMENTED,
            recipient_user=post.author.user,
            community=self.request.community,
            actor=membership,
            target=comment,
            space=self._space,
        )

    # Notify parent comment author (reply)
    if comment.parent and comment.parent.author.user != membership.user:
        notify(
            notification_type=Notification.Type.COMMENT_REPLIED,
            recipient_user=comment.parent.author.user,
            community=self.request.community,
            actor=membership,
            target=comment,
            space=self._space,
        )

    award_points(membership, PointTransaction.Action.COMMENT_CREATED, comment)

[ ] Step 3: Reaction → notify content author

In apps/spaces/views/reactions.py (or wherever the reaction toggle view lives), add after creating a reaction:

python

from apps.communities.notifications import notify, Notification

# After reaction is created (not on delete):
target_author = getattr(target, "author", None) or getattr(target, "sender", None)
if target_author and target_author.user != membership.user:
    notify(
        notification_type=Notification.Type.REACTION,
        recipient_user=target_author.user,
        community=request.community,
        actor=membership,
        target=target,
        space=space,
    )

[ ] Step 4: DM message → notify recipient

In apps/spaces/views/dm_messages.py, add to DMMessageListCreateView.perform_create:

python

from apps.communities.notifications import notify, Notification

def perform_create(self, serializer):
    thread = self._thread
    sender = self.request.membership

    message = serializer.save(
        sender=sender,
        community=self.request.community,
        thread=thread,
    )

    # Notify the other participant
    recipient = thread.membership_b if thread.membership_a == sender else thread.membership_a
    if recipient.user != sender.user:
        notify(
            notification_type=Notification.Type.DM_RECEIVED,
            recipient_user=recipient.user,
            community=self.request.community,
            actor=sender,
            target=message,
        )

[ ] Step 5: Lesson completion → self-notify

In apps/spaces/views/lesson_progress.py, the LessonProgressView already awards points. Add a self-notification:

python

from apps.communities.notifications import notify, Notification

# After marking lesson complete:
notify(
    notification_type=Notification.Type.LESSON_COMPLETED,
    recipient_user=request.user,
    community=request.community,
    actor=request.membership,
    target=lesson,
    space=lesson.space,
)

[ ] Step 6: Commit

bash

git add apps/spaces/views/posts.py apps/spaces/views/comments.py apps/spaces/views/reactions.py apps/spaces/views/dm_messages.py apps/spaces/views/lesson_progress.py
git commit -m "feat: wire up notification triggers in content creation views"

Task 6: Wire Up Headless API Stubs

Covers: Replace all notification stub responses in the headless API with real queries.

Files:

Modify: apps/headless/views.py
[ ] Step 1: Replace NotificationsNewCountView

python

# apps/headless/views.py — replace NotificationsNewCountView

class NotificationsNewCountView(APIView):
    """GET /api/headless/v1/notifications/new_notifications_count."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def get(self, request, *args, **kwargs):
        from apps.communities.models import Notification

        unread = Notification.objects.filter(
            community=request.community,
            recipient=request.user,
            status=Notification.Status.UNREAD,
        ).count()

        return Response({
            "new_notifications_count": unread,
            "new_mentions_count": 0,
            "new_inbox_count": 0,
        })

[ ] Step 2: Replace NotificationsListView

python

class NotificationsListView(APIView):
    """GET /api/headless/v1/notifications."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def get(self, request, *args, **kwargs):
        from apps.communities.models import Notification
        from django.core.paginator import Paginator

        page = int(request.query_params.get("page", 1))
        per_page = int(request.query_params.get("per_page", 20))
        status_filter = request.query_params.get("status", "")

        qs = Notification.objects.filter(
            community=request.community,
            recipient=request.user,
        )

        if status_filter == "unread":
            qs = qs.filter(status=Notification.Status.UNREAD)
        elif status_filter == "read":
            qs = qs.filter(status=Notification.Status.READ)

        paginator = Paginator(qs, per_page)
        page_obj = paginator.get_page(page)

        records = []
        for n in page_obj:
            records.append({
                "id": n.id,
                "notification_type": n.notification_type,
                "title": n.title,
                "status": n.status,
                "space_name": n.space_name,
                "actor": {
                    "id": n.actor.id,
                    "display_name": n.actor.user.get_display_name(),
                    "avatar_url": n.actor.user.avatar_url,
                } if n.actor else None,
                "created_at": n.created_at.isoformat(),
                "read_at": n.read_at.isoformat() if n.read_at else None,
            })

        return Response({
            "page": page_obj.number,
            "per_page": per_page,
            "has_next_page": page_obj.has_next(),
            "count": paginator.count,
            "page_count": paginator.num_pages,
            "records": records,
        })

[ ] Step 3: Replace NotificationsMarkAllReadView

python

class NotificationsMarkAllReadView(APIView):
    """POST /api/headless/v1/notifications/mark_all_as_read."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def post(self, request, *args, **kwargs):
        from apps.communities.notifications import mark_all_read

        mark_all_read(request.user, request.community)
        return Response({"success": True})

[ ] Step 4: Replace NotificationsMarkReadView

python

class NotificationsMarkReadView(APIView):
    """POST /api/headless/v1/notifications/{id}/mark_as_read."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def post(self, request, id, *args, **kwargs):
        from apps.communities.models import Notification

        try:
            notification = Notification.objects.get(
                id=id,
                community=request.community,
                recipient=request.user,
            )
            notification.mark_read()
        except Notification.DoesNotExist:
            pass

        return Response({"success": True})

[ ] Step 5: Replace NotificationsArchiveView

python

class NotificationsArchiveView(APIView):
    """POST /api/headless/v1/notifications/{id}/archive."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def post(self, request, id, *args, **kwargs):
        from apps.communities.models import Notification

        try:
            notification = Notification.objects.get(
                id=id,
                community=request.community,
                recipient=request.user,
            )
            notification.archive()
        except Notification.DoesNotExist:
            pass

        return Response({"success": True})

[ ] Step 6: Replace NotificationsDeleteView

python

class NotificationsDeleteView(APIView):
    """DELETE /api/headless/v1/notifications/{id}."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def delete(self, request, id, *args, **kwargs):
        from apps.communities.models import Notification

        Notification.objects.filter(
            id=id,
            community=request.community,
            recipient=request.user,
        ).delete()

        return Response({"success": True})

[ ] Step 7: Replace SpaceNotificationDetailsView

python

class SpaceNotificationDetailsView(APIView):
    """GET /api/headless/v1/space_notification_details."""

    permission_classes = [RequiresCommunity, permissions.IsAuthenticated, RequiresActiveMembership]

    def get(self, request, *args, **kwargs):
        from apps.communities.models import Notification
        from apps.spaces.models import Space

        spaces = Space.objects.visible_to(request.membership)
        space_ids = list(spaces.values_list("id", flat=True))

        unread_by_space = (
            Notification.objects.filter(
                community=request.community,
                recipient=request.user,
                status=Notification.Status.UNREAD,
                space_id__in=space_ids,
            )
            .values("space_id")
            .annotate(count=Count("id"))
        )
        unread_map = {item["space_id"]: item["count"] for item in unread_by_space}

        return Response([
            {
                "id": s.id,
                "unread_content_count": 0,
                "unread_notifications_count": unread_map.get(s.id, 0),
            }
            for s in spaces
        ])

[ ] Step 8: Replace notification preference stubs

Replace NotificationPreferencesDetailView, NotificationPreferencesUpdateView, NotificationPreferenceSpacesView, NotificationPreferenceSpacesUpdateView, NotificationPreferenceSpaceDetailView with imports from the new views:

python

# At the top of apps/headless/views.py, add:
from apps.communities.views.notifications import (
    NotificationPreferenceDetailView as _NPDetail,
    SpaceNotificationPreferenceListView as _SpaceNPList,
    SpaceNotificationPreferenceBulkUpdateView as _SpaceNPBulk,
    SpaceNotificationPreferenceDetailView as _SpaceNPDetail,
)

# Then replace the stub classes to delegate:
class NotificationPreferencesDetailView(_NPDetail):
    """GET/PUT /api/headless/v1/notification_preferences/{medium}."""
    pass

class NotificationPreferenceSpacesView(_SpaceNPList):
    """GET/POST /api/headless/v1/notification_preferences/{medium}/spaces."""
    pass

class NotificationPreferenceSpacesUpdateView(_SpaceNPBulk):
    """PUT /api/headless/v1/notification_preferences/{medium}/spaces."""
    pass

class NotificationPreferenceSpaceDetailView(_SpaceNPDetail):
    """PUT /api/headless/v1/notification_preferences/{medium}/spaces/{id}."""
    pass

[ ] Step 9: Commit

bash

git add apps/headless/views.py
git commit -m "feat: wire up headless notification API stubs to real data"

Task 7: Notification Seed Data

Covers: Add notification seed data for the seed management command.

Files:

Modify: apps/web/management/commands/seed.py
[ ] Step 1: Add notification seeding

In apps/web/management/commands/seed.py, add after existing seed data:

python

from apps.communities.models import Notification

# Seed sample notifications
sample_notifications = [
    (Notification.Type.POST_CREATED, "Welcome to the community!"),
    (Notification.Type.POST_COMMENTED, "Someone commented on your post"),
    (Notification.Type.REACTION, "Someone liked your post"),
]

for i, (ntype, title) in enumerate(sample_notifications):
    Notification.objects.get_or_create(
        community=community,
        recipient=owner,
        notification_type=ntype,
        title=title,
        defaults={"actor": owner_membership},
    )

[ ] Step 2: Commit

bash

git add apps/web/management/commands/seed.py
git commit -m "feat: add notification seed data"

Task 8: Tests

Covers: Unit tests for models, notify() helper, and API endpoints.

Files:

Create: apps/communities/tests/test_notifications.py
[ ] Step 1: Write model tests

python

# apps/communities/tests/test_notifications.py
from django.utils import timezone

from apps.communities.models import (
    Notification,
    NotificationPreference,
    SpaceNotificationPreference,
)
from apps.communities.notifications import notify, get_unread_count, mark_all_read
from apps.communities.tests.base import CommunityTestCase


class NotificationModelTests(CommunityTestCase):
    def test_create_notification(self):
        notification = Notification.objects.create(
            community=self.community,
            recipient=self.owner,
            actor=self.owner_membership,
            notification_type=Notification.Type.POST_CREATED,
            title="Test notification",
        )
        self.assertEqual(notification.status, Notification.Status.UNREAD)
        self.assertIsNone(notification.read_at)

    def test_mark_read(self):
        notification = Notification.objects.create(
            community=self.community,
            recipient=self.owner,
            notification_type=Notification.Type.POST_CREATED,
            title="Test",
        )
        notification.mark_read()
        notification.refresh_from_db()
        self.assertEqual(notification.status, Notification.Status.READ)
        self.assertIsNotNone(notification.read_at)

    def test_archive(self):
        notification = Notification.objects.create(
            community=self.community,
            recipient=self.owner,
            notification_type=Notification.Type.POST_CREATED,
            title="Test",
        )
        notification.archive()
        notification.refresh_from_db()
        self.assertEqual(notification.status, Notification.Status.ARCHIVED)
        self.assertIsNotNone(notification.archived_at)


class NotifyHelperTests(CommunityTestCase):
    def test_creates_notification(self):
        from apps.spaces.tests.base import SpaceFixtureMixin

        class _SF(SpaceFixtureMixin, CommunityTestCase):
            pass

        sf = _SF()
        sf.setUpClass()
        space = sf.create_space(community=self.community)

        post = sf.create_post(space=space, author=self.owner_membership)

        notify(
            notification_type=Notification.Type.POST_COMMENTED,
            recipient_user=self.owner,
            community=self.community,
            actor=self.owner_membership,
            target=post,
            space=space,
        )

        self.assertEqual(
            Notification.objects.filter(
                recipient=self.owner,
                notification_type=Notification.Type.POST_COMMENTED,
            ).count(),
            1,
        )

    def test_skip_self_notification(self):
        result = notify(
            notification_type=Notification.Type.POST_CREATED,
            recipient_user=self.owner,
            community=self.community,
            actor=self.owner_membership,
        )
        self.assertIsNone(result)
        self.assertEqual(Notification.objects.count(), 0)

    def test_respects_preference_disabled(self):
        NotificationPreference.objects.create(
            community=self.community,
            user=self.owner,
            medium=NotificationPreference.Medium.IN_APP,
            notification_type=Notification.Type.POST_COMMENTED,
            enabled=False,
        )

        notify(
            notification_type=Notification.Type.POST_COMMENTED,
            recipient_user=self.owner,
            community=self.community,
            actor=self.owner_membership,
        )
        self.assertEqual(Notification.objects.count(), 0)


class NotificationAPITests(CommunityTestCase):
    def _login(self):
        self.client.login(username=self.owner.email, password="testpass123")

    def test_list_notifications(self):
        self._login()
        Notification.objects.create(
            community=self.community,
            recipient=self.owner,
            notification_type=Notification.Type.POST_CREATED,
            title="Test",
        )
        response = self.client.get(
            "/api/headless/v1/notifications",
            HTTP_HOST=self.community.slug + ".localhost",
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data["count"], 1)

    def test_unread_count(self):
        self._login()
        Notification.objects.create(
            community=self.community,
            recipient=self.owner,
            notification_type=Notification.Type.POST_CREATED,
            title="Test",
        )
        response = self.client.get(
            "/api/headless/v1/notifications/new_notifications_count",
            HTTP_HOST=self.community.slug + ".localhost",
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data["new_notifications_count"], 1)

    def test_mark_all_read(self):
        self._login()
        Notification.objects.create(
            community=self.community,
            recipient=self.owner,
            notification_type=Notification.Type.POST_CREATED,
            title="Test",
        )
        response = self.client.post(
            "/api/headless/v1/notifications/mark_all_as_read",
            HTTP_HOST=self.community.slug + ".localhost",
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(
            Notification.objects.filter(
                recipient=self.owner,
                status=Notification.Status.UNREAD,
            ).count(),
            0,
        )

[ ] Step 2: Run tests

Run: make test ARGS='apps.communities.tests.test_notifications' Expected: All tests pass

[ ] Step 3: Commit

bash

git add apps/communities/tests/test_notifications.py
git commit -m "feat: add notification system tests"

Task 9: Verify & Lint

Covers: Final verification that everything works together.

[ ] Step 1: Run full test suite

Run: make test Expected: All tests pass (including existing tests)

[ ] Step 2: Run linter

Run: make ruff Expected: No linting errors

[ ] Step 3: Verify migration

Run: make manage ARGS='showmigrations' Expected: 0004_notification_notificationpreference is applied

[ ] Step 4: Verify headless API manually

Run: make dev Test in browser or curl:

bash

# Get JWT token
curl -X POST http://localhost:8000/api/v1/auth/token/ \
  -H "Content-Type: application/json" \
  -d '{"email":"admin@example.com","password":"testpass123"}'

# List notifications
curl http://localhost:8000/api/headless/v1/notifications \
  -H "Authorization: Bearer <token>" \
  -H "Host: acme.localhost"

# Get unread count
curl http://localhost:8000/api/headless/v1/notifications/new_notifications_count \
  -H "Authorization: Bearer <token>" \
  -H "Host: acme.localhost"

[ ] Step 5: Commit any fixes

bash

git add -A
git commit -m "fix: notification system lint and test fixes"

Notification System — Backend Django Implementation Plan ​

File Structure ​

Task 1: Notification Model + Migration ​

Task 2: Notification Helper + Types ​

Task 3: Notification Consumer (WebSocket) ​

Task 4: Notification Preference Endpoints ​

Task 5: Wire Up Notification Triggers in Views ​

Task 6: Wire Up Headless API Stubs ​

Task 7: Notification Seed Data ​

Task 8: Tests ​

Task 9: Verify & Lint ​

Notification System — Backend Django Implementation Plan

File Structure

Task 1: Notification Model + Migration

Task 2: Notification Helper + Types

Task 3: Notification Consumer (WebSocket)

Task 4: Notification Preference Endpoints

Task 5: Wire Up Notification Triggers in Views

Task 6: Wire Up Headless API Stubs

Task 7: Notification Seed Data

Task 8: Tests

Task 9: Verify & Lint