feat(discovery): index open_contest submissions into contest_submissions

packages/discovery-provider/ddl/migrations/0179_open_contest_event_type.sql

@@ -0,0 +1,17 @@
+begin;
+
+-- Adds the 'open_contest' value to the event_type enum used by the events
+-- table. Open contests differ from remix_contest in that entries do not
+-- require a remix-parent track; submissions land in api-land's
+-- contest_submissions table (see api repo migration 0203). The events
+-- row may carry a NULL entity_id for open_contest since there is no
+-- parent track to point at — the entity_id column was already nullable
+-- when the events table was created.
+
+DO $$ BEGIN
+    ALTER TYPE event_type ADD VALUE IF NOT EXISTS 'open_contest';
+EXCEPTION
+    WHEN duplicate_object THEN null;
+END $$;
+
+commit;

packages/discovery-provider/src/models/events/contest_submission.py

@@ -0,0 +1,27 @@
+from sqlalchemy import Column, DateTime, Integer, text
+
+from src.models.base import Base
+from src.models.model_utils import RepresentableMixin
+
+
+class ContestSubmission(Base, RepresentableMixin):
+    """Tracks submitted to an open_contest event.
+
+    Table lives in api-land (see api repo migration 0203); discovery's
+    entity manager writes a row here when it processes a SubmitToContest
+    ManageEntity action.
+    """
+
+    __tablename__ = "contest_submissions"
+
+    contest_id = Column(Integer, primary_key=True)
+    track_id = Column(Integer, primary_key=True)
+    user_id = Column(Integer, nullable=False)
+    created_at = Column(
+        DateTime,
+        nullable=False,
+        server_default=text("CURRENT_TIMESTAMP"),
+    )
+
+    def get_attributes_dict(self):
+        return {col.name: getattr(self, col.name) for col in self.__table__.columns}

packages/discovery-provider/src/models/events/event.py

@@ -22,6 +22,7 @@ class EventType(str, enum.Enum):
     remix_contest = "remix_contest"
     live_event = "live_event"
     new_release = "new_release"
+    open_contest = "open_contest"
 
 
 class EventEntityType(str, enum.Enum):

packages/discovery-provider/src/queries/get_remixes_of.py

@@ -2,6 +2,7 @@
 from sqlalchemy.orm import aliased
 
 from src import exceptions
+from src.models.events.contest_submission import ContestSubmission
 from src.models.events.event import Event, EventType
 from src.models.social.aggregate_plays import AggregatePlay
 from src.models.social.repost import Repost, RepostType
@@ -29,6 +30,10 @@ def get_remixes_of(args):
     sort_method = args.get("sort_method", "recent")
     only_cosigns = args.get("only_cosigns", False)
     only_contest_entries = args.get("only_contest_entries", False)
+    # When provided, return tracks submitted to this open_contest event
+    # (rows in contest_submissions). Open contests have no parent track,
+    # so the classic remixes-table join doesn't apply.
+    contest_id = args.get("contest_id")
 
     db = get_db_read_replica()
 
@@ -151,7 +156,54 @@ def get_unpopulated_remixes():
             track_ids = list(map(lambda track: track["track_id"], tracks))
             return (tracks, track_ids, count)
 
-        (tracks, track_ids, count) = get_unpopulated_remixes()
+        def get_unpopulated_open_contest_submissions():
+            submission_query = (
+                session.query(Track)
+                .join(
+                    ContestSubmission,
+                    ContestSubmission.track_id == Track.track_id,
+                )
+                .outerjoin(
+                    AggregateTrack, AggregateTrack.track_id == Track.track_id
+                )
+                .outerjoin(
+                    AggregatePlay, AggregatePlay.play_item_id == Track.track_id
+                )
+                .filter(
+                    ContestSubmission.contest_id == contest_id,
+                    Track.is_current == True,
+                    Track.is_delete == False,
+                    Track.is_unlisted == False,
+                )
+            )
+
+            if sort_method == RemixesSortMethod.likes:
+                submission_query = submission_query.order_by(
+                    desc(func.coalesce(AggregateTrack.save_count, 0)),
+                    desc(Track.track_id),
+                )
+            elif sort_method == RemixesSortMethod.plays:
+                submission_query = submission_query.order_by(
+                    desc(func.coalesce(AggregatePlay.count, 0)),
+                    desc(Track.track_id),
+                )
+            else:
+                submission_query = submission_query.order_by(
+                    desc(ContestSubmission.created_at), desc(Track.track_id)
+                )
+
+            (tracks, count) = add_query_pagination(
+                submission_query, limit, offset, True, True
+            )
+            tracks = tracks.all()
+            tracks = helpers.query_result_to_list(tracks)
+            track_ids = list(map(lambda track: track["track_id"], tracks))
+            return (tracks, track_ids, count)
+
+        if contest_id is not None and not track_id:
+            (tracks, track_ids, count) = get_unpopulated_open_contest_submissions()
+        else:
+            (tracks, track_ids, count) = get_unpopulated_remixes()
         tracks = populate_track_metadata(session, track_ids, tracks, current_user_id)
         if args.get("with_users", False):
             add_users_to_tracks(session, tracks, current_user_id)

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

