fix(version-history): FTRS-1627 Update entity id to use field name not document

Author: prjh2-nhs

services/data-migration/src/version_history/stream_processor.py

@@ -34,7 +34,15 @@
 
 def _extract_record_metadata(
     record: DynamoDBRecord,
-) -> tuple[str, str, Optional[Dict[str, Any]], Optional[Dict[str, Any]], str]:
+) -> tuple[
+    str,
+    str,
+    Optional[Dict[str, Any]],
+    Optional[Dict[str, Any]],
+    str,
+    str,
+    Dict[str, Any],
+]:
     """Extract metadata from stream record."""
     event_source_arn = record.event_source_arn or ""
     full_table_name = extract_table_name_from_arn(event_source_arn)
@@ -45,49 +53,52 @@ def _extract_record_metadata(
     new_image = record.dynamodb.new_image
 
     record_id = keys.get("id")
+    # Field name for composite key: "document" for gp org entities, may differ for pharmacy
+    field_name = keys.get("field", "document")
     event_name = record.get("eventName", "MODIFY")
 
-    LOGGER.log(
-        VersionHistoryLogBase.VH_PROCESSOR_006,
-        entity_name=entity_name,
-        record_id=record_id,
-        field_name="document",
-        has_old_image=old_image is not None,
-        has_new_image=new_image is not None,
-        keys=keys,
-    )
+    return entity_name, event_name, old_image, new_image, record_id, field_name, keys
 
-    LOGGER.log(
-        VersionHistoryLogBase.VH_PROCESSOR_003,
-        entity_name=entity_name,
-        record_id=record_id,
-        field_name="document",
-        event_type=event_name,
-    )
-
-    return entity_name, event_name, old_image, new_image, record_id
 
-
-def _extract_document_values(
+def _extract_field_values(
     old_image: Optional[Dict[str, Any]],
     new_image: Optional[Dict[str, Any]],
+    field_name: str,
 ) -> tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
-    """Extract document values, excluding system fields."""
-    old_value = (
-        {k: v for k, v in old_image.items() if k not in SYSTEM_FIELDS}
-        if old_image
-        else None
-    )
-    new_value = (
-        {k: v for k, v in new_image.items() if k not in SYSTEM_FIELDS}
-        if new_image
-        else None
-    )
+    """Extract field values from DynamoDB images.
+
+    For 'document' field: extracts entire record excluding system fields.
+    For specific fields: could extract individual field values.
+
+    Args:
+        old_image: Previous DynamoDB image
+        new_image: New DynamoDB image
+        field_name: Name of field to extract (e.g., 'document')
+
+    Returns:
+        Tuple of (old_value, new_value)
+    """
+    # For 'document' field type, extract entire record minus system fields
+    if field_name == "document":
+        old_value = (
+            {k: v for k, v in old_image.items() if k not in SYSTEM_FIELDS}
+            if old_image
+            else None
+        )
+        new_value = (
+            {k: v for k, v in new_image.items() if k not in SYSTEM_FIELDS}
+            if new_image
+            else None
+        )
+    else:
+        # For specific field tracking, extract just that field
+        old_value = old_image.get(field_name) if old_image else None
+        new_value = new_image.get(field_name) if new_image else None
 
     LOGGER.log(
         VersionHistoryLogBase.VH_PROCESSOR_017,
-        field_name="document",
-        is_document_field=True,
+        field_name=field_name,
+        is_document_field=(field_name == "document"),
         system_fields_count=len(SYSTEM_FIELDS),
     )
 
@@ -128,6 +139,7 @@ def _should_skip_update(
     field_delta: Dict[str, Any],
     entity_name: str,
     record_id: str,
+    field_name: str,
 ) -> bool:
     """Check if UPDATE should be skipped due to no meaningful changes."""
     if change_type != "UPDATE":
@@ -139,7 +151,7 @@ def _should_skip_update(
             VersionHistoryLogBase.VH_PROCESSOR_015,
             entity_name=entity_name,
             record_id=record_id,
-            field_name="document",
+            field_name=field_name,
         )
         return True
     else:
@@ -158,6 +170,7 @@ def _create_version_item(
     field_delta: Dict[str, Any],
     new_image: Optional[Dict[str, Any]],
     old_image: Optional[Dict[str, Any]],
+    field_name: str,
 ) -> Dict[str, Any]:
     """Build version history item for DynamoDB."""
     changed_by = extract_changed_by(new_image or old_image or {})
@@ -169,7 +182,7 @@ def _create_version_item(
         changed_by_display=changed_by.get("display"),
     )
 
-    entity_id = f"{entity_name}|{record_id}|details"
+    entity_id = f"{entity_name}#{record_id}#{field_name}"
     timestamp = datetime.now(UTC)
 
     LOGGER.log(
@@ -182,7 +195,7 @@ def _create_version_item(
         "entity_id": entity_id,
         "timestamp": timestamp.isoformat(),
         "change_type": change_type,
-        "changed_fields": {"details": field_delta},
+        "changed_fields": {field_name: field_delta},
         "changed_by": changed_by,
     }
 
@@ -192,24 +205,44 @@ def process_stream_record(
     version_history_table: "Table",
 ) -> None:
     """Process DynamoDB stream record and write to version history."""
-    entity_name, event_name, old_image, new_image, record_id = _extract_record_metadata(
-        record
+    entity_name, event_name, old_image, new_image, record_id, field_name, keys = (
+        _extract_record_metadata(record)
+    )
+
+    LOGGER.log(
+        VersionHistoryLogBase.VH_PROCESSOR_006,
+        entity_name=entity_name,
+        record_id=record_id,
+        field_name=field_name,
+        has_old_image=old_image is not None,
+        has_new_image=new_image is not None,
+        keys=keys,
+    )
+
+    LOGGER.log(
+        VersionHistoryLogBase.VH_PROCESSOR_003,
+        entity_name=entity_name,
+        record_id=record_id,
+        field_name=field_name,
+        event_type=event_name,
     )
 
-    old_value, new_value = _extract_document_values(old_image, new_image)
+    old_value, new_value = _extract_field_values(old_image, new_image, field_name)
     change_type = _determine_change_type(event_name, old_value, new_value)
     field_delta = compute_field_delta(old_value, new_value)
 
     LOGGER.log(
         VersionHistoryLogBase.VH_PROCESSOR_008,
-        field_name="document",
+        field_name=field_name,
         delta_keys=list(field_delta.keys()),
         has_diff="diff" in field_delta,
         delta=field_delta,
         values_equal=field_delta.get("old") == field_delta.get("new"),
     )
 
-    if _should_skip_update(change_type, field_delta, entity_name, record_id):
+    if _should_skip_update(
+        change_type, field_delta, entity_name, record_id, field_name
+    ):
         return
 
     LOGGER.log(
@@ -225,6 +258,7 @@ def process_stream_record(
         field_delta,
         new_image,
         old_image,
+        field_name,
     )
 
     LOGGER.log(
@@ -238,5 +272,5 @@ def process_stream_record(
         VersionHistoryLogBase.VH_PROCESSOR_002,
         entity_id=version_item["entity_id"],
         change_type=change_type,
-        changed_fields=["details"],
+        changed_fields=[field_name],
     )

services/data-migration/tests/unit/version_history/test_stream_processor.py

@@ -26,13 +26,15 @@ def test_process_organisation_document_field_change(
         call_args = mock_version_history_table.put_item.call_args
         item = call_args.kwargs["Item"]
 
-        expected_entity_id = "organisation|d0d6af8a-1138-5a2f-a4e2-5f489fb44653|details"
+        expected_entity_id = (
+            "organisation#d0d6af8a-1138-5a2f-a4e2-5f489fb44653#document"
+        )
         assert item["entity_id"] == expected_entity_id
         assert item["change_type"] == "UPDATE"
-        assert "details" in item["changed_fields"]
+        assert "document" in item["changed_fields"]
 
-        # Verify that the details field contains a diff of the changes
-        document_delta = item["changed_fields"]["details"]
+        # Verify that the document field contains a diff of the changes
+        document_delta = item["changed_fields"]["document"]
         assert "diff" in document_delta
         assert "old" in document_delta
         assert "new" in document_delta
@@ -172,7 +174,7 @@ def test_process_document_field_handles_insert_event(
 
         assert item["change_type"] == "CREATE"
         expected_changed_fields = {
-            "details": {
+            "document": {
                 "old": None,
                 "new": {
                     "name": "New Practice Name",
@@ -237,7 +239,7 @@ def test_process_document_field_handles_remove_event(
 
         assert item["change_type"] == "DELETE"
         expected_changed_fields = {
-            "details": {
+            "document": {
                 "old": {
                     "name": "Old Location Name",
                     "status": "active",

tests/service_automation/tests/features/data_migration_features/version_history_tracking.feature

@@ -19,10 +19,10 @@ Feature: Version History Tracking
       | name        | Updated Organisation Name            |
     Then a version history record should exist with
       | attribute      | value                                                 |
-      | entity_id      | organisation\|aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa\|details |
+      | entity_id      | organisation#aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa#document |
       | change_type    | UPDATE                                                |
-    And the version history "changed_fields.details.diff.values_changed.root['name'].new_value" should be "Updated Organisation Name"
-    And the version history "changed_fields.details.diff.values_changed.root['name'].old_value" should be "Original Organisation Name"
+    And the version history "changed_fields.document.diff.values_changed.root['name'].new_value" should be "Updated Organisation Name"
+    And the version history "changed_fields.document.diff.values_changed.root['name'].old_value" should be "Original Organisation Name"
     And the version history "changed_by.type" should be "app"
     And the version history record should have a valid timestamp
 
@@ -35,4 +35,4 @@ Feature: Version History Tracking
     When the Organisation document is updated with changes
       | field       | value                                |
       | status      | active                               |
-    Then no version history record should be created for entity "organisation|dddddddd-dddd-dddd-dddd-dddddddddddd|details"
+    Then no version history record should be created for entity "organisation#dddddddd-dddd-dddd-dddd-dddddddddddd#document"