insights/products/notifications/backend/presentation/views.py

from datetime import datetime
from typing import cast

from django.db.models import Exists, OuterRef, QuerySet, Subquery
from django.shortcuts import get_object_or_404
from django.utils.dateparse import parse_datetime

import hanzo_insights
from drf_spectacular.utils import OpenApiParameter, extend_schema
from rest_framework import serializers as drf_serializers
from rest_framework.exceptions import ValidationError
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet

from insights.api.mixins import ValidatedRequest, validated_request
from insights.api.routing import TeamAndOrgViewSetMixin
from insights.api.utils import action
from insights.models import User
from insights.rbac.user_access_control import UserAccessControl

from products.notifications.backend.cache import get_unread_count, invalidate_unread_count, set_unread_count
from products.notifications.backend.facade.enums import AC_RESOURCE_TYPES
from products.notifications.backend.models import NotificationArchiveState, NotificationEvent, NotificationReadState
from products.notifications.backend.presentation.serializers import NotificationEventSerializer

_BULK_NOTIFICATION_IDS_MAX = 500


class BulkNotificationIdsRequestSerializer(drf_serializers.Serializer):
    notification_ids = drf_serializers.ListField(
        child=drf_serializers.UUIDField(),
        max_length=_BULK_NOTIFICATION_IDS_MAX,
        help_text=(
            f"UUIDs of notification events to mark in bulk (max {_BULK_NOTIFICATION_IDS_MAX}). "
            "Events the user is not a recipient of are silently skipped."
        ),
    )


class NotificationPagination(LimitOffsetPagination):
    default_limit = 20
    max_limit = 100


