Merge pull request #5853 from opsmill/pog-group-event-ancestors-IFC-1073

ogenstad · web-flow · commit 9508b80634dc · 2025-02-27T11:20:57.000+01:00
Add ancestors to group events
diff --git a/backend/infrahub/core/manager.py b/backend/infrahub/core/manager.py
@@ -451,7 +451,7 @@ async def query_hierarchy(
         limit: Optional[int] = None,
         at: Optional[Union[Timestamp, str]] = None,
         branch: Optional[Union[Branch, str]] = None,
-    ) -> dict[str, Any]:
+    ) -> dict[str, Node]:
         branch = await registry.get_branch(branch=branch, db=db)
         at = Timestamp(at)
 
diff --git a/backend/infrahub/events/group_action.py b/backend/infrahub/events/group_action.py
@@ -16,6 +16,9 @@ class GroupMutatedEvent(InfrahubEvent):
     node_id: str = Field(..., description="The ID of the updated group")
     action: MutationAction = Field(..., description="The action taken on the node")
     members: list[EventNode] = Field(default_factory=list, description="Updated members during this event.")
+    ancestors: list[EventNode] = Field(
+        default_factory=list, description="A list of groups that are ancestors of this group."
+    )
 
     def get_related(self) -> list[dict[str, str]]:
         related = super().get_related()
@@ -26,7 +29,22 @@ def get_related(self) -> list[dict[str, str]]:
                 "infrahub.node.kind": self.kind,
             }
         )
+        related.append(
+            {
+                "prefect.resource.id": self.node_id,
+                "prefect.resource.role": "infrahub.group.update",
+                "infrahub.node.kind": self.kind,
+            }
+        )
+
         for member in self.members:
+            related.append(
+                {
+                    "prefect.resource.id": member.id,
+                    "prefect.resource.role": "infrahub.group.member",
+                    "infrahub.node.kind": member.kind,
+                }
+            )
             related.append(
                 {
                     "prefect.resource.id": member.id,
@@ -35,6 +53,29 @@ def get_related(self) -> list[dict[str, str]]:
                 }
             )
 
+        for ancestor in self.ancestors:
+            related.append(
+                {
+                    "prefect.resource.id": ancestor.id,
+                    "prefect.resource.role": "infrahub.group.ancestor",
+                    "infrahub.node.kind": ancestor.kind,
+                }
+            )
+            related.append(
+                {
+                    "prefect.resource.id": ancestor.id,
+                    "prefect.resource.role": "infrahub.related.node",
+                    "infrahub.node.kind": ancestor.kind,
+                }
+            )
+            related.append(
+                {
+                    "prefect.resource.id": ancestor.id,
+                    "prefect.resource.role": "infrahub.group.update",
+                    "infrahub.node.kind": ancestor.kind,
+                }
+            )
+
         return related
 
     @computed_field
@@ -51,7 +92,10 @@ def get_resource(self) -> dict[str, str]:
         }
 
     def get_payload(self) -> dict[str, Any]:
-        return {"members": [member.model_dump() for member in self.members]}
+        return {
+            "ancestors": [ancestor.model_dump() for ancestor in self.ancestors],
+            "members": [member.model_dump() for member in self.members],
+        }
 
     def get_messages(self) -> list[InfrahubMessage]:
         return []
diff --git a/backend/infrahub/graphql/mutations/relationship.py b/backend/infrahub/graphql/mutations/relationship.py
@@ -15,6 +15,7 @@
     PermissionAction,
     PermissionDecision,
     RelationshipCardinality,
+    RelationshipHierarchyDirection,
 )
 from infrahub.core.manager import NodeManager
 from infrahub.core.query.node import NodeGetKindQuery
@@ -23,6 +24,7 @@
     RelationshipPeerData,
 )
 from infrahub.core.relationship import Relationship
+from infrahub.core.schema import NodeSchema
 from infrahub.database import retry_db_transaction
 from infrahub.events import EventMeta, NodeMutatedEvent
 from infrahub.events.group_action import GroupMemberAddedEvent, GroupMemberRemovedEvent
