use actix::Addr; use sqlx::{Pool, Sqlite}; use sqlx::pool::PoolConnection; use tokio_tungstenite::tungstenite; use tokio_tungstenite::tungstenite::Message; use emgauwa_common::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule}; use emgauwa_common::errors::{DatabaseError, EmgauwaError}; use emgauwa_common::models::{Controller, Relay}; use emgauwa_common::types::{ControllerWsAction, ScheduleUid}; use crate::app_state::AppState; use crate::utils; use crate::utils::app_state_get_this; pub async fn handle_message( pool: Pool<Sqlite>, app_state: &Addr<AppState>, message_result: Result<Message, tungstenite::Error>, ) { let msg = match message_result { Ok(msg) => msg, Err(err) => { log::error!("Error reading message: {}", err); return; } }; if let Message::Text(text) = msg { 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; if let Err(e) = action_res { log::error!("Error handling action: {:?}", e); } } Err(e) => { log::error!("Error deserializing action: {:?}", e); } } } } pub async fn handle_action( conn: &mut PoolConnection<Sqlite>, app_state: &Addr<AppState>, action: ControllerWsAction, ) -> Result<(), EmgauwaError> { let this = app_state_get_this(app_state).await?; match action { ControllerWsAction::Controller(controller) => { handle_controller(conn, &this, controller).await? } ControllerWsAction::Relays(relays) => handle_relays(conn, app_state, &this, relays).await?, ControllerWsAction::Schedules(schedules) => handle_schedules(conn, schedules).await?, ControllerWsAction::RelayPulse((relay_num, duration)) => { handle_relay_pulse(app_state, relay_num, duration).await? } _ => return Ok(()), }; utils::app_state_reload(app_state).await } async fn handle_controller( conn: &mut PoolConnection<Sqlite>, this: &Controller, controller: Controller, ) -> Result<(), EmgauwaError> { if controller.c.uid != this.c.uid { return Err(EmgauwaError::Other(String::from( "Controller UID mismatch during update", ))); } DbController::get_by_uid(conn, &controller.c.uid) .await? .ok_or(DatabaseError::NotFound)? .update(conn, controller.c.name.as_str(), this.c.relay_count) .await?; Ok(()) } async fn handle_schedules( conn: &mut PoolConnection<Sqlite>, schedules: Vec<DbSchedule>, ) -> Result<(), EmgauwaError> { let mut handled_uids = vec![ // on and off schedules are always present and should not be updated ScheduleUid::On, ScheduleUid::Off, ]; for schedule in schedules { if handled_uids.contains(&schedule.uid) { continue; } handled_uids.push(schedule.uid.clone()); log::debug!("Handling schedule: {:?}", schedule); let schedule_db = DbSchedule::get_by_uid(conn, &schedule.uid).await?; if let Some(schedule_db) = schedule_db { schedule_db .update(conn, schedule.name.as_str(), &schedule.periods) .await?; } else { DbSchedule::create( conn, schedule.uid.clone(), schedule.name.as_str(), &schedule.periods, ) .await?; } } Ok(()) } async fn handle_relays( conn: &mut PoolConnection<Sqlite>, app_state: &Addr<AppState>, this: &Controller, relays: Vec<Relay>, ) -> Result<(), EmgauwaError> { for relay in relays { if relay.controller.uid != this.c.uid { return Err(EmgauwaError::Other(String::from( "Controller UID mismatch during relay update", ))); } let db_relay = DbRelay::get_by_controller_and_num(conn, &this.c, relay.r.number) .await? .ok_or(DatabaseError::NotFound)?; db_relay.update(conn, relay.r.name.as_str()).await?; handle_schedules(conn, 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) .await? .ok_or(DatabaseError::NotFound)?, ); } utils::app_state_relay_override_schedule( app_state, relay.r.number, relay.override_schedule.clone(), relay.override_schedule_weekday, ) .await?; DbJunctionRelaySchedule::set_schedules(conn, &db_relay, schedules.iter().collect()).await?; } Ok(()) } async fn handle_relay_pulse( app_state: &Addr<AppState>, relay_num: i64, duration: Option<u32>, ) -> Result<(), EmgauwaError> { utils::app_state_relay_pulse(app_state, relay_num, duration).await }