Add CheckpointService to resolve active checkpoints

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,46 @@
+from datetime import datetime
+
+from sqlalchemy import or_, select
+
+from h.models import Checkpoint, Document
+
+
+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 factory(_context, request) -> CheckpointService:
+    """Return a CheckpointService instance for the passed context and request."""
+    return CheckpointService(db=request.db)

tests/unit/h/services/checkpoint_test.py

@@ -0,0 +1,94 @@
+from datetime import datetime, timedelta
+from unittest import mock
+
+import pytest
+
+from h.services.checkpoint import CheckpointService, factory
+
+
+class TestActiveCheckpoint:
+    def test_it_returns_an_unrevealed_checkpoint(self, svc, group, document):
+        checkpoint = self.checkpoint(group, document, reveal_date=None)
+
+        assert svc.active_checkpoint(group.id, "http://example.com/page") == checkpoint
+
+    def test_it_returns_a_checkpoint_with_a_future_reveal_date(
+        self, svc, group, document
+    ):
+        checkpoint = self.checkpoint(
+            group,
+            document,
+            reveal_date=datetime.utcnow() + timedelta(days=1),  # noqa: DTZ003
+        )
+
+        assert svc.active_checkpoint(group.id, "http://example.com/page") == checkpoint
+
+    def test_it_returns_None_when_the_checkpoint_is_revealed(
+        self, svc, group, document
+    ):
+        self.checkpoint(
+            group,
+            document,
+            reveal_date=datetime.utcnow() - timedelta(days=1),  # noqa: DTZ003
+        )
+
+        assert svc.active_checkpoint(group.id, "http://example.com/page") is None
+
+    @pytest.mark.usefixtures("document")
+    def test_it_returns_None_when_there_is_no_checkpoint(self, svc, group):
+        assert svc.active_checkpoint(group.id, "http://example.com/page") is None
+
+    def test_it_returns_None_for_a_different_group(
+        self, svc, group, document, factories
+    ):
+        self.checkpoint(group, document, reveal_date=None)
+        other_group = factories.Group()
+
+        assert svc.active_checkpoint(other_group.id, "http://example.com/page") is None
+
+    def test_it_resolves_the_uri_to_the_document(self, svc, group, document, factories):
+        # A second URI on the same document (e.g. a PDF fingerprint) must
+        # resolve to the same checkpoint.
+        factories.DocumentURI(document=document, uri="urn:x-pdf:the-fingerprint")
+        checkpoint = self.checkpoint(group, document, reveal_date=None)
+
+        assert (
+            svc.active_checkpoint(group.id, "urn:x-pdf:the-fingerprint") == checkpoint
+        )
+
+    def test_it_returns_None_for_an_unknown_uri(self, svc, group, document):
+        self.checkpoint(group, document, reveal_date=None)
+
+        assert svc.active_checkpoint(group.id, "http://example.com/other") is None
+
+    def checkpoint(self, group, document, reveal_date):
+        return self.factories.Checkpoint(
+            group=group, document=document, reveal_date=reveal_date
+        )
+
+    @pytest.fixture(autouse=True)
+    def _factories(self, factories):
+        self.factories = factories
+
+    @pytest.fixture
+    def group(self, factories):
+        return factories.Group()
+
+    @pytest.fixture
+    def document(self, factories):
+        document = factories.Document()
+        factories.DocumentURI(document=document, uri="http://example.com/page")
+        return document
+
+
+class TestFactory:
+    def test_it(self, pyramid_request):
+        svc = factory(mock.sentinel.context, pyramid_request)
+
+        assert isinstance(svc, CheckpointService)
+        assert svc.db == pyramid_request.db
+
+
+@pytest.fixture
+def svc(db_session):
+    return CheckpointService(db=db_session)