Feature: Rejecting unexpected recent activities

connect/common/models.py

@@ -2445,6 +2445,32 @@ class RecentActivity(models.Model):
         (NEXUS, "Artificial Intelligence Entity"),
     ]
 
+    ACTIONS = dict(
+        ADD=dict(USER="joined-project"),
+        CREATE=dict(
+            TRIGGER="created-trigger",
+            CAMPAIGN="created-campaign",
+            FLOW="created-flow",
+            CHANNEL="created-channel",
+            AI="created-ai",
+            NEXUS="created-ai",
+        ),
+        UPDATE=dict(
+            TRIGGER="edited-trigger",
+            CAMPAIGN="edited-campaign",
+            FLOW="edited-flow",
+            CHANNEL="edited-channel",
+            NEXUS="edited-ai",
+        ),
+        INTEGRATE=dict(AI="integrated-ai"),
+        TRAIN=dict(AI="trained-ai"),
+        DELETE=dict(
+            FLOW="deleted-flow",
+            AI="deleted-ai",
+            NEXUS="deleted-ai",
+        ),
+    )
+
     project = models.ForeignKey(
         Project, on_delete=models.CASCADE, related_name="project_recent_activity"
     )
@@ -2458,31 +2484,7 @@ class RecentActivity(models.Model):
 
     @property
     def action_description_key(self) -> str:
-        actions = dict(
-            ADD=dict(USER="joined-project"),
-            CREATE=dict(
-                TRIGGER="created-trigger",
-                CAMPAIGN="created-campaign",
-                FLOW="created-flow",
-                CHANNEL="created-channel",
-                AI="created-ai",
-                NEXUS="created-ai",
-            ),
-            UPDATE=dict(
-                TRIGGER="edited-trigger",
-                CAMPAIGN="edited-campaign",
-                FLOW="edited-flow",
-                CHANNEL="edited-channel",
-            ),
-            INTEGRATE=dict(AI="integrated-ai"),
-            TRAIN=dict(AI="trained-ai"),
-            DELETE=dict(
-                FLOW="deleted-flow",
-                AI="deleted-ai",
-                NEXUS="deleted-ai",
-            ),
-        )
-        return actions[self.action][self.entity]
+        return self.ACTIONS[self.action][self.entity]
 
     @property
     def user_name(self):

connect/internals/event_driven/consumer/recent_activities.py

@@ -6,10 +6,10 @@
 from connect.usecases.recent_activities.create import RecentActivityUseCase
 
 from .parsers.json_parser import JSONParser
+from connect.usecases.recent_activities.exceptions import InvalidActionEntityCombination
 
 
 class RecentActivitiesConsumer(EDAConsumer):
-
     def consume(self, message: amqp.Message):
         try:
             msg_body = JSONParser.parse(message.body)
@@ -19,6 +19,12 @@ def consume(self, message: amqp.Message):
 
             message.channel.basic_ack(message.delivery_tag)
             print("[RecentActivitiesConsumer] - Recent activity created.")
+        except InvalidActionEntityCombination as exception:
+            capture_exception(exception)
+            message.channel.basic_reject(message.delivery_tag, requeue=False)
+            print(
+                f"[RecentActivitiesConsumer] - Message rejected due to invalid action/entity combination: {exception}"
+            )
         except Exception as exception:
             capture_exception(exception)
             message.channel.basic_reject(message.delivery_tag, requeue=False)

connect/usecases/recent_activities/create.py

@@ -1,15 +1,22 @@
 from connect.authentication.models import User
 from connect.common.models import RecentActivity
+from connect.usecases.recent_activities.exceptions import InvalidActionEntityCombination
 
 
 class RecentActivityUseCase:
-    def create_recent_activity(
-        self,
-        msg_body
-    ) -> RecentActivity:
-
+    def create_recent_activity(self, msg_body) -> RecentActivity:
         user = User.objects.get(email=msg_body.get("user"))
 
-        new_activity = RecentActivity.create_recent_activities(msg_body, user)
+        action = msg_body.get("action")
+        entity = msg_body.get("entity")
+
+        if not self._is_valid_action_entity_combination(action, entity):
+            raise InvalidActionEntityCombination(
+                f"Invalid combination of action '{action}' and entity '{entity}'"
+            )
 
+        new_activity = RecentActivity.create_recent_activities(msg_body, user)
         return new_activity
+
+    def _is_valid_action_entity_combination(self, action, entity):
+        return entity in RecentActivity.ACTIONS.get(action, [])

connect/usecases/recent_activities/exceptions.py

@@ -0,0 +1,2 @@
+class InvalidActionEntityCombination(Exception):
+    pass