Hide annotations behind active checkpoints 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/checkpoint.py

@@ -1,8 +1,27 @@
+from dataclasses import dataclass
 from datetime import datetime
 
 from sqlalchemy import or_, select
 
-from h.models import Checkpoint, Document
+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:
@@ -40,6 +59,68 @@ def active_checkpoint(self, group_id: int, uri: str) -> Checkpoint | None:
             .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."""

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)

tests/unit/h/search/query_test.py

@@ -5,7 +5,9 @@
 import pytest
 from webob.multidict import MultiDict
 
+from h.models import GroupMembership
 from h.models.annotation import ModerationStatus
+from h.models.group import LMSRole
 from h.search import Search, query
 from h.security.identity import Identity
 from h.security.permissions import Permission
@@ -1173,6 +1175,145 @@ def test_it_returns_annotation_counts_by_user(
         assert result.aggregations["users"] == expected
 
 
+class TestHideRevealFilter:
+    def test_a_student_sees_only_their_visible_annotations(
+        self, search, Annotation, make_own, other_student, instructor
+    ):
+        # The student's own annotation is persisted (not just indexed) so the
+        # resolver can find it when matching instructor replies.
+        own = make_own()
+        instructor_note = Annotation(userid=instructor.userid)
+        instructor_reply_to_me = Annotation(
+            userid=instructor.userid, references=[own.id]
+        )
+        peer = Annotation(userid=other_student.userid)
+        Annotation(userid=instructor.userid, references=[peer.id])
+
+        result = search.run(MultiDict({}))
+
+        assert sorted(result.annotation_ids) == sorted(
+            [own.id, instructor_note.id, instructor_reply_to_me.id]
+        )
+
+    def test_it_does_not_hide_annotations_on_other_documents(
+        self, search, Annotation, other_student
+    ):
+        elsewhere = Annotation(
+            userid=other_student.userid, target_uri="http://example.com/other"
+        )
+
+        result = search.run(MultiDict({}))
+
+        assert elsewhere.id in result.annotation_ids
+
+    def test_it_does_not_hide_annotations_in_other_groups(
+        self, search, Annotation, other_student
+    ):
+        other = Annotation(userid=other_student.userid, groupid="__world__")
+
+        result = search.run(MultiDict({}))
+
+        assert other.id in result.annotation_ids
+
+    @pytest.mark.usefixtures("revealed_checkpoint")
+    def test_a_revealed_checkpoint_hides_nothing(
+        self, search, Annotation, other_student
+    ):
+        peer = Annotation(userid=other_student.userid)
+
+        result = search.run(MultiDict({}))
+
+        assert peer.id in result.annotation_ids
+
+    @pytest.mark.usefixtures("as_instructor")
+    def test_an_instructor_sees_everything(self, search, Annotation, other_student):
+        peer = Annotation(userid=other_student.userid)
+
+        result = search.run(MultiDict({}))
+
+        assert peer.id in result.annotation_ids
+
+    def membership(self, db_session, group, user, lms_role):
+        db_session.add(GroupMembership(user=user, group=group, lms_role=lms_role.value))
+        db_session.flush()
+
+    @pytest.fixture
+    def search(self, search, pyramid_request):
+        search.append_modifier(query.HideRevealFilter(pyramid_request))
+        return search
+
+    @pytest.fixture(autouse=True)
+    def group(self, factories):
+        return factories.Group()
+
+    @pytest.fixture(autouse=True)
+    def document(self, factories):
+        document = factories.Document()
+        factories.DocumentURI(document=document, uri="http://example.com/page")
+        return document
+
+    @pytest.fixture(autouse=True)
+    def checkpoint(self, factories, group, document):
+        return factories.Checkpoint(group=group, document=document, reveal_date=None)
+
+    @pytest.fixture
+    def revealed_checkpoint(self, checkpoint):
+        checkpoint.reveal_date = datetime.datetime(2000, 1, 1)  # noqa: DTZ001
+        return checkpoint
+
+    @pytest.fixture
+    def student(self, factories, group, db_session):
+        student = factories.User()
+        self.membership(db_session, group, student, LMSRole.LMS_STUDENT)
+        return student
+
+    @pytest.fixture
+    def other_student(self, factories, group, db_session):
+        other_student = factories.User()
+        self.membership(db_session, group, other_student, LMSRole.LMS_STUDENT)
+        return other_student
+
+    @pytest.fixture
+    def instructor(self, factories, group, db_session):
+        instructor = factories.User()
+        self.membership(db_session, group, instructor, LMSRole.LMS_INSTRUCTOR)
+        return instructor
+
+    @pytest.fixture(autouse=True)
+    def as_student(self, pyramid_request, pyramid_config, student):
+        pyramid_request.user = student
+        pyramid_config.testing_securitypolicy(student.userid)
+
+    @pytest.fixture
+    def as_instructor(self, pyramid_request, pyramid_config, instructor):
+        pyramid_request.user = instructor
+        pyramid_config.testing_securitypolicy(instructor.userid)
+
+    @pytest.fixture
+    def Annotation(self, Annotation, group):
+        def HideRevealAnnotation(**kwargs):
+            kwargs.setdefault("groupid", group.pubid)
+            kwargs.setdefault("target_uri", "http://example.com/page")
+            kwargs.setdefault("shared", True)
+            return Annotation(**kwargs)
+
+        return HideRevealAnnotation
+
+    @pytest.fixture
+    def make_own(self, factories, index_annotations, group, student):
+        def make_own():
+            annotation = factories.Annotation(
+                userid=student.userid,
+                groupid=group.pubid,
+                target_uri="http://example.com/page",
+                shared=True,
+            )
+            index_annotations(annotation)
+            return annotation
+
+        return make_own
+
+
 @pytest.fixture
 def search(pyramid_request, group_service):  # noqa: ARG001
     search = Search(pyramid_request)