Always flush entity manager challenges (#12018)

Author: rickyrombo

packages/discovery-provider/src/challenges/challenge_event_bus.py

@@ -1,5 +1,4 @@
 import json
-import logging
 from collections import defaultdict
 from contextlib import contextmanager
 from datetime import datetime
@@ -49,8 +48,9 @@
     trending_underground_track_challenge_manager,
 )
 from src.utils.redis_connection import get_redis
+from src.utils.structured_logger import StructuredLogger
 
-logger = logging.getLogger(__name__)
+logger = StructuredLogger(__name__)
 REDIS_QUEUE_PREFIX = "challenges-event-queue"
 
 

packages/discovery-provider/src/tasks/entity_manager/entities/comment.py

@@ -175,14 +175,13 @@ def create_comment(params: ManageEntityParameters):
 
     params.add_record(comment_id, comment_record, EntityType.COMMENT)
 
-    with challenge_bus.use_scoped_dispatch_queue():
-        challenge_bus.dispatch(
-            ChallengeEvent.first_weekly_comment,
-            params.block_number,
-            params.block_datetime,
-            user_id,
-            {"created_at": params.block_datetime.timestamp()},
-        )
+    challenge_bus.dispatch(
+        ChallengeEvent.first_weekly_comment,
+        params.block_number,
+        params.block_datetime,
+        user_id,
+        {"created_at": params.block_datetime.timestamp()},
+    )
 
     if (
         not is_reply
@@ -746,18 +745,17 @@ def pin_comment(params: ManageEntityParameters):
         )
 
         if artist_is_verified:
-            with params.challenge_bus.use_scoped_dispatch_queue():
-                params.challenge_bus.dispatch(
-                    ChallengeEvent.pinned_comment,
-                    params.block_number,
-                    params.block_datetime,
-                    comment_user_id,
-                    {
-                        "track_id": track_id,
-                        "comment_id": comment_id,
-                        "track_owner_id": track_owner_id,
-                    },
-                )
+            params.challenge_bus.dispatch(
+                ChallengeEvent.pinned_comment,
+                params.block_number,
+                params.block_datetime,
+                comment_user_id,
+                {
+                    "track_id": track_id,
+                    "comment_id": comment_id,
+                    "track_owner_id": track_owner_id,
+                },
+            )
 
     params.add_record(track_id, track, EntityType.TRACK)
 

packages/discovery-provider/src/tasks/entity_manager/entities/social_features.py

@@ -94,50 +94,47 @@ def create_social_record(params: ManageEntityParameters):
                 record_type=record_type,
             )
 
-    with params.challenge_bus.use_scoped_dispatch_queue():
-        # dispatch repost, favorite, follow challenges
-        if create_record and params.action in action_to_challenge_event:
-            challenge_event = action_to_challenge_event[params.action]
-            params.challenge_bus.dispatch(
-                challenge_event,
-                params.block_number,
-                params.block_datetime,
-                params.user_id,
-            )
+    # dispatch repost, favorite, follow challenges
+    if create_record and params.action in action_to_challenge_event:
+        challenge_event = action_to_challenge_event[params.action]
+        params.challenge_bus.dispatch(
+            challenge_event,
+            params.block_number,
+            params.block_datetime,
+            params.user_id,
+        )
+
+    # dispatch verified cosign challenge
+    if (
+        params.action in [Action.SAVE, Action.REPOST]
+        and params.entity_type == EntityType.TRACK
+        and params.existing_records[EntityType.TRACK][params.entity_id].remix_of
+    ):
+        remixed_track = (
+            params.session.query(Track)
+            .join(Remix, Remix.parent_track_id == Track.track_id)
+            .filter(Remix.child_track_id == params.entity_id)
+            .first()
+        )
+        remixer = params.existing_records[EntityType.TRACK][params.entity_id].owner_id
+        cosigner = params.existing_records[EntityType.USER][params.user_id]
 
-        # dispatch verified cosign challenge
         if (
-            params.action in [Action.SAVE, Action.REPOST]
-            and params.entity_type == EntityType.TRACK
-            and params.existing_records[EntityType.TRACK][params.entity_id].remix_of
+            remixed_track
+            and remixed_track.owner_id == params.user_id
+            and cosigner.is_verified
         ):
-            remixed_track = (
-                params.session.query(Track)
-                .join(Remix, Remix.parent_track_id == Track.track_id)
-                .filter(Remix.child_track_id == params.entity_id)
-                .first()
+            params.challenge_bus.dispatch(
+                ChallengeEvent.cosign,
+                params.block_number,
+                params.block_datetime,
+                remixer,
+                {
+                    "original_track_owner_id": params.user_id,
+                    "remix_track_id": params.entity_id,
+                    "cosign_date": params.block_datetime.timestamp(),
+                },
             )
-            remixer = params.existing_records[EntityType.TRACK][
-                params.entity_id
-            ].owner_id
-            cosigner = params.existing_records[EntityType.USER][params.user_id]
-
-            if (
-                remixed_track
-                and remixed_track.owner_id == params.user_id
-                and cosigner.is_verified
-            ):
-                params.challenge_bus.dispatch(
-                    ChallengeEvent.cosign,
-                    params.block_number,
-                    params.block_datetime,
-                    remixer,
-                    {
-                        "original_track_owner_id": params.user_id,
-                        "remix_track_id": params.entity_id,
-                        "cosign_date": params.block_datetime.timestamp(),
-                    },
-                )
 
 
 def get_attribute_from_record_metadata(params, attribute):