@@ -1,7 +1,9 @@
+import json
 from datetime import datetime
 
 from src.challenges.challenge_event_bus import ChallengeEvent
 from src.exceptions import IndexingValidationError
+from src.models.events.contest_submission import ContestSubmission
 from src.models.events.event import Event, EventEntityType, EventType
 from src.models.tracks.track import Track
 from src.tasks.entity_manager.utils import (
@@ -33,6 +35,8 @@ def validate_create_event_tx(params: ManageEntityParameters):
         if field not in metadata:
             raise IndexingValidationError(f"Missing required field: {field}")
 
+    is_open_contest = metadata["event_type"] == EventType.open_contest
+
     if params.metadata.get("end_date"):
         # Validate end_date is a valid iso format
         try:
@@ -57,28 +61,31 @@ def validate_create_event_tx(params: ManageEntityParameters):
             f"Invalid entity_type: {params.metadata['entity_type']}"
         )
 
-    # Validate entity type is correct and entity exists
-    # TODO: Update this to validate that the entity_type is correct
-    if (
-        params.metadata["entity_id"]
-        and params.metadata["entity_type"] == EventEntityType.track.value
-        and params.metadata["entity_id"]
-        not in params.existing_records[EntityType.TRACK.value]
-    ):
-        raise IndexingValidationError(
-            f"Track {params.metadata['entity_id']} does not exist"
-        )
-
-    # Validate user is the owner of the entity
-    if params.metadata["entity_type"] == EventEntityType.track.value:
-        track_owner = params.existing_records[EntityType.TRACK.value][
+    # Open contests intentionally have no parent track, so skip the
+    # entity_id / entity_type / track-owner checks below.
+    if not is_open_contest:
+        # Validate entity type is correct and entity exists
+        # TODO: Update this to validate that the entity_type is correct
+        if (
             params.metadata["entity_id"]
-        ].owner_id
-        if track_owner != params.user_id:
+            and params.metadata["entity_type"] == EventEntityType.track.value
+            and params.metadata["entity_id"]
+            not in params.existing_records[EntityType.TRACK.value]
+        ):
             raise IndexingValidationError(
-                f"User {params.user_id} is not the owner of the track {params.metadata['entity_id']}"
+                f"Track {params.metadata['entity_id']} does not exist"
             )
 
+        # Validate user is the owner of the entity
+        if params.metadata["entity_type"] == EventEntityType.track.value:
+            track_owner = params.existing_records[EntityType.TRACK.value][
+                params.metadata["entity_id"]
+            ].owner_id
+            if track_owner != params.user_id:
+                raise IndexingValidationError(
+                    f"User {params.user_id} is not the owner of the track {params.metadata['entity_id']}"
+                )
+
     # Validate user exists
     if params.user_id not in params.existing_records[EntityType.USER.value]:
         raise IndexingValidationError(f"User {params.user_id} does not exist")
@@ -262,3 +269,68 @@ def delete_event(params: ManageEntityParameters):
     existing_event.blocknumber = params.block_number
     existing_event.updated_at = params.block_datetime
     existing_event.is_deleted = True
