Feat: Add message status filtering to messages API (#806)

* feat: filter message list by status and include status field

* fix: exclude message status field from message_list response

* chore: linting issues

* fix: tests

* fix: linting

Author: amalcaraz

src/aleph/db/accessors/messages.py

@@ -5,7 +5,7 @@
 from aleph_message.models import Chain, ItemHash, MessageType
 from sqlalchemy import delete, func, nullsfirst, nullslast, select, text, update
 from sqlalchemy.dialects.postgresql import array, insert
-from sqlalchemy.orm import load_only, selectinload
+from sqlalchemy.orm import contains_eager, load_only, selectinload
 from sqlalchemy.sql import Insert, Select
 from sqlalchemy.sql.elements import literal
 
@@ -64,6 +64,7 @@ def make_matching_messages_query(
     chains: Optional[Sequence[Chain]] = None,
     message_type: Optional[MessageType] = None,
     message_types: Optional[Sequence[MessageType]] = None,
+    message_statuses: Optional[Sequence[MessageStatus]] = None,
     start_date: Optional[Union[float, dt.datetime]] = None,
     end_date: Optional[Union[float, dt.datetime]] = None,
     start_block: Optional[int] = None,
@@ -82,6 +83,19 @@ def make_matching_messages_query(
 ) -> Select:
     select_stmt = select(MessageDb)
 
+    if message_statuses:
+        select_stmt = (
+            select_stmt.join(
+                MessageStatusDb, MessageDb.item_hash == MessageStatusDb.item_hash
+            )
+            .where(MessageStatusDb.status.in_(message_statuses))
+            .options(
+                contains_eager(MessageDb.status).options(
+                    load_only(MessageStatusDb.status)
+                )
+            )
+        )
+
     if include_confirmations:
         # Note: we assume this is only used for the API, so we only load the fields
         # returned by the API. If additional fields are required, add them here to

src/aleph/db/models/messages.py

@@ -104,6 +104,13 @@ class MessageDb(Base):
         "ChainTxDb", secondary=message_confirmations
     )
 
+    status: Optional["MessageStatusDb"] = relationship(
+        "MessageStatusDb",
+        primaryjoin="MessageDb.item_hash == MessageStatusDb.item_hash",
+        foreign_keys=MessageStatusDb.item_hash,
+        uselist=False,  # Critical: Makes it one-to-one
+    )
+
     _parsed_content: Optional[BaseContent] = None
 
     @property

src/aleph/web/controllers/messages.py

@@ -89,6 +89,11 @@ class BaseMessageQueryParams(BaseModel):
     message_types: Optional[List[MessageType]] = Field(
         default=None, alias="msgTypes", description="Accepted message types."
     )
+    message_statuses: Optional[List[MessageStatus]] = Field(
+        default=[MessageStatus.PROCESSED, MessageStatus.REMOVING],
+        alias="msgStatuses",
+        description="Accepted values for the 'status' field.",
+    )
     addresses: Optional[List[str]] = Field(
         default=None, description="Accepted values for the 'sender' field."
     )
@@ -176,6 +181,7 @@ def validate_field_dependencies(self):
         "chains",
         "channels",
         "message_types",
+        "message_statuses",
         "tags",
         mode="before",
     )
@@ -266,6 +272,10 @@ def message_to_dict(message: MessageDb) -> Dict[str, Any]:
     ]
     message_dict["confirmations"] = confirmations
     message_dict["confirmed"] = bool(confirmations)
+
+    # TODO: Add this field in the response when we make sure it won't break any sdk schema checking
+    # message_dict["status"] = message.status.status
+
     return message_dict
 
 

tests/api/conftest.py

@@ -18,12 +18,14 @@
     MessageDb,
     message_confirmations,
 )
+from aleph.db.models.messages import MessageStatusDb
 from aleph.db.models.posts import PostDb
 from aleph.handlers.message_handler import MessageHandler
 from aleph.jobs.process_pending_messages import PendingMessageProcessor
 from aleph.storage import StorageService
-from aleph.toolkit.timestamp import timestamp_to_datetime
+from aleph.toolkit.timestamp import timestamp_to_datetime, utc_now
 from aleph.types.db_session import DbSessionFactory
+from aleph.types.message_status import MessageStatus
 
 
 # TODO: remove the raw parameter, it's just to avoid larger refactorings
@@ -44,6 +46,7 @@ async def _load_fixtures(
             message_db = MessageDb.from_message_dict(message_dict)
             messages.append(message_db)
             session.add(message_db)
+
             for confirmation in message_dict.get("confirmations", []):
                 if (tx_hash := confirmation["hash"]) not in tx_hashes:
                     chain_tx_db = ChainTxDb.from_dict(confirmation)
@@ -56,6 +59,14 @@ async def _load_fixtures(
                         item_hash=message_db.item_hash, tx_hash=tx_hash
                     )
                 )
