Added transaction commits to multi sql functions in session_manager (#5693)
Signed-off-by: Vincent Huang <vhuang@squareup.com>
This commit is contained in:
@@ -970,6 +970,8 @@ impl SessionStorage {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn add_message(&self, session_id: &str, message: &Message) -> Result<()> {
|
async fn add_message(&self, session_id: &str, message: &Message) -> Result<()> {
|
||||||
|
let mut tx = self.pool.begin().await?;
|
||||||
|
|
||||||
let metadata_json = serde_json::to_string(&message.metadata)?;
|
let metadata_json = serde_json::to_string(&message.metadata)?;
|
||||||
|
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
@@ -983,14 +985,15 @@ impl SessionStorage {
|
|||||||
.bind(serde_json::to_string(&message.content)?)
|
.bind(serde_json::to_string(&message.content)?)
|
||||||
.bind(message.created)
|
.bind(message.created)
|
||||||
.bind(metadata_json)
|
.bind(metadata_json)
|
||||||
.execute(&self.pool)
|
.execute(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
sqlx::query("UPDATE sessions SET updated_at = datetime('now') WHERE id = ?")
|
sqlx::query("UPDATE sessions SET updated_at = datetime('now') WHERE id = ?")
|
||||||
.bind(session_id)
|
.bind(session_id)
|
||||||
.execute(&self.pool)
|
.execute(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
tx.commit().await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1049,10 +1052,12 @@ impl SessionStorage {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn delete_session(&self, session_id: &str) -> Result<()> {
|
async fn delete_session(&self, session_id: &str) -> Result<()> {
|
||||||
|
let mut tx = self.pool.begin().await?;
|
||||||
|
|
||||||
let exists =
|
let exists =
|
||||||
sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM sessions WHERE id = ?)")
|
sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM sessions WHERE id = ?)")
|
||||||
.bind(session_id)
|
.bind(session_id)
|
||||||
.fetch_one(&self.pool)
|
.fetch_one(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
if !exists {
|
if !exists {
|
||||||
@@ -1061,14 +1066,15 @@ impl SessionStorage {
|
|||||||
|
|
||||||
sqlx::query("DELETE FROM messages WHERE session_id = ?")
|
sqlx::query("DELETE FROM messages WHERE session_id = ?")
|
||||||
.bind(session_id)
|
.bind(session_id)
|
||||||
.execute(&self.pool)
|
.execute(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
sqlx::query("DELETE FROM sessions WHERE id = ?")
|
sqlx::query("DELETE FROM sessions WHERE id = ?")
|
||||||
.bind(session_id)
|
.bind(session_id)
|
||||||
.execute(&self.pool)
|
.execute(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
tx.commit().await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user