+
+
+def submit_to_contest(params: ManageEntityParameters):
+    """Index a SubmitToContest ManageEntity tx into contest_submissions.
+
+    Shape: entity_type=Event, entity_id=contest event_id, metadata is a
+    raw JSON string {"track_id": <id>} (no CID wrapper). The
+    contest_submissions table lives in api-land (api repo migration
+    0203); discovery is the writer.
+    """
+    validate_signer(params)
+
+    contest_id = params.entity_id
+    existing_event = params.existing_records[EntityType.EVENT.value].get(contest_id)
+    if not existing_event:
+        raise IndexingValidationError(
+            f"Cannot submit to contest {contest_id} — event does not exist"
+        )
+    if existing_event.is_deleted:
+        raise IndexingValidationError(
+            f"Cannot submit to deleted contest {contest_id}"
+        )
+    if existing_event.event_type != EventType.open_contest:
+        raise IndexingValidationError(
+            f"Event {contest_id} is not an open_contest"
+        )
+    if existing_event.end_date and existing_event.end_date < params.block_datetime:
+        raise IndexingValidationError(f"Contest {contest_id} has ended")
+
+    try:
+        metadata = json.loads(params.metadata)
+    except Exception:
+        raise IndexingValidationError("SubmitToContest metadata is not valid JSON")
+
+    track_id = metadata.get("track_id")
+    if not isinstance(track_id, int):
+        raise IndexingValidationError("SubmitToContest metadata.track_id is required")
+
+    track = params.existing_records[EntityType.TRACK.value].get(track_id)
+    if not track:
+        raise IndexingValidationError(f"Track {track_id} does not exist")
+    if track.owner_id != params.user_id:
+        raise IndexingValidationError(
+            f"User {params.user_id} is not the owner of track {track_id}"
+        )
+
+    existing_submission = (
+        params.session.query(ContestSubmission)
+        .filter(
+            ContestSubmission.contest_id == contest_id,
+            ContestSubmission.track_id == track_id,
+        )
+        .first()
+    )
+    if existing_submission:
+        return
+
+    params.session.add(
+        ContestSubmission(
+            contest_id=contest_id,
+            track_id=track_id,
+            user_id=params.user_id,
+            created_at=params.block_datetime,
+        )
+    )

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

@@ -66,6 +66,7 @@
 from src.tasks.entity_manager.entities.event import (
     create_event,
     delete_event,
+    submit_to_contest,
     update_event,
 )
 from src.tasks.entity_manager.entities.grant import (
@@ -480,6 +481,11 @@ def entity_manager_update(
                             and params.entity_type == EntityType.EVENT
                         ):
                             delete_event(params)
+                        elif (
+                            params.action == Action.SUBMIT_TO_CONTEST
+                            and params.entity_type == EntityType.EVENT
+                        ):
+                            submit_to_contest(params)
 
                         logger.debug("process transaction")  # log event context
                     except IndexingValidationError as e:
@@ -679,6 +685,19 @@ def collect_entities_to_fetch(update_task, entity_manager_txs):
                     if event_entity_type == EventEntityType.track:
                         event_entity_id = json_metadata.get("data", {}).get("entity_id")
                         entities_to_fetch[EntityType.TRACK].add(event_entity_id)
+                elif action == Action.SUBMIT_TO_CONTEST:
+                    # SubmitToContest metadata is bare JSON {"track_id": <id>},
+                    # not a CID-wrapped envelope, so the data field doesn't apply.
+                    try:
+                        json_metadata = json.loads(metadata)
+                    except Exception as e:
+                        logger.error(
+                            f"tasks | entity_manager.py | Exception deserializing SubmitToContest metadata: {e}"
+                        )
+                        continue
+                    track_id = json_metadata.get("track_id")
+                    if isinstance(track_id, int):
+                        entities_to_fetch[EntityType.TRACK].add(track_id)
 
             if entity_type == EntityType.COMMENT:
                 if (

packages/discovery-provider/src/tasks/entity_manager/utils.py

@@ -98,6 +98,7 @@ class Action(str, Enum):
     ADD_EMAIL = "AddEmail"
     REPORT = "Report"
     SHARE = "Share"
+    SUBMIT_TO_CONTEST = "SubmitToContest"
 
     def __str__(self) -> str:
         return str.__str__(self)
@@ -359,6 +360,7 @@ def expect_cid_metadata_json(metadata, action, entity_type):
         Action.UNSUBSCRIBE,
         Action.APPROVE,
         Action.REJECT,
+        Action.SUBMIT_TO_CONTEST,
     ]:
         return False
     if not metadata: