Fetch Orders by Transaction Hash (#1194)

* enable API endpoint to query orders by txHash

* add todo once we start having multiple settlements in a single block

* update test to have multiple settlements and trades over a few different blocks

orderbook/openapi.yml

@@ -161,6 +161,24 @@ paths:
           description: Invalid signature
         404:
           description: Order was not found
+  /api/v1/transactions/{txHash}/orders:
+    get:
+      summary: Get orders by settlement transaction hash.
+      parameters:
+        - in: path
+          name: txHash
+          schema:
+            $ref: "#/components/schemas/TransactionHash"
+          required: true
+      responses:
+        200:
+          description: Order
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/Order"
   /api/v1/tokens/{sellToken}/fee:
     get:
       description: |

orderbook/src/api.rs

@@ -5,6 +5,7 @@ mod get_fee_info;
 mod get_markets;
 mod get_order_by_uid;
 mod get_orders;
+mod get_orders_by_tx;
 mod get_solvable_orders;
 mod get_trades;
 mod get_user_orders;
@@ -45,7 +46,8 @@ pub fn handle_all_routes(
         get_fee_and_quote::get_fee_and_quote_sell(fee_calculator.clone(), price_estimator.clone());
     let get_fee_and_quote_buy =
         get_fee_and_quote::get_fee_and_quote_buy(fee_calculator, price_estimator.clone());
-    let get_user_orders = get_user_orders::get_user_orders(orderbook);
+    let get_user_orders = get_user_orders::get_user_orders(orderbook.clone());
+    let get_orders_by_tx = get_orders_by_tx::get_orders_by_tx(orderbook);
     let post_quote = post_quote::post_quote();
     let cors = warp::cors()
         .allow_any_origin()
@@ -64,6 +66,7 @@ pub fn handle_all_routes(
             .or(get_fee_and_quote_sell.with(handle_metrics("get_fee_and_quote_sell")))
             .or(get_fee_and_quote_buy.with(handle_metrics("get_fee_and_quote_buy")))
             .or(get_user_orders.with(handle_metrics("get_user_orders")))
+            .or(get_orders_by_tx.with(handle_metrics("get_orders_by_tx")))
             .or(post_quote.with(handle_metrics("get_user_orders"))),
     );
 

orderbook/src/api/get_orders_by_tx.rs

@@ -0,0 +1,61 @@
+use crate::orderbook::Orderbook;
+use anyhow::Result;
+use ethcontract::H256;
+use model::order::Order;
+use std::{convert::Infallible, sync::Arc};
+use warp::{hyper::StatusCode, reply, Filter, Rejection, Reply};
+
+pub fn get_orders_by_tx_request() -> impl Filter<Extract = (H256,), Error = Rejection> + Clone {
+    warp::path!("transactions" / H256 / "orders").and(warp::get())
+}
+
+pub fn get_orders_by_tx_response(result: Result<Vec<Order>>) -> impl Reply {
+    match result {
+        Ok(orders) => reply::with_status(reply::json(&orders), StatusCode::OK),
+        Err(err) => {
+            tracing::error!(?err, "get_orders_by_tx error");
+            reply::with_status(super::internal_error(), StatusCode::INTERNAL_SERVER_ERROR)
+        }
+    }
+}
+
+pub fn get_orders_by_tx(
+    orderbook: Arc<Orderbook>,
+) -> impl Filter<Extract = (impl Reply,), Error = Rejection> + Clone {
+    get_orders_by_tx_request().and_then(move |hash: H256| {
+        let orderbook = orderbook.clone();
+        async move {
+            let result = orderbook.get_orders_for_tx(&hash).await;
+            Result::<_, Infallible>::Ok(get_orders_by_tx_response(result))
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::api::response_body;
+    use std::str::FromStr;
+
+    #[tokio::test]
+    async fn request_ok() {
+        let hash_str = "0x0191dbb560e936bd3320d5a505c9c05580a0ebb7e12fe117551ac26e484f295e";
+        let result = warp::test::request()
+            .path(&format!("/transactions/{:}/orders", hash_str))
+            .method("GET")
+            .filter(&get_orders_by_tx_request())
+            .await
+            .unwrap();
+        assert_eq!(result.0, H256::from_str(hash_str).unwrap().0);
+    }
+
+    #[tokio::test]
+    async fn response_ok() {
+        let orders = vec![Order::default()];
+        let response = get_orders_by_tx_response(Ok(orders.clone())).into_response();
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = response_body(response).await;
+        let response_orders: Vec<Order> = serde_json::from_slice(body.as_slice()).unwrap();
+        assert_eq!(response_orders, orders);
+    }
+}

orderbook/src/database/instrumented.rs

@@ -1,5 +1,7 @@
 use super::{orders::OrderStoring, trades::TradeRetrieving, Postgres};
 use crate::fee::MinFeeStoring;
+use ethcontract::H256;
+use model::order::Order;
 use prometheus::Histogram;
 use shared::{event_handling::EventStoring, maintenance::Maintaining};
 use std::sync::Arc;
@@ -129,6 +131,14 @@ impl OrderStoring for Instrumented {
         self.inner.orders(filter).await
     }
 
+    async fn orders_for_tx(&self, tx_hash: &H256) -> anyhow::Result<Vec<Order>> {
+        let _timer = self
+            .metrics
+            .database_query_histogram("orders_for_tx")
+            .start_timer();
+        self.inner.orders_for_tx(tx_hash).await
+    }
+
     async fn single_order(
         &self,
         uid: &model::order::OrderUid,

orderbook/src/database/orders.rs

@@ -4,6 +4,7 @@ use anyhow::{anyhow, Context, Result};
 use bigdecimal::{BigDecimal, Zero};
 use chrono::{DateTime, Utc};
 use const_format::concatcp;
+use ethcontract::H256;
 use futures::stream::TryStreamExt;
 use model::{
     app_id::AppId,
@@ -23,6 +24,7 @@ pub trait OrderStoring: Send + Sync {
     async fn cancel_order(&self, order_uid: &OrderUid, now: DateTime<Utc>) -> Result<()>;
     // Legacy generic orders route that we are phasing out.
     async fn orders(&self, filter: &OrderFilter) -> Result<Vec<Order>>;
+    async fn orders_for_tx(&self, tx_hash: &H256) -> Result<Vec<Order>>;
     async fn single_order(&self, uid: &OrderUid) -> Result<Option<Order>>;
     /// Orders that are solvable: minimum valid to, not fully executed, not invalidated.
     async fn solvable_orders(&self, min_valid_to: u32) -> Result<Vec<Order>>;
@@ -322,6 +324,50 @@ impl OrderStoring for Postgres {
             .await
     }
 
+    async fn orders_for_tx(&self, tx_hash: &H256) -> Result<Vec<Order>> {
+        // TODO - This query assumes there is only one settlement per block.
+        //  when there are two, we would want all trades for which the log index is between
+        //  that of the correct settlement and the next. For this we would have to
+        //  - fetch all settlements for the block containing the specified txHash
+        //  - sort them by log index
+        //  - pick out the target settlement and get all trades with log index between target's and next.
+        //  I believe this would require a string of queries something like
+        // with target_block_number as (
+        //     SELECT block_number from settlements where tx_hash = $1
+        // ),
+        // with next_log_index as (
+        //     SELECT log_index from settlements
+        //     WHERE block_number > target_block_number
+        //     ORDER BY block_number asc
+        //     LIMIT 1
+        // )
+        // "SELECT ", ORDERS_SELECT,
+        // "FROM ", ORDERS_FROM,
+        // "JOIN trades t \
+        //     ON t.order_uid = o.uid \
+        //  JOIN settlements s \
+        //     ON s.block_number = t.block_number \
+        //  WHERE s.tx_hash = $1 \
+        //  AND t.log_index BETWEEN s.log_index AND next_log_index"
+        #[rustfmt::skip]
+        const QUERY: &str = concatcp!(
+            "SELECT ", ORDERS_SELECT,
+            "FROM ", ORDERS_FROM,
+            "JOIN trades t \
+                ON t.order_uid = o.uid \
+             JOIN settlements s \
+                ON s.block_number = t.block_number \
+             WHERE s.tx_hash = $1 ",
+        );
+        sqlx::query_as(QUERY)
+            .bind(tx_hash.0.as_ref())
+            .fetch(&self.pool)
+            .err_into()
+            .and_then(|row: OrdersQueryRow| async move { row.into_order() })
+            .try_collect()
+            .await
+    }
+
     async fn single_order(&self, uid: &OrderUid) -> Result<Option<Order>> {
         #[rustfmt::skip]
         const QUERY: &str = concatcp!(
@@ -1621,4 +1667,59 @@ mod tests {
         let result = db.user_orders(&owners[0], 2, Some(1)).await.unwrap();
         assert_eq!(result, vec![]);
     }
+
+    #[tokio::test]
+    #[ignore]
+    async fn postgres_returns_expected_orders_for_tx_hash_request() {
+        let db = Postgres::new("postgresql://").unwrap();
+        db.clear().await.unwrap();
+
+        let orders: Vec<Order> = (0..=3)
+            .map(|i| Order {
+                order_meta_data: OrderMetaData {
+                    uid: OrderUid::from_integer(i),
+                    ..Default::default()
+                },
+                order_creation: Default::default(),
+            })
+            .collect();
+
+        // Each order was traded in the consecutive blocks.
+        for (i, order) in orders.clone().iter().enumerate() {
+            db.insert_order(order).await.unwrap();
+            db.append_events_(vec![
+                // Add settlement
+                (
+                    EventIndex {
+                        block_number: i as u64,
+                        log_index: 0,
+                    },
+                    Event::Settlement(Settlement {
+                        solver: Default::default(),
+                        transaction_hash: H256::from_low_u64_be(i as u64),
+                    }),
+                ),
+                // Add trade
+                (
+                    EventIndex {
+                        block_number: i as u64,
+                        log_index: 1,
+                    },
+                    Event::Trade(Trade {
+                        order_uid: order.order_meta_data.uid,
+                        ..Default::default()
+                    }),
+                ),
+            ])
+            .await
+            .unwrap();
+        }
+        for (i, order) in orders.into_iter().enumerate() {
+            let res = db
+                .orders_for_tx(&H256::from_low_u64_be(i as u64))
+                .await
+                .unwrap();
+            assert_eq!(res, vec![order]);
+        }
+    }
 }

orderbook/src/orderbook.rs

@@ -5,6 +5,7 @@ use crate::{
 };
 use anyhow::{ensure, Result};
 use chrono::Utc;
+use ethcontract::H256;
 use model::{
     order::{Order, OrderCancellation, OrderCreationPayload, OrderStatus, OrderUid},
     signature::SigningScheme,
@@ -178,6 +179,12 @@ impl Orderbook {
         Ok(Some(order))
     }
 
+    pub async fn get_orders_for_tx(&self, hash: &H256) -> Result<Vec<Order>> {
+        let mut orders = self.database.orders_for_tx(hash).await?;
+        set_available_balances(orders.as_mut_slice(), &self.solvable_orders);
+        Ok(orders)
+    }
+
     pub async fn get_solvable_orders(&self) -> Result<Vec<Order>> {
         let (orders, timestamp) = self.solvable_orders.cached_solvable_orders();
         ensure!(