Fix observed_by filter to use event_observers junction table

The observed_by filter on messages, advertisements, telemetry, and
trace_paths matched only the first observer (stored in observer_node_id),
silently excluding events whose secondary observers appear only in the
event_observers junction table. This caused filtered lists to appear
'several hours behind' when a dominant observer consistently won the
first-insert race for recent events.

Replace the ObserverNode.public_key predicate with an IN subquery against
the event_observers junction table (the canonical multi-observer source
already used for display). Add a shared observed_by_filter_clause() helper
in observer_utils.py to avoid duplication across all four routes.

Add regression tests proving a secondary observer (present only in
event_observers) sees events via the filter. Update existing fixtures and
inline test data to seed event_hash and EventObserver rows.

Fixes #239

tests/test_api/conftest.py

@@ -22,6 +22,7 @@ from meshcore_hub.common.models import (
     Advertisement,
     Base,
     Channel,
+    EventObserver,
     Message,
     Node,
     NodeTag,
@@ -310,6 +311,7 @@ def receiver_node(api_db_session):
 @pytest.fixture
 def sample_message_with_receiver(api_db_session, receiver_node):
     """Create a message with a receiver node."""
+    event_hash = "deadbeefdeadbeefdeadbeefdeadbe01"
     message = Message(
         message_type="channel",
         channel_idx=17,
@@ -317,16 +319,28 @@ def sample_message_with_receiver(api_db_session, receiver_node):
         text="Channel message with receiver",
         received_at=datetime.now(timezone.utc),
         observer_node_id=receiver_node.id,
+        event_hash=event_hash,
     )
     api_db_session.add(message)
     api_db_session.commit()
     api_db_session.refresh(message)
+
+    api_db_session.add(
+        EventObserver(
+            event_type="message",
+            event_hash=event_hash,
+            observer_node_id=receiver_node.id,
+            observed_at=datetime.now(timezone.utc),
+        )
+    )
+    api_db_session.commit()
     return message
 
 
 @pytest.fixture
 def sample_advertisement_with_receiver(api_db_session, sample_node, receiver_node):
     """Create an advertisement with source and receiver nodes."""
+    event_hash = "deadbeefdeadbeefdeadbeefdeadbe02"
     advert = Advertisement(
         public_key=sample_node.public_key,
         name="SourceNode",
@@ -334,41 +348,76 @@ def sample_advertisement_with_receiver(api_db_session, sample_node, receiver_nod
         received_at=datetime.now(timezone.utc),
         node_id=sample_node.id,
         observer_node_id=receiver_node.id,
+        event_hash=event_hash,
     )
     api_db_session.add(advert)
     api_db_session.commit()
     api_db_session.refresh(advert)
+
+    api_db_session.add(
+        EventObserver(
+            event_type="advertisement",
+            event_hash=event_hash,
+            observer_node_id=receiver_node.id,
+            observed_at=datetime.now(timezone.utc),
+        )
+    )
+    api_db_session.commit()
     return advert
 
 
 @pytest.fixture
 def sample_telemetry_with_receiver(api_db_session, receiver_node):
     """Create a telemetry record with a receiver node."""
+    event_hash = "deadbeefdeadbeefdeadbeefdeadbe03"
     telemetry = Telemetry(
         node_public_key="xyz789xyz789xyz789xyz789xyz789xy",
         parsed_data={"battery_level": 50.0},
         received_at=datetime.now(timezone.utc),
         observer_node_id=receiver_node.id,
+        event_hash=event_hash,
     )
     api_db_session.add(telemetry)
     api_db_session.commit()
     api_db_session.refresh(telemetry)
+
+    api_db_session.add(
+        EventObserver(
+            event_type="telemetry",
+            event_hash=event_hash,
+            observer_node_id=receiver_node.id,
+            observed_at=datetime.now(timezone.utc),
+        )
+    )
+    api_db_session.commit()
     return telemetry
 
 
 @pytest.fixture
 def sample_trace_path_with_receiver(api_db_session, receiver_node):
     """Create a trace path with a receiver node."""
+    event_hash = "deadbeefdeadbeefdeadbeefdeadbe04"
     trace = TracePath(
         initiator_tag=99999,
         path_hashes=["aaa111", "bbb222"],
         hop_count=2,
         received_at=datetime.now(timezone.utc),
         observer_node_id=receiver_node.id,
+        event_hash=event_hash,
     )
     api_db_session.add(trace)
     api_db_session.commit()
     api_db_session.refresh(trace)
+
+    api_db_session.add(
+        EventObserver(
+            event_type="trace",
+            event_hash=event_hash,
+            observer_node_id=receiver_node.id,
+            observed_at=datetime.now(timezone.utc),
+        )
+    )
+    api_db_session.commit()
     return trace
 
 

tests/test_api/test_advertisements.py

@@ -2,6 +2,7 @@
 
 from datetime import datetime, timedelta, timezone
 
+from meshcore_hub.common.hash_utils import compute_advertisement_hash
 from meshcore_hub.common.models import Advertisement, EventObserver
 
 
@@ -215,23 +216,56 @@ class TestListAdvertisementsFilters:
         api_db_session.commit()
 
         # Create two advertisements, each observed by a different receiver
+        now = datetime.now(timezone.utc)
+        ad1_hash = compute_advertisement_hash(
+            public_key="ad1pubad1pubad1pubad1pubad1pubad",
+            name="AD1",
+            adv_type="CLIENT",
+            received_at=now,
+        )
+        ad2_hash = compute_advertisement_hash(
+            public_key="ad2pubad2pubad2pubad2pubad2pubad",
+            name="AD2",
+            adv_type="CLIENT",
+            received_at=now,
+        )
         ad1 = Advertisement(
             public_key="ad1pubad1pubad1pubad1pubad1pubad",
             name="AD1",
             adv_type="CLIENT",
-            received_at=datetime.now(timezone.utc),
+            received_at=now,
             observer_node_id=receiver_node.id,
+            event_hash=ad1_hash,
         )
         ad2 = Advertisement(
             public_key="ad2pubad2pubad2pubad2pubad2pubad",
             name="AD2",
             adv_type="CLIENT",
-            received_at=datetime.now(timezone.utc),
+            received_at=now,
             observer_node_id=second_receiver.id,
+            event_hash=ad2_hash,
         )
         api_db_session.add_all([ad1, ad2])
         api_db_session.commit()
 
+        api_db_session.add_all(
+            [
+                EventObserver(
+                    event_type="advertisement",
+                    event_hash=ad1_hash,
+                    observer_node_id=receiver_node.id,
+                    observed_at=datetime.now(timezone.utc),
+                ),
+                EventObserver(
+                    event_type="advertisement",
+                    event_hash=ad2_hash,
+                    observer_node_id=second_receiver.id,
+                    observed_at=datetime.now(timezone.utc),
+                ),
+            ]
+        )
+        api_db_session.commit()
+
         # Filter by both receivers
         response = client_no_auth.get(
             f"/api/v1/advertisements?observed_by={receiver_node.public_key}&observed_by={second_receiver.public_key}"
@@ -249,6 +283,64 @@ class TestListAdvertisementsFilters:
         assert len(data["items"]) == 1
         assert data["items"][0]["name"] == "AD1"
 
+    def test_filter_by_observed_by_secondary_observer(
+        self,
+        client_no_auth,
+        api_db_session,
+    ):
+        """Secondary observer (only in event_observers) sees the ad."""
+        from meshcore_hub.common.models import Node
+
+        primary_node = Node(
+            public_key="p1advp1advp1advp1advp1advp1advp",
+            name="PrimaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        secondary_node = Node(
+            public_key="s1advs1advs1advs1advs1advs1advs1",
+            name="SecondaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        api_db_session.add_all([primary_node, secondary_node])
+        api_db_session.commit()
+
+        now = datetime.now(timezone.utc)
+        event_hash = compute_advertisement_hash(
+            public_key="secobssecobssecobssecobssecobsse",
+            name="SecondaryObsAd",
+            adv_type="CLIENT",
+            received_at=now,
+        )
+        advert = Advertisement(
+            public_key="secobssecobssecobssecobssecobsse",
+            name="SecondaryObsAd",
+            adv_type="CLIENT",
+            received_at=now,
+            observer_node_id=primary_node.id,
+            event_hash=event_hash,
+        )
+        api_db_session.add(advert)
+        api_db_session.commit()
+
+        api_db_session.add(
+            EventObserver(
+                event_type="advertisement",
+                event_hash=event_hash,
+                observer_node_id=secondary_node.id,
+                observed_at=datetime.now(timezone.utc),
+            )
+        )
+        api_db_session.commit()
+
+        response = client_no_auth.get(
+            f"/api/v1/advertisements?observed_by={secondary_node.public_key}"
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["name"] == "SecondaryObsAd"
+        assert data["items"][0]["observed_by"] == primary_node.public_key
+
     def test_filter_by_since(self, client_no_auth, api_db_session):
         """Test filtering advertisements by since timestamp."""
         now = datetime.now(timezone.utc)

tests/test_api/test_messages.py

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta, timezone
 
 import pytest
 
+from meshcore_hub.common.hash_utils import compute_message_hash
 from meshcore_hub.common.models import EventObserver, Message, Node, NodeTag, Channel
 
 
@@ -289,12 +290,15 @@ class TestListMessagesFilters:
         api_db_session.commit()
 
         # Create two messages, each observed by a different receiver
+        msg1_hash = compute_message_hash(text="Msg from receiver A", channel_idx=17)
+        msg2_hash = compute_message_hash(text="Msg from receiver B", channel_idx=17)
         msg1 = Message(
             message_type="channel",
             channel_idx=17,
             text="Msg from receiver A",
             received_at=datetime.now(timezone.utc),
             observer_node_id=receiver_node.id,
+            event_hash=msg1_hash,
         )
         msg2 = Message(
             message_type="channel",
@@ -302,10 +306,29 @@ class TestListMessagesFilters:
             text="Msg from receiver B",
             received_at=datetime.now(timezone.utc),
             observer_node_id=second_receiver.id,
+            event_hash=msg2_hash,
         )
         api_db_session.add_all([msg1, msg2])
         api_db_session.commit()
 
+        api_db_session.add_all(
+            [
+                EventObserver(
+                    event_type="message",
+                    event_hash=msg1_hash,
+                    observer_node_id=receiver_node.id,
+                    observed_at=datetime.now(timezone.utc),
+                ),
+                EventObserver(
+                    event_type="message",
+                    event_hash=msg2_hash,
+                    observer_node_id=second_receiver.id,
+                    observed_at=datetime.now(timezone.utc),
+                ),
+            ]
+        )
+        api_db_session.commit()
+
         # Filter by both receivers
         response = client_no_auth.get(
             f"/api/v1/messages?observed_by={receiver_node.public_key}&observed_by={second_receiver.public_key}"
@@ -323,6 +346,58 @@ class TestListMessagesFilters:
         assert len(data["items"]) == 1
         assert data["items"][0]["text"] == "Msg from receiver A"
 
+    def test_filter_by_observed_by_secondary_observer(
+        self,
+        client_no_auth,
+        api_db_session,
+    ):
+        """Secondary observer (only in event_observers) sees the message."""
+        primary_node = Node(
+            public_key="p1msgp1msgp1msgp1msgp1msgp1msgp",
+            name="PrimaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        secondary_node = Node(
+            public_key="s1msgs1msgs1msgs1msgs1msgs1msgs1",
+            name="SecondaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        api_db_session.add_all([primary_node, secondary_node])
+        api_db_session.commit()
+
+        event_hash = compute_message_hash(
+            text="Secondary observer test", channel_idx=17
+        )
+        msg = Message(
+            message_type="channel",
+            channel_idx=17,
+            text="Secondary observer test",
+            received_at=datetime.now(timezone.utc),
+            observer_node_id=primary_node.id,
+            event_hash=event_hash,
+        )
+        api_db_session.add(msg)
+        api_db_session.commit()
+
+        api_db_session.add(
+            EventObserver(
+                event_type="message",
+                event_hash=event_hash,
+                observer_node_id=secondary_node.id,
+                observed_at=datetime.now(timezone.utc),
+            )
+        )
+        api_db_session.commit()
+
+        response = client_no_auth.get(
+            f"/api/v1/messages?observed_by={secondary_node.public_key}"
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["text"] == "Secondary observer test"
+        assert data["items"][0]["observed_by"] == primary_node.public_key
+
     def test_filter_by_since(self, client_no_auth, api_db_session):
         """Test filtering messages by since timestamp."""
         now = datetime.now(timezone.utc)

tests/test_api/test_telemetry.py

@@ -73,6 +73,63 @@ class TestListTelemetryFilters:
         data = response.json()
         assert len(data["items"]) == 1
 
+    def test_filter_by_observed_by_secondary_observer(
+        self,
+        client_no_auth,
+        api_db_session,
+    ):
+        """Secondary observer (only in event_observers) sees the telemetry."""
+        from meshcore_hub.common.hash_utils import compute_telemetry_hash
+        from meshcore_hub.common.models import EventObserver, Node, Telemetry
+
+        primary_node = Node(
+            public_key="p1telp1telp1telp1telp1telp1telp1",
+            name="PrimaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        secondary_node = Node(
+            public_key="s1tels1tels1tels1tels1tels1tels1",
+            name="SecondaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        api_db_session.add_all([primary_node, secondary_node])
+        api_db_session.commit()
+
+        now = datetime.now(timezone.utc)
+        node_pk = "secobstelsecobstelsecobsteltel"
+        event_hash = compute_telemetry_hash(
+            node_public_key=node_pk,
+            parsed_data={"temperature": 42.0},
+            received_at=now,
+        )
+        telemetry = Telemetry(
+            node_public_key=node_pk,
+            parsed_data={"temperature": 42.0},
+            received_at=now,
+            observer_node_id=primary_node.id,
+            event_hash=event_hash,
+        )
+        api_db_session.add(telemetry)
+        api_db_session.commit()
+
+        api_db_session.add(
+            EventObserver(
+                event_type="telemetry",
+                event_hash=event_hash,
+                observer_node_id=secondary_node.id,
+                observed_at=datetime.now(timezone.utc),
+            )
+        )
+        api_db_session.commit()
+
+        response = client_no_auth.get(
+            f"/api/v1/telemetry?observed_by={secondary_node.public_key}"
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["observed_by"] == primary_node.public_key
+
     def test_filter_by_since(self, client_no_auth, api_db_session):
         """Test filtering telemetry by since timestamp."""
         from meshcore_hub.common.models import Telemetry

tests/test_api/test_trace_paths.py

@@ -106,6 +106,58 @@ class TestListTracePathsFilters:
         data = response.json()
         assert len(data["items"]) == 1
 
+    def test_filter_by_observed_by_secondary_observer(
+        self,
+        client_no_auth,
+        api_db_session,
+    ):
+        """Secondary observer (only in event_observers) sees the trace path."""
+        from meshcore_hub.common.hash_utils import compute_trace_hash
+        from meshcore_hub.common.models import EventObserver, Node, TracePath
+
+        primary_node = Node(
+            public_key="p1trcp1trcp1trcp1trcp1trcp1trcp",
+            name="PrimaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        secondary_node = Node(
+            public_key="s1trcs1trcs1trcs1trcs1trcs1trcs1",
+            name="SecondaryObserver",
+            first_seen=datetime.now(timezone.utc),
+        )
+        api_db_session.add_all([primary_node, secondary_node])
+        api_db_session.commit()
+
+        event_hash = compute_trace_hash(initiator_tag=55555)
+        trace = TracePath(
+            initiator_tag=55555,
+            path_hashes=["cc", "dd"],
+            hop_count=2,
+            received_at=datetime.now(timezone.utc),
+            observer_node_id=primary_node.id,
+            event_hash=event_hash,
+        )
+        api_db_session.add(trace)
+        api_db_session.commit()
+
+        api_db_session.add(
+            EventObserver(
+                event_type="trace",
+                event_hash=event_hash,
+                observer_node_id=secondary_node.id,
+                observed_at=datetime.now(timezone.utc),
+            )
+        )
+        api_db_session.commit()
+
+        response = client_no_auth.get(
+            f"/api/v1/trace-paths?observed_by={secondary_node.public_key}"
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["observed_by"] == primary_node.public_key
+
     def test_filter_by_since(self, client_no_auth, api_db_session):
         """Test filtering trace paths by since timestamp."""
         from meshcore_hub.common.models import TracePath