feat: introduce updated at field in top level torii query

crates/torii/core/src/error.rs

@@ -51,4 +51,6 @@ pub enum QueryError {
     SqliteJoinLimit,
     #[error("Invalid namespaced model: {0}")]
     InvalidNamespacedModel(String),
+    #[error("Invalid timestamp: {0}. Expected valid number of seconds since unix epoch.")]
+    InvalidTimestamp(u64),
 }

crates/torii/core/src/model.rs

@@ -116,6 +116,7 @@
 }
 
 /// Creates a query that fetches all models and their nested data.
+#[allow(clippy::too_many_arguments)]
 pub fn build_sql_query(
     schemas: &Vec<Ty>,
     table_name: &str,
@@ -124,6 +125,7 @@
     order_by: Option<&str>,
     limit: Option<u32>,
     offset: Option<u32>,
+    internal_updated_at: u64,
 ) -> Result<(String, String), Error> {
     fn collect_columns(table_prefix: &str, path: &str, ty: &Ty, selections: &mut Vec<String>) {
         match ty {
@@ -172,6 +174,7 @@
     selections.push(format!("{}.id", table_name));
     selections.push(format!("{}.keys", table_name));
 
+    let mut internal_updated_at_clause = Vec::with_capacity(schemas.len());
     // Process each model schema
     for model in schemas {
         let model_table = model.name();
@@ -180,6 +183,10 @@
              [{model_table}].{entity_relation_column}",
         ));
 
+        if internal_updated_at > 0 {
+            internal_updated_at_clause.push(format!("[{model_table}].internal_updated_at >= ?"));
+        }
+
         // Collect columns with table prefix
         collect_columns(&model_table, "", model, &mut selections);
     }
@@ -197,6 +204,18 @@
         count_query += &format!(" WHERE {}", where_clause);
     }
 
+    if !internal_updated_at_clause.is_empty() {
+        if where_clause.is_none() {
+            query += " WHERE ";
+            count_query += " WHERE ";
+        } else {
+            query += " AND ";
+            count_query += " AND ";
+        }
+        query += &format!(" {}", internal_updated_at_clause.join(" AND "));
+        count_query += &format!(" {}", internal_updated_at_clause.join(" AND "));
+    }
+
     // Use custom order by if provided, otherwise default to event_id DESC
     if let Some(order_clause) = order_by {
         query += &format!(" ORDER BY {}", order_clause);
@@ -494,6 +513,7 @@
             None,
             None,
             None,
+            0,
         )
         .unwrap();
 

crates/torii/grpc/proto/types.proto

@@ -76,6 +76,7 @@ message Query {
     bool dont_include_hashed_keys = 4;
     repeated OrderBy order_by = 5;
     repeated string entity_models = 6;
+	uint64 internal_updated_at = 7;
 }
 
 message EventQuery {

crates/torii/grpc/src/server/mod.rs

@@ -27,6 +27,7 @@
 use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use sqlx::prelude::FromRow;
 use sqlx::sqlite::SqliteRow;
+use sqlx::types::chrono::{DateTime, Utc};
 use sqlx::{Pool, Row, Sqlite};
 use starknet::core::types::Felt;
 use starknet::providers::jsonrpc::HttpTransport;
@@ -229,6 +230,7 @@
         dont_include_hashed_keys: bool,
         order_by: Option<&str>,
         entity_models: Vec<String>,
+        internal_updated_at: u64,
     ) -> Result<(Vec<proto::types::Entity>, u32), Error> {
         self.query_by_hashed_keys(
             table,
@@ -240,6 +242,7 @@
             dont_include_hashed_keys,
             order_by,
             entity_models,
+            internal_updated_at,
         )
         .await
     }
@@ -258,6 +261,7 @@
         row_events.iter().map(map_row_to_event).collect()
     }
 
+    #[allow(clippy::too_many_arguments)]
     async fn fetch_entities(
         &self,
         table: &str,
@@ -266,7 +270,8 @@
         dont_include_hashed_keys: bool,
         order_by: Option<&str>,
         entity_models: Vec<String>,
-    ) -> Result<Vec<proto::types::Entity>, Error> {
+        internal_updated_at: u64,
+    ) -> Result<(Vec<proto::types::Entity>, usize), Error> {
         let entity_models =
             entity_models.iter().map(|tag| compute_selector_from_tag(tag)).collect::<Vec<Felt>>();
 
@@ -319,6 +324,8 @@
             insert_start.elapsed()
         );
 
+        let mut query_result_count: usize = 0;
+
         let query_start = std::time::Instant::now();
         for (models_str, entity_ids) in &model_groups {
             tracing::debug!("Processing model group with {} entities", entity_ids.len());
@@ -344,7 +351,7 @@
                 continue;
             }
 
