feat(torii-core): add entities deletions to susbcription broker (#2106)

* feat(torii-core): add entities deletions to susbcription broker

* fmt

* chore: return empty entities when count 0

crates/torii/core/src/sql.rs

@@ -171,14 +171,16 @@ impl Sql {
                                ?, ?, ?) ON CONFLICT(id) DO UPDATE SET \
                                executed_at=EXCLUDED.executed_at, event_id=EXCLUDED.event_id \
                                RETURNING *";
-        let entity_updated: EntityUpdated = sqlx::query_as(insert_entities)
+        let mut entity_updated: EntityUpdated = sqlx::query_as(insert_entities)
             .bind(&entity_id)
             .bind(&keys_str)
             .bind(event_id)
             .bind(utc_dt_string_from_timestamp(block_timestamp))
             .fetch_one(&self.pool)
             .await?;
 
+        entity_updated.updated_model = Some(entity.clone());
+
         let path = vec![entity.name()];
         self.build_set_entity_queries_recursive(
             path,
@@ -253,8 +255,18 @@ impl Sql {
     pub async fn delete_entity(&mut self, keys: Vec<FieldElement>, entity: Ty) -> Result<()> {
         let entity_id = format!("{:#x}", poseidon_hash_many(&keys));
         let path = vec![entity.name()];
+        // delete entity models data
         self.build_delete_entity_queries_recursive(path, &entity_id, &entity);
         self.query_queue.execute_all().await?;
+
+        // delete entity
+        let entity_deleted =
+            sqlx::query_as::<_, EntityUpdated>("DELETE FROM entities WHERE id = ? RETURNING *")
+                .bind(entity_id)
+                .fetch_one(&self.pool)
+                .await?;
+
+        SimpleBroker::publish(entity_deleted);
         Ok(())
     }
 

crates/torii/core/src/types.rs

@@ -1,6 +1,7 @@
 use core::fmt;
 
 use chrono::{DateTime, Utc};
+use dojo_types::schema::Ty;
 use serde::{Deserialize, Serialize};
 use sqlx::FromRow;
 use starknet::core::types::FieldElement;
@@ -37,6 +38,9 @@ pub struct Entity {
     pub executed_at: DateTime<Utc>,
     pub created_at: DateTime<Utc>,
     pub updated_at: DateTime<Utc>,
+    // if updated_model is None, then the entity has been deleted
+    #[sqlx(skip)]
+    pub updated_model: Option<Ty>,
 }
 
 #[derive(FromRow, Deserialize, Debug, Clone)]

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

@@ -239,6 +239,10 @@ impl DojoWorld {
         // total count of rows without limit and offset
         let total_count: u32 = sqlx::query_scalar(&count_query).fetch_one(&self.pool).await?;
 
+        if total_count == 0 {
+            return Ok((Vec::new(), 0));
+        }
+
         // query to filter with limit and offset
         let query = format!(
             r#"
@@ -335,6 +339,10 @@ impl DojoWorld {
         let total_count =
             sqlx::query_scalar(&count_query).bind(&keys_pattern).fetch_one(&self.pool).await?;
 
+        if total_count == 0 {
+            return Ok((Vec::new(), 0));
+        }
+
         let models_query = format!(
             r#"
             SELECT group_concat({model_relation_table}.model_id) as model_ids