fix: Preserve Version when Migrating Legacy Library Block

cms/djangoapps/contentstore/views/library.py

@@ -285,6 +285,8 @@ def library_blocks_view(library, user, response_format):
 
     Assumes that read permissions have been checked before calling this.
     """
+    if migration_info := migration_api.get_migration(library):
+        redirect_here(migration_info.target)
     assert isinstance(library.location.library_key, LibraryLocator)
     assert isinstance(library.location, LibraryUsageLocator)
 

cms/djangoapps/modulestore_migrator/api.py

@@ -108,17 +108,15 @@ def start_bulk_migration_to_library(
     )
 
 
-def is_successfully_migrated(
-    source_key: CourseKey | LibraryLocator,
-    source_version: str | None = None,
-) -> bool:
+def is_successfully_migrated(source_key: CourseKey | LibraryLocator) -> bool:
     """
     Check if the source course/library has been migrated successfully.
     """
-    filters = {"task_status__state": UserTaskStatus.SUCCEEDED}
-    if source_version is not None:
-        filters["source_version"] = source_version
-    return ModulestoreSource.objects.get_or_create(key=str(source_key))[0].migrations.filter(**filters).exists()
+    return ModulestoreSource.objects.get_or_create(
+        key=str(source_key)
+    )[0].migrations.filter(
+        task_status__state=UserTaskStatus.SUCCEEDED
+    ).exists()
 
 
 def get_migration_info(source_keys: list[CourseKey | LibraryLocator]) -> dict:
@@ -163,26 +161,32 @@ def get_all_migrations_info(source_keys: list[CourseKey | LibraryLocator]) -> di
     return dict(results)
 
 
-def get_target_block_usage_keys(source_key: CourseKey | LibraryLocator) -> dict[UsageKey, LibraryUsageLocatorV2 | None]:
+def get_target_block_usage_keys(
+        source_key: CourseKey | LibraryLocator,
+) -> dict[
+    UsageKey,  # Each source usage key, within the legacy library or course.
+    tuple[LibraryUsageLocatorV2, int]  # The target component usage key and version num, if migrated.
+ ]:
     """
-    For given source_key, get a map of legacy block key and its new location in migrated v2 library.
+    Given a source context, get a mapping from its blocks to their migrated components in a v2 library.
     """
-    query_set = ModulestoreBlockMigration.objects.filter(overall_migration__source__key=source_key).select_related(
-        'source', 'target__component__component_type', 'target__learning_package'
-    )
-
-    def construct_usage_key(lib_key_str: str, component: Component) -> LibraryUsageLocatorV2 | None:
-        try:
-            lib_key = LibraryLocatorV2.from_string(lib_key_str)
-        except InvalidKeyError:
-            return None
-        return library_component_usage_key(lib_key, component)
-
-    # Use LibraryUsageLocatorV2 and construct usage key
+    try:
+        source: ModulestoreSource = ModulestoreSource.objects.get(key=source_key)
+    except ModulestoreSource.DoesNotExist:
+        return {}
+    if not (migration := source.forwarded):
+        return {}
+    try:
+        lib_key = LibraryLocatorV2.from_string(migration.target.key)
+    except InvalidKeyError:
+        return {}
     return {
-        obj.source.key: construct_usage_key(obj.target.learning_package.key, obj.target.component)
-        for obj in query_set
-        if obj.source.key is not None and obj.target is not None
+        block_migration.source.key: (
+            library_component_usage_key(lib_key, block_migration.target.component),
+            block_migration.change_log_record.new_version_num if block_migration.change_log_record else None,
+        )
+        for block_migration in migration.block_migrations
+        if block_migration.target
     }
 
 

xmodule/library_content_block.py

@@ -16,7 +16,8 @@
 
 import nh3
 from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
-from opaque_keys.edx.locator import LibraryLocator
+from opaque_keys.edx.locator import LibraryLocator, LibraryLocatorV2
+from opaque_keys.edx.keys import UsageKey
 from web_fragments.fragment import Fragment
 from webob import Response
 from xblock.core import XBlock
@@ -127,7 +128,7 @@ def is_source_lib_migrated_to_v2(self):
         return (
             self.source_library_id
             and self.source_library_version
-            and is_successfully_migrated(self.source_library_key, source_version=self.source_library_version)
+            and is_successfully_migrated(self.source_library_key)
         )
 
     @property
@@ -316,15 +317,14 @@ def _v2_update_children_upstream_version(self):
         library blocks. This essentially converts this legacy block to new ItemBankBlock.
         """
         from cms.djangoapps.modulestore_migrator.api import get_target_block_usage_keys
-        blocks = get_target_block_usage_keys(self.source_library_key)
+        migration_map: dict[UsageKey, tuple[LibraryLocatorV2, int]] = get_target_block_usage_keys(self.source_library_key)
         store = modulestore()
         with store.bulk_operations(self.course_id):
             for child in self.get_children():
                 source_key, _ = self.runtime.modulestore.get_block_original_usage(child.usage_key)
-                child.upstream = str(blocks.get(source_key, ""))
-                # Since after migration, the component in library is in draft state, we want to make sure that sync icon
-                # appears when it is published
-                child.upstream_version = 0
+                target_key, target_version = migration_map.get(source_key, (None, None))
+                child.upstream = str(target_key) if target_key else None
+                child.upstream_version = target_version if (target_key and target_version) else 0
                 # Use `modulestore()` instead of `self.runtime.modulestore` to make sure that the XBLOCK_UPDATED signal
                 # is triggered
                 store.update_item(child, None)