-            let (entity_query, _) = build_sql_query(
+            let (entity_query, count_query) = build_sql_query(
                 &schemas,
                 table,
                 entity_relation_column,
@@ -354,9 +361,27 @@
                 order_by,
                 None,
                 None,
+                internal_updated_at,
             )?;
 
-            let rows = sqlx::query(&entity_query).bind(models_str).fetch_all(&mut *tx).await?;
+            let mut query = sqlx::query(&entity_query).bind(models_str);
+            let mut count_query = sqlx::query(&count_query).bind(models_str);
+            if internal_updated_at > 0 {
+                for _ in 0..schemas.len() {
+                    let time = DateTime::<Utc>::from_timestamp(internal_updated_at as i64, 0)
+                        .ok_or_else(|| {
+                            Error::from(QueryError::InvalidTimestamp(internal_updated_at))
+                        })?
+                        .to_rfc3339();
+                    query = query.bind(time.clone());
+                    count_query = count_query.bind(time);
+                }
+            }
+            let rows = query.fetch_all(&mut *tx).await?;
+            let fetch_one_result = count_query.fetch_one(&mut *tx).await?;
+            let query_count: u32 = fetch_one_result.get(0);
+            query_result_count += query_count as usize;
+
             let schemas = Arc::new(schemas);
 
             let group_entities: Result<Vec<_>, Error> = rows
@@ -376,7 +401,7 @@
 
         tracing::debug!("Total fetch_entities operation took {:?}", start.elapsed());
 
-        Ok(all_entities)
+        Ok((all_entities, query_result_count))
     }
 
     async fn fetch_historical_event_messages(
@@ -430,6 +455,7 @@
         dont_include_hashed_keys: bool,
         order_by: Option<&str>,
         entity_models: Vec<String>,
+        internal_updated_at: u64,
     ) -> Result<(Vec<proto::types::Entity>, u32), Error> {
         // TODO: use prepared statement for where clause
         let filter_ids = match hashed_keys {
@@ -502,17 +528,18 @@
         let db_entities: Vec<(String, String)> =
             sqlx::query_as(&query).bind(limit).bind(offset).fetch_all(&self.pool).await?;
 
-        let entities = self
+        let (entities, query_result_count) = self
             .fetch_entities(
                 table,
                 entity_relation_column,
                 db_entities,
                 dont_include_hashed_keys,
                 order_by,
                 entity_models,
+                internal_updated_at,
             )
             .await?;
-        Ok((entities, total_count))
+        Ok((entities, query_result_count as u32))
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -527,6 +554,7 @@
         dont_include_hashed_keys: bool,
         order_by: Option<&str>,
         entity_models: Vec<String>,
+        internal_updated_at: u64,
     ) -> Result<(Vec<proto::types::Entity>, u32), Error> {
         let keys_pattern = build_keys_pattern(keys_clause)?;
 
@@ -647,17 +675,18 @@
             .fetch_all(&self.pool)
             .await?;
 
-        let entities = self
+        let (entities, query_result_count) = self
             .fetch_entities(
                 table,
                 entity_relation_column,
                 db_entities,
                 dont_include_hashed_keys,
                 order_by,
                 entity_models,
+                internal_updated_at,
             )
             .await?;
-        Ok((entities, total_count))
+        Ok((entities, query_result_count as u32))
     }
 
     pub(crate) async fn events_by_keys(
@@ -699,6 +728,7 @@
         dont_include_hashed_keys: bool,
         order_by: Option<&str>,
         entity_models: Vec<String>,
+        internal_updated_at: u64,
     ) -> Result<(Vec<proto::types::Entity>, u32), Error> {
         let entity_models =
             entity_models.iter().map(|model| compute_selector_from_tag(model)).collect::<Vec<_>>();
@@ -765,6 +795,7 @@
             order_by,
             limit,
             offset,
+            internal_updated_at,
         )?;
 
         let total_count = sqlx::query_scalar(&count_query)
@@ -798,6 +829,7 @@
         dont_include_hashed_keys: bool,
         order_by: Option<&str>,
         entity_models: Vec<String>,
+        internal_updated_at: u64,
     ) -> Result<(Vec<proto::types::Entity>, u32), Error> {
         let (where_clause, having_clause, join_clause, bind_values) =
             build_composite_clause(table, model_relation_table, &composite)?;
@@ -860,17 +892,18 @@
 
         let db_entities: Vec<(String, String)> = db_query.fetch_all(&self.pool).await?;
 
-        let entities = self
+        let (entities, query_result_count) = self
             .fetch_entities(
                 table,
                 entity_relation_column,
                 db_entities,
                 dont_include_hashed_keys,
                 order_by,
                 entity_models,
+                internal_updated_at,
             )
             .await?;
-        Ok((entities, total_count))
+        Ok((entities, query_result_count as u32))
     }
 
     pub async fn model_metadata(
@@ -1036,6 +1069,7 @@
                     query.dont_include_hashed_keys,
                     order_by,
                     query.entity_models,
+                    query.internal_updated_at,
                 )
                 .await?
             }
@@ -1059,6 +1093,7 @@
                             query.dont_include_hashed_keys,
                             order_by,
                             query.entity_models,
+                            query.internal_updated_at,
                         )
                         .await?
                     }
@@ -1073,6 +1108,7 @@
                             query.dont_include_hashed_keys,
                             order_by,
                             query.entity_models,
+                            query.internal_updated_at,
                         )
                         .await?
                     }
@@ -1087,6 +1123,7 @@
                             query.dont_include_hashed_keys,
                             order_by,
                             query.entity_models,
+                            query.internal_updated_at,
                         )
                         .await?
                     }
@@ -1101,6 +1138,7 @@
                             query.dont_include_hashed_keys,
                             order_by,
                             query.entity_models,
+                            query.internal_updated_at,
                         )
                         .await?
                     }

crates/torii/grpc/src/server/tests/entities_test.rs

@@ -143,6 +143,7 @@ async fn test_entities_queries(sequencer: &RunnerCtx) {
             false,
             None,
             vec![],
+            0,
         )
         .await
         .unwrap()

crates/torii/grpc/src/types/mod.rs

@@ -106,6 +106,7 @@
     pub dont_include_hashed_keys: bool,
     pub order_by: Vec<OrderBy>,
     pub entity_models: Vec<String>,
+    pub internal_updated_at: u64,
 }
 
 #[derive(Debug, Serialize, Deserialize, PartialEq, Hash, Eq, Clone)]
@@ -271,6 +272,7 @@
             dont_include_hashed_keys: value.dont_include_hashed_keys,
             order_by: value.order_by.into_iter().map(|o| o.into()).collect(),
             entity_models: value.entity_models,
+            internal_updated_at: value.internal_updated_at,
         }
     }
 }