chore: bump up deps

Author: 0xgleb

Cargo.lock

@@ -34,15 +34,16 @@ redundant_pub_crate = "allow"
 
 [workspace.dependencies]
 async-trait = "0.1.89"
-chrono = { version = "0.4.44", features = ["serde"] }
+chrono = { version = "0.4.45", features = ["serde"] }
 cqrs-es = "0.4.12"
 serde = { version = "1.0.228", features = ["derive"] }
-serde_json = "1.0.149"
+serde_json = "1.0.150"
 sqlite-es = { path = "crates/sqlite-es" }
-sqlx = { version = "0.8.6", features = [
+sqlx = { version = "0.9.0", features = [
   "sqlite",
   "chrono",
-  "runtime-tokio-rustls",
+  "runtime-tokio",
+  "tls-rustls",
 ] }
 thiserror = "2.0.18"
 tokio = { version = "1.52.3", features = ["full"] }

Cargo.toml

@@ -94,6 +94,7 @@ use cqrs_es::EventStore;
 use cqrs_es::persist::PersistedEventStore;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
+use sqlx::AssertSqlSafe;
 use sqlx::SqlitePool;
 use std::fmt::{Debug, Display};
 use std::str::FromStr;
@@ -506,7 +507,7 @@ pub async fn vacuum(pool: &SqlitePool) -> Result<(), sqlx::Error> {
 ///
 /// Returns database errors from SQLite `PRAGMA incremental_vacuum`.
 pub async fn incremental_vacuum(pool: &SqlitePool, pages: u32) -> Result<(), sqlx::Error> {
-    sqlx::query(&format!("PRAGMA incremental_vacuum({pages})"))
+    sqlx::query(AssertSqlSafe(format!("PRAGMA incremental_vacuum({pages})")))
         .execute(pool)
         .await?;
     Ok(())

crates/event-sorcery/src/lib.rs

@@ -9,6 +9,7 @@ use async_trait::async_trait;
 use cqrs_es::Aggregate;
 use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
 use sqlite_es::SqliteViewRepository;
+use sqlx::AssertSqlSafe;
 use sqlx::SqlitePool;
 use sqlx::sqlite::Sqlite;
 use std::fmt::Debug;
@@ -162,7 +163,8 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
              ORDER BY view_id ASC"
         );
 
-        let rows: Vec<(String, String)> = sqlx::query_as(&query).fetch_all(pool).await?;
+        let rows: Vec<(String, String)> =
+            sqlx::query_as(AssertSqlSafe(query)).fetch_all(pool).await?;
 
         Ok(Self::parse_rows(rows))
     }
@@ -202,8 +204,10 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
              ORDER BY view_id ASC"
         );
 
-        let rows: Vec<(String, String)> =
-            sqlx::query_as(&query).bind(value).fetch_all(pool).await?;
+        let rows: Vec<(String, String)> = sqlx::query_as(AssertSqlSafe(query))
+            .bind(value)
+            .fetch_all(pool)
+            .await?;
 
         Ok(Self::parse_rows(rows))
     }
@@ -224,8 +228,9 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
         // Drive from events table (LEFT JOIN) so we also detect aggregates
         // with persisted events but no view row (crash before initial view write).
         // view_version is NULL when the view row is missing.
-        let stale_aggregates: Vec<(String, Option<i64>, i64)> = sqlx::query_as(&format!(
-            "SELECT e.aggregate_id, v.version, e.max_seq \
+        let stale_aggregates: Vec<(String, Option<i64>, i64)> =
+            sqlx::query_as(AssertSqlSafe(format!(
+                "SELECT e.aggregate_id, v.version, e.max_seq \
              FROM ( \
                  SELECT aggregate_id, MAX(sequence) as max_seq \
                  FROM events \
@@ -234,10 +239,10 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
              ) e \
              LEFT JOIN {table} v ON v.view_id = e.aggregate_id \
              WHERE v.version IS NULL OR e.max_seq > v.version"
-        ))
-        .bind(&aggregate_type)
-        .fetch_all(pool)
-        .await?;
+            )))
+            .bind(&aggregate_type)
+            .fetch_all(pool)
+            .await?;
 
         if stale_aggregates.is_empty() {
             debug!(target: "cqrs", %aggregate_type, "All views up to date, nothing to replay");
@@ -275,10 +280,12 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
 
         info!(target: "cqrs", %view_id, %table, "Rebuilding view from event history");
 
-        sqlx::query(&format!("DELETE FROM {table} WHERE view_id = ?1"))
-            .bind(&view_id)
-            .execute(pool)
-            .await?;
+        sqlx::query(AssertSqlSafe(format!(
+            "DELETE FROM {table} WHERE view_id = ?1"
+        )))
+        .bind(&view_id)
+        .execute(pool)
+        .await?;
 
         self.catch_up().await
     }
