Add sql transactions

src/ws/mod.rs

@@ -5,7 +5,7 @@ use emgauwa_common::errors::{DatabaseError, EmgauwaError};
 use emgauwa_common::models::{Controller, Relay};
 use emgauwa_common::types::{ControllerWsAction, ScheduleUid};
 use futures::{future, pin_mut, SinkExt, StreamExt};
-use sqlx::pool::PoolConnection;
+use sqlx::Transaction;
 use sqlx::{Pool, Sqlite};
 use tokio::time;
 use tokio_tungstenite::tungstenite::Message;
@@ -108,14 +108,7 @@ async fn handle_message(
 		match serde_json::from_str(&text) {
 			Ok(action) => {
 				log::debug!("Received action: {:?}", action);
-				let mut pool_conn = match pool.acquire().await {
-					Ok(conn) => conn,
-					Err(err) => {
-						log::error!("Failed to acquire database connection: {:?}", err);
-						return;
-					}
-				};
-				let action_res = handle_action(&mut pool_conn, app_state, action).await;
+				let action_res = handle_action(pool, app_state, action).await;
 				if let Err(e) = action_res {
 					log::error!("Error handling action: {:?}", e);
 				}
@@ -128,29 +121,31 @@ async fn handle_message(
 }
 
 pub async fn handle_action(
-	conn: &mut PoolConnection<Sqlite>,
+	pool: Pool<Sqlite>,
 	app_state: &Addr<AppState>,
 	action: ControllerWsAction,
 ) -> Result<(), EmgauwaError> {
 	let this = app_state_get_this(app_state).await?;
+	let mut tx = pool.begin().await?;
 
 	match action {
 		ControllerWsAction::Controller(controller) => {
-			handle_controller(conn, &this, controller).await?
+			handle_controller(&mut tx, &this, controller).await?
 		}
-		ControllerWsAction::Relays(relays) => handle_relays(conn, &this, relays).await?,
-		ControllerWsAction::Schedules(schedules) => handle_schedules(conn, schedules).await?,
+		ControllerWsAction::Relays(relays) => handle_relays(&mut tx, &this, relays).await?,
+		ControllerWsAction::Schedules(schedules) => handle_schedules(&mut tx, schedules).await?,
 		ControllerWsAction::RelayPulse((relay_num, duration)) => {
 			handle_relay_pulse(app_state, relay_num, duration).await?
 		}
 		_ => return Ok(()),
 	};
+	tx.commit().await?;
 
 	utils::app_state_reload(app_state).await
 }
 
 async fn handle_controller(
-	conn: &mut PoolConnection<Sqlite>,
+	tx: &mut Transaction<'_, Sqlite>,
 	this: &Controller,
 	controller: Controller,
 ) -> Result<(), EmgauwaError> {
@@ -159,17 +154,17 @@ async fn handle_controller(
 			"Controller UID mismatch during update",
 		)));
 	}
-	DbController::get_by_uid(conn, &controller.c.uid)
+	DbController::get_by_uid(tx, &controller.c.uid)
 		.await?
 		.ok_or(DatabaseError::NotFound)?
-		.update(conn, controller.c.name.as_str(), this.c.relay_count)
+		.update(tx, controller.c.name.as_str(), this.c.relay_count)
 		.await?;
 
 	Ok(())
 }
 
 async fn handle_schedules(
-	conn: &mut PoolConnection<Sqlite>,
+	tx: &mut Transaction<'_, Sqlite>,
 	schedules: Vec<DbSchedule>,
 ) -> Result<(), EmgauwaError> {
 	let mut handled_uids = vec![
@@ -184,15 +179,15 @@ async fn handle_schedules(
 		handled_uids.push(schedule.uid.clone());
 
 		log::debug!("Handling schedule: {:?}", schedule);
-		let schedule_db = DbSchedule::get_by_uid(conn, &schedule.uid).await?;
+		let schedule_db = DbSchedule::get_by_uid(tx, &schedule.uid).await?;
 
 		if let Some(schedule_db) = schedule_db {
 			schedule_db
-				.update(conn, schedule.name.as_str(), &schedule.periods)
+				.update(tx, schedule.name.as_str(), &schedule.periods)
 				.await?;
 		} else {
 			DbSchedule::create(
-				conn,
+				tx,
 				schedule.uid.clone(),
 				schedule.name.as_str(),
 				&schedule.periods,
@@ -205,7 +200,7 @@ async fn handle_schedules(
 }
 
 async fn handle_relays(
-	conn: &mut PoolConnection<Sqlite>,
+	tx: &mut Transaction<'_, Sqlite>,
 	this: &Controller,
 	relays: Vec<Relay>,
 ) -> Result<(), EmgauwaError> {
@@ -215,24 +210,24 @@ async fn handle_relays(
 				"Controller UID mismatch during relay update",
 			)));
 		}
-		let db_relay = DbRelay::get_by_controller_and_num(conn, &this.c, relay.r.number)
+		let db_relay = DbRelay::get_by_controller_and_num(tx, &this.c, relay.r.number)
 			.await?
 			.ok_or(DatabaseError::NotFound)?;
 
-		db_relay.update(conn, relay.r.name.as_str()).await?;
+		db_relay.update(tx, relay.r.name.as_str()).await?;
 
-		handle_schedules(conn, relay.schedules.clone()).await?;
+		handle_schedules(tx, relay.schedules.clone()).await?;
 
 		let mut schedules = Vec::new(); // We need to get the schedules from the database to have the right IDs
 		for schedule in relay.schedules {
 			schedules.push(
-				DbSchedule::get_by_uid(conn, &schedule.uid)
+				DbSchedule::get_by_uid(tx, &schedule.uid)
 					.await?
 					.ok_or(DatabaseError::NotFound)?,
 			);
 		}
 
-		DbJunctionRelaySchedule::set_schedules(conn, &db_relay, schedules.iter().collect()).await?;
+		DbJunctionRelaySchedule::set_schedules(tx, &db_relay, schedules.iter().collect()).await?;
 	}
 
 	Ok(())