fix(grpc): require updated model for deletions

Author: ponderingdemocritus

crates/grpc/server/src/subscriptions/entity.rs

@@ -115,6 +115,10 @@ impl Service {
     // Process updates synchronously - no async overhead
     fn process_entity_update(subs: &EntityManager, entity: &EntityWithMetadata) {
         let mut closed_stream = Vec::new();
+        let updated_model = entity
+            .updated_model
+            .clone()
+            .or_else(|| entity.entity.models.first().map(|m| Ty::Struct(m.clone())));
 
         for sub in subs.subscribers.iter() {
             let idx = sub.key();
@@ -131,7 +135,7 @@ impl Service {
                 if !match_entity(
                     entity.entity.hashed_keys,
                     &entity.keys,
-                    &entity.entity.models.first().map(|m| Ty::Struct(m.clone())),
+                    &updated_model,
                     clause,
                 ) {
                     continue;

crates/grpc/server/src/subscriptions/event_message.rs

@@ -118,6 +118,10 @@ impl Service {
     // Process updates synchronously - no async overhead
     fn process_event_message(subs: &EventMessageManager, event: &EntityWithMetadata<true>) {
         let mut closed_stream = Vec::new();
+        let updated_model = event
+            .updated_model
+            .clone()
+            .or_else(|| event.entity.models.first().map(|m| Ty::Struct(m.clone())));
 
         for sub in subs.subscribers.iter() {
             let idx = sub.key();
@@ -134,7 +138,7 @@ impl Service {
                 if !match_entity(
                     event.entity.hashed_keys,
                     &event.keys,
-                    &event.entity.models.first().map(|m| Ty::Struct(m.clone())),
+                    &updated_model,
                     clause,
                 ) {
                     continue;

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

@@ -1402,6 +1402,66 @@ async fn test_historical_query(sequencer: &RunnerCtx) {
     );
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_entity_broker_delete_clause_matches_updated_model() {
+    use crate::subscriptions::entity::{EntityManager, Service};
+    use chrono::Utc;
+    use dojo_types::schema::{Struct, Ty};
+    use std::time::Duration;
+    use tokio::time::timeout;
+    use torii_broker::{types::EntityUpdate, MemoryBroker};
+    use torii_proto::schema::{Entity, EntityWithMetadata};
+
+    let config = GrpcConfig::default();
+    let entity_manager = Arc::new(EntityManager::new(config));
+
+    let service = Service::new(entity_manager.clone());
+    tokio::spawn(service);
+
+    let clause = Clause::Keys(KeysClause {
+        keys: vec![],
+        pattern_matching: PatternMatching::FixedLen,
+        models: vec!["ns-Model".to_string()],
+    });
+    let mut receiver = entity_manager.add_subscriber(Some(clause), vec![]).await;
+
+    // Skip the initial empty response
+    if let Ok(Some(response)) = timeout(Duration::from_secs(1), receiver.recv()).await {
+        match response {
+            Ok(resp) => assert!(resp.entity.is_none()),
+            Err(e) => panic!("Subscriber received error: {:?}", e),
+        }
+    }
+
+    let now = Utc::now();
+    let entity = Entity {
+        hashed_keys: Felt::from(1_u64),
+        world_address: Felt::ZERO,
+        models: vec![], // Simulate deletion payload where models are empty.
+        created_at: now,
+        updated_at: now,
+        executed_at: now,
+    };
+    let updated_model = Ty::Struct(Struct {
+        name: "ns-Model".to_string(),
+        children: vec![],
+    });
+    let entity_with_metadata = EntityWithMetadata {
+        entity,
+        event_id: "delete_event".to_string(),
+        keys: vec![],
+        updated_model: Some(updated_model),
+    };
+
+    MemoryBroker::publish(EntityUpdate::new(entity_with_metadata, false));
+
+    let response = timeout(Duration::from_secs(2), receiver.recv())
+        .await
+        .expect("timed out waiting for broker update");
+    let response = response.expect("receiver closed").expect("subscriber error");
+    assert!(response.entity.is_some());
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_entity_broker_multiple_subscriptions() {
     use crate::subscriptions::entity::{EntityManager, Service};
@@ -1453,6 +1513,7 @@ async fn test_entity_broker_multiple_subscriptions() {
             entity,
             event_id: format!("event_{}", update_id),
             keys,
+            updated_model: None,
         };
 
         // Publish the update to the broker
@@ -1721,6 +1782,7 @@ async fn test_entity_broker_stress_test() {
                 entity,
                 event_id: format!("stress_event_{}", update_id),
                 keys,
+                updated_model: None,
             };
 
             // Publish the update to the broker

crates/proto/src/schema.rs

@@ -13,6 +13,8 @@ pub struct EntityWithMetadata<const EVENT_MESSAGE: bool = false> {
     pub entity: Entity<EVENT_MESSAGE>,
     pub event_id: String,
     pub keys: Vec<Felt>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub updated_model: Option<Ty>,
 }
 
 #[derive(Debug, Default, Serialize, Deserialize, PartialEq, Hash, Eq, Clone)]

crates/sqlite/types/src/lib.rs

@@ -54,11 +54,18 @@ pub struct Entity {
 
 impl<const EVENT_MESSAGE: bool> From<Entity> for torii_proto::schema::Entity<EVENT_MESSAGE> {
     fn from(value: Entity) -> Self {
-        let models = if value.deleted {
-            vec![]
-        } else {
-            vec![value.updated_model.unwrap().as_struct().unwrap().clone()]
-        };
+        // Always include model info so clause matching works for deletions too.
+        // The `deleted` flag indicates deletion status - no need to empty models.
+        let updated_model = value.updated_model.as_ref().unwrap_or_else(|| {
+            if value.deleted {
+                panic!("deleted entity missing updated_model: {}", value.id);
+            }
+            panic!("entity missing updated_model: {}", value.id);
+        });
+        let model_struct = updated_model.as_struct().unwrap_or_else(|| {
+            panic!("entity updated_model is not a struct: {}", value.id);
+        });
+        let models = vec![model_struct.clone()];
 
         // Use the dedicated entity_id column (no parsing needed!)
         Self {
@@ -74,6 +81,8 @@ impl<const EVENT_MESSAGE: bool> From<Entity> for torii_proto::schema::Entity<EVE
 
 impl<const EVENT_MESSAGE: bool> From<Entity> for EntityWithMetadata<EVENT_MESSAGE> {
     fn from(value: Entity) -> Self {
+        let updated_model = value.updated_model.clone();
+        let event_id = value.event_id.clone();
         let keys = value
             .keys
             .split('/')
@@ -86,9 +95,10 @@ impl<const EVENT_MESSAGE: bool> From<Entity> for EntityWithMetadata<EVENT_MESSAG
             })
             .collect();
         Self {
-            event_id: value.event_id.clone(),
+            event_id,
             entity: value.into(),
             keys,
+            updated_model,
         }
     }
 }
@@ -204,6 +214,58 @@ impl From<Event> for torii_proto::Event {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use dojo_types::schema::{Struct, Ty};
+
+    fn sample_entity(updated_model: Option<Ty>, deleted: bool) -> Entity {
+        let now = Utc::now();
+        Entity {
+            id: "0x1:0x2".to_string(),
+            entity_id: "0x2".to_string(),
+            world_address: "0x1".to_string(),
+            keys: "0x1/0x2".to_string(),
+            event_id: "event_1".to_string(),
+            executed_at: now,
+            created_at: now,
+            updated_at: now,
+            updated_model,
+            deleted,
+        }
+    }
+
+    #[test]
+    #[should_panic(expected = "deleted entity missing updated_model")]
+    fn deleted_entity_requires_updated_model() {
+        let entity = sample_entity(None, true);
+        let _: torii_proto::schema::Entity = entity.into();
+    }
+
+    #[test]
+    fn deleted_entity_includes_model_info() {
+        let model = Struct {
+            name: "ns-Model".to_string(),
+            children: vec![],
+        };
+        let entity = sample_entity(Some(Ty::Struct(model.clone())), true);
+
+        let proto_entity: torii_proto::schema::Entity = entity.clone().into();
+        assert_eq!(proto_entity.models.len(), 1);
+        assert_eq!(proto_entity.models[0].name, model.name);
+
+        let entity_with_metadata: EntityWithMetadata = entity.into();
+        let updated_model = entity_with_metadata
+            .updated_model
+            .expect("updated_model missing");
+        match updated_model {
+            Ty::Struct(struct_ty) => assert_eq!(struct_ty.name, "ns-Model"),
+            _ => panic!("expected updated_model to be a struct"),
+        }
+    }
+}
+
 #[derive(FromRow, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct Token {