Migrate RequestDiffRefresh to prefect

backend/infrahub/core/diff/models.py

@@ -13,3 +13,10 @@ class RequestDiffUpdate(BaseModel):
     name: str | None = None
     from_time: str | None = None
     to_time: str | None = None
+
+
+class RequestDiffRefresh(BaseModel):
+    """Request diff be recalculated from scratch."""
+
+    branch_name: str = Field(..., description="The branch associated with the diff")
+    diff_id: str = Field(..., description="The id for this diff")

backend/infrahub/core/diff/tasks.py

@@ -2,7 +2,7 @@
 
 from infrahub.core import registry
 from infrahub.core.diff.coordinator import DiffCoordinator
-from infrahub.core.diff.models import RequestDiffUpdate
+from infrahub.core.diff.models import RequestDiffRefresh, RequestDiffUpdate
 from infrahub.dependencies.registry import get_component_registry
 from infrahub.log import get_logger
 from infrahub.services import services
@@ -26,3 +26,15 @@ async def update_diff(model: RequestDiffUpdate) -> None:
         to_time=model.to_time,
         name=model.name,
     )
+
+
+@flow(name="diff-refresh")
+async def refresh_diff(model: RequestDiffRefresh) -> None:
+    service = services.service
+
+    component_registry = get_component_registry()
+    base_branch = await registry.get_branch(db=service.database, branch=registry.default_branch)
+    diff_branch = await registry.get_branch(db=service.database, branch=model.branch_name)
+
+    diff_coordinator = await component_registry.get_component(DiffCoordinator, db=service.database, branch=diff_branch)
+    await diff_coordinator.recalculate(base_branch=base_branch, diff_branch=diff_branch, diff_id=model.diff_id)

backend/infrahub/graphql/mutations/artifact_definition.py

@@ -6,10 +6,10 @@
 from typing_extensions import Self
 
 from infrahub.core.schema import NodeSchema
+from infrahub.git.models import RequestArtifactDefinitionGenerate
 from infrahub.log import get_logger
+from infrahub.workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE
 
-from ...git.models import RequestArtifactDefinitionGenerate
-from ...workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE
 from .main import InfrahubMutationMixin, InfrahubMutationOptions
 
 if TYPE_CHECKING:

backend/infrahub/graphql/mutations/diff.py

@@ -5,10 +5,9 @@
 
 from infrahub.core import registry
 from infrahub.core.diff.coordinator import DiffCoordinator
+from infrahub.core.diff.models import RequestDiffUpdate
 from infrahub.dependencies.registry import get_component_registry
-
-from ...core.diff.models import RequestDiffUpdate
-from ...workflows.catalogue import REQUEST_DIFF_UPDATE
+from infrahub.workflows.catalogue import REQUEST_DIFF_UPDATE
 
 if TYPE_CHECKING:
     from ..initialization import GraphqlContext

backend/infrahub/message_bus/messages/__init__.py

@@ -31,7 +31,6 @@
 from .refresh_registry_rebasedbranch import RefreshRegistryRebasedBranch
 from .refresh_webhook_configuration import RefreshWebhookConfiguration
 from .request_artifactdefinition_check import RequestArtifactDefinitionCheck
-from .request_diff_refresh import RequestDiffRefresh
 from .request_generatordefinition_check import RequestGeneratorDefinitionCheck
 from .request_generatordefinition_run import RequestGeneratorDefinitionRun
 from .request_graphqlquerygroup_update import RequestGraphQLQueryGroupUpdate
@@ -75,7 +74,6 @@
     "refresh.registry.rebased_branch": RefreshRegistryRebasedBranch,
     "refresh.webhook.configuration": RefreshWebhookConfiguration,
     "request.artifact_definition.check": RequestArtifactDefinitionCheck,
-    "request.diff.refresh": RequestDiffRefresh,
     "request.generator_definition.check": RequestGeneratorDefinitionCheck,
     "request.generator_definition.run": RequestGeneratorDefinitionRun,
     "request.graphql_query_group.update": RequestGraphQLQueryGroupUpdate,

backend/infrahub/message_bus/operations/__init__.py

@@ -44,7 +44,6 @@
     "refresh.registry.branches": refresh.registry.branches,
     "refresh.registry.rebased_branch": refresh.registry.rebased_branch,
     "refresh.webhook.configuration": refresh.webhook.configuration,