@@ -293,7 +300,7 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
 
         info!(target: "cqrs", %table, "Rebuilding all views from event history");
 
-        sqlx::query(&format!("DELETE FROM {table}"))
+        sqlx::query(AssertSqlSafe(format!("DELETE FROM {table}")))
             .execute(pool)
             .await?;
 
@@ -375,11 +382,11 @@ impl<Entity: EventSourced<Materialized = Table>> Projection<Entity> {
         // Write directly with version = max_seq, bypassing the view repo's
         // optimistic lock (which expects version + 1 increments). This is
         // safe because catch_up runs once at startup before the main loop.
-        sqlx::query(&format!(
+        sqlx::query(AssertSqlSafe(format!(
             "INSERT INTO {table} (view_id, version, payload) \
              VALUES (?1, ?2, ?3) \
              ON CONFLICT(view_id) DO UPDATE SET version = ?2, payload = ?3"
-        ))
+        )))
         .bind(view_id)
         .bind(max_seq)
         .bind(&payload)
@@ -557,10 +564,11 @@ async fn validate_column<Entity: EventSourced>(
 ) -> Result<(), ProjectionError<Entity>> {
     let column_name = column.0;
 
-    let columns: Vec<(String,)> =
-        sqlx::query_as(&format!("SELECT name FROM pragma_table_xinfo('{table}')"))
-            .fetch_all(pool)
-            .await?;
+    let columns: Vec<(String,)> = sqlx::query_as(AssertSqlSafe(format!(
+        "SELECT name FROM pragma_table_xinfo('{table}')"
+    )))
+    .fetch_all(pool)
+    .await?;
 
     if !columns.iter().any(|(name,)| name == column_name) {
         warn!(
@@ -575,15 +583,15 @@ async fn validate_column<Entity: EventSourced>(
         });
     }
 
-    let row_count: (i64,) = sqlx::query_as(&format!("SELECT COUNT(*) FROM {table}"))
+    let row_count: (i64,) = sqlx::query_as(AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}")))
         .fetch_one(pool)
         .await?;
 
     if row_count.0 > 0 {
-        let non_null_count: (i64,) = sqlx::query_as(&format!(
+        let non_null_count: (i64,) = sqlx::query_as(AssertSqlSafe(format!(
             "SELECT COUNT(*) FROM {table}
              WHERE {column_name} IS NOT NULL"
-        ))
+        )))
         .fetch_one(pool)
         .await?;
 

crates/event-sorcery/src/projection.rs

@@ -10,8 +10,12 @@ serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
-sqlx = { workspace = true, features = ["sqlite", "runtime-tokio-rustls"] }
-chrono = "0.4.44"
+sqlx = { workspace = true, features = [
+  "sqlite",
+  "runtime-tokio",
+  "tls-rustls",
+] }
+chrono = "0.4.45"
 tokio = { version = "1.52.3", features = ["rt", "sync"] }
 
 [lints]

crates/sqlite-es/Cargo.toml

@@ -5,7 +5,8 @@ use cqrs_es::persist::{
 };
 use serde_json::Value;
 use sqlx::sqlite::SqliteRow;
-use sqlx::{Pool, Row, Sqlite};
+use sqlx::{AssertSqlSafe, Pool, Row, Sqlite};
+use std::sync::Arc;
 
 use crate::sql_query::SqlQueryFactory;
 
@@ -148,7 +149,7 @@ impl SqliteEventRepository {
         aggregate_id: &str,
     ) -> Result<Vec<SerializedEvent>, SqliteAggregateError> {
         let query = self.query_factory.select_events();
-        let rows = sqlx::query(&query)
+        let rows = sqlx::query(AssertSqlSafe(query))
             .bind(A::aggregate_type())
             .bind(aggregate_id)
             .fetch_all(&self.pool)
@@ -165,7 +166,7 @@ impl SqliteEventRepository {
         let query = self.query_factory.get_last_events();
         let last_sequence_i64 = i64::try_from(last_sequence)?;
 
-        let rows = sqlx::query(&query)
+        let rows = sqlx::query(AssertSqlSafe(query))
             .bind(A::aggregate_type())
             .bind(aggregate_id)
             .bind(last_sequence_i64)
@@ -180,7 +181,7 @@ impl SqliteEventRepository {
         aggregate_id: &str,
     ) -> Result<Option<SerializedSnapshot>, SqliteAggregateError> {
         let query = self.query_factory.select_snapshot();
-        let row = sqlx::query(&query)
+        let row = sqlx::query(AssertSqlSafe(query))
             .bind(A::aggregate_type())
             .bind(aggregate_id)
             .fetch_optional(&self.pool)
@@ -193,14 +194,14 @@ impl SqliteEventRepository {
     }
 
     async fn insert_events(&self, events: &[SerializedEvent]) -> Result<(), SqliteAggregateError> {
-        let insert_query = self.query_factory.insert_event();
+        let insert_query: Arc<str> = self.query_factory.insert_event().into();
 
         let mut tx = self.pool.begin().await?;
 
         for event in events {
             let sequence_i64 = i64::try_from(event.sequence)?;
 
-            let result = sqlx::query(&insert_query)
+            let result = sqlx::query(AssertSqlSafe(Arc::clone(&insert_query)))
                 .bind(&event.aggregate_type)
                 .bind(&event.aggregate_id)
                 .bind(sequence_i64)
@@ -235,7 +236,7 @@ impl SqliteEventRepository {
 
         let timestamp = chrono::Utc::now().to_rfc3339();
 
-        sqlx::query(&query)
+        sqlx::query(AssertSqlSafe(query))
             .bind(A::aggregate_type())
             .bind(aggregate_id)
             .bind(last_sequence_i64)
@@ -261,14 +262,14 @@ impl SqliteEventRepository {
         tokio::spawn(async move {
             let rows = match &aggregate_id {
                 Some(id) => {
-                    sqlx::query(&query)
+                    sqlx::query(AssertSqlSafe(query))
                         .bind(&aggregate_type)
                         .bind(id)
                         .fetch_all(&pool)
                         .await
                 }
                 None => {
-                    sqlx::query(&query)
+                    sqlx::query(AssertSqlSafe(query))
                         .bind(&aggregate_type)
                         .fetch_all(&pool)
                         .await

crates/sqlite-es/src/event_repository.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
 use cqrs_es::{Aggregate, View};
-use sqlx::{Pool, Row, Sqlite};
+use sqlx::{AssertSqlSafe, Pool, Row, Sqlite};
 use std::marker::PhantomData;
 
 use crate::sql_query::SqlQueryFactory;
@@ -101,7 +101,7 @@ where
     async fn load(&self, view_id: &str) -> Result<Option<V>, PersistenceError> {
         let query = SqlQueryFactory::select_view(&self.view_table);
 
-        let row = sqlx::query(&query)
+        let row = sqlx::query(AssertSqlSafe(query))
             .bind(view_id)
             .fetch_optional(&self.pool)
             .await
@@ -127,7 +127,7 @@ where
     ) -> Result<Option<(V, ViewContext)>, PersistenceError> {
         let query = SqlQueryFactory::select_view(&self.view_table);
 
-        let row = sqlx::query(&query)
+        let row = sqlx::query(AssertSqlSafe(query))
             .bind(view_id)
             .fetch_optional(&self.pool)
             .await
@@ -169,7 +169,7 @@ where
         if context.version == 0 {
             let insert_query = SqlQueryFactory::insert_view(&self.view_table);
 
-            sqlx::query(&insert_query)
+            sqlx::query(AssertSqlSafe(insert_query))
                 .bind(&context.view_instance_id)
                 .bind(new_version)
                 .bind(&payload)
@@ -179,7 +179,7 @@ where
         } else {
             let update_query = SqlQueryFactory::update_view(&self.view_table);
 
-            let result = sqlx::query(&update_query)
+            let result = sqlx::query(AssertSqlSafe(update_query))
                 .bind(new_version)
                 .bind(&payload)
                 .bind(&context.view_instance_id)