feat(torii-graphql): event messages update & subscription

crates/torii/core/src/sql_test.rs

@@ -94,13 +94,13 @@ async fn test_load_from_remote() {
 
     let world = WorldContract::new(strat.world_address, &account);
 
-    world
+    let res = world
         .grant_writer(&compute_bytearray_hash("dojo_examples"), &ContractAddress(actions_address))
         .send_with_cfg(&TxnConfig::init_wait())
         .await
         .unwrap();
 
-    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
     // spawn
     let tx = &account
@@ -237,16 +237,16 @@ async fn test_load_from_remote_del() {
 
     let world = WorldContract::new(strat.world_address, &account);
 
-    world
+    let res = world
         .grant_writer(&compute_bytearray_hash("dojo_examples"), &ContractAddress(actions_address))
         .send_with_cfg(&TxnConfig::init_wait())
         .await
         .unwrap();
 
-    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
     // spawn
-    account
+    let res = account
         .execute_v1(vec![Call {
             to: actions_address,
             selector: get_selector_from_name("spawn").unwrap(),
@@ -256,10 +256,10 @@ async fn test_load_from_remote_del() {
         .await
         .unwrap();
 
-    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
     // Set player config.
-    account
+    let res = account
         .execute_v1(vec![Call {
             to: actions_address,
             selector: get_selector_from_name("set_player_config").unwrap(),
@@ -270,9 +270,9 @@ async fn test_load_from_remote_del() {
         .await
         .unwrap();
 
-    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
-    account
+    let res = account
         .execute_v1(vec![Call {
             to: actions_address,
             selector: get_selector_from_name("reset_player_config").unwrap(),
@@ -282,7 +282,7 @@ async fn test_load_from_remote_del() {
         .await
         .unwrap();
 
-    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
     let world_reader = WorldContractReader::new(strat.world_address, account.provider());
 
@@ -349,15 +349,16 @@ async fn test_get_entity_keys() {
 
     let world = WorldContract::new(strat.world_address, &account);
 
-    world
+    let res = world
         .grant_writer(&compute_bytearray_hash("dojo_examples"), &ContractAddress(actions_address))
         .send_with_cfg(&TxnConfig::init_wait())
         .await
         .unwrap();
-    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
+
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
     // spawn
-    account
+    let res = account
         .execute_v1(vec![Call {
             to: actions_address,
             selector: get_selector_from_name("spawn").unwrap(),
@@ -367,7 +368,7 @@ async fn test_get_entity_keys() {
         .await
         .unwrap();
 
-    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
+    TransactionWaiter::new(res.transaction_hash, &account.provider()).await.unwrap();
 
     let world_reader = WorldContractReader::new(strat.world_address, account.provider());
 

crates/torii/graphql/src/constants.rs

@@ -13,6 +13,7 @@ pub const METADATA_TABLE: &str = "metadata";
 pub const ID_COLUMN: &str = "id";
 pub const EVENT_ID_COLUMN: &str = "event_id";
 pub const ENTITY_ID_COLUMN: &str = "entity_id";
+pub const EVENT_MESSAGE_ID_COLUMN: &str = "event_message_id";
 pub const JSON_COLUMN: &str = "json";
 pub const TRANSACTION_HASH_COLUMN: &str = "transaction_hash";
 

crates/torii/graphql/src/object/entity.rs

@@ -13,7 +13,8 @@ use torii_core::types::Entity;
 use super::inputs::keys_input::keys_argument;
 use super::{BasicObject, ResolvableObject, TypeMapping, ValueMapping};
 use crate::constants::{
-    DATETIME_FORMAT, ENTITY_NAMES, ENTITY_TABLE, ENTITY_TYPE_NAME, EVENT_ID_COLUMN, ID_COLUMN,
+    DATETIME_FORMAT, ENTITY_ID_COLUMN, ENTITY_NAMES, ENTITY_TABLE, ENTITY_TYPE_NAME,
+    EVENT_ID_COLUMN, ID_COLUMN,
 };
 use crate::mapping::ENTITY_TYPE_MAPPING;
 use crate::object::{resolve_many, resolve_one};
@@ -143,6 +144,7 @@ fn model_union_field() -> Field {
                         // but the table name for the model data is the unhashed model name
                         let data: ValueMapping = match model_data_recursive_query(
                             &mut conn,
+                            ENTITY_ID_COLUMN,
                             vec![format!("{namespace}-{name}")],
                             &entity_id,
                             &[],
@@ -173,6 +175,7 @@ fn model_union_field() -> Field {
 #[async_recursion]
 pub async fn model_data_recursive_query(
     conn: &mut PoolConnection<Sqlite>,
+    entity_id_column: &str,
     path_array: Vec<String>,
     entity_id: &str,
     indexes: &[i64],
@@ -182,7 +185,8 @@ pub async fn model_data_recursive_query(
     // For nested types, we need to remove prefix in path array
     let namespace = format!("{}_", path_array[0]);
     let table_name = &path_array.join("$").replace(&namespace, "");
-    let mut query = format!("SELECT * FROM [{}] WHERE entity_id = '{}' ", table_name, entity_id);
+    let mut query =
+        format!("SELECT * FROM [{}] WHERE {entity_id_column} = '{}' ", table_name, entity_id);
     for (column_idx, index) in indexes.iter().enumerate() {
         query.push_str(&format!("AND idx_{} = {} ", column_idx, index));
     }
@@ -205,6 +209,7 @@ pub async fn model_data_recursive_query(
 
                 let nested_values = model_data_recursive_query(
                     conn,
+                    entity_id_column,
                     nested_path,
                     entity_id,
                     &if is_list {
@@ -226,6 +231,7 @@ pub async fn model_data_recursive_query(
 
                 let data = match model_data_recursive_query(
                     conn,
+                    entity_id_column,
                     nested_path,
                     entity_id,
                     // this might need to be changed to support 2d+ arrays

crates/torii/graphql/src/object/event_message.rs

@@ -3,23 +3,22 @@ use async_graphql::dynamic::{
     Field, FieldFuture, FieldValue, InputValue, SubscriptionField, SubscriptionFieldFuture, TypeRef,
 };
 use async_graphql::{Name, Value};
-use async_recursion::async_recursion;
-use sqlx::pool::PoolConnection;
 use sqlx::{Pool, Sqlite};
 use tokio_stream::StreamExt;
 use torii_core::simple_broker::SimpleBroker;
 use torii_core::types::EventMessage;
 
+use super::entity::model_data_recursive_query;
 use super::inputs::keys_input::keys_argument;
 use super::{BasicObject, ResolvableObject, TypeMapping, ValueMapping};
 use crate::constants::{
-    EVENT_ID_COLUMN, EVENT_MESSAGE_NAMES, EVENT_MESSAGE_TABLE, EVENT_MESSAGE_TYPE_NAME, ID_COLUMN,
+    DATETIME_FORMAT, EVENT_ID_COLUMN, EVENT_MESSAGE_ID_COLUMN, EVENT_MESSAGE_NAMES,
+    EVENT_MESSAGE_TABLE, EVENT_MESSAGE_TYPE_NAME, ID_COLUMN,
 };
 use crate::mapping::ENTITY_TYPE_MAPPING;
 use crate::object::{resolve_many, resolve_one};
-use crate::query::{type_mapping_query, value_mapping_from_row};
-use crate::types::TypeData;
-use crate::utils::extract;
+use crate::query::type_mapping_query;
+use crate::utils;
 
 #[derive(Debug)]
 pub struct EventMessageObject;
@@ -106,11 +105,15 @@ impl EventMessageObject {
             (Name::new("eventId"), Value::from(entity.event_id)),
             (
                 Name::new("createdAt"),
-                Value::from(entity.created_at.format("%Y-%m-%d %H:%M:%S").to_string()),
+                Value::from(entity.created_at.format(DATETIME_FORMAT).to_string()),
             ),
             (
                 Name::new("updatedAt"),
-                Value::from(entity.updated_at.format("%Y-%m-%d %H:%M:%S").to_string()),
+                Value::from(entity.updated_at.format(DATETIME_FORMAT).to_string()),
+            ),
+            (
+                Name::new("executedAt"),
+                Value::from(entity.executed_at.format(DATETIME_FORMAT).to_string()),
             ),
         ])
     }
@@ -123,13 +126,13 @@ fn model_union_field() -> Field {
                 Value::Object(indexmap) => {
                     let mut conn = ctx.data::<Pool<Sqlite>>()?.acquire().await?;
 
-                    let entity_id = extract::<String>(indexmap, "id")?;
+                    let entity_id = utils::extract::<String>(indexmap, "id")?;
                     // fetch name from the models table
                     // using the model id (hashed model name)
-                    let model_ids: Vec<(String, String)> = sqlx::query_as(
-                        "SELECT id, name
+                    let model_ids: Vec<(String, String, String)> = sqlx::query_as(
+                        "SELECT id, namespace, name
                         FROM models
-                        WHERE id IN (
+                        WHERE id IN (    
                             SELECT model_id
                             FROM event_model
                             WHERE entity_id = ?
@@ -140,20 +143,30 @@ fn model_union_field() -> Field {
                     .await?;
 
                     let mut results: Vec<FieldValue<'_>> = Vec::new();
-                    for (id, name) in model_ids {
-                        // the model id is used as the id for the model members
+                    for (id, namespace, name) in model_ids {
+                        // the model id in the model mmeebrs table is the hashed model name (id)
                         let type_mapping = type_mapping_query(&mut conn, &id).await?;
 
-                        // but the model data tables use the unhashed model name as the table name
-                        let data = model_data_recursive_query(
+                        // but the table name for the model data is the unhashed model name
+                        let data: ValueMapping = match model_data_recursive_query(
                             &mut conn,
-                            vec![name.clone()],
+                            EVENT_MESSAGE_ID_COLUMN,
+                            vec![format!("{namespace}-{name}")],
                             &entity_id,
+                            &[],
                             &type_mapping,
+                            false,
                         )
-                        .await?;
+                        .await?
+                        {
+                            Value::Object(map) => map,
+                            _ => unreachable!(),
+                        };
 
-                        results.push(FieldValue::with_type(FieldValue::owned_any(data), name));
+                        results.push(FieldValue::with_type(
+                            FieldValue::owned_any(data),
+                            utils::type_name_from_names(&namespace, &name),
+                        ))
                     }
 
                     Ok(Some(FieldValue::list(results)))
@@ -163,33 +176,3 @@ fn model_union_field() -> Field {
         })
     })
 }
-
-// TODO: flatten query
-#[async_recursion]
-pub async fn model_data_recursive_query(
-    conn: &mut PoolConnection<Sqlite>,
-    path_array: Vec<String>,
-    entity_id: &str,
-    type_mapping: &TypeMapping,
-) -> sqlx::Result<ValueMapping> {
-    // For nested types, we need to remove prefix in path array
-    let namespace = format!("{}_", path_array[0]);
-    let table_name = &path_array.join("$").replace(&namespace, "");
-    let query = format!("SELECT * FROM {} WHERE event_message_id = '{}'", table_name, entity_id);
-    let row = sqlx::query(&query).fetch_one(conn.as_mut()).await?;
-    let mut value_mapping = value_mapping_from_row(&row, type_mapping, true)?;
-
-    for (field_name, type_data) in type_mapping {
-        if let TypeData::Nested((_, nested_mapping)) = type_data {
-            let mut nested_path = path_array.clone();
-            nested_path.push(field_name.to_string());
-
-            let nested_values =
-                model_data_recursive_query(conn, nested_path, entity_id, nested_mapping).await?;
-
-            value_mapping.insert(Name::new(field_name), Value::Object(nested_values));
-        }
-    }
-
-    Ok(value_mapping)
-}

crates/torii/graphql/src/object/model_data.rs

@@ -10,7 +10,8 @@ use super::inputs::where_input::{parse_where_argument, where_argument, WhereInpu
 use super::inputs::InputObjectTrait;
 use super::{BasicObject, ResolvableObject, TypeMapping, ValueMapping};
 use crate::constants::{
-    ENTITY_ID_COLUMN, ENTITY_TABLE, EVENT_ID_COLUMN, ID_COLUMN, INTERNAL_ENTITY_ID_KEY,
+    ENTITY_ID_COLUMN, ENTITY_TABLE, ENTITY_TYPE_NAME, EVENT_ID_COLUMN, EVENT_MESSAGE_TABLE,
+    EVENT_MESSAGE_TYPE_NAME, ID_COLUMN, INTERNAL_ENTITY_ID_KEY,
 };
 use crate::mapping::ENTITY_TYPE_MAPPING;
 use crate::query::data::{count_rows, fetch_multiple_rows, fetch_single_row};
@@ -77,6 +78,7 @@ impl BasicObject for ModelDataObject {
         // root object requires entity_field association
         let mut root = objects.pop().unwrap();
         root = root.field(entity_field());
+        root = root.field(event_message_field());
 
         objects.push(root);
         objects
@@ -245,7 +247,7 @@ pub fn object(type_name: &str, type_mapping: &TypeMapping, path_array: Vec<Strin
 }
 
 fn entity_field() -> Field {
-    Field::new("entity", TypeRef::named("World__Entity"), |ctx| {
+    Field::new("entity", TypeRef::named(ENTITY_TYPE_NAME), |ctx| {
         FieldFuture::new(async move {
             match ctx.parent_value.try_to_value()? {
                 Value::Object(indexmap) => {
@@ -262,3 +264,23 @@ fn entity_field() -> Field {
         })
     })
 }
+
+fn event_message_field() -> Field {
+    Field::new("eventMessage", TypeRef::named(EVENT_MESSAGE_TYPE_NAME), |ctx| {
+        FieldFuture::new(async move {
+            match ctx.parent_value.try_to_value()? {
+                Value::Object(indexmap) => {
+                    let mut conn = ctx.data::<Pool<Sqlite>>()?.acquire().await?;
+                    let entity_id = utils::extract::<String>(indexmap, INTERNAL_ENTITY_ID_KEY)?;
+                    let data =
+                        fetch_single_row(&mut conn, EVENT_MESSAGE_TABLE, ID_COLUMN, &entity_id)
+                            .await?;
+                    let event_message = value_mapping_from_row(&data, &ENTITY_TYPE_MAPPING, false)?;
+
+                    Ok(Some(Value::Object(event_message)))
+                }
+                _ => Err("incorrect value, requires Value::Object".into()),
+            }
+        })
+    })
+}

crates/torii/graphql/src/query/mod.rs

@@ -10,7 +10,9 @@ use sqlx::sqlite::SqliteRow;
 use sqlx::{Row, SqliteConnection};
 use torii_core::sql::FELT_DELIMITER;
 
-use crate::constants::{BOOLEAN_TRUE, ENTITY_ID_COLUMN, INTERNAL_ENTITY_ID_KEY};
+use crate::constants::{
+    BOOLEAN_TRUE, ENTITY_ID_COLUMN, EVENT_MESSAGE_ID_COLUMN, INTERNAL_ENTITY_ID_KEY,
+};
 use crate::object::model_data::ModelMember;
 use crate::types::{TypeData, TypeMapping, ValueMapping};
 
@@ -195,6 +197,8 @@ pub fn value_mapping_from_row(
     // entity_id is not part of a model's type_mapping but needed to relate to parent entity
     if let Ok(entity_id) = row.try_get::<String, &str>(ENTITY_ID_COLUMN) {
         value_mapping.insert(Name::new(INTERNAL_ENTITY_ID_KEY), Value::from(entity_id));
+    } else if let Ok(event_message_id) = row.try_get::<String, &str>(EVENT_MESSAGE_ID_COLUMN) {
+        value_mapping.insert(Name::new(INTERNAL_ENTITY_ID_KEY), Value::from(event_message_id));
     }
 
     Ok(value_mapping)