add tempo fields

Author: ryansmith3136

be/src/query.rs

@@ -460,6 +460,7 @@ impl UserQuery {
         if let Some(lit) = expr_to_bytes(right) {
             if let Some(param) = self.get_param(left) {
                 *right = match param {
+                    abi::Parameter::Tuple { .. } => right.clone(),
                     abi::Parameter::Address { .. } | abi::Parameter::Uint { .. } => {
                         ast::Expr::Value(ast::Value::SingleQuotedString(format!(
                             r#"\x{}"#,
@@ -596,24 +597,15 @@ impl UserQuery {
             ast::Expr::IsNotFalse(_) => Ok(()),
             ast::Expr::IsTrue(_) => Ok(()),
             ast::Expr::IsNotTrue(_) => Ok(()),
-            ast::Expr::IsNull(_) => Ok(()),
-            ast::Expr::IsNotNull(_) => Ok(()),
+            ast::Expr::IsNull(expr) => self.validate_expression(expr),
+            ast::Expr::IsNotNull(expr) => self.validate_expression(expr),
             ast::Expr::Ceil { expr, field: _ } => self.validate_expression(expr),
             ast::Expr::Floor { expr, field: _ } => self.validate_expression(expr),
             ast::Expr::Value(_) => Ok(()),
             ast::Expr::Exists { subquery, .. } => self.validate_query(subquery),
             ast::Expr::Subquery(subquery) => self.validate_query(subquery),
             ast::Expr::Tuple(exprs) => self.validate_expressions(exprs),
             ast::Expr::UnaryOp { expr, .. } => self.validate_expression(expr),
-            ast::Expr::BinaryOp {
-                left,
-                right,
-                op: ast::BinaryOperator::LongArrow | ast::BinaryOperator::Arrow,
-                ..
-            } => {
-                self.validate_expression(left)?;
-                self.validate_expression(right)
-            }
             ast::Expr::BinaryOp { left, right, .. } => {
                 self.set_chain(left, right);
                 self.rewrite_binary_expr(left, right)?;
@@ -658,7 +650,7 @@ impl UserQuery {
 
     fn validate_function(&mut self, function: &mut ast::Function) -> Result<(), api::Error> {
         let name = function.name.to_string().to_lowercase();
-        const VALID_FUNCS: [&str; 14] = [
+        const VALID_FUNCS: [&str; 15] = [
             "coalesce",
             "min",
             "max",
@@ -673,6 +665,7 @@ impl UserQuery {
             "abi_uint",
             "abi_int",
             "abi_string",
+            "jsonb_path_query_array",
         ];
         if !VALID_FUNCS.contains(&name.as_str()) {
             return no!(format!(r#"'{}' function"#, name));
@@ -810,6 +803,9 @@ fn bytes_to_expr(bytes: Vec<u8>, to: ast::DataType) -> Result<ast::Expr, api::Er
                 value: s,
             })
         }
+        ast::DataType::JSONB => Ok(ast::Expr::Value(ast::Value::SingleQuotedString(
+            String::from_utf8(bytes).map_err(|_| api::Error::User("invalid JSONB".to_string()))?,
+        ))),
         _ => Err(api::Error::User(format!(
             "unable to convert {bytes:?} to {to:?}",
         ))),
@@ -831,9 +827,10 @@ fn base_column_type(id: &Ident) -> Option<ast::DataType> {
 
         // Txs
         "idx" | "type" => Some(ast::DataType::Int64),
-        "from" | "to" | "input" => Some(ast::DataType::Bytea),
+        "from" | "to" | "input" | "fee_token" => Some(ast::DataType::Bytea),
         "value" => Some(ast::DataType::Numeric(ast::ExactNumberInfo::None)),
         "gas" | "gas_price" => Some(ast::DataType::Numeric(ast::ExactNumberInfo::None)),
+        "calls" => Some(ast::DataType::JSONB),
 
         // Logs
         "tx_hash" | "address" | "topics" | "data" => Some(ast::DataType::Bytea),

be/src/sql/schema.sql

@@ -30,7 +30,9 @@ create table if not exists txs (
     "from" bytea not null,
     "to" bytea not null,
     input bytea not null,
-    value numeric not null
+    value numeric not null,
+    fee_token bytea,
+    calls jsonb
 ) partition by list(chain);
 
 create table if not exists logs (

be/src/sync.rs

@@ -551,7 +551,9 @@ pub async fn copy_txs(
             "from",
             "to",
             input,
-            value
+            value,
+            fee_token,
+            calls
         )
         from stdin binary
     "#;
@@ -572,6 +574,8 @@ pub async fn copy_txs(
             tokio_postgres::types::Type::BYTEA,
             tokio_postgres::types::Type::BYTEA,
             tokio_postgres::types::Type::NUMERIC,
+            tokio_postgres::types::Type::BYTEA,
+            tokio_postgres::types::Type::JSONB,
         ],
     );
     pin_mut!(writer);
@@ -595,6 +599,11 @@ pub async fn copy_txs(
                     &tx.to.unwrap_or_default().to_vec(),
                     &tx.input.to_vec(),
                     &tx.value,
+                    &tx.fee_token.as_ref().map(|a| a.to_vec()),
+                    &tx.calls
+                        .as_ref()
+                        .filter(|v| !v.is_empty())
+                        .map(|v| tokio_postgres::types::Json(v.as_slice())),
                 ])
                 .await?;
         }

shared/src/jrpc.rs

@@ -1,6 +1,6 @@
 use alloy::primitives::{Address, BlockHash, Bytes, FixedBytes, U256, U64};
 use itertools::Itertools;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
 use std::{fmt, time::Duration};
 
@@ -41,6 +41,14 @@ pub struct Log {
     pub data: Bytes,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Call {
+    pub from: Option<Address>,
+    pub to: Option<Address>,
+    pub value: Option<U256>,
+    pub data: Option<Bytes>,
+}
+
 #[derive(Deserialize, Debug)]
 pub struct Tx {
     #[serde(rename = "type")]
@@ -61,6 +69,9 @@ pub struct Tx {
     pub gas: U256,
     #[serde(rename = "gasPrice")]
     pub gas_price: Option<U256>,
+    pub calls: Option<Vec<Call>>,
+    #[serde(rename = "feeToken")]
+    pub fee_token: Option<Address>,
 }
 
 #[derive(Deserialize, Debug)]
@@ -115,6 +126,7 @@ impl Client {
         }
     }
 
+    #[tracing::instrument(level="info" skip_all)]
     pub async fn chain_id(&self) -> Result<U64, Error> {
         let request = serde_json::json!({
             "id": 1,
@@ -196,13 +208,13 @@ impl Client {
             .collect())
     }
 
-    #[tracing::instrument(level="info" skip_all, fields(number))]
-    pub async fn block(&self, param: String) -> Result<Block, Error> {
+    #[tracing::instrument(level="info" skip_all, fields(number = %number))]
+    pub async fn block(&self, number: String) -> Result<Block, Error> {
         let request = serde_json::json!({
             "id": "1",
             "jsonrpc": "2.0",
             "method": "eth_getBlockByNumber",
-            "params": [param, true],
+            "params": [number, true],
         });
         let response: RpcEither<Block> = self
             .http_client