+
+            message_status = MessageStatusDb(
+                item_hash=message_dict["item_hash"],
+                status=MessageStatus.PROCESSED,
+                reception_time=utc_now(),
+            )
+            session.add(message_status)
+
         session.commit()
 
     return messages_json if raw else messages
@@ -133,7 +144,7 @@ async def fixture_posts(
 
 
 @pytest.fixture
-def post_with_refs_and_tags() -> Tuple[MessageDb, PostDb]:
+def post_with_refs_and_tags() -> Tuple[MessageDb, PostDb, MessageStatusDb]:
     message = MessageDb(
         item_hash="1234",
         sender="0xdeadbeef",
@@ -160,12 +171,20 @@ def post_with_refs_and_tags() -> Tuple[MessageDb, PostDb]:
         latest_amend=None,
     )
 
-    return message, post
+    message_status = MessageStatusDb(
+        item_hash=message.item_hash,
+        status=MessageStatus.PROCESSED,
+        reception_time=utc_now(),
+    )
+
+    return message, post, message_status
 
 
 @pytest.fixture
-def amended_post_with_refs_and_tags(post_with_refs_and_tags: Tuple[MessageDb, PostDb]):
-    original_message, original_post = post_with_refs_and_tags
+def amended_post_with_refs_and_tags(
+    post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb]
+):
+    original_message, original_post, _ = post_with_refs_and_tags
 
     amend_message = MessageDb(
         item_hash="5678",
@@ -193,7 +212,13 @@ def amended_post_with_refs_and_tags(post_with_refs_and_tags: Tuple[MessageDb, Po
         latest_amend=None,
     )
 
-    return amend_message, amend_post
+    message_status = MessageStatusDb(
+        item_hash=amend_message.item_hash,
+        status=MessageStatus.PROCESSED,
+        reception_time=utc_now(),
+    )
+
+    return amend_message, amend_post, message_status
 
 
 @pytest.fixture

tests/api/test_list_messages.py

@@ -19,9 +19,11 @@
 )
 
 from aleph.db.models import MessageDb, PostDb
-from aleph.toolkit.timestamp import timestamp_to_datetime
+from aleph.db.models.messages import MessageStatusDb
+from aleph.toolkit.timestamp import timestamp_to_datetime, utc_now
 from aleph.types.channel import Channel
 from aleph.types.db_session import DbSessionFactory
+from aleph.types.message_status import MessageStatus
 
 from .utils import get_messages_by_keys
 
@@ -171,19 +173,21 @@ async def test_get_messages_filter_by_tags(
     fixture_messages,
     ccn_api_client,
     session_factory: DbSessionFactory,
-    post_with_refs_and_tags: Tuple[MessageDb, PostDb],
-    amended_post_with_refs_and_tags: Tuple[MessageDb, PostDb],
+    post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
+    amended_post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
 ):
     """
     Tests getting messages by tags.
     There's no example in the fixtures, we just test that the endpoint returns a 200.
     """
 
-    message_db, _ = post_with_refs_and_tags
-    amend_message_db, _ = amended_post_with_refs_and_tags
+    message_db, _, message_status_db = post_with_refs_and_tags
+    amend_message_db, _, amend_message_status_db = amended_post_with_refs_and_tags
 
     with session_factory() as session:
-        session.add_all([message_db, amend_message_db])
+        session.add_all(
+            [message_db, message_status_db, amend_message_db, amend_message_status_db]
+        )
         session.commit()
 
     # Matching tag for both messages
@@ -499,8 +503,8 @@ def get_confirmed_time(msg: Dict) -> Tuple[float, float]:
 
 
 @pytest.fixture
-def instance_message_fixture() -> MessageDb:
-    return MessageDb(
+def instance_message_fixture() -> Tuple[MessageDb, MessageStatusDb]:
+    message = MessageDb(
         item_hash="9f29cdb6579d94be1053b1e1400ee3440958da4cf4feb9b44b674746fdb17c9c",
         chain=Chain.ETH,
         sender="0xB68B9D4f3771c246233823ed1D3Add451055F9Ef",
@@ -547,24 +551,34 @@ def instance_message_fixture() -> MessageDb:
         channel=Channel("TEST"),
     )
 
+    message_status = MessageStatusDb(
+        item_hash=message.item_hash,
+        status=MessageStatus.PROCESSED,
+        reception_time=utc_now(),
+    )
+
+    return message, message_status
+
 
 @pytest.mark.asyncio
 async def test_get_instance(
     ccn_api_client,
-    instance_message_fixture: MessageDb,
+    instance_message_fixture: Tuple[MessageDb, MessageStatusDb],
     session_factory: DbSessionFactory,
 ):
+
+    message_db, status_db = instance_message_fixture
     with session_factory() as session:
-        session.add(instance_message_fixture)
+        session.add_all([message_db, status_db])
         session.commit()
 
     response = await ccn_api_client.get(
-        MESSAGES_URI, params={"hashes": instance_message_fixture.item_hash}
+        MESSAGES_URI, params={"hashes": message_db.item_hash}
     )
     assert response.status == 200, await response.text()
 
     messages = (await response.json())["messages"]
     assert len(messages) == 1
 
     message = messages[0]
-    assert message["item_hash"] == instance_message_fixture.item_hash
+    assert message["item_hash"] == message_db.item_hash

tests/api/test_posts.py

@@ -4,6 +4,7 @@
 import pytest
 
 from aleph.db.models import MessageDb
+from aleph.db.models.messages import MessageStatusDb
 from aleph.db.models.posts import PostDb
 from aleph.types.db_session import DbSessionFactory
 
@@ -58,14 +59,13 @@ async def test_get_posts_refs(
     ccn_api_client,
     session_factory: DbSessionFactory,
     fixture_posts: Sequence[PostDb],
-    post_with_refs_and_tags: Tuple[MessageDb, PostDb],
+    post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
 ):
-    message_db, post_db = post_with_refs_and_tags
+    message_db, post_db, message_status_db = post_with_refs_and_tags
 
     with session_factory() as session:
         session.add_all(fixture_posts)
-        session.add(message_db)
-        session.add(post_db)
+        session.add_all([message_db, post_db, message_status_db])
         session.commit()
 
     # Match the ref
@@ -113,20 +113,24 @@ async def test_get_amended_posts_refs(
     ccn_api_client,
     session_factory: DbSessionFactory,
     fixture_posts: Sequence[PostDb],
-    post_with_refs_and_tags: Tuple[MessageDb, PostDb],
-    amended_post_with_refs_and_tags: Tuple[MessageDb, PostDb],
+    post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
+    amended_post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
 ):
-    original_message_db, original_post_db = post_with_refs_and_tags
-    amend_message_db, amend_post_db = amended_post_with_refs_and_tags
+    original_message_db, original_post_db, original_message_status_db = (
+        post_with_refs_and_tags
+    )
+    amend_message_db, amend_post_db, amend_message_status_db = (
+        amended_post_with_refs_and_tags
+    )
 
     original_post_db.latest_amend = amend_post_db.item_hash
 
     with session_factory() as session:
         session.add_all(fixture_posts)
-        session.add(original_message_db)
-        session.add(original_post_db)
-        session.add(amend_message_db)
-        session.add(amend_post_db)
+        session.add_all(
+            [original_message_db, original_post_db, original_message_status_db]
+        )
+        session.add_all([amend_message_db, amend_post_db, amend_message_status_db])
         session.commit()
 
     # Match the ref
@@ -174,14 +178,13 @@ async def test_get_posts_tags(
     ccn_api_client,
     session_factory: DbSessionFactory,
     fixture_posts: Sequence[PostDb],
-    post_with_refs_and_tags: Tuple[MessageDb, PostDb],
+    post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
 ):
-    message_db, post_db = post_with_refs_and_tags
+    message_db, post_db, message_status_db = post_with_refs_and_tags
 
     with session_factory() as session:
         session.add_all(fixture_posts)
-        session.add(message_db)
-        session.add(post_db)
+        session.add_all([message_db, post_db, message_status_db])
         session.commit()
 
     # Match one tag
@@ -244,20 +247,24 @@ async def test_get_amended_posts_tags(
     ccn_api_client,
     session_factory: DbSessionFactory,
     fixture_posts: Sequence[PostDb],
-    post_with_refs_and_tags: Tuple[MessageDb, PostDb],
-    amended_post_with_refs_and_tags: Tuple[MessageDb, PostDb],
+    post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
+    amended_post_with_refs_and_tags: Tuple[MessageDb, PostDb, MessageStatusDb],
 ):
-    original_message_db, original_post_db = post_with_refs_and_tags
-    amend_message_db, amend_post_db = amended_post_with_refs_and_tags
+    original_message_db, original_post_db, original_message_status_db = (
+        post_with_refs_and_tags
+    )
+    amend_message_db, amend_post_db, amend_message_status_db = (
+        amended_post_with_refs_and_tags
+    )
 
     original_post_db.latest_amend = amend_post_db.item_hash
 
     with session_factory() as session:
         session.add_all(fixture_posts)
-        session.add(original_message_db)
-        session.add(original_post_db)
-        session.add(amend_message_db)
-        session.add(amend_post_db)
+        session.add_all(
+            [original_message_db, original_post_db, original_message_status_db]
+        )
+        session.add_all([amend_message_db, amend_post_db, amend_message_status_db])
         session.commit()
 
     # Match one tag