feat: update content libraries API to use upstream entity published events

Author: bradenmacdonald

cms/djangoapps/modulestore_migrator/tasks.py

@@ -895,13 +895,9 @@ def _migrate_container(
     ).publishable_entity_version
 
     # Publish the container
-    # Call post publish events synchronously to avoid
-    # an error when calling `wait_for_post_publish_events`
-    # inside a celery task.
     libraries_api.publish_container_changes(
         container.container_key,
         context.created_by,
-        call_post_publish_events_sync=True,
     )
     context.used_container_slugs.add(container.container_key.container_id)
     return container_publishable_entity_version, None

openedx/core/djangoapps/content_libraries/api/blocks.py

@@ -37,7 +37,6 @@
 )
 from openedx.core.types import User as UserType
 
-from .. import tasks
 from ..models import ContentLibrary
 from .block_metadata import LibraryXBlockMetadata, LibraryXBlockStaticFile
 from .container_metadata import container_subclass_for_olx_tag
@@ -818,13 +817,7 @@ def publish_component_changes(usage_key: LibraryUsageLocatorV2, user_id: int):
     # The core publishing API is based on draft objects, so find the draft that corresponds to this component:
     drafts_to_publish = content_api.get_all_drafts(learning_package.id).filter(entity__entity_ref=component.entity_ref)
     # Publish the component and update anything that needs to be updated (e.g. search index):
-    publish_log = content_api.publish_from_drafts(
-        learning_package.id, draft_qset=drafts_to_publish, published_by=user_id,
-    )
-    # Since this is a single component, it should be safe to process synchronously and in-process:
-    tasks.send_events_after_publish(publish_log.pk, str(library_key))
-    # IF this is found to be a performance issue, we could instead make it async where necessary:
-    # tasks.wait_for_post_publish_events(publish_log, library_key=library_key)
+    content_api.publish_from_drafts(learning_package.id, draft_qset=drafts_to_publish, published_by=user_id)
 
 
 def _component_exists(usage_key: UsageKeyV2) -> bool:

openedx/core/djangoapps/content_libraries/api/containers.py

@@ -17,7 +17,6 @@
 from openedx_events.content_authoring.data import LibraryContainerData
 from openedx_events.content_authoring.signals import LIBRARY_CONTAINER_DELETED
 
-from .. import tasks
 from ..models import ContentLibrary
 from .block_metadata import LibraryXBlockMetadata
 from .container_metadata import (
@@ -249,7 +248,6 @@ def get_containers_contains_item(key: LibraryUsageLocatorV2 | LibraryContainerLo
 def publish_container_changes(
     container_key: LibraryContainerLocator,
     user_id: int | None,
-    call_post_publish_events_sync=False,
 ) -> None:
     """
     [ 🛑 UNSTABLE ] Publish all unpublished changes in a container and all its child
@@ -263,17 +261,7 @@ def publish_container_changes(
     # The core publishing API is based on draft objects, so find the draft that corresponds to this container:
     drafts_to_publish = content_api.get_all_drafts(learning_package.id).filter(entity__pk=container.id)
     # Publish the container, which will also auto-publish any unpublished child components:
-    publish_log = content_api.publish_from_drafts(
-        learning_package.id,
-        draft_qset=drafts_to_publish,
-        published_by=user_id,
-    )
-    # Update the search index (and anything else) for the affected container + blocks
-    # This is mostly synchronous but may complete some work asynchronously if there are a lot of changes.
-    if call_post_publish_events_sync:
-        tasks.send_events_after_publish(publish_log.pk, str(library_key))
-    else:
-        tasks.wait_for_post_publish_events(publish_log, library_key)
+    content_api.publish_from_drafts(learning_package.id, draft_qset=drafts_to_publish, published_by=user_id)
 
 
 def copy_container(container_key: LibraryContainerLocator, user_id: int) -> UserClipboardData:

openedx/core/djangoapps/content_libraries/api/libraries.py

@@ -72,7 +72,7 @@
 
 from openedx.core.types import User as UserType
 
-from .. import permissions, tasks
+from .. import permissions
 from ..constants import ALL_RIGHTS_RESERVED
 from ..models import ContentLibrary, ContentLibraryPermission
 from .exceptions import LibraryAlreadyExists, LibraryPermissionIntegrityError
@@ -758,15 +758,7 @@ def publish_changes(library_key: LibraryLocatorV2, user_id: int | None = None):
     """
     learning_package = ContentLibrary.objects.get_by_key(library_key).learning_package
     assert learning_package is not None  # shouldn't happen but it's technically possible.
-    publish_log = content_api.publish_all_drafts(learning_package.id, published_by=user_id)
-
-    # Update the search index (and anything else) for the affected blocks
-    # This is mostly synchronous but may complete some work asynchronously if there are a lot of changes.
-    tasks.wait_for_post_publish_events(publish_log, library_key)
-
-    # Unlike revert_changes below, we do not have to re-index collections,
-    # because publishing changes does not affect the component counts, and
-    # collections themselves don't have draft/published/unpublished status.
+    content_api.publish_all_drafts(learning_package.id, published_by=user_id)
 
 
 def revert_changes(library_key: LibraryLocatorV2, user_id: int | None = None) -> None:

openedx/core/djangoapps/content_libraries/signal_handlers.py

@@ -56,6 +56,39 @@ def entities_updated(
     fn(learning_package_id=learning_package.id, change_list=[asdict(chg) for chg in change_log.changes])
 
 
+@receiver(content_signals.LEARNING_PACKAGE_ENTITIES_PUBLISHED)
+def entities_published(
+    learning_package: content_signals.LearningPackageEventData,
+    change_log: content_signals.PublishLogEventData,
+    **kwargs,
+) -> None:
+    """
+    Entities (containers/components) have been published - handle that as needed.
+
+    We receive this low-level event from `openedx_content`, and check if it
+    happened in a library. If so, we emit more detailed library-specific events.
+
+    💾 This event is only received after the transaction has committed.
+    ⏳ This event is emitted synchronously and this handler is called
+       synchronously. If a lot of entities were published, we need to dispatch
+       an asynchronous handler to deal with them to avoid slowdowns. If only one
+       entity was published, we want to deal with that synchronously so that we
+       can show the user correct data when the current requests completes.
+    """
+    try:
+        library = ContentLibrary.objects.get(learning_package_id=learning_package.id)
+    except ContentLibrary.DoesNotExist:
+        return  # We don't care about non-library events.
+
+    if len(change_log.changes) == 1:
+        fn = tasks.send_events_after_publish
+    else:
+        # More than one entity was published at once. Handle asynchronously:
+        fn = tasks.send_events_after_publish.delay
+
+    fn(publish_log_id=change_log.publish_log_id, library_key_str=str(library.library_key))
+
+
 @receiver(content_signals.LEARNING_PACKAGE_COLLECTION_CHANGED)
 def collection_updated(
     learning_package: content_signals.LearningPackageEventData,

openedx/core/djangoapps/content_libraries/tasks.py

@@ -21,6 +21,7 @@
 import logging
 import os
 import shutil
+from collections.abc import Iterable
 from datetime import datetime
 from io import StringIO
 from tempfile import NamedTemporaryFile, mkdtemp
@@ -204,14 +205,16 @@ def check_container_content_changes(
     else:
         old_entity_list_id = None
     if new_version_id:
-        new_version = content_api.get_container_version(new_version_id) if new_version_id else None
+        new_version = content_api.get_container_version(new_version_id)
         new_entity_list_id = new_version.entity_list_id
     else:
         new_entity_list_id = None
 
+    old_child_ids: Iterable[PublishableEntity.ID]
+    new_child_ids: Iterable[PublishableEntity.ID]
     # If the title has changed, we notify ALL children that their parent container(s) have changed, e.g. to update the
     # list of "units this component is used in", "sections this subsection is used in", etc. in the search index
-    title_changed: bool = old_version and new_version and old_version.title != new_version.title
+    title_changed: bool = bool(old_version and new_version) and (old_version.title != new_version.title)
     if title_changed:
         # TODO: there is no "get entity list for container version" API in openedx_content
         new_child_ids = new_version.entity_list.entitylistrow_set.values_list("entity_id", flat=True)
@@ -297,7 +300,7 @@ def send_collections_changed_events(
 
 @shared_task(base=LoggedTask)
 @set_code_owner_attribute
-def send_events_after_publish(publish_log_pk: int, library_key_str: str) -> None:
+def send_events_after_publish(publish_log_id: int, library_key_str: str) -> None:
     """
     Send events to trigger actions like updating the search index, after we've
     published some items in a library.
@@ -311,12 +314,11 @@ def send_events_after_publish(publish_log_pk: int, library_key_str: str) -> None
     event handlers like updating the search index may a while to complete in
     that case.
     """
-    publish_log = PublishLog.objects.get(pk=publish_log_pk)
+    publish_log = PublishLog.objects.get(id=publish_log_id)
     library_key = LibraryLocatorV2.from_string(library_key_str)
     affected_entities = publish_log.records.select_related(
         "entity", "entity__container", "entity__container__container_type", "entity__component",
     ).all()
-    affected_containers: set[LibraryContainerLocator] = set()
 
     # Update anything that needs to be updated (e.g. search index):
     for record in affected_entities:
@@ -330,61 +332,21 @@ def send_events_after_publish(publish_log_pk: int, library_key_str: str) -> None
             LIBRARY_BLOCK_PUBLISHED.send_event(
                 library_block=LibraryBlockData(library_key=library_key, usage_key=usage_key)
             )
-            # Publishing a container will auto-publish its children, but publishing a single component or all changes
-            # in the library will NOT usually include any parent containers. But we do need to notify listeners that the
-            # parent container(s) have changed, e.g. so the search index can update the "has_unpublished_changes"
-            try:
-                for parent_container in api.get_containers_contains_item(usage_key):
-                    affected_containers.add(parent_container.container_key)
-                    # TODO: should this be a CONTAINER_CHILD_PUBLISHED event instead of CONTAINER_PUBLISHED ?
-            except api.ContentLibraryBlockNotFound:
-                # The component has been deleted.
-                pass
         elif hasattr(record.entity, "container"):
             container_key = api.library_container_locator(library_key, record.entity.container)
-            affected_containers.add(container_key)
-
-            try:
-                # We do need to notify listeners that the parent container(s) have changed,
-                # e.g. so the search index can update the "has_unpublished_changes"
-                for parent_container in api.get_containers_contains_item(container_key):
-                    affected_containers.add(parent_container.container_key)
-            except api.ContentLibraryContainerNotFound:
-                # The deleted children remains in the entity, so, in this case, the container may not be found.
-                pass
+            # Note: this container may have been directly published, or perhaps one of its children was published and
+            # it hasn't technically changed. Such ancestors of published entities are still included in the publish log.
+            # .. event_implemented_name: LIBRARY_CONTAINER_PUBLISHED
+            # .. event_type: org.openedx.content_authoring.content_library.container.published.v1
+            LIBRARY_CONTAINER_PUBLISHED.send_event(
+                library_container=LibraryContainerData(container_key=container_key)
+            )
         else:
             log.warning(
                 f"PublishableEntity {record.entity.pk} / {record.entity.entity_ref} "
                 "was modified during publish operation but is of unknown type."
             )
 
-    for container_key in affected_containers:
-        # .. event_implemented_name: LIBRARY_CONTAINER_PUBLISHED
-        # .. event_type: org.openedx.content_authoring.content_library.container.published.v1
-        LIBRARY_CONTAINER_PUBLISHED.send_event(
-            library_container=LibraryContainerData(container_key=container_key)
-        )
-
-
-def wait_for_post_publish_events(publish_log: PublishLog, library_key: LibraryLocatorV2):
-    """
-    After publishing some changes, trigger the required event handlers (e.g.
-    update the search index). Try to wait for that to complete before returning,
-    up to some reasonable timeout, and then finish anything remaining
-    asynchonrously.
-    """
-    # Update the search index (and anything else) for the affected blocks
-    result = send_events_after_publish.apply_async(args=(publish_log.pk, str(library_key)))
-    # Try waiting a bit for those post-publish events to be handled:
-    try:
-        result.get(timeout=15)
-    except TimeoutError:
-        pass
-        # This is fine! The search index is still being updated, and/or other
-        # event handlers are still following up on the results, but the publish
-        # already *did* succeed, and the events will continue to be processed in
-        # the background by the celery worker until everything is updated.
-
 
 def _filter_child(store, usage_key, capa_type):
     """