refactor(torii-core): enqueue models & events #2471
Conversation
WalkthroughOhayo, sensei! This pull request introduces changes to the Changes
Possibly related PRs
Suggested reviewers
Recent review detailsConfiguration used: .coderabbit.yaml Files selected for processing (1)
Files skipped from review as they are similar to previous changes (1)
TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
Documentation and Community
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Files selected for processing (2)
- crates/torii/core/src/query_queue.rs (2 hunks)
- crates/torii/core/src/sql.rs (4 hunks)
Additional comments not posted (6)
crates/torii/core/src/query_queue.rs (2)
52-54: Ohayo, sensei! The newQueryTypevariants look great.Adding
EventMessage(Ty),RegisterModel, andStoreEventto theQueryTypeenum appropriately extends the functionality of the query handling system.
150-158: Verify thatEventMessageUpdatedreceives the correctupdated_model.Ohayo, sensei! In the
QueryType::EventMessage(entity)variant,event_message.updated_modelis set using the providedentity. Please verify thatentitycontains the correct data and thatupdated_modelis being used as intended in subsequent processing.You can confirm this by checking where
EventMessageUpdatedis published and howupdated_modelis utilized:crates/torii/core/src/sql.rs (4)
18-19: Ohayo, sensei! The new imports enhance the functionality.The addition of
Argument,DeleteEntityQuery,QueryQueue, andQueryTypefromquery_queue, as well asutc_dt_string_from_timestampfromutils, is appropriate for the refactoring efforts.
174-184: Ohayo, sensei! The arguments vector is correctly constructed for theinsert_modelsquery.The
argumentsvector now includes all necessary parameters, ensuring proper insertion into the database. The use ofutc_dt_string_from_timestamp(block_timestamp)correctly formats the timestamp.
315-324: Ohayo, sensei! Theinsert_entitiesquery is accurately enqueued with the right arguments.Enqueuing the
insert_entitiesquery with the provided arguments ensures that event messages are stored correctly. The inclusion ofQueryType::Otheris appropriate here.
502-502: Ohayo, sensei! The use ofQueryType::StoreEventenhances event handling.Assigning
QueryType::StoreEventto the event storage query improves the clarity and management of different query types within the queue.
crates/torii/core/src/query_queue.rs
Outdated
| QueryType::RegisterModel => { | ||
| let row = query.fetch_one(&mut *tx).await.with_context(|| { | ||
| format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) | ||
| })?; | ||
| let model_registered = ModelRegistered::from_row(&row)?; | ||
| self.push_publish(BrokerMessage::ModelRegistered(model_registered)); | ||
| } | ||
| QueryType::EventMessage(entity) => { | ||
| let row = query.fetch_one(&mut *tx).await.with_context(|| { | ||
| format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) | ||
| })?; | ||
| let mut event_message = EventMessageUpdated::from_row(&row)?; | ||
| event_message.updated_model = Some(entity); | ||
| let broker_message = BrokerMessage::EventMessageUpdated(event_message); | ||
| self.push_publish(broker_message); | ||
| } | ||
| QueryType::StoreEvent => { | ||
| let row = query.fetch_one(&mut *tx).await.with_context(|| { | ||
| format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) | ||
| })?; | ||
| let event = EventEmitted::from_row(&row)?; | ||
| self.push_publish(BrokerMessage::EventEmitted(event)); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider refactoring to reduce code duplication in query execution and error handling.
Sensei, the QueryType::RegisterModel, QueryType::EventMessage, and QueryType::StoreEvent variants share similar logic for executing a query, handling errors, fetching a row, and publishing a broker message. Refactoring this repeated code into a helper function would enhance maintainability and reduce redundancy.
Here's how you might implement the refactor:
-
Create a helper function for executing queries and publishing messages:
async fn execute_query_and_publish<T, F>( &mut self, query: sqlx::Query<'_, sqlx::Sqlite, sqlx::SqliteArguments<'_>>, tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, statement: &str, arguments: &Vec<Argument>, map_row: F, ) -> Result<()> where T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow> + Send, F: Fn(T) -> BrokerMessage, { let row = query.fetch_one(tx).await.with_context(|| { format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) })?; let item = T::from_row(&row)?; self.push_publish(map_row(item)); Ok(()) }
-
Use the helper function in each variant:
QueryType::RegisterModel => { self.execute_query_and_publish( query, &mut tx, &statement, &arguments, |model_registered| BrokerMessage::ModelRegistered(model_registered), ).await?; } QueryType::EventMessage(entity) => { self.execute_query_and_publish( query, &mut tx, &statement, &arguments, move |mut event_message| { event_message.updated_model = Some(entity.clone()); BrokerMessage::EventMessageUpdated(event_message) }, ).await?; } QueryType::StoreEvent => { self.execute_query_and_publish( query, &mut tx, &statement, &arguments, |event| BrokerMessage::EventEmitted(event), ).await?; }
This refactor reduces duplication and makes the code more modular and easier to maintain.
crates/torii/core/src/query_queue.rs
Outdated
| QueryType::StoreEvent => { | ||
| let row = query.fetch_one(&mut *tx).await.with_context(|| { | ||
| format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) | ||
| })?; | ||
| let event = EventEmitted::from_row(&row)?; | ||
| self.push_publish(BrokerMessage::EventEmitted(event)); | ||
| } |
There was a problem hiding this comment.
Ensure proper handling when no rows are returned in StoreEvent.
Sensei, in the QueryType::StoreEvent variant, if fetch_one does not find a matching event, it will result in an error. Consider using fetch_optional and handling the case where no event is found to prevent potential runtime errors.
Here is a suggested change to handle this scenario:
let row = query.fetch_one(&mut *tx).await.with_context(|| {
format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
})?;
let event = EventEmitted::from_row(&row)?;
self.push_publish(BrokerMessage::EventEmitted(event));Apply this diff:
-let row = query.fetch_one(&mut *tx).await.with_context(|| {
+let row = query.fetch_optional(&mut *tx).await.with_context(|| {
format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
})?;
-let event = EventEmitted::from_row(&row)?;
-self.push_publish(BrokerMessage::EventEmitted(event));
+if let Some(row) = row {
+ let event = EventEmitted::from_row(&row)?;
+ self.push_publish(BrokerMessage::EventEmitted(event));
+} else {
+ // Optionally handle the case where no event is found
+ // For example, log a warning or take alternative action
+}This change ensures that your code gracefully handles situations where the query returns no results.
Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| QueryType::StoreEvent => { | |
| let row = query.fetch_one(&mut *tx).await.with_context(|| { | |
| format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) | |
| })?; | |
| let event = EventEmitted::from_row(&row)?; | |
| self.push_publish(BrokerMessage::EventEmitted(event)); | |
| } | |
| QueryType::StoreEvent => { | |
| let row = query.fetch_optional(&mut *tx).await.with_context(|| { | |
| format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) | |
| })?; | |
| if let Some(row) = row { | |
| let event = EventEmitted::from_row(&row)?; | |
| self.push_publish(BrokerMessage::EventEmitted(event)); | |
| } else { | |
| // Optionally handle the case where no event is found | |
| // For example, log a warning or take alternative action | |
| } | |
| } |
There was a problem hiding this comment.
Actionable comments posted: 0
Outside diff range and nitpick comments (2)
crates/torii/core/src/query_queue.rs (2)
49-51: LGTM! Consider adding documentation for new variants.Ohayo, sensei! The new variants in the
QueryTypeenum look good and expand the functionality of the query system. To improve clarity and maintainability, consider adding documentation comments for these new variants to explain their purpose and usage.Here's a suggestion for documenting the new variants:
pub enum QueryType { // ... existing variants ... /// Represents an event message query with associated type information. EventMessage(Ty), /// Represents a query to register a new model. RegisterModel, /// Represents a query to store an event. StoreEvent, // ... other variants ... }
138-160: LGTM! Consider refactoring to reduce duplication.Ohayo, sensei! The new cases for
RegisterModel,EventMessage, andStoreEventare well-implemented and consistent with the new enum variants. The error handling and broker message creation look good.However, there's some code duplication in the error handling and row fetching logic. Consider refactoring this common pattern into a helper function to improve maintainability and reduce the chance of errors.
Here's a suggestion for a helper function:
async fn fetch_and_publish<T, F>( query: sqlx::Query<'_, sqlx::Sqlite>, tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, statement: &str, arguments: &[Argument], publish_queue: &mut VecDeque<BrokerMessage>, transform: F, ) -> Result<()> where T: for<'r> FromRow<'r, sqlx::sqlite::SqliteRow>, F: FnOnce(T) -> BrokerMessage, { let row = query.fetch_one(&mut *tx).await.with_context(|| { format!("Failed to execute query: {:?}, args: {:?}", statement, arguments) })?; let item = T::from_row(&row)?; publish_queue.push_back(transform(item)); Ok(()) }You could then use this helper function in each case:
QueryType::RegisterModel => { fetch_and_publish::<ModelRegistered, _>( query, &mut tx, &statement, &arguments, &mut publish_queue, BrokerMessage::ModelRegistered, ).await?; }This refactoring would make the code more concise and easier to maintain.
Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Files selected for processing (2)
- crates/torii/core/src/query_queue.rs (5 hunks)
- crates/torii/core/src/sql.rs (4 hunks)
Files skipped from review as they are similar to previous changes (1)
- crates/torii/core/src/sql.rs
Additional comments not posted (4)
crates/torii/core/src/query_queue.rs (4)
57-57: LGTM! Simplified struct definition.Ohayo, sensei! The removal of the
publish_queuefield from theQueryQueuestruct simplifies its definition. This change looks good, but it's important to ensure that theexecute_allmethod has been updated accordingly to handle publish operations without relying on this field.To verify that the publish operations are still handled correctly, please check the
execute_allmethod implementation.
71-73: LGTM! Local publish queue introduced.Ohayo, sensei! The introduction of a local
publish_queuevariable is a good approach to handle publish operations after removing the field from the struct. This change maintains the functionality while improving the design by localizing the queue to the method where it's used.
97-97: LGTM! Consistent use of local publish queue.Ohayo, sensei! The direct push to the local
publish_queueis consistent with the earlier changes. This modification maintains the functionality while simplifying the code by removing the need for a separatepush_publishmethod.
172-174: LGTM! Consistent use of local publish queue.Ohayo, sensei! The publish loop at the end of the
execute_allmethod has been updated to use the localpublish_queue, which is consistent with the earlier changes. This modification maintains the original functionality while adapting to the new structure of the code.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #2471 +/- ##
==========================================
- Coverage 68.45% 68.44% -0.02%
==========================================
Files 368 368
Lines 48162 48185 +23
==========================================
+ Hits 32971 32980 +9
- Misses 15191 15205 +14 ☔ View full report in Codecov by Sentry. |
| Argument::String(event_id.to_string()), | ||
| Argument::String(utc_dt_string_from_timestamp(block_timestamp)), | ||
| ], | ||
| QueryType::Other, |
There was a problem hiding this comment.
think this should be QueryType::EntityMessage
Summary by CodeRabbit
New Features
EventMessage,RegisterModel, andStoreEvent, enhancing the query handling capabilities.Refactor