Fix solver competition reading (#3171)

# Description
Fixes a bug where transaction hashes from different environment
(prod/staging) are shown on solver competition endpoint.

Example: https://api.cow.fi/mainnet/api/v1/solver_competition/9861815

Here, transaction
`0xb608f1e2a644befda65a6a87ce6b93fc8a3441d9729b8ce4b2a2db02c1247e60` is
from production enviroment and valid one, while
`0xc85cd0ad8e3aac389836a71da9795bdd1bedce7b21da1950b48f93165c9f6be8` is
from staging environment and mistakenly added to result.

The reason why this happens is:
1. The solver competition with the same `auction_id` can be saved in
both environments, regardless if the competition ended up with mined
settlement (which is fine).
2. `settlement::Observer` saves the relationship <auction, settlement>
even for settlements from another environment (this is a decision made
long time ago, before refactoring, [reference code
here](https://github.com/cowprotocol/services/blob/main/crates/autopilot/src/infra/persistence/mod.rs#L654-L660)
)

But, what this PR introduces is leveraging the fact that, for settlement
from another environment, there is no matching entry in
`settlement_observations` table, [reference code
here](https://github.com/cowprotocol/services/blob/main/crates/autopilot/src/infra/persistence/mod.rs#L682-L693).
Note how the saving is executed only if the `settlement` exists, which
if
[false](https://github.com/cowprotocol/services/blob/main/crates/autopilot/src/domain/settlement/observer.rs#L90-L92)
for settlements from another environment.

# Changes
<!-- List of detailed changes (how the change is accomplished) -->

- [ ] Update reading functions for solver competition, to exclude
transaction hashes for settlements from another environment.

## How to test
Updated unit test.
Existing e2e tests.

Manually tested all updated functions on production database, works as
expected.

crates/database/src/solver_competition.rs

@@ -38,10 +38,14 @@ pub async fn load_by_id(
     id: AuctionId,
 ) -> Result<Option<LoadCompetition>, sqlx::Error> {
     const QUERY: &str = r#"
-SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE s.tx_hash IS NOT NULL), '{}') AS tx_hashes
+SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE so.block_number IS NOT NULL), '{}') AS tx_hashes
 FROM solver_competitions sc
 -- outer joins because the data might not have been indexed yet
 LEFT OUTER JOIN settlements s ON sc.id = s.auction_id
+-- exclude settlements from another environment for which observation is guaranteed to not exist
+LEFT OUTER JOIN settlement_observations so 
+    ON s.block_number = so.block_number 
+    AND s.log_index = so.log_index
 WHERE sc.id = $1
 GROUP BY sc.id
     ;"#;
@@ -52,10 +56,14 @@ pub async fn load_latest_competition(
     ex: &mut PgConnection,
 ) -> Result<Option<LoadCompetition>, sqlx::Error> {
     const QUERY: &str = r#"
-SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE s.tx_hash IS NOT NULL), '{}') AS tx_hashes
+SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE so.block_number IS NOT NULL), '{}') AS tx_hashes
 FROM solver_competitions sc
 -- outer joins because the data might not have been indexed yet
 LEFT OUTER JOIN settlements s ON sc.id = s.auction_id
+-- exclude settlements from another environment for which observation is guaranteed to not exist
+LEFT OUTER JOIN settlement_observations so 
+    ON s.block_number = so.block_number 
+    AND s.log_index = so.log_index
 GROUP BY sc.id
 ORDER BY sc.id DESC
 LIMIT 1
@@ -72,11 +80,17 @@ WITH competition AS (
     SELECT sc.id
     FROM solver_competitions sc
     JOIN settlements s ON sc.id = s.auction_id
+    JOIN settlement_observations so 
+        ON s.block_number = so.block_number 
+        AND s.log_index = so.log_index
     WHERE s.tx_hash = $1
 )
-SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE s.tx_hash IS NOT NULL), '{}') AS tx_hashes
+SELECT sc.json, sc.id, COALESCE(ARRAY_AGG(s.tx_hash) FILTER (WHERE so.block_number IS NOT NULL), '{}') AS tx_hashes
 FROM solver_competitions sc
 JOIN settlements s ON sc.id = s.auction_id
+JOIN settlement_observations so 
+    ON s.block_number = so.block_number 
+    AND s.log_index = so.log_index
 WHERE sc.id = (SELECT id FROM competition)
 GROUP BY sc.id
     ;"#;
@@ -336,7 +350,9 @@ mod tests {
         // non-existent auction returns none
         assert!(load_by_id(&mut db, 1).await.unwrap().is_none());
 
-        // insert two settlement events for the same auction id
+        // insert three settlement events for the same auction id, with one of them not
+        // having observation (in practice usually meaning it's from different
+        // environment)
         crate::events::insert_settlement(
             &mut db,
             &EventIndex {
@@ -350,6 +366,16 @@ mod tests {
         )
         .await
         .unwrap();
+        crate::settlement_observations::upsert(
+            &mut db,
+            crate::settlement_observations::Observation {
+                block_number: 0,
+                log_index: 0,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
         crate::events::insert_settlement(
             &mut db,
             &EventIndex {
@@ -363,12 +389,38 @@ mod tests {
         )
         .await
         .unwrap();
+        crate::settlement_observations::upsert(
+            &mut db,
+            crate::settlement_observations::Observation {
+                block_number: 0,
+                log_index: 1,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+        crate::events::insert_settlement(
+            &mut db,
+            &EventIndex {
+                block_number: 0,
+                log_index: 2,
+            },
+            &Settlement {
+                solver: Default::default(),
+                transaction_hash: ByteArray([2u8; 32]),
+            },
+        )
+        .await
+        .unwrap();
         crate::settlements::update_settlement_auction(&mut db, 0, 0, 0)
             .await
             .unwrap();
         crate::settlements::update_settlement_auction(&mut db, 0, 1, 0)
             .await
             .unwrap();
+        crate::settlements::update_settlement_auction(&mut db, 0, 2, 0)
+            .await
+            .unwrap();
 
         // load by id works, and finds two hashes
         let value_ = load_by_id(&mut db, 0).await.unwrap().unwrap();
@@ -389,6 +441,11 @@ mod tests {
             .unwrap()
             .unwrap();
         assert!(value_.tx_hashes.len() == 2);
+        // this one should not find any hashes since it's from another environment
+        let value_ = load_by_tx_hash(&mut db, &ByteArray([2u8; 32]))
+            .await
+            .unwrap();
+        assert!(value_.is_none());
     }
 
     #[tokio::test]