Get token first trade block API (#3197)

# Description
This is a pre-requisite for improving `internal buffers at risk` alert
by eliminating false positives. The idea is to provide the token's first
trade timestamp so https://github.com/cowprotocol/tenderly-alerts/ can
then decide whether the token should be ignored. Currently, there is no
way to return the timestamp since the `order_events` table gets cleaned
up periodically. Technically, it can be used to get a timestamp, but the
result will be useless if a token was deployed more than 30 days ago and
wasn't traded for the last 30 days. Instead, the block number is
returned, so Tenderly RPC must be used to fetch the block's timestamp.
Otherwise, this would require access to an archive node from the
orderbook side.

## Changes
For some reason, this query takes around 20s on mainnet-prod for
avg-popular tokens. Unfortunately, this couldn't be reproduced locally.
I assume this is related to available resources on prod. I added indexes
that improved the query speed ~x2.5 times.

## Caching
To avoid querying the DB for the same tokens too often, I would consider
introducing caching on the NGINX side rather than in memory since we
often have multiple orderbook instances. Also, the first trade block
never changes and can be kept in the NGINX cache forever. All the errors
won't be kept in the cache. Requires an infra PR.

## How to test
The query is tested on prod and locally. This would require further
testing on prod by collecting metrics and adjusting resources.
Potentially, `work_mem`, `max_parallel_workers_per_gather`, etc.

crates/database/src/trades.rs

@@ -106,6 +106,31 @@ AND t.log_index BETWEEN (SELECT * from previous_settlement) AND $2
         .await
 }
 
+pub async fn token_first_trade_block(
+    ex: &mut PgConnection,
+    token: Address,
+) -> Result<Option<i64>, sqlx::Error> {
+    const QUERY: &str = r#"
+SELECT MIN(sub.block_number) AS earliest_block
+FROM (
+    SELECT MIN(t.block_number) AS block_number
+    FROM trades t
+    JOIN orders o ON t.order_uid = o.uid
+    WHERE o.sell_token = $1 OR o.buy_token = $1
+
+    UNION ALL
+
+    SELECT MIN(t.block_number) AS block_number
+    FROM trades t
+    JOIN jit_orders j ON t.order_uid = j.uid
+    WHERE j.sell_token = $1 OR j.buy_token = $1
+) AS sub
+"#;
+
+    let (block_number,) = sqlx::query_as(QUERY).bind(token).fetch_one(ex).await?;
+    Ok(block_number)
+}
+
 #[cfg(test)]
 mod tests {
     use {
@@ -579,4 +604,31 @@ mod tests {
             }]
         );
     }
+
+    #[tokio::test]
+    #[ignore]
+    async fn postgres_token_first_trade_block() {
+        let mut db = PgConnection::connect("postgresql://").await.unwrap();
+        let mut db = db.begin().await.unwrap();
+        crate::clear_DANGER_(&mut db).await.unwrap();
+
+        let token = Default::default();
+        assert_eq!(token_first_trade_block(&mut db, token).await.unwrap(), None);
+
+        let (owners, order_ids) = generate_owners_and_order_ids(2, 2).await;
+        let event_index_a = EventIndex {
+            block_number: 123,
+            log_index: 0,
+        };
+        let event_index_b = EventIndex {
+            block_number: 124,
+            log_index: 0,
+        };
+        add_order_and_trade(&mut db, owners[0], order_ids[0], event_index_a, None, None).await;
+        add_order_and_trade(&mut db, owners[1], order_ids[1], event_index_b, None, None).await;
+        assert_eq!(
+            token_first_trade_block(&mut db, token).await.unwrap(),
+            Some(123)
+        );
+    }
 }

crates/orderbook/src/api.rs

@@ -23,6 +23,7 @@ mod get_order_by_uid;
 mod get_order_status;
 mod get_orders_by_tx;
 mod get_solver_competition;
+mod get_token_metadata;
 mod get_total_surplus;
 mod get_trades;
 mod get_user_orders;
@@ -105,7 +106,11 @@ pub fn handle_all_routes(
         ),
         (
             "v1/get_total_surplus",
-            box_filter(get_total_surplus::get(database)),
+            box_filter(get_total_surplus::get(database.clone())),
+        ),
+        (
+            "v1/get_token_metadata",
+            box_filter(get_token_metadata::get_token_metadata(database)),
         ),
     ];
 

crates/orderbook/src/api/get_token_metadata.rs

@@ -0,0 +1,31 @@
+use {
+    crate::database::Postgres,
+    hyper::StatusCode,
+    primitive_types::H160,
+    std::convert::Infallible,
+    warp::{reply, Filter, Rejection},
+};
+
+fn get_native_prices_request() -> impl Filter<Extract = (H160,), Error = Rejection> + Clone {
+    warp::path!("v1" / "token" / H160 / "metadata").and(warp::get())
+}
+
+pub fn get_token_metadata(
+    db: Postgres,
+) -> impl Filter<Extract = (super::ApiReply,), Error = Rejection> + Clone {
+    get_native_prices_request().and_then(move |token: H160| {
+        let db = db.clone();
+        async move {
+            let result = db.token_metadata(&token).await;
+            let response = match result {
+                Ok(metadata) => reply::with_status(reply::json(&metadata), StatusCode::OK),
+                Err(err) => {
+                    tracing::error!(?err, ?token, "Failed to fetch token's first trade block");
+                    crate::api::internal_error_reply()
+                }
+            };
+
+            Result::<_, Infallible>::Ok(response)
+        }
+    })
+}

crates/orderbook/src/database/orders.rs

@@ -1,6 +1,6 @@
 use {
     super::Postgres,
-    crate::orderbook::AddOrderError,
+    crate::{dto::TokenMetadata, orderbook::AddOrderError},
     anyhow::{Context as _, Result},
     app_data::AppDataHash,
     async_trait::async_trait,
@@ -492,6 +492,27 @@ impl Postgres {
             .map(full_order_into_model_order)
             .collect::<Result<Vec<_>>>()
     }
+
+    pub async fn token_metadata(&self, token: &H160) -> Result<TokenMetadata> {
+        let timer = super::Metrics::get()
+            .database_queries
+            .with_label_values(&["token_first_trade_block"])
+            .start_timer();
+
+        let mut ex = self.pool.acquire().await?;
+        let block_number = database::trades::token_first_trade_block(&mut ex, ByteArray(token.0))
+            .await
+            .map_err(anyhow::Error::from)?
+            .map(u32::try_from)
+            .transpose()
+            .map_err(anyhow::Error::from)?;
+
+        timer.stop_and_record();
+
+        Ok(TokenMetadata {
+            first_trade_block: block_number,
+        })
+    }
 }
 
 #[async_trait]

crates/orderbook/src/dto/mod.rs

@@ -5,3 +5,11 @@ pub use {
     auction::{Auction, AuctionId, AuctionWithId},
     order::Order,
 };
+use {serde::Serialize, serde_with::serde_as};
+
+#[serde_as]
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TokenMetadata {
+    pub first_trade_block: Option<u32>,
+}

database/sql/V077__orders_token_indexes.sql

@@ -0,0 +1,3 @@
+CREATE INDEX orders_sell_buy_tokens ON orders (sell_token, buy_token);
+
+CREATE INDEX jit_orders_sell_buy_tokens ON jit_orders (sell_token, buy_token);