fix cursor for v1 and v2 compat

Author: ryansmith3136

be/src/api_sql.rs

@@ -189,10 +189,7 @@ async fn query(
         .iter()
         .map(|r| {
             query::sql(
-                Some(query::Cursor::new(
-                    r.chain.unwrap_or_default(),
-                    r.block_height,
-                )),
+                &mut query::Cursor::new(r.chain.unwrap_or_default(), r.block_height),
                 r.event_signatures.iter().map(|s| s.as_str()).collect(),
                 &r.query,
             )
@@ -224,8 +221,8 @@ async fn query(
         .get::<usize, U64>(0)
         .to::<u64>();
     let mut result: Vec<Rows> = Vec::new();
-    for eq in queries {
-        result.push(handle_rows(pgtx.query(&eq.query, &[]).await?)?);
+    for q in queries {
+        result.push(handle_rows(pgtx.query(&q, &[]).await?)?);
     }
     Ok(Response {
         block_height,

be/src/api_sql2.rs

@@ -35,7 +35,8 @@ use crate::{
 pub struct Request {
     #[serde(alias = "api-key")]
     pub api_key: Option<api::Key>,
-    pub cursor: Option<query::Cursor>,
+    #[serde(default)]
+    pub cursor: query::Cursor,
     #[serde(default)]
     pub signatures: Vec<String>,
     pub query: String,
@@ -101,7 +102,7 @@ pub async fn handle_sse(
             .await
             {
                 Ok(resp) if resp.len() == 1 => {
-                    req.cursor = Some(resp[0].cursor.clone());
+                    req.cursor = resp[0].cursor.clone();
                     yield Ok(SSEvent::default().json_data(&resp).unwrap());
                 }
                 Err(err) => {
@@ -120,7 +121,7 @@ pub async fn handle_sse(
                 match val {
                     Ok(broadcast::Message::Close) => return,
                     Ok(broadcast::Message::Block(new_block)) => {
-                        if req.cursor.as_ref().unwrap().contains(new_block.chain)  {
+                        if req.cursor.contains(new_block.chain)  {
                             break;
                         }
                     },
@@ -163,7 +164,7 @@ impl RequestLog {
         for req in &log.requests {
             gafe.log_query(
                 req.api_key.clone(),
-                req.cursor.clone().unwrap_or_default(),
+                req.cursor.clone(),
                 req.signatures.clone(),
                 req.query.clone(),
                 latency,
@@ -208,14 +209,16 @@ async fn query(
     .await?;
     let mut result: Vec<Response> = Vec::new();
     for r in requests {
-        let eq = query::sql(
-            r.cursor.clone(),
+        let mut cursor = r.cursor.clone();
+        let q = query::sql(
+            &mut cursor,
             r.signatures.iter().map(|s| s.as_str()).collect(),
             &r.query,
         )?;
-        let rows = pgtx.query(&eq.query, &[]).await?;
+        let rows = pgtx.query(&q, &[]).await?;
+        update_cursor(&pgtx, &mut cursor).await?;
         result.push(Response {
-            cursor: update_cursor(&pgtx, eq.cusror).await?,
+            cursor,
             columns: get_columns(&rows),
             rows: get_rows(&rows),
         });
@@ -225,8 +228,8 @@ async fn query(
 
 async fn update_cursor(
     pgtx: &tokio_postgres::Transaction<'_>,
-    mut cursor: query::Cursor,
-) -> Result<query::Cursor, api::Error> {
+    cursor: &mut query::Cursor,
+) -> Result<(), api::Error> {
     for c in cursor.chains() {
         let row = pgtx
             .query_one(
@@ -236,7 +239,7 @@ async fn update_cursor(
             .await?;
         cursor.set_block_height(c, row.get::<usize, U64>(0).to());
     }
-    Ok(cursor)
+    Ok(())
 }
 
 fn get_columns(rows: &[tokio_postgres::Row]) -> HashMap<String, String> {

be/src/query.rs

@@ -35,21 +35,18 @@ impl Cursor {
         Cursor(map)
     }
 
-    pub fn from_chains(chains: &HashSet<u64>) -> Self {
-        let map = chains.iter().map(|&c| (c, None)).collect();
-        Cursor(map)
+    pub fn add_chains(&mut self, chains: &HashSet<u64>) {
+        chains.iter().for_each(|&c| {
+            self.0.entry(c).or_insert(None);
+        })
     }
 
     pub fn contains(&self, chain: u64) -> bool {
         self.0.keys().any(|c| *c == chain)
     }
 
     pub fn chains(&self) -> Vec<u64> {
-        self.0.keys().cloned().collect()
-    }
-
-    pub fn set_block_height(&mut self, chain: u64, n: u64) {
-        self.0.insert(chain, Some(n));
+        self.0.keys().sorted().cloned().collect()
     }
 
     pub fn chain(&self) -> u64 {
@@ -59,36 +56,42 @@ impl Cursor {
         }
     }
 
-    pub fn to_sql(&self) -> String {
-        self.0
+    pub fn set_block_height(&mut self, chain: u64, n: u64) {
+        self.0.insert(chain, Some(n));
+    }
+
+    pub fn to_sql(&self, col_name: &str) -> String {
+        let predicates = self
+            .0
             .iter()
+            .sorted_by_key(|(chain, _)| *chain)
             .map(|(chain, block_num)| match block_num {
-                Some(n) => format!("(chain = {} and block_num > {})", chain, n),
+                Some(n) => format!("(chain = {} and {} > {})", chain, col_name, n),
                 None => format!("chain = {}", chain),
             })
-            .collect::<Vec<_>>()
-            .join(" or ")
+            .collect::<Vec<_>>();
+        if predicates.len() == 1 {
+            predicates[0].clone()
+        } else {
+            format!("({})", predicates.join(" or "))
+        }
     }
 }
 
-pub struct EnhancedQuery {
-    pub query: String,
-    pub cusror: Cursor,
-}
-
 const PG: &sqlparser::dialect::PostgreSqlDialect = &sqlparser::dialect::PostgreSqlDialect {};
 
 /// Parses the user supplied query into a SQL AST
 /// and validates the query against the provided event signatures.
 /// The SQL API implements onlny a subset of SQL so un-supported
 /// SQL results in an error.
 pub fn sql(
-    cursor: Option<Cursor>,
+    cursor: &mut Cursor,
     signatures: Vec<&str>,
     user_query: &str,
-) -> Result<EnhancedQuery, api::Error> {
+) -> Result<String, api::Error> {
     let mut q = UserQuery::new(signatures)?;
-    let new_query = q.process(user_query)?;
+    let rewritten_query = q.process(user_query)?;
+    cursor.add_chains(&q.chains);
     let query = [
         "with".to_string(),
         q.relations
@@ -97,13 +100,10 @@ pub fn sql(
             .sorted_by_key(|s| s.table_name.to_string())
             .map(|rel| rel.to_sql(cursor.clone()))
             .join(","),
-        new_query.to_string(),
+        rewritten_query.to_string(),
     ]
     .join(" ");
-    Ok(EnhancedQuery {
-        query,
-        cusror: Cursor::from_chains(&q.chains),
-    })
+    Ok(query)
 }
 
 /*
@@ -153,7 +153,7 @@ impl Relation {
             || self.table_alias.contains(other)
     }
 
-    fn to_sql(&self, cursor: Option<Cursor>) -> String {
+    fn to_sql(&self, cursor: Cursor) -> String {
         let mut res: Vec<String> = Vec::new();
         res.push(format!("{} as not materialized (", self.table_name));
         res.push("select".to_string());
@@ -182,8 +182,10 @@ impl Relation {
         res.push(select_list.join(","));
 
         let mut predicates = vec![];
-        if let Some(cursor) = cursor {
-            predicates.push(cursor.to_sql());
+        if self.table_name.value.to_lowercase() == "blocks" {
+            predicates.push(cursor.to_sql("num"));
+        } else {
+            predicates.push(cursor.to_sql("block_num"));
         }
         if let Some(event) = self.event.as_ref() {
             predicates.push(event.sighash_sql_predicate());
@@ -950,10 +952,10 @@ mod tests {
     }
 
     async fn check_sql(event_sigs: Vec<&str>, user_query: &str, want: &str) {
-        let got = sql(Some(Cursor::new(1, None)), event_sigs, user_query)
+        let got = sql(&mut Cursor::new(1, None), event_sigs, user_query)
             .unwrap_or_else(|e| panic!("unable to create sql for:\n{} error: {:?}", user_query, e));
         let (got, want) = (
-            fmt_sql(&got.query).unwrap_or_else(|_| panic!("unable to format got: {}", got.query)),
+            fmt_sql(&got).unwrap_or_else(|_| panic!("unable to format got: {}", got)),
             fmt_sql(want).unwrap_or_else(|_| panic!("unable to format want: {}", want)),
         );
         if got.to_lowercase().ne(&want.to_lowercase()) {
@@ -964,6 +966,24 @@ mod tests {
         pg.query(&got, &[]).await.expect("issue with query");
     }
 
+    #[test]
+    fn test_cursor() {
+        let mut cursor = Cursor::default();
+        cursor.set_block_height(8453, 100);
+        cursor.set_block_height(10, 42);
+        let _ = sql(
+            &mut cursor,
+            vec![],
+            "select hash from txs where chain in (8453, 10, 1)",
+        )
+        .unwrap();
+        assert_eq!(cursor.chains(), vec![1, 10, 8453]);
+        assert_eq!(
+            cursor.to_sql("foo"),
+            "(chain = 1 or (chain = 10 and foo > 42) or (chain = 8453 and foo > 100))"
+        );
+    }
+
     #[tokio::test]
     async fn test_blocks_table() {
         check_sql(

fe/src/god_mode.rs

@@ -19,7 +19,7 @@ time::serde::format_description!(
 #[derive(Clone, Debug, Serialize)]
 struct UserQuery {
     owner_email: Option<String>,
-    cursor: Option<query::Cursor>,
+    cursor: query::Cursor,
     events: Vec<String>,
     sql: String,
     latency: Option<u64>,
@@ -34,12 +34,11 @@ struct UserQuery {
 impl UserQuery {
     pub fn gen_sql(mut self) -> UserQuery {
         self.generated_sql = be::query::sql(
-            self.cursor.clone(),
+            &mut self.cursor.clone(),
             self.events.iter().map(AsRef::as_ref).collect(),
             &self.sql,
         )
-        .ok()
-        .map(|eq| eq.query);
+        .ok();
         self
     }
 }
@@ -117,7 +116,7 @@ async fn log(
         .await?
         .into_iter()
         .map(|row| UserQuery {
-            cursor: Some(query::Cursor::default()),
+            cursor: query::Cursor::default(),
             owner_email: row.get("owner_email"),
             events: row.get("events"),
             sql: row.get("user_query"),
@@ -164,7 +163,7 @@ async fn top(
         .await?
         .into_iter()
         .map(|row| UserQuery {
-            cursor: Some(query::Cursor::default()),
+            cursor: query::Cursor::default(),
             owner_email: row.get("owner_email"),
             events: row.get("events"),
             sql: row.get("user_query"),