-    "request.diff.refresh": requests.diff.refresh,
     "request.generator_definition.check": requests.generator_definition.check,
     "request.generator_definition.run": requests.generator_definition.run,
     "request.graphql_query_group.update": requests.graphql_query_group.update,

backend/infrahub/message_bus/operations/event/branch.py

@@ -4,14 +4,15 @@
 
 from infrahub.core import registry
 from infrahub.core.diff.model.path import BranchTrackingId
-from infrahub.core.diff.models import RequestDiffUpdate
+from infrahub.core.diff.models import RequestDiffRefresh, RequestDiffUpdate
 from infrahub.core.diff.repository.repository import DiffRepository
 from infrahub.dependencies.registry import get_component_registry
 from infrahub.log import get_logger
 from infrahub.message_bus import InfrahubMessage, messages
 from infrahub.services import InfrahubServices
 from infrahub.workflows.catalogue import (
     GIT_REPOSITORIES_CREATE_BRANCH,
+    REQUEST_DIFF_REFRESH,
     REQUEST_DIFF_UPDATE,
     TRIGGER_ARTIFACT_DEFINITION_GENERATE,
 )
@@ -100,7 +101,12 @@ async def rebased(message: messages.EventBranchRebased, service: InfrahubService
 
     for diff_root in diff_roots_to_refresh:
         if diff_root.base_branch_name != diff_root.diff_branch_name:
-            events.append(messages.RequestDiffRefresh(branch_name=diff_root.diff_branch_name, diff_id=diff_root.uuid))
+            request_diff_refresh_model = RequestDiffRefresh(
+                branch_name=diff_root.diff_branch_name, diff_id=diff_root.uuid
+            )
+            await service.workflow.submit_workflow(
+                workflow=REQUEST_DIFF_REFRESH, parameters={"model": request_diff_refresh_model}
+            )
 
     for event in events:
         event.assign_meta(parent=message)

backend/infrahub/message_bus/operations/requests/__init__.py

@@ -1,6 +1,5 @@
 from . import (
     artifact_definition,
-    diff,
     generator_definition,
     graphql_query_group,
     proposed_change,
@@ -9,7 +8,6 @@
 
 __all__ = [
     "artifact_definition",
-    "diff",
     "generator_definition",
     "graphql_query_group",
     "proposed_change",

backend/infrahub/workflows/catalogue.py

@@ -99,6 +99,13 @@
     function="update_diff",
 )
 
+REQUEST_DIFF_REFRESH = WorkflowDefinition(
+    name="diff-refresh",
+    type=WorkflowType.INTERNAL,
+    module="infrahub.core.diff.tasks",
+    function="refresh_diff",
+)
+
 GIT_REPOSITORIES_SYNC = WorkflowDefinition(
     name="git_repositories_sync",
     type=WorkflowType.INTERNAL,
@@ -152,4 +159,5 @@
     REQUEST_ARTIFACT_DEFINITION_GENERATE,
     REQUEST_GENERATOR_RUN,
     REQUEST_DIFF_UPDATE,
+    REQUEST_DIFF_REFRESH,
 ]

backend/tests/unit/message_bus/operations/event/test_branch.py

@@ -5,15 +5,15 @@
 
 from infrahub.core.branch import Branch
 from infrahub.core.diff.model.path import BranchTrackingId, EnrichedDiffRoot
-from infrahub.core.diff.models import RequestDiffUpdate
+from infrahub.core.diff.models import RequestDiffRefresh, RequestDiffUpdate
 from infrahub.core.diff.repository.repository import DiffRepository
 from infrahub.core.timestamp import Timestamp
 from infrahub.dependencies.component.registry import ComponentDependencyRegistry
 from infrahub.message_bus import messages
 from infrahub.message_bus.operations.event.branch import delete, merge, rebased
 from infrahub.services import InfrahubServices, services
 from infrahub.services.adapters.workflow.local import WorkflowLocalExecution
-from infrahub.workflows.catalogue import REQUEST_DIFF_UPDATE, TRIGGER_ARTIFACT_DEFINITION_GENERATE
+from infrahub.workflows.catalogue import REQUEST_DIFF_REFRESH, REQUEST_DIFF_UPDATE, TRIGGER_ARTIFACT_DEFINITION_GENERATE
 from tests.adapters.message_bus import BusRecorder
 
 
@@ -131,7 +131,7 @@ async def test_rebased(default_branch: Branch, prefect_test_fixture):
 
     recorder = BusRecorder()
     database = MagicMock()
-    service = InfrahubServices(message_bus=recorder, database=database)
+    service = InfrahubServices(message_bus=recorder, database=database, workflow=WorkflowLocalExecution())
     diff_roots = [
         EnrichedDiffRoot(
             base_branch_name="main",
@@ -149,14 +149,30 @@ async def test_rebased(default_branch: Branch, prefect_test_fixture):
     mock_get_component_registry = MagicMock(return_value=mock_component_registry)
     mock_component_registry.get_component.return_value = diff_repo
 
-    with patch("infrahub.message_bus.operations.event.branch.get_component_registry", new=mock_get_component_registry):
+    with (
+        patch("infrahub.message_bus.operations.event.branch.get_component_registry", new=mock_get_component_registry),
+        patch(
+            "infrahub.services.adapters.workflow.local.WorkflowLocalExecution.submit_workflow"
+        ) as mock_submit_workflow,
+    ):
         await rebased(message=message, service=service)
 
+        expected_calls = [
+            call(
+                workflow=REQUEST_DIFF_REFRESH,
+                parameters={"model": RequestDiffRefresh(branch_name=branch_name, diff_id=diff_roots[0].uuid)},
+            ),
+            call(
+                workflow=REQUEST_DIFF_REFRESH,
+                parameters={"model": RequestDiffRefresh(branch_name=branch_name, diff_id=diff_roots[1].uuid)},
+            ),
+        ]
+        mock_submit_workflow.assert_has_calls(expected_calls)
+        assert mock_submit_workflow.call_count == len(expected_calls)
+
     mock_component_registry.get_component.assert_awaited_once_with(DiffRepository, db=database, branch=default_branch)
     diff_repo.get_empty_roots.assert_awaited_once_with(diff_branch_names=[branch_name])
-    assert len(recorder.messages) == 3
+    assert len(recorder.messages) == 1
     assert isinstance(recorder.messages[0], messages.RefreshRegistryRebasedBranch)
     refresh_message: messages.RefreshRegistryRebasedBranch = recorder.messages[0]
     assert refresh_message.branch == "cr1234"
-    assert recorder.messages[1] == messages.RequestDiffRefresh(branch_name=branch_name, diff_id=diff_roots[0].uuid)
-    assert recorder.messages[2] == messages.RequestDiffRefresh(branch_name=branch_name, diff_id=diff_roots[1].uuid)

docs/docs/reference/message-bus-events.mdx

@@ -586,26 +586,6 @@ For more detailed explanations on how to use these events within Infrahub, see t
 | **destination_branch** | The target branch | string | None |
 <!-- vale on -->
 
-<!-- vale off -->
-### Request Diff
-<!-- vale on -->
-
-<!-- vale off -->
-#### Event request.diff.refresh
-<!-- vale on -->
-
-**Description**: Request diff be recalculated from scratch.
-
-**Priority**: 3
-
-<!-- vale off -->
-| Key | Description | Type | Default Value |
-|-----|-------------|------|---------------|
-| **meta** | Meta properties for the message | N/A | None |
-| **branch_name** | The branch associated with the diff | string | None |
-| **diff_id** | The id for this diff | string | None |
-<!-- vale on -->
-
 <!-- vale off -->
 ### Request Generator Definition
 <!-- vale on -->
@@ -1561,27 +1541,6 @@ For more detailed explanations on how to use these events within Infrahub, see t
 | **destination_branch** | The target branch | string | None |
 <!-- vale on -->
 
-<!-- vale off -->
-### Request Diff
-<!-- vale on -->
-
-<!-- vale off -->
-#### Event request.diff.refresh
-<!-- vale on -->
-
-**Description**: Request diff be recalculated from scratch.
-
-**Priority**: 3
-
-
-<!-- vale off -->
-| Key | Description | Type | Default Value |
-|-----|-------------|------|---------------|
-| **meta** | Meta properties for the message | N/A | None |
-| **branch_name** | The branch associated with the diff | string | None |
-| **diff_id** | The id for this diff | string | None |
-<!-- vale on -->
-
 <!-- vale off -->
 ### Request Generator Definition
 <!-- vale on -->