fix: historical entities with mutli world support (#359)

* fix: historical entities with mutli world support

* fmt

Author: Larkooo

crates/sqlite/sqlite/src/error.rs

@@ -58,6 +58,8 @@ pub enum ParseError {
     InvalidTyEntity,
     #[error("Invalid FixedSizeArray format: {0}")]
     InvalidFixedSizeArray(String),
+    #[error("Invalid world-scoped ID format: {0}")]
+    InvalidWorldScopedId(String),
 }
 
 #[derive(Debug, thiserror::Error)]

crates/sqlite/sqlite/src/executor/mod.rs

@@ -585,31 +585,39 @@ impl<P: Provider + Sync + Send + Clone + 'static> Executor<'_, P> {
                 if entity.is_historical {
                     entity_counter += 1;
 
+                    let (world_address, entity_id) = entity
+                        .entity_id
+                        .split_once(':')
+                        .expect("Invalid world-scoped ID format");
                     let data = serde_json::to_string(&entity.ty.to_json_value()?)
                         .map_err(|e| ExecutorQueryError::Parse(ParseError::FromJsonStr(e)))?;
                     if let Some(keys) = entity.keys_str {
                         sqlx::query(
                             "INSERT INTO entities_historical (id, keys, event_id, data, model_id, \
-                             executed_at) VALUES (?, ?, ?, ?, ?, ?) RETURNING *",
+                             executed_at, world_address, entity_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING *",
                         )
                         .bind(entity.entity_id.clone())
                         .bind(keys)
                         .bind(entity.event_id.clone())
                         .bind(data)
                         .bind(entity.model_id.clone())
                         .bind(entity.block_timestamp.clone())
+                        .bind(world_address)
+                        .bind(entity_id)
                         .fetch_one(&mut **tx)
                         .await?;
                     } else {
                         sqlx::query(
                             "INSERT INTO entities_historical (id, event_id, data, model_id, \
-                             executed_at) VALUES (?, ?, ?, ?, ?) RETURNING *",
+                             executed_at, world_address, entity_id) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING *",
                         )
                         .bind(entity.entity_id.clone())
                         .bind(entity.event_id.clone())
                         .bind(data)
                         .bind(entity.model_id.clone())
                         .bind(entity.block_timestamp.clone())
+                        .bind(world_address)
+                        .bind(entity_id)
                         .fetch_one(&mut **tx)
                         .await?;
                     }
@@ -749,18 +757,24 @@ impl<P: Provider + Sync + Send + Clone + 'static> Executor<'_, P> {
                 if em_query.is_historical {
                     event_counter += 1;
 
+                    let (world_address, entity_id) = em_query
+                        .entity_id
+                        .split_once(':')
+                        .expect("Invalid world-scoped ID format");
                     let data = serde_json::to_string(&em_query.ty.to_json_value()?)
                         .map_err(|e| ExecutorQueryError::Parse(ParseError::FromJsonStr(e)))?;
                     sqlx::query(
                         "INSERT INTO event_messages_historical (id, keys, event_id, data, \
-                         model_id, executed_at) VALUES (?, ?, ?, ?, ?, ?) RETURNING *",
+                         model_id, executed_at, world_address, entity_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING *",
                     )
                     .bind(em_query.entity_id.clone())
                     .bind(em_query.keys_str.clone())
                     .bind(em_query.event_id.clone())
                     .bind(data)
                     .bind(em_query.model_id.clone())
                     .bind(em_query.block_timestamp.clone())
+                    .bind(world_address)
+                    .bind(entity_id)
                     .fetch_one(&mut **tx)
                     .await?;
                 }

crates/sqlite/sqlite/src/model.rs

@@ -687,7 +687,12 @@ impl Sql {
                 .map(|_| {
                     format!(
                         "{}.model_id = {}.world_address || ':' || ?",
-                        model_relation_table, table
+                        if historical {
+                            table
+                        } else {
+                            model_relation_table
+                        },
+                        table
                     )
                 })
                 .collect();
@@ -704,15 +709,8 @@ impl Sql {
         };
 
         let page = if historical {
-            self.fetch_historical_entities(
-                table,
-                model_relation_table,
-                &where_clause,
-                bind_values,
-                pagination,
-                &schemas,
-            )
-            .await?
+            self.fetch_historical_entities(table, &where_clause, bind_values, pagination, &schemas)
+                .await?
         } else {
             let page = self
                 .fetch_entities(
@@ -746,7 +744,6 @@ impl Sql {
     async fn fetch_historical_entities(
         &self,
         table: &str,
-        model_relation_table: &str,
         where_clause: &str,
         bind_values: Vec<String>,
         pagination: Pagination,
@@ -756,20 +753,16 @@ impl Sql {
 
         let mut query_builder = QueryBuilder::new(table)
             .select(&[
-                format!("{}.world_address", table),
-                format!("{}.id", table),
-                format!("{}.entity_id", table),
-                format!("{}.data", table),
-                format!("{}.model_id", table),
-                format!("{}.event_id", table),
-                format!("{}.created_at", table),
-                format!("{}.updated_at", table),
-                format!("{}.executed_at", table),
-                format!("group_concat({model_relation_table}.model_id) as model_ids"),
+                format!("{table}.world_address"),
+                format!("{table}.id"),
+                format!("{table}.entity_id"),
+                format!("{table}.data"),
+                format!("{table}.model_id"),
+                format!("{table}.event_id"),
+                format!("{table}.created_at"),
+                format!("{table}.updated_at"),
+                format!("{table}.executed_at"),
             ])
-            .join(&format!(
-                "JOIN {model_relation_table} ON {table}.id = {model_relation_table}.entity_id",
-            ))
             .group_by(&format!("{table}.event_id"));
 
         // Add user where clause if provided (applies to already-filtered set)