[HnR:2] CheckpointService + hide annotations in search

h/search/core.py

@@ -53,6 +53,7 @@ def __init__(
             query.UserFilter(),
             query.NIPSAFilter(request),
             query.GroupAndModerationFilter(request),
+            query.HideRevealFilter(request),
             query.AnyMatcher(),
             query.TagsMatcher(),
             query.UriCombinedWildcardFilter(

h/search/query.py

@@ -8,6 +8,7 @@
 from h.models import Group, User
 from h.search.util import add_default_scheme, wildcard_uri_is_valid
 from h.security import Identity, Permission, identity_permits
+from h.services.checkpoint import CheckpointService
 from h.traversal import GroupContext
 from h.util import uri
 from h.util.uri import build_scope_key, parse_uri_versions
@@ -263,6 +264,69 @@ def __call__(self, search, params):
         return search.filter(Q("bool", should=query_clauses))
 
 
+class HideRevealFilter:
+    """
+    Hide annotations covered by an active Hide & Reveal checkpoint.
+
+    Until a checkpoint's reveal_date passes, a student must not see their peers'
+    annotations on the checkpointed document. The scope is resolved from the
+    requesting user's own memberships (see CheckpointService.hidden_scopes). An
+    instructor sees everything, while everyone else sees only their own
+    annotations, instructor notes, and instructor replies to them.
+
+    This is a no-op for the common case of a user with no active checkpoints.
+    """
+
+    def __init__(self, request):
+        self.userid = request.authenticated_userid
+        self.checkpoint_service = request.find_service(CheckpointService)
+        self._user = request.user
+
+    def __call__(self, search, params):  # noqa: ARG002
+        for scope in self.checkpoint_service.hidden_scopes(self._user):
+            search = search.exclude(self._hidden_query(scope))
+        return search
+
+    def _hidden_query(self, scope):
+        in_scope = Q(
+            "bool",
+            must=[
+                Q("term", group=scope.group_pubid),
+                Q("bool", should=self._uri_queries(scope.uris), minimum_should_match=1),
+            ],
+        )
+
+        visible = Q(
+            "bool",
+            minimum_should_match=1,
+            should=[
+                # user's own annotations.
+                Q("term", user_raw=self.userid),
+                # Instructor notes and instructor replies to user.
+                Q(
+                    "bool",
+                    must=[Q("terms", user_raw=scope.instructor_userids)],
+                    minimum_should_match=1,
+                    should=[
+                        Q("bool", must_not=[Q("exists", field="references")]),
+                        Q("terms", references=scope.own_annotation_ids),
+                    ],
+                ),
+            ],
+        )
+
+        # Hide annotations that are in scope but not in the visible set.
+        return Q("bool", must=[in_scope], must_not=[visible])
+
+    @staticmethod
+    def _uri_queries(uris):
+        queries = []
+        for normalized in uris:
+            queries.append(Q("term", **{"target.scope": normalized}))
+            queries.append(Q("wildcard", **{"target.scope": f"{normalized}__v*"}))
+        return queries
+
+
 class UriCombinedWildcardFilter:
     """
     A filter that selects only annotations where the uri matches.

h/services/__init__.py

@@ -11,6 +11,7 @@
     BulkGroupService,
     BulkLMSStatsService,
 )
+from h.services.checkpoint import CheckpointService
 from h.services.email import EmailService
 from h.services.http import HTTPService
 from h.services.job_queue import JobQueueService
@@ -51,6 +52,9 @@ def includeme(config):  # pragma: no cover  # noqa: PLR0915
     config.register_service_factory(
         "h.services.annotation_write.service_factory", iface=AnnotationWriteService
     )
+    config.register_service_factory(
+        "h.services.checkpoint.factory", iface=CheckpointService
+    )
     config.register_service_factory("h.services.mention.factory", iface=MentionService)
     config.register_service_factory(
         "h.services.notification.factory", iface=NotificationService

h/services/checkpoint.py

@@ -0,0 +1,127 @@
+from dataclasses import dataclass
+from datetime import datetime
+
+from sqlalchemy import or_, select
+
+from h.models import (
+    Annotation,
+    Checkpoint,
+    Document,
+    DocumentURI,
+    GroupMembership,
+    User,
+)
+from h.models.group import LMSRole
+
+
+@dataclass
+class HiddenScope:
+    """A (group, document) under an active checkpoint, with its visibility data."""
+
+    group_pubid: str
+    uris: list[str]
+    instructor_userids: list[str]
+    own_annotation_ids: list[str]
+
+
+class CheckpointService:
+    """Resolve Hide & Reveal checkpoints for annotation-search authorization."""
+
+    def __init__(self, db):
+        self.db = db
+
+    def active_checkpoint(self, group_id: int, uri: str) -> Checkpoint | None:
+        """
+        Return an active (unrevealed) checkpoint for `(group_id, uri)`, or None.
+
+        The `uri` is resolved to its Document(s) the same way the search layer
+        resolves the request's `uri` param, so the checkpoint lookup matches the
+        annotations the search will return even when the same document is
+        addressed by an equivalent URI (e.g. a PDF fingerprint).
+
+        A checkpoint is "active" (still hiding annotations) when its reveal_date
+        has not yet passed: it is NULL (never revealed) or in the future.
+        """
+        document_ids = [doc.id for doc in Document.find_by_uris(self.db, [uri])]
+        if not document_ids:
+            return None
+
+        return self.db.scalar(
+            select(Checkpoint)
+            .where(Checkpoint.group_id == group_id)
+            .where(Checkpoint.document_id.in_(document_ids))
+            .where(
+                or_(
+                    Checkpoint.reveal_date.is_(None),
+                    Checkpoint.reveal_date > datetime.utcnow(),  # noqa: DTZ003
+                )
+            )
+            .limit(1)
+        )
+
+    def hidden_scopes(self, user: User | None) -> list[HiddenScope]:
+        """
+        Return the scopes whose annotations must be hidden from user.
+
+        An empty list (the common case: a user with no active checkpoints) means
+        search behaves normally.
+        """
+        if user is None:
+            return []
+
+        checkpoints = self.db.scalars(
+            select(Checkpoint)
+            .join(GroupMembership, GroupMembership.group_id == Checkpoint.group_id)
+            .where(GroupMembership.user_id == user.id)
+            .where(
+                or_(
+                    GroupMembership.lms_role.is_(None),
+                    GroupMembership.lms_role != LMSRole.LMS_INSTRUCTOR.value,
+                )
+            )
+            .where(
+                or_(
+                    Checkpoint.reveal_date.is_(None),
+                    Checkpoint.reveal_date > datetime.utcnow(),  # noqa: DTZ003
+                )
+            )
+        ).all()
+
+        return [self._hidden_scope(user, checkpoint) for checkpoint in checkpoints]
+
+    def _hidden_scope(self, user: User, checkpoint: Checkpoint) -> HiddenScope:
+        group_pubid = checkpoint.group.pubid
+
+        uris = self.db.scalars(
+            select(DocumentURI.uri_normalized).where(
+                DocumentURI.document_id == checkpoint.document_id
+            )
+        ).all()
+
+        # User.userid is a hybrid that compiles to a tuple, so it can't be
+        # SELECTed directly: load the users and read it in Python.
+        instructors = self.db.scalars(
+            select(User)
+            .join(GroupMembership, GroupMembership.user_id == User.id)
+            .where(GroupMembership.group_id == checkpoint.group_id)
+            .where(GroupMembership.lms_role == LMSRole.LMS_INSTRUCTOR.value)
+        ).all()
+        instructor_userids = [instructor.userid for instructor in instructors]
+
+        own_annotation_ids = self.db.scalars(
+            select(Annotation.id)
+            .where(Annotation.userid == user.userid)
+            .where(Annotation.groupid == group_pubid)
+        ).all()
+
+        return HiddenScope(
+            group_pubid=group_pubid,
+            uris=list(uris),
+            instructor_userids=instructor_userids,
+            own_annotation_ids=list(own_annotation_ids),
+        )
+
+
+def factory(_context, request) -> CheckpointService:
+    """Return a CheckpointService instance for the passed context and request."""
+    return CheckpointService(db=request.db)

tests/unit/h/search/conftest.py

@@ -3,6 +3,7 @@
 import pytest
 
 from h.models import Group
+from h.services.checkpoint import CheckpointService
 from h.services.group import GroupService
 from h.services.search_index import SearchIndexService
 
@@ -12,6 +13,13 @@ def world_group(db_session):
     return db_session.query(Group).filter_by(pubid="__world__").one()
 
 
+@pytest.fixture(autouse=True)
+def checkpoint_service(pyramid_config, db_session):
+    service = CheckpointService(db_session)
+    pyramid_config.register_service(service, iface=CheckpointService)
+    return service
+
+
 @pytest.fixture
 def group_service(pyramid_config, world_group):
     group_service = mock.create_autospec(GroupService, instance=True, spec_set=True)