@@ -89,6 +91,10 @@ async def mutate(
 
         existing_peers = await _collect_current_peers(info=info, data=data, source_node=source)
 
+        group_event_type = _get_group_event_type(
+            node=source, relationship_schema=rel_schema, relationship_name=relationship_name
+        )
+
         async with graphql_context.db.start_transaction() as db:
             peers: list[EventNode] = []
             for node_data in data.get("nodes"):
@@ -99,19 +105,19 @@ async def mutate(
                 await rel.resolve(db=db)
                 # Save it only if it does not exist
                 if rel.get_peer_id() not in existing_peers.keys():
-                    peers.append(EventNode(id=rel.get_peer_id(), kind=rel.get_peer_kind()))
+                    if group_event_type != GroupUpdateType.NONE:
+                        peers.append(EventNode(id=rel.get_peer_id(), kind=nodes[rel.get_peer_id()].get_kind()))
                     node_changelog.create_relationship(relationship=rel)
                     await rel.save(db=db)
 
         if config.SETTINGS.broker.enable and graphql_context.background and node_changelog.has_changes:
-            group_event_type = _get_group_event_type(
-                node=source, relationship_schema=rel_schema, relationship_name=relationship_name
-            )
             if group_event_type == GroupUpdateType.MEMBERS:
+                ancestors = await _collect_ancestors(info=info, node_kind=source.get_schema().kind, node_id=source.id)
                 group_add_event = GroupMemberAddedEvent(
                     node_id=source.id,
                     kind=source.get_schema().kind,
                     members=peers,
+                    ancestors=ancestors,
                     meta=EventMeta(branch=graphql_context.branch, context=graphql_context.get_context()),
                 )
                 graphql_context.background.add_task(graphql_context.active_service.event.send, group_add_event)
@@ -124,9 +130,11 @@ async def mutate(
                     node_kind_map = await node_kind_query.get_node_kind_map()
 
                     for node_id, node_kind in node_kind_map.items():
+                        ancestors = await _collect_ancestors(info=info, node_kind=node_kind, node_id=node_id)
                         group_add_event = GroupMemberAddedEvent(
                             node_id=node_id,
                             kind=node_kind,
+                            ancestors=ancestors,
                             members=[EventNode(id=source.get_id(), kind=source.get_kind())],
                             meta=EventMeta(branch=graphql_context.branch, context=graphql_context.get_context()),
                         )
@@ -175,6 +183,9 @@ async def mutate(
         )
 
         existing_peers = await _collect_current_peers(info=info, data=data, source_node=source)
+        group_event_type = _get_group_event_type(
+            node=source, relationship_schema=rel_schema, relationship_name=relationship_name
+        )
 
         async with graphql_context.db.start_transaction() as db:
             peers: list[EventNode] = []
@@ -186,19 +197,19 @@ async def mutate(
                     # it would be more query efficient
                     rel = Relationship(schema=rel_schema, branch=graphql_context.branch, node=source)
                     await rel.load(db=db, data=existing_peers[node_data.get("id")])
-                    peers.append(EventNode(id=rel.get_peer_id(), kind=rel.get_peer_kind()))
+                    if group_event_type != GroupUpdateType.NONE:
+                        peers.append(EventNode(id=rel.get_peer_id(), kind=nodes[rel.get_peer_id()].get_kind()))
                     node_changelog.delete_relationship(relationship=rel)
                     await rel.delete(db=db)
 
         if config.SETTINGS.broker.enable and graphql_context.background and node_changelog.has_changes:
-            group_event_type = _get_group_event_type(
-                node=source, relationship_schema=rel_schema, relationship_name=relationship_name
-            )
             if group_event_type == GroupUpdateType.MEMBERS:
+                ancestors = await _collect_ancestors(info=info, node_kind=source.get_schema().kind, node_id=source.id)
                 group_remove_event = GroupMemberRemovedEvent(
                     node_id=source.id,
                     kind=source.get_schema().kind,
                     members=peers,
+                    ancestors=ancestors,
                     meta=EventMeta(branch=graphql_context.branch, context=graphql_context.get_context()),
                 )
                 graphql_context.background.add_task(graphql_context.active_service.event.send, group_remove_event)
@@ -210,6 +221,7 @@ async def mutate(
                     node_kind_map = await node_kind_query.get_node_kind_map()
 
                     for node_id, node_kind in node_kind_map.items():
+                        ancestors = await _collect_ancestors(info=info, node_kind=node_kind, node_id=node_id)
                         group_remove_event = GroupMemberRemovedEvent(
                             node_id=node_id,
                             kind=node_kind,
@@ -332,6 +344,24 @@ async def _validate_peer_types(
                     )
 
 
+async def _collect_ancestors(info: GraphQLResolveInfo, node_kind: str, node_id: str) -> list[EventNode]:
+    graphql_context: GraphqlContext = info.context
+    schema = graphql_context.db.schema.get(name=node_kind, branch=graphql_context.branch)
+
+    if not isinstance(schema, NodeSchema):
+        return []
+
+    ancestors = await NodeManager.query_hierarchy(
+        db=graphql_context.db,
+        branch=graphql_context.branch,
+        direction=RelationshipHierarchyDirection.ANCESTORS,
+        id=node_id,
+        node_schema=schema,
+        filters={"id": None},
+    )
+    return [EventNode(id=ancestor.get_id(), kind=ancestor.get_kind()) for ancestor in ancestors.values()]
+
+
 async def _collect_current_peers(
     info: GraphQLResolveInfo, data: RelationshipNodesInput, source_node: Node
 ) -> dict[str, RelationshipPeerData]:
diff --git a/backend/infrahub/graphql/types/event.py b/backend/infrahub/graphql/types/event.py
@@ -113,6 +113,14 @@ class Meta:
     attributes = Field(List(of_type=NonNull(InfrahubMutatedAttribute), required=True), required=True)
 
 
+class GroupEvent(ObjectType):
+    class Meta:
+        interfaces = (EventNodeInterface,)
+
+    members = List(NonNull(RelatedNode), required=True, description="Group members modified in this event")
+    ancestors = List(NonNull(RelatedNode), required=True, description="Ancestor groups of this impacted group")
+
+
 class StandardEvent(ObjectType):
     class Meta:
         interfaces = (EventNodeInterface,)
@@ -128,5 +136,7 @@ class Meta:
     "infrahub.branch.merged": BranchMergedEvent,
     "infrahub.branch.rebased": BranchRebasedEvent,
     "infrahub.branch.deleted": BranchDeletedEvent,
+    "infrahub.group.member_added": GroupEvent,
+    "infrahub.group.member_removed": GroupEvent,
     "undefined": StandardEvent,
 }
diff --git a/backend/infrahub/task_manager/event.py b/backend/infrahub/task_manager/event.py
@@ -125,21 +125,38 @@ def _return_branch_merged(self) -> dict[str, Any]:
     def _return_branch_rebased(self) -> dict[str, Any]:
         return {"rebased_branch": self._get_branch_name_from_resource()}
 
+    def _return_group_event(self) -> dict[str, Any]:
+        members = []
+        ancestors = []
+
+        for resource in self.related:
+            if resource.role == "infrahub.group.member" and resource.get("infrahub.node.kind"):
+                members.append({"id": resource.id, "kind": resource.get("infrahub.node.kind")})
+            elif resource.role == "infrahub.group.ancestor" and resource.get("infrahub.node.kind"):
+                ancestors.append({"id": resource.id, "kind": resource.get("infrahub.node.kind")})
+
+        return {"members": members, "ancestors": ancestors}
+
     def _return_event_specifics(self) -> dict[str, Any]:
         """Return event specific data based on the type of event being processed"""
+
+        event_specifics = {}
+
         match self.event:
             case "infrahub.node.created" | "infrahub.node.updated" | "infrahub.node.deleted":
-                return self._return_node_mutation()
+                event_specifics = self._return_node_mutation()
             case "infrahub.branch.created":
-                return self._return_branch_created()
+                event_specifics = self._return_branch_created()
             case "infrahub.branch.deleted":
-                return self._return_branch_deleted()
+                event_specifics = self._return_branch_deleted()
             case "infrahub.branch.merged":
-                return self._return_branch_merged()
+                event_specifics = self._return_branch_merged()
             case "infrahub.branch.rebased":
-                return self._return_branch_rebased()
+                event_specifics = self._return_branch_rebased()
+            case "infrahub.group.member_added" | "infrahub.group.member_removed":
+                event_specifics = self._return_group_event()
 
-        return {}
+        return event_specifics
 
     def to_graphql(self) -> dict[str, Any]:
         response = {
diff --git a/backend/tests/unit/graphql/queries/test_event.py b/backend/tests/unit/graphql/queries/test_event.py
diff --git a/backend/tests/unit/graphql/test_mutation_relationship.py b/backend/tests/unit/graphql/test_mutation_relationship.py
