⏪️(backend) reintroduce the reset connection mechanism

When an access change or is deleted or a link configuration changes, we
call the yhub server to reset connections and remove them if needed. The
YHubService is used for this.

CHANGELOG.md

@@ -98,6 +98,8 @@ and this project adheres to
   `503` instead of denying access like a permission failure, so clients retry
   instead of giving up. The built-in endpoints can also answer JSON on
   `Accept: application/json`
+- ✨(backend) reset the yhub connections of a document and its descendants
+  when an access or the link configuration changes
 - ✨(backend) add a service to call the yhub REST API
 - ✨(backend) add a service generating cached RS256 JWT tokens
 - ✨(backend) publish the JWT public key on a JWKS endpoint

src/backend/core/api/viewsets.py

@@ -71,6 +71,7 @@ from core.services.search_indexers import (
     get_document_indexer,
     get_visited_document_ids_of,
 )
+from core.tasks.access import reset_service_connections_in_cascade
 from core.tasks.mail import send_ask_for_access_mail
 from core.utils.analytics import PosthogEventName, posthog_capture
 from core.utils.dicts import lowercase_keys
@@ -1754,6 +1755,9 @@ class DocumentViewSet(
 
         serializer.save()
 
+        # Notify collaboration server about the link updated
+        reset_service_connections_in_cascade.delay(str(document.id))
+
         return drf.response.Response(serializer.data, status=drf.status.HTTP_200_OK)
 
     @drf.decorators.action(detail=True, methods=["post", "delete"], url_path="favorite")
@@ -2767,12 +2771,28 @@ class DocumentAccessViewSet(
                 or settings.LANGUAGE_CODE,
             )
 
+    def perform_update(self, serializer):
+        """Update an access to the document and notify the collaboration server."""
+        access = serializer.save()
+
+        access_user_id = None
+        if access.user:
+            access_user_id = str(access.user.id)
+
+        # Notify collaboration server about the access change
+        reset_service_connections_in_cascade.delay(
+            str(access.document.id), access_user_id
+        )
+
     def perform_destroy(self, instance):
-        """Delete an access to the document."""
+        """Delete an access to the document and notify the collaboration server."""
         # Snapshot the identifiers before deletion as Django resets the primary key
         # on the instance once it is deleted.
         access_id = str(instance.id)
         document_id = str(instance.document_id)
+        # an access is granted either to a user or to a team, only a user has
+        # connections of their own to reset
+        user_id = str(instance.user.id) if instance.user else None
 
         instance.delete()
 
@@ -2782,6 +2802,9 @@ class DocumentAccessViewSet(
             {"access_id": access_id, "document_id": document_id},
         )
 
+        # Notify collaboration server about the access removed
+        reset_service_connections_in_cascade.delay(document_id, user_id)
+
 
 class InvitationViewset(
     drf.mixins.CreateModelMixin,

src/backend/core/tasks/access.py

@@ -0,0 +1,42 @@
+"""Tasks dedicated to document's accesses."""
+
+from logging import getLogger
+
+from core import models
+from core.services.yhub_services import YHubError, YHubService
+
+from impress.celery_app import app
+
+logger = getLogger(__name__)
+
+
+@app.task
+def reset_service_connections_in_cascade(document_id, user_id=None):
+    """
+    Reset the connections of a document and all its descendants on the
+    collaboration server.
+
+    A document inherits the accesses of its ancestors, so a change on one of
+    them can revoke the access to the whole subtree: yhub re-checks every
+    connection of each document and disconnects the ones that lost their
+    access. The endpoint is document scoped, hence the walk down the tree.
+
+    A document failing is logged and does not stop the ones after it, its
+    clients keep the rights they connected with until they reconnect.
+    """
+    try:
+        document = models.Document.objects.get(pk=document_id)
+    except models.Document.DoesNotExist:
+        logger.error("Document %s does not exists anymore", document_id)
+        return
+
+    documents = models.Document.objects.filter(
+        path__startswith=document.path, depth__gte=document.depth
+    ).order_by("path")
+
+    service = YHubService()
+    for doc in documents:
+        try:
+            service.reset_connections(doc, user_id)
+        except YHubError:
+            logger.exception("impossible to reset connections for document %s", doc.id)

src/backend/core/tests/documents/test_api_document_accesses.py

@@ -4,6 +4,7 @@ Test document accesses API endpoints for users in impress's core app.
 # pylint: disable=too-many-lines
 
 import random
+from contextlib import contextmanager
 from unittest import mock
 from uuid import uuid4
 
@@ -18,6 +19,25 @@ from core.utils.analytics import PosthogEventName
 pytestmark = pytest.mark.django_db
 
 
+@pytest.fixture(name="mock_reset_connections")
+def mock_reset_connections_fixture():
+    """
+    Provide a context manager that patches the ``reset_service_connections_in_cascade``
+    Celery task and asserts its ``delay`` method is called exactly once for the given
+    document and user when leaving the context.
+    """
+
+    @contextmanager
+    def _mock_reset_connections(document_id, user_id=None):
+        with mock.patch(
+            "core.api.viewsets.reset_service_connections_in_cascade.delay"
+        ) as mock_delay:
+            yield mock_delay
+            mock_delay.assert_called_once_with(str(document_id), user_id)
+
+    return _mock_reset_connections
+
+
 def test_api_document_accesses_list_anonymous():
     """Anonymous users should not be allowed to list document accesses."""
     document = factories.DocumentFactory()
@@ -734,6 +754,7 @@ def test_api_document_accesses_update_administrator_except_owner(
     create_for,
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     A user who is a direct administrator in a document should be allowed to update a user
@@ -772,12 +793,13 @@ def test_api_document_accesses_update_administrator_except_owner(
 
     for field, value in new_values.items():
         new_data = {**old_values, field: value}
-        response = client.put(
-            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-            data=new_data,
-            format="json",
-        )
-        assert response.status_code == 200
+        with mock_reset_connections(document.id, str(access.user_id)):
+            response = client.put(
+                f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+                data=new_data,
+                format="json",
+            )
+            assert response.status_code == 200
 
         access.refresh_from_db()
         updated_values = serializers.DocumentAccessSerializer(instance=access).data
@@ -842,6 +864,7 @@ def test_api_document_accesses_update_administrator_from_owner(via, mock_user_te
 def test_api_document_accesses_update_administrator_to_owner(
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     A user who is an administrator in a document, should not be allowed to update
@@ -889,12 +912,13 @@ def test_api_document_accesses_update_administrator_to_owner(
 
             assert response.status_code == 403
         else:
-            response = client.put(
-                f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-                data=new_data,
-                format="json",
-            )
-            assert response.status_code == 200
+            with mock_reset_connections(document.id, str(access.user_id)):
+                response = client.put(
+                    f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+                    data=new_data,
+                    format="json",
+                )
+                assert response.status_code == 200
 
         access.refresh_from_db()
         updated_values = serializers.DocumentAccessSerializer(instance=access).data
@@ -907,6 +931,7 @@ def test_api_document_accesses_update_owner(
     create_for,
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     A user who is an owner in a document should be allowed to update
@@ -943,13 +968,14 @@ def test_api_document_accesses_update_owner(
 
     for field, value in new_values.items():
         new_data = {**old_values, field: value}
-        response = client.put(
-            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-            data=new_data,
-            format="json",
-        )
+        with mock_reset_connections(document.id, str(access.user_id)):
+            response = client.put(
+                f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+                data=new_data,
+                format="json",
+            )
 
-        assert response.status_code == 200
+            assert response.status_code == 200
 
         access.refresh_from_db()
         updated_values = serializers.DocumentAccessSerializer(instance=access).data
@@ -968,6 +994,7 @@ def test_api_document_accesses_update_owner(
 def test_api_document_accesses_update_owner_self_root(
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     A user who is owner of a document should be allowed to update
@@ -1006,27 +1033,30 @@ def test_api_document_accesses_update_owner_self_root(
     # Add another owner and it should now work
     factories.UserDocumentAccessFactory(document=document, role="owner")
 
-    response = client.put(
-        f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-        data={
-            **old_values,
-            "role": new_role,
-            "user_id": old_values.get("user", {}).get("id")
-            if old_values.get("user") is not None
-            else None,
-        },
-        format="json",
-    )
+    user_id = str(access.user_id) if via == USER else None
+    with mock_reset_connections(document.id, user_id):
+        response = client.put(
+            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+            data={
+                **old_values,
+                "role": new_role,
+                "user_id": old_values.get("user", {}).get("id")
+                if old_values.get("user") is not None
+                else None,
+            },
+            format="json",
+        )
 
-    assert response.status_code == 200
-    access.refresh_from_db()
-    assert access.role == new_role
+        assert response.status_code == 200
+        access.refresh_from_db()
+        assert access.role == new_role
 
 
 @pytest.mark.parametrize("via", VIA)
 def test_api_document_accesses_update_owner_self_child(
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     A user who is owner of a document should be allowed to update
@@ -1054,11 +1084,13 @@ def test_api_document_accesses_update_owner_self_child(
     old_values = serializers.DocumentAccessSerializer(instance=access).data
     new_role = random.choice(["administrator", "editor", "reader"])
 
-    response = client.put(
-        f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-        data={**old_values, "role": new_role},
-        format="json",
-    )
+    user_id = str(access.user_id) if via == USER else None
+    with mock_reset_connections(document.id, user_id):
+        response = client.put(
+            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+            data={**old_values, "role": new_role},
+            format="json",
+        )
 
     assert response.status_code == 200
     access.refresh_from_db()
@@ -1138,6 +1170,7 @@ def test_api_document_accesses_delete_reader_or_editor(via, role, mock_user_team
 def test_api_document_accesses_delete_administrators_except_owners(
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     Users who are administrators in a document should be allowed to delete an access
@@ -1166,13 +1199,14 @@ def test_api_document_accesses_delete_administrators_except_owners(
     assert models.DocumentAccess.objects.count() == 2
     assert models.DocumentAccess.objects.filter(user=access.user).exists()
 
-    with mock.patch("core.api.viewsets.posthog_capture") as mock_capture:
-        response = client.delete(
-            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-        )
+    with mock_reset_connections(document.id, str(access.user_id)):
+        with mock.patch("core.api.viewsets.posthog_capture") as mock_capture:
+            response = client.delete(
+                f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+            )
 
-    assert response.status_code == 204
-    assert models.DocumentAccess.objects.count() == 1
+        assert response.status_code == 204
+        assert models.DocumentAccess.objects.count() == 1
 
     # The access deletion should be tracked in PostHog
     mock_capture.assert_called_once_with(
@@ -1221,6 +1255,7 @@ def test_api_document_accesses_delete_administrator_on_owners(via, mock_user_tea
 def test_api_document_accesses_delete_owners(
     via,
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     Users should be able to delete the document access of another user
@@ -1245,10 +1280,11 @@ def test_api_document_accesses_delete_owners(
     assert models.DocumentAccess.objects.count() == 2
     assert models.DocumentAccess.objects.filter(user=access.user).exists()
 
-    with mock.patch("core.api.viewsets.posthog_capture") as mock_capture:
-        response = client.delete(
-            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-        )
+    with mock_reset_connections(document.id, str(access.user_id)):
+        with mock.patch("core.api.viewsets.posthog_capture") as mock_capture:
+            response = client.delete(
+                f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+            )
 
     assert response.status_code == 204
     assert models.DocumentAccess.objects.count() == 1
@@ -1291,7 +1327,9 @@ def test_api_document_accesses_delete_owners_last_owner_root(via, mock_user_team
     assert models.DocumentAccess.objects.count() == 2
 
 
-def test_api_document_accesses_delete_owners_last_owner_child_user():
+def test_api_document_accesses_delete_owners_last_owner_child_user(
+    mock_reset_connections,
+):
     """
     It should be possible to delete the last owner access from a document that is not a root.
     """
@@ -1307,9 +1345,10 @@ def test_api_document_accesses_delete_owners_last_owner_child_user():
     )
 
     assert models.DocumentAccess.objects.count() == 2
-    response = client.delete(
-        f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-    )
+    with mock_reset_connections(document.id, str(access.user_id)):
+        response = client.delete(
+            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+        )
 
     assert response.status_code == 204
     assert models.DocumentAccess.objects.count() == 1
@@ -1320,6 +1359,7 @@ def test_api_document_accesses_delete_owners_last_owner_child_user():
 )
 def test_api_document_accesses_delete_owners_last_owner_child_team(
     mock_user_teams,
+    mock_reset_connections,
 ):
     """
     It should be possible to delete the last owner access from a document that
@@ -1338,9 +1378,10 @@ def test_api_document_accesses_delete_owners_last_owner_child_team(
     )
 
     assert models.DocumentAccess.objects.count() == 2
-    response = client.delete(
-        f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
-    )
+    with mock_reset_connections(document.id, str(access.user_id)):
+        response = client.delete(
+            f"/api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
+        )
 
     assert response.status_code == 204
     assert models.DocumentAccess.objects.count() == 1

src/backend/core/tests/documents/test_api_documents_link_configuration.py

@@ -1,5 +1,8 @@
 """Tests for link configuration of documents on API endpoint"""
 
+from contextlib import contextmanager
+from unittest import mock
+
 import pytest
 from rest_framework.test import APIClient
 
@@ -10,6 +13,25 @@ from core.tests.conftest import TEAM, USER, VIA
 pytestmark = pytest.mark.django_db
 
 
+@pytest.fixture(name="mock_reset_connections")
+def mock_reset_connections_fixture():
+    """
+    Provide a context manager that patches the ``reset_service_connections_in_cascade``
+    Celery task and asserts its ``delay`` method is called exactly once for the given
+    document when leaving the context.
+    """
+
+    @contextmanager
+    def _mock_reset_connections(document_id):
+        with mock.patch(
+            "core.api.viewsets.reset_service_connections_in_cascade.delay"
+        ) as mock_delay:
+            yield mock_delay
+            mock_delay.assert_called_once_with(str(document_id))
+
+    return _mock_reset_connections
+
+
 @pytest.mark.parametrize("role", models.LinkRoleChoices.values)
 @pytest.mark.parametrize("reach", models.LinkReachChoices.values)
 def test_api_documents_link_configuration_update_anonymous(reach, role):
@@ -119,6 +141,7 @@ def test_api_documents_link_configuration_update_authenticated_related_success(
     via,
     role,
     mock_user_teams,
+    mock_reset_connections,  # pylint: disable=redefined-outer-name
 ):
     """
     A user who is administrator or owner of a document should be allowed to update
@@ -148,17 +171,18 @@ def test_api_documents_link_configuration_update_authenticated_related_success(
         )
     ).data
 
-    response = client.put(
-        f"/api/v1.0/documents/{document.id!s}/link-configuration/",
-        new_document_values,
-        format="json",
-    )
-    assert response.status_code == 200
+    with mock_reset_connections(document.id):
+        response = client.put(
+            f"/api/v1.0/documents/{document.id!s}/link-configuration/",
+            new_document_values,
+            format="json",
+        )
+        assert response.status_code == 200
 
-    document = models.Document.objects.get(pk=document.pk)
-    document_values = serializers.LinkDocumentSerializer(instance=document).data
-    for key, value in document_values.items():
-        assert value == new_document_values[key]
+        document = models.Document.objects.get(pk=document.pk)
+        document_values = serializers.LinkDocumentSerializer(instance=document).data
+        for key, value in document_values.items():
+            assert value == new_document_values[key]
 
 
 def test_api_documents_link_configuration_update_role_restricted_forbidden():
@@ -230,7 +254,9 @@ def test_api_documents_link_configuration_update_link_reach_required():
     assert "This field is required" in response.json()["link_reach"][0]
 
 
-def test_api_documents_link_configuration_update_restricted_without_role_success():
+def test_api_documents_link_configuration_update_restricted_without_role_success(
+    mock_reset_connections,  # pylint: disable=redefined-outer-name
+):
     """
     Test that setting link_reach to restricted without specifying link_role succeeds.
     """
@@ -252,15 +278,16 @@ def test_api_documents_link_configuration_update_restricted_without_role_success
         "link_reach": models.LinkReachChoices.RESTRICTED,
     }
 
-    response = client.put(
-        f"/api/v1.0/documents/{document.id!s}/link-configuration/",
-        new_data,
-        format="json",
-    )
+    with mock_reset_connections(document.id):
+        response = client.put(
+            f"/api/v1.0/documents/{document.id!s}/link-configuration/",
+            new_data,
+            format="json",
+        )
 
-    assert response.status_code == 200
-    document.refresh_from_db()
-    assert document.link_reach == models.LinkReachChoices.RESTRICTED
+        assert response.status_code == 200
+        document.refresh_from_db()
+        assert document.link_reach == models.LinkReachChoices.RESTRICTED
 
 
 @pytest.mark.parametrize(
@@ -270,6 +297,7 @@ def test_api_documents_link_configuration_update_restricted_without_role_success
 def test_api_documents_link_configuration_update_non_restricted_with_valid_role_success(
     reach,
     role,
+    mock_reset_connections,  # pylint: disable=redefined-outer-name
 ):
     """
     Test that setting non-restricted link_reach with valid link_role succeeds.
@@ -292,16 +320,17 @@ def test_api_documents_link_configuration_update_non_restricted_with_valid_role_
         "link_role": role,
     }
 
-    response = client.put(
-        f"/api/v1.0/documents/{document.id!s}/link-configuration/",
-        new_data,
-        format="json",
-    )
+    with mock_reset_connections(document.id):
+        response = client.put(
+            f"/api/v1.0/documents/{document.id!s}/link-configuration/",
+            new_data,
+            format="json",
+        )
 
-    assert response.status_code == 200
-    document.refresh_from_db()
-    assert document.link_reach == reach
-    assert document.link_role == role
+        assert response.status_code == 200
+        document.refresh_from_db()
+        assert document.link_reach == reach
+        assert document.link_role == role
 
 
 def test_api_documents_link_configuration_update_with_ancestor_constraints():

src/backend/core/tests/external_api/test_external_api_documents_link_configuration.py

@@ -6,6 +6,8 @@ because the resource server viewsets inherit from the api viewsets.
 
 """
 
+from unittest.mock import patch
+
 from django.test import override_settings
 
 import pytest
@@ -59,8 +61,9 @@ def test_external_api_documents_link_configuration_not_allowed(
         },
     },
 )
+@patch("core.api.viewsets.reset_service_connections_in_cascade.delay")
 def test_external_api_documents_link_configuration_can_be_allowed(
-    user_token, resource_server_backend, user_specific_sub
+    mock_reset, user_token, resource_server_backend, user_specific_sub
 ):
     """
     Connected users SHOULD be allowed to update the link configuration of a document
@@ -98,3 +101,6 @@ def test_external_api_documents_link_configuration_can_be_allowed(
     document.refresh_from_db()
     assert document.link_reach == models.LinkReachChoices.PUBLIC
     assert document.link_role == models.LinkRoleChoices.EDITOR
+
+    # the collaboration server should be notified through the Celery task
+    mock_reset.assert_called_once_with(str(document.id))

src/backend/core/tests/test_tasks_access.py

@@ -0,0 +1,83 @@
+"""
+Tests for the `reset_service_connections_in_cascade` Celery task in the
+core.tasks.access module.
+"""
+
+from unittest import mock
+
+import pytest
+
+from core import factories
+from core.services.yhub_services import ServiceUnavailableError
+from core.tasks.access import reset_service_connections_in_cascade
+
+pytestmark = pytest.mark.django_db
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_resets_the_document(mock_service):
+    """The task should reset the connections of the document it is given."""
+    document = factories.DocumentFactory()
+
+    reset_service_connections_in_cascade(str(document.id))
+
+    mock_service.return_value.reset_connections.assert_called_once_with(document, None)
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_forwards_the_user_id(mock_service):
+    """The user whose access changed should be forwarded to the service."""
+    document = factories.DocumentFactory()
+
+    reset_service_connections_in_cascade(str(document.id), "user-id")
+
+    mock_service.return_value.reset_connections.assert_called_once_with(
+        document, "user-id"
+    )
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_in_cascade(mock_service):
+    """
+    A document inherits the accesses of its ancestors, so the whole subtree
+    should be reset, the document itself included and its ancestors left out.
+    """
+    parent = factories.DocumentFactory()
+    document = factories.DocumentFactory(parent=parent)
+    child = factories.DocumentFactory(parent=document)
+    grand_child = factories.DocumentFactory(parent=child)
+    factories.DocumentFactory()  # a document of another tree
+
+    reset_service_connections_in_cascade(str(document.id))
+
+    assert mock_service.return_value.reset_connections.call_args_list == [
+        mock.call(document, None),
+        mock.call(child, None),
+        mock.call(grand_child, None),
+    ]
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_unknown_document(mock_service):
+    """A document deleted in the meantime should not reach the service."""
+    reset_service_connections_in_cascade("d43ea3c5-b8ee-4a4a-9c60-2ad7a1d9e6cf")
+
+    mock_service.return_value.reset_connections.assert_not_called()
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_keeps_going_on_failure(mock_service):
+    """A document failing should not deprive the ones after it of their reset."""
+    document = factories.DocumentFactory()
+    child = factories.DocumentFactory(parent=document)
+    mock_service.return_value.reset_connections.side_effect = [
+        ServiceUnavailableError("yhub is down"),
+        None,
+    ]
+
+    reset_service_connections_in_cascade(str(document.id))
+
+    assert mock_service.return_value.reset_connections.call_args_list == [
+        mock.call(document, None),
+        mock.call(child, None),
+    ]