class NotificationsViewSet(TeamAndOrgViewSetMixin, GenericViewSet):
    queryset = NotificationEvent.objects.all()
    serializer_class = NotificationEventSerializer
    scope_object = "INTERNAL"
    pagination_class = NotificationPagination

    def _get_user(self) -> User:
        return cast(User, self.request.user)

    def _is_feature_enabled(self) -> bool:
        user = self._get_user()
        if not user.distinct_id:
            return False
        org_id = str(self.team.organization_id)
        return bool(
            hanzo_insights.feature_enabled(
                "real-time-notifications",
                user.distinct_id,
                groups={"organization": org_id},
                group_properties={"organization": {"id": org_id}},
                only_evaluate_locally=False,
                send_feature_flag_events=False,
            )
        )

    def _get_base_queryset(self) -> QuerySet:
        user = self._get_user()
        team = self.team
        return (
            NotificationEvent.objects.filter(
                organization_id=team.organization_id,
                resolved_user_ids__contains=[user.id],
            )
            .annotate(
                read=Exists(
                    NotificationReadState.objects.filter(
                        notification_event_id=OuterRef("id"),
                        user_id=user.id,
                    )
                ),
                read_at=Subquery(
                    NotificationReadState.objects.filter(
                        notification_event_id=OuterRef("id"),
                        user_id=user.id,
                    ).values("created_at")[:1]
                ),
                archived=Exists(
                    NotificationArchiveState.objects.filter(
                        notification_event_id=OuterRef("id"),
                        user_id=user.id,
                    )
                ),
            )
            .order_by("-created_at")
        )

    def _filter_by_access_control(self, queryset: QuerySet) -> QuerySet:
        resource_types_in_set = set(
            queryset.exclude(resource_type__isnull=True)
            .exclude(resource_type="")
            .values_list("resource_type", flat=True)
            .distinct()
        )

        ac_types_to_check = resource_types_in_set & AC_RESOURCE_TYPES
        if not ac_types_to_check:
            return queryset

        user_ac = UserAccessControl(self._get_user(), self.team)
        if not user_ac.access_controls_supported:
            return queryset

        denied_types = set()
        for resource_type in ac_types_to_check:
            if not user_ac.check_access_level_for_resource(resource_type, "viewer"):
                denied_types.add(resource_type)

        if denied_types:
            queryset = queryset.exclude(resource_type__in=denied_types)

        return queryset

    def _apply_filters(self, queryset: QuerySet, request: Request) -> QuerySet:
        # Explicit per-field filters (no dict-unpack) so semgrep's no-request-param-orm-filter
        # taint analysis is satisfied — every field name is hard-coded at the call site.
        params = request.query_params
        if value := params.get("notification_type"):
            queryset = queryset.filter(notification_type=value)
        if value := params.get("target_type"):
            queryset = queryset.filter(target_type=value)
        if value := params.get("target_id"):
            queryset = queryset.filter(target_id=value)
        if value := params.get("resource_type"):
            queryset = queryset.filter(resource_type=value)
        if value := params.get("resource_id"):
            queryset = queryset.filter(resource_id=value)

        def _parse_iso(raw: str, field: str) -> datetime:
            # Django decodes "+" timezone offsets in query strings as " "; restore before parsing.
            parsed = parse_datetime(raw.replace(" ", "+"))
            if parsed is None:
                raise ValidationError({field: f"Invalid ISO 8601 timestamp: {raw}"})
            return parsed

        if raw := params.get("created_after"):
            queryset = queryset.filter(created_at__gte=_parse_iso(raw, "created_after"))
        if raw := params.get("created_before"):
            queryset = queryset.filter(created_at__lt=_parse_iso(raw, "created_before"))
        return queryset

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="notification_type",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Filter by notification type",
            ),
            OpenApiParameter(
                name="target_type",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Filter by recipient target type (e.g. `user`, `team`)",
            ),
            OpenApiParameter(
                name="target_id",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Filter by recipient target ID (e.g. a user ID)",
            ),
            OpenApiParameter(
                name="resource_type",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Filter by the type of the resource the notification refers to (e.g. `insight`, `dashboard`)",
            ),
            OpenApiParameter(
                name="resource_id",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="Filter by the ID of the resource the notification refers to",
            ),
            OpenApiParameter(
                name="created_after",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="ISO 8601 timestamp; only events at or after this time",
            ),
            OpenApiParameter(
                name="created_before",
                type=str,
                location=OpenApiParameter.QUERY,
                required=False,
                description="ISO 8601 timestamp; only events strictly before this time",
            ),
            OpenApiParameter(
                name="archived",
                type=bool,
                location=OpenApiParameter.QUERY,
                required=False,
                description="When true, return only notifications the recipient has archived; "
                "otherwise return only non-archived notifications (the default)",
            ),
        ]
    )
    def list(self, request: Request, *args, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"results": [], "next": None, "previous": None, "count": 0})

        archived = request.query_params.get("archived", "").lower() in ("true", "1")
        queryset = self._get_base_queryset().filter(archived=archived)
        queryset = self._apply_filters(queryset, request)
        queryset = self._filter_by_access_control(queryset)
        page = self.paginate_queryset(queryset)
        if page is not None:
            serializer = NotificationEventSerializer(page, many=True)
            return self.get_paginated_response(serializer.data)
        serializer = NotificationEventSerializer(queryset, many=True)
        return Response(serializer.data)

    @action(methods=["GET"], detail=False)
    def unread_count(self, request: Request, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"count": 0})

        user = self._get_user()
        org_id = self.team.organization_id
        count = get_unread_count(user.id, org_id)
        if count is None:
            queryset = self._get_base_queryset().filter(archived=False)
            queryset = self._filter_by_access_control(queryset)
            count = queryset.filter(read=False).count()
            set_unread_count(user.id, org_id, count)
        return Response({"count": count})

    @extend_schema(request=None)
    @action(methods=["POST"], detail=False)
    def mark_all_read(self, request: Request, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"status": "ok"})

        user = self._get_user()
        queryset = self._get_base_queryset().filter(read=False, archived=False)
        event_ids = list(queryset.values_list("id", flat=True))
        if event_ids:
            read_states = [NotificationReadState(notification_event_id=eid, user=user) for eid in event_ids]
            NotificationReadState.objects.bulk_create(read_states, ignore_conflicts=True)
        set_unread_count(user.id, self.team.organization_id, 0)
        return Response({"updated": len(event_ids)})

    def _get_recipient_event_or_404(self) -> NotificationEvent:
        # NotificationEvent is org-scoped + user-targeted, not team-scoped. self.get_object() applies
        # the team_id URL filter from TeamAndOrgViewSetMixin and 404s on org-level notifications
        # (team_id=NULL). (organization_id, resolved_user_ids__contains=[user.id]) is the real
        # authorization boundary — the same pair the bulk mark endpoints use.
        user = self._get_user()
        return get_object_or_404(
            NotificationEvent.objects.filter(  # nosemgrep: idor-lookup-without-team
                organization_id=self.team.organization_id,
                resolved_user_ids__contains=[user.id],
            ),
            pk=self.kwargs["pk"],  # nosemgrep: idor-taint-user-input-to-model-get
        )

    @extend_schema(request=None)
    @action(methods=["POST"], detail=True, url_path="mark_read")
    def mark_read(self, request: Request, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"status": "ok"})

        user = self._get_user()
        event = self._get_recipient_event_or_404()
        NotificationReadState.objects.get_or_create(notification_event=event, user=user)
        invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"status": "ok"})

    @extend_schema(request=None)
    @action(methods=["POST"], detail=True, url_path="mark_unread")
    def mark_unread(self, request: Request, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"status": "ok"})

        user = self._get_user()
        event = self._get_recipient_event_or_404()
        NotificationReadState.objects.filter(notification_event=event, user=user).delete()
        invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"status": "ok"})

    @validated_request(request_serializer=BulkNotificationIdsRequestSerializer)
    @action(methods=["POST"], detail=False, url_path="mark_read_bulk")
    def mark_read_bulk(self, request: ValidatedRequest, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"updated": 0})

        user = self._get_user()
        ids = request.validated_data["notification_ids"]
        if not ids:
            return Response({"updated": 0})

        # NotificationEvent is org-scoped + user-targeted (notifications can span teams within an org),
        # not team-scoped. The (organization_id, resolved_user_ids__contains=[user.id]) pair already
        # ensures the caller can only touch notifications they were a recipient of — adding team_id
        # would silently drop notifications addressed to this user but with a different team_id.
        eligible_ids = list(
            NotificationEvent.objects.filter(  # nosemgrep: idor-lookup-without-team
                id__in=ids,  # nosemgrep: idor-taint-user-input-to-model-get
                organization_id=self.team.organization_id,
                resolved_user_ids__contains=[user.id],
            ).values_list("id", flat=True)
        )
        if eligible_ids:
            read_states = [NotificationReadState(notification_event_id=eid, user=user) for eid in eligible_ids]
            NotificationReadState.objects.bulk_create(read_states, ignore_conflicts=True)
            invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"updated": len(eligible_ids)})

    @validated_request(request_serializer=BulkNotificationIdsRequestSerializer)
    @action(methods=["POST"], detail=False, url_path="mark_unread_bulk")
    def mark_unread_bulk(self, request: ValidatedRequest, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"updated": 0})

        user = self._get_user()
        ids = request.validated_data["notification_ids"]
        if not ids:
            return Response({"updated": 0})

        # NotificationEvent is org-scoped + user-targeted (notifications can span teams within an org),
        # not team-scoped. The (organization_id, resolved_user_ids__contains=[user.id]) pair already
        # ensures the caller can only touch notifications they were a recipient of — adding team_id
        # would silently drop notifications addressed to this user but with a different team_id.
        eligible_ids = list(
            NotificationEvent.objects.filter(  # nosemgrep: idor-lookup-without-team
                id__in=ids,  # nosemgrep: idor-taint-user-input-to-model-get
                organization_id=self.team.organization_id,
                resolved_user_ids__contains=[user.id],
            ).values_list("id", flat=True)
        )
        if eligible_ids:
            NotificationReadState.objects.filter(
                notification_event_id__in=eligible_ids,
                user=user,
            ).delete()
            invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"updated": len(eligible_ids)})

    @extend_schema(request=None)
    @action(methods=["POST"], detail=True, url_path="archive")
    def archive(self, request: Request, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"status": "ok"})

        user = self._get_user()
        event = self._get_recipient_event_or_404()
        NotificationArchiveState.objects.get_or_create(notification_event=event, user=user)
        invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"status": "ok"})

    @validated_request(request_serializer=BulkNotificationIdsRequestSerializer)
    @action(methods=["POST"], detail=False, url_path="archive_bulk")
    def archive_bulk(self, request: ValidatedRequest, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"updated": 0})

        user = self._get_user()
        ids = request.validated_data["notification_ids"]
        if not ids:
            return Response({"updated": 0})

        eligible_ids = list(
            NotificationEvent.objects.filter(  # nosemgrep: idor-lookup-without-team
                id__in=ids,  # nosemgrep: idor-taint-user-input-to-model-get
                organization_id=self.team.organization_id,
                resolved_user_ids__contains=[user.id],
            ).values_list("id", flat=True)
        )
        if eligible_ids:
            archive_states = [NotificationArchiveState(notification_event_id=eid, user=user) for eid in eligible_ids]
            NotificationArchiveState.objects.bulk_create(archive_states, ignore_conflicts=True)
            invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"updated": len(eligible_ids)})

    @extend_schema(request=None)
    @action(methods=["POST"], detail=False, url_path="archive_all")
    def archive_all(self, request: Request, **kwargs) -> Response:
        if not self._is_feature_enabled():
            return Response({"updated": 0})

        user = self._get_user()
        queryset = self._get_base_queryset().filter(archived=False)
        event_ids = list(queryset.values_list("id", flat=True))
        if event_ids:
            archive_states = [NotificationArchiveState(notification_event_id=eid, user=user) for eid in event_ids]
            NotificationArchiveState.objects.bulk_create(archive_states, ignore_conflicts=True)
            invalidate_unread_count(user.id, self.team.organization_id)
        return Response({"updated": len(event_ids)})