Split project (keep controller)

src/app_state.rs

@@ -0,0 +1,173 @@
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use actix::{Actor, Context, Handler, Message};
+use emgauwa_common::constants;
+use emgauwa_common::drivers::RelayDriver;
+use emgauwa_common::errors::EmgauwaError;
+use emgauwa_common::models::Controller;
+use emgauwa_common::types::RelayStates;
+use futures::executor::block_on;
+use sqlx::{Pool, Sqlite};
+use tokio::sync::Notify;
+
+use crate::settings::Settings;
+
+#[derive(Message)]
+#[rtype(result = "Result<(), EmgauwaError>")]
+pub struct Reload {}
+
+#[derive(Message)]
+#[rtype(result = "()")]
+pub struct UpdateRelayStates {
+	pub relay_states: RelayStates,
+}
+
+#[derive(Message)]
+#[rtype(result = "Result<(), EmgauwaError>")]
+pub struct RelayPulse {
+	pub relay_number: i64,
+	pub duration: Option<u32>,
+}
+
+#[derive(Message)]
+#[rtype(result = "Controller")]
+pub struct GetThis {}
+
+#[derive(Message)]
+#[rtype(result = "Arc<Notify>")]
+pub struct GetControllerNotifier {}
+
+#[derive(Message)]
+#[rtype(result = "Arc<Notify>")]
+pub struct GetRelayNotifier {}
+
+pub struct AppState {
+	pub pool: Pool<Sqlite>,
+	pub this: Controller,
+	pub settings: Settings,
+	pub drivers: Vec<Box<dyn RelayDriver>>,
+	pub controller_notifier: Arc<Notify>,
+	pub relay_notifier: Arc<Notify>,
+}
+
+impl AppState {
+	pub fn new(
+		pool: Pool<Sqlite>,
+		this: Controller,
+		settings: Settings,
+		drivers: Vec<Box<dyn RelayDriver>>,
+	) -> AppState {
+		AppState {
+			pool,
+			this,
+			settings,
+			drivers,
+			controller_notifier: Arc::new(Notify::new()),
+			relay_notifier: Arc::new(Notify::new()),
+		}
+	}
+
+	pub fn notify_controller_change(&self) {
+		self.controller_notifier.notify_one();
+	}
+
+	pub fn notify_relay_change(&self) {
+		self.relay_notifier.notify_one();
+	}
+}
+
+impl Actor for AppState {
+	type Context = Context<Self>;
+}
+
+impl Handler<Reload> for AppState {
+	type Result = Result<(), EmgauwaError>;
+
+	fn handle(&mut self, _msg: Reload, _ctx: &mut Self::Context) -> Self::Result {
+		log::debug!("Reloading controller");
+		let mut pool_conn = block_on(self.pool.acquire())?;
+
+		self.this.reload(&mut pool_conn)?;
+
+		self.notify_controller_change();
+
+		Ok(())
+	}
+}
+
+impl Handler<UpdateRelayStates> for AppState {
+	type Result = ();
+
+	fn handle(&mut self, msg: UpdateRelayStates, _ctx: &mut Self::Context) -> Self::Result {
+		self.this.apply_relay_states(&msg.relay_states);
+		self.drivers
+			.iter_mut()
+			.zip(msg.relay_states.iter())
+			.for_each(|(driver, state)| {
+				if let Err(e) = driver.set(state.unwrap_or(false)) {
+					log::error!("Error setting relay: {}", e);
+				}
+			});
+
+		self.notify_relay_change();
+	}
+}
+
+impl Handler<RelayPulse> for AppState {
+	type Result = Result<(), EmgauwaError>;
+
+	fn handle(&mut self, msg: RelayPulse, _ctx: &mut Self::Context) -> Self::Result {
+		let relay_num = msg.relay_number;
+
+		let duration = Duration::from_secs(
+			match msg.duration {
+				None => {
+					self.settings
+						.get_relay(relay_num)
+						.ok_or(EmgauwaError::Other(String::from(
+							"Relay not found in settings",
+						)))?
+						.pulse
+				}
+				Some(dur) => Some(dur as u64),
+			}
+			.unwrap_or(constants::RELAY_PULSE_DURATION),
+		);
+		let now = Instant::now();
+		let until = now + duration;
+
+		self.this.relay_pulse(relay_num, until)?;
+		log::debug!(
+			"Pulsing relay {} for {} seconds until {:?}",
+			relay_num,
+			duration.as_secs(),
+			until
+		);
+		Ok(())
+	}
+}
+
+impl Handler<GetThis> for AppState {
+	type Result = Controller;
+
+	fn handle(&mut self, _msg: GetThis, _ctx: &mut Self::Context) -> Self::Result {
+		self.this.clone()
+	}
+}
+
+impl Handler<GetControllerNotifier> for AppState {
+	type Result = Arc<Notify>;
+
+	fn handle(&mut self, _msg: GetControllerNotifier, _ctx: &mut Self::Context) -> Self::Result {
+		Arc::clone(&self.controller_notifier)
+	}
+}
+
+impl Handler<GetRelayNotifier> for AppState {
+	type Result = Arc<Notify>;
+
+	fn handle(&mut self, _msg: GetRelayNotifier, _ctx: &mut Self::Context) -> Self::Result {
+		Arc::clone(&self.relay_notifier)
+	}
+}

src/driver.rs

@@ -0,0 +1,22 @@
+use serde::{Deserialize, Deserializer};
+
+#[derive(Debug, Clone, Copy)]
+pub enum Driver {
+	Null,
+	Gpio,
+	PiFace,
+}
+
+impl<'de> Deserialize<'de> for Driver {
+	fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+	where
+		D: Deserializer<'de>,
+	{
+		match String::deserialize(deserializer)?.as_str() {
+			"null" => Ok(Driver::Null),
+			"gpio" => Ok(Driver::Gpio),
+			"piface" => Ok(Driver::PiFace),
+			_ => Err(serde::de::Error::custom("invalid driver")),
+		}
+	}
+}

src/main.rs

@@ -0,0 +1,124 @@
+use actix::Actor;
+use emgauwa_common::db;
+use emgauwa_common::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule};
+use emgauwa_common::errors::EmgauwaError;
+use emgauwa_common::models::{Controller, FromDbModel};
+use emgauwa_common::types::EmgauwaUid;
+use emgauwa_common::utils::{drop_privileges, init_logging};
+use rppal_pfd::PiFaceDigital;
+use sqlx::pool::PoolConnection;
+use sqlx::Sqlite;
+
+use crate::relay_loop::run_relays_loop;
+use crate::settings::Settings;
+use crate::ws::run_ws_loop;
+
+mod app_state;
+mod driver;
+mod relay_loop;
+mod settings;
+mod utils;
+mod ws;
+
+async fn create_this_controller(
+	conn: &mut PoolConnection<Sqlite>,
+	settings: &Settings,
+) -> Result<DbController, EmgauwaError> {
+	DbController::create(
+		conn,
+		&EmgauwaUid::default(),
+		&settings.name,
+		settings.relays.len() as i64,
+	)
+	.await
+	.map_err(EmgauwaError::from)
+}
+
+async fn create_this_relay(
+	conn: &mut PoolConnection<Sqlite>,
+	this_controller: &DbController,
+	settings_relay: &settings::Relay,
+) -> Result<DbRelay, EmgauwaError> {
+	let relay = DbRelay::create(
+		conn,
+		&settings_relay.name,
+		settings_relay.number.ok_or(EmgauwaError::Internal(
+			"Relay number is missing".to_string(),
+		))?,
+		this_controller,
+	)
+	.await?;
+
+	let off = DbSchedule::get_off(conn).await?;
+	for weekday in 0..7 {
+		DbJunctionRelaySchedule::set_schedule(conn, &relay, &off, weekday).await?;
+	}
+
+	Ok(relay)
+}
+
+#[actix::main]
+async fn main() -> Result<(), std::io::Error> {
+	let settings = settings::init()?;
+
+	drop_privileges(&settings.permissions)?;
+
+	init_logging(&settings.logging.level)?;
+
+	let mut pfd: Option<PiFaceDigital> = None;
+	let drivers = settings.relays_make_drivers(&mut pfd)?;
+
+	let pool = db::init(&settings.database)
+		.await
+		.map_err(EmgauwaError::from)?;
+
+	let mut conn = pool.acquire().await.map_err(EmgauwaError::from)?;
+
+	let db_controller = match DbController::get_all(&mut conn)
+		.await
+		.map_err(EmgauwaError::from)?
+		.pop()
+	{
+		None => futures::executor::block_on(create_this_controller(&mut conn, &settings))?,
+		Some(c) => c,
+	};
+
+	for relay in &settings.relays {
+		if DbRelay::get_by_controller_and_num(
+			&mut conn,
+			&db_controller,
+			relay.number.ok_or(EmgauwaError::Internal(
+				"Relay number is missing".to_string(),
+			))?,
+		)
+		.await
+		.map_err(EmgauwaError::from)?
+		.is_none()
+		{
+			create_this_relay(&mut conn, &db_controller, relay)
+				.await
+				.map_err(EmgauwaError::from)?;
+		}
+	}
+
+	let db_controller = db_controller
+		.update(&mut conn, &db_controller.name, settings.relays.len() as i64)
+		.await
+		.map_err(EmgauwaError::from)?;
+
+	let this = Controller::from_db_model(&mut conn, db_controller).map_err(EmgauwaError::from)?;
+
+	let url = format!(
+		"ws://{}:{}/api/v1/ws/controllers",
+		settings.server.host, settings.server.port
+	);
+
+	let app_state = app_state::AppState::new(pool.clone(), this, settings, drivers).start();
+
+	let _ = tokio::join!(
+		tokio::spawn(run_relays_loop(app_state.clone())),
+		tokio::spawn(run_ws_loop(pool.clone(), app_state.clone(), url)),
+	);
+
+	Ok(())
+}

src/relay_loop.rs

@@ -0,0 +1,150 @@
+use std::time::{Duration, Instant};
+
+use actix::Addr;
+use chrono::{Local, Timelike};
+use emgauwa_common::constants::RELAYS_RETRY_TIMEOUT;
+use emgauwa_common::errors::EmgauwaError;
+use emgauwa_common::models::Controller;
+use emgauwa_common::types::{RelayStates, Weekday};
+use emgauwa_common::utils::printable_relay_states;
+use futures::pin_mut;
+use tokio::time;
+use tokio::time::timeout;
+use utils::app_state_get_controller_notifier;
+
+use crate::app_state::AppState;
+use crate::utils;
+
+pub async fn run_relays_loop(app_state: Addr<AppState>) {
+	log::debug!("Spawned relays loop");
+	loop {
+		let run_result = run_relays(&app_state).await;
+		if let Err(err) = run_result {
+			log::error!("Error running relays: {}", err);
+		}
+		time::sleep(RELAYS_RETRY_TIMEOUT).await;
+	}
+}
+
+async fn run_relays(app_state: &Addr<AppState>) -> Result<(), EmgauwaError> {
+	let notifier = &*app_state_get_controller_notifier(app_state).await?;
+
+	let mut last_weekday = emgauwa_common::utils::get_weekday();
+	let mut this = utils::app_state_get_this(app_state).await?;
+	let mut relay_states: RelayStates = Vec::new();
+	init_relay_states(&mut relay_states, &this);
+	calc_relay_states(&mut relay_states, &mut this, app_state).await?;
+
+	let mut duration_override = None;
+
+	loop {
+		log::debug!(
+			"Relay loop at {}: {}",
+			Local::now().naive_local().time(),
+			printable_relay_states(&this.get_relay_states())
+		);
+
+		let notifier_future = notifier.notified();
+		pin_mut!(notifier_future);
+		let mut changed = timeout(
+			get_next_duration(&this, &mut duration_override),
+			&mut notifier_future,
+		)
+		.await
+		.is_ok();
+
+		check_weekday(app_state, &mut last_weekday, &mut changed).await?;
+
+		if changed {
+			log::debug!("Reloading controller in relay loop");
+			this = utils::app_state_get_this(app_state).await?;
+		}
+
+		let now_pulse = Instant::now();
+		duration_override = this
+			.relays
+			.iter_mut()
+			.filter_map(|relay| match relay.check_pulsing(&now_pulse) {
+				None => None,
+				Some(pulse) => {
+					let dur = pulse - now_pulse;
+					log::debug!(
+						"Pulsing relay {} for {}s until {:?} ",
+						relay.r.number,
+						dur.as_secs(),
+						pulse
+					);
+					Some(dur)
+				}
+			})
+			.min();
+
+		calc_relay_states(&mut relay_states, &mut this, app_state).await?;
+	}
+}
+
+fn init_relay_states(relay_states: &mut RelayStates, this: &Controller) {
+	relay_states.clear();
+	for _ in 0..this.c.relay_count {
+		relay_states.push(None);
+	}
+}
+
+async fn calc_relay_states(
+	relay_states: &mut RelayStates,
+	this: &mut Controller,
+	app_state: &Addr<AppState>,
+) -> Result<(), EmgauwaError> {
+	let now = Local::now().time();
+	let now_pulse = Instant::now();
+
+	this.relays
+		.iter_mut()
+		.zip(relay_states.iter_mut())
+		.for_each(|(relay, state)| {
+			relay.is_on = Some(
+				relay.active_schedule.is_on(&now) || relay.check_pulsing(&now_pulse).is_some(),
+			);
+			*state = relay.is_on;
+		});
+	utils::app_state_update_relays_on(app_state, relay_states.clone()).await
+}
+
+fn get_next_duration(this: &Controller, duration_override: &mut Option<Duration>) -> Duration {
+	if let Some(duration) = duration_override {
+		log::debug!("Duration override. Waiting for {}s", duration.as_secs());
+		return *duration;
+	}
+
+	let now = Local::now().time();
+	let now_in_s = now.num_seconds_from_midnight();
+	let next_timestamp = this
+		.get_next_time(&now)
+		.map_or(86400, |t| t.num_seconds_from_midnight());
+
+	let duration_to_next = Duration::from_secs((next_timestamp - now_in_s) as u64);
+
+	log::debug!(
+		"Next timestamp: {}; Waiting for {}s",
+		next_timestamp,
+		duration_to_next.as_secs()
+	);
+
+	duration_to_next
+}
+
+async fn check_weekday(
+	app_state: &Addr<AppState>,
+	last_weekday: &mut Weekday,
+	changed: &mut bool,
+) -> Result<(), EmgauwaError> {
+	let current_weekday = emgauwa_common::utils::get_weekday();
+	if current_weekday.ne(last_weekday) {
+		log::debug!("Weekday changed");
+		*last_weekday = current_weekday;
+		utils::app_state_reload(app_state).await?;
+		*changed = true;
+	}
+
+	Ok(())
+}

src/settings.rs

@@ -0,0 +1,106 @@
+use emgauwa_common::errors::EmgauwaError;
+use emgauwa_common::{drivers, settings};
+use rppal_pfd::PiFaceDigital;
+use serde_derive::Deserialize;
+
+use crate::driver::Driver;
+
+#[derive(Clone, Debug, Deserialize)]
+#[serde(default)]
+#[allow(unused)]
+pub struct Relay {
+	pub driver: Driver,
+	pub name: String,
+	pub number: Option<i64>,
+	pub pin: u8,
+	pub inverted: bool,
+	pub pulse: Option<u64>,
+}
+
+#[derive(Clone, Debug, Deserialize)]
+#[serde(default)]
+#[allow(unused)]
+pub struct Settings {
+	pub server: settings::Server,
+	pub database: String,
+	pub permissions: settings::Permissions,
+	pub logging: settings::Logging,
+
+	pub name: String,
+	pub relays: Vec<Relay>,
+}
+
+impl Default for Settings {
+	fn default() -> Self {
+		Settings {
+			server: settings::Server::default(),
+			database: String::from("sqlite://emgauwa-controller.sqlite"),
+			permissions: settings::Permissions::default(),
+			logging: settings::Logging::default(),
+
+			name: String::from("Emgauwa Controller"),
+			relays: Vec::new(),
+		}
+	}
+}
+
+impl Default for Relay {
+	fn default() -> Self {
+		Relay {
+			driver: Driver::Gpio,
+			number: None,
+			name: String::from("Relay"),
+			pin: 0,
+			inverted: false,
+			pulse: None,
+		}
+	}
+}
+
+pub fn init() -> Result<Settings, EmgauwaError> {
+	let mut settings: Settings = settings::load("controller", "CONTROLLER")?;
+
+	for (num, relay) in settings.relays.iter_mut().enumerate() {
+		if relay.number.is_none() {
+			relay.number = Some(num as i64);
+		}
+	}
+
+	Ok(settings)
+}
+
+impl Settings {
+	pub fn get_relay(&self, number: i64) -> Option<&Relay> {
+		self.relays.iter().find(|r| r.number == Some(number))
+	}
+
+	pub fn relays_make_drivers(
+		&self,
+		pfd: &mut Option<PiFaceDigital>,
+	) -> Result<Vec<Box<dyn drivers::RelayDriver>>, EmgauwaError> {
+		let mut drivers = Vec::new();
+		for relay in &self.relays {
+			drivers.push(relay.make_driver(pfd)?);
+		}
+		Ok(drivers)
+	}
+}
+
+impl Relay {
+	pub fn make_driver(
+		&self,
+		pfd: &mut Option<PiFaceDigital>,
+	) -> Result<Box<dyn drivers::RelayDriver>, EmgauwaError> {
+		let driver: Box<dyn drivers::RelayDriver> = match self.driver {
+			Driver::Null => Box::new(drivers::NullDriver::new(self.pin)),
+			Driver::Gpio => Box::new(drivers::GpioDriver::new(self.pin, self.inverted)?),
+			Driver::PiFace => {
+				if pfd.is_none() {
+					*pfd = Some(drivers::PiFaceDriver::init_piface()?);
+				}
+				Box::new(drivers::PiFaceDriver::new(self.pin, pfd)?)
+			}
+		};
+		Ok(driver)
+	}
+}

src/utils.rs

@@ -0,0 +1,66 @@
+use std::sync::Arc;
+
+use actix::Addr;
+use emgauwa_common::errors::EmgauwaError;
+use emgauwa_common::models::Controller;
+use emgauwa_common::types::RelayStates;
+use tokio::sync::Notify;
+
+use crate::app_state;
+use crate::app_state::AppState;
+
+pub async fn app_state_get_this(app_state: &Addr<AppState>) -> Result<Controller, EmgauwaError> {
+	app_state
+		.send(app_state::GetThis {})
+		.await
+		.map_err(EmgauwaError::from)
+}
+
+pub async fn app_state_get_relay_notifier(
+	app_state: &Addr<AppState>,
+) -> Result<Arc<Notify>, EmgauwaError> {
+	app_state
+		.send(app_state::GetRelayNotifier {})
+		.await
+		.map_err(EmgauwaError::from)
+}
+
+pub async fn app_state_get_controller_notifier(
+	app_state: &Addr<AppState>,
+) -> Result<Arc<Notify>, EmgauwaError> {
+	app_state
+		.send(app_state::GetControllerNotifier {})
+		.await
+		.map_err(EmgauwaError::from)
+}
+
+pub async fn app_state_reload(app_state: &Addr<AppState>) -> Result<(), EmgauwaError> {
+	app_state
+		.send(app_state::Reload {})
+		.await
+		.map_err(EmgauwaError::from)?
+}
+
+pub async fn app_state_update_relays_on(
+	app_state: &Addr<AppState>,
+	relay_states: RelayStates,
+) -> Result<(), EmgauwaError> {
+	app_state
+		.send(app_state::UpdateRelayStates { relay_states })
+		.await
+		.map_err(EmgauwaError::from)
+}
+
+pub async fn app_state_relay_pulse(
+	app_state: &Addr<AppState>,
+	relay_number: i64,
+	duration: Option<u32>,
+) -> Result<(), EmgauwaError> {
+	app_state
+		.send(app_state::RelayPulse {
+			relay_number,
+			duration,
+		})
+		.await
+		.map_err(EmgauwaError::from)?
+}

src/ws/mod.rs

@@ -0,0 +1,247 @@
+use actix::Addr;
+use emgauwa_common::constants::WEBSOCKET_RETRY_TIMEOUT;
+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 futures::{future, pin_mut, SinkExt, StreamExt};
+use sqlx::pool::PoolConnection;
+use sqlx::{Pool, Sqlite};
+use tokio::time;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::{connect_async, tungstenite};
+
+use crate::app_state::AppState;
+use crate::utils;
+use crate::utils::{app_state_get_relay_notifier, app_state_get_this};
+
+pub async fn run_ws_loop(pool: Pool<Sqlite>, app_state: Addr<AppState>, url: String) {
+	log::debug!("Spawned ws loop");
+	loop {
+		let run_result = run_websocket(pool.clone(), &app_state, &url).await;
+		if let Err(err) = run_result {
+			log::error!("Error running websocket: {}", err);
+		}
+
+		log::info!(
+			"Retrying to connect in {} seconds...",
+			WEBSOCKET_RETRY_TIMEOUT.as_secs()
+		);
+		time::sleep(WEBSOCKET_RETRY_TIMEOUT).await;
+	}
+}
+
+async fn run_websocket(
+	pool: Pool<Sqlite>,
+	app_state: &Addr<AppState>,
+	url: &str,
+) -> Result<(), EmgauwaError> {
+	log::debug!("Trying to connect to {}", url);
+	match connect_async(url).await {
+		Ok(connection) => {
+			log::info!("Websocket connected");
+			let (ws_stream, _) = connection;
+
+			let (mut write, read) = ws_stream.split();
+
+			let ws_action = ControllerWsAction::Register(app_state_get_this(app_state).await?);
+
+			let ws_action_json = serde_json::to_string(&ws_action)?;
+			if let Err(err) = write.send(Message::text(ws_action_json)).await {
+				log::error!("Failed to register at websocket: {}", err);
+				return Ok(());
+			}
+
+			let (app_state_tx, app_state_rx) = futures_channel::mpsc::unbounded::<Message>();
+			tokio::spawn(read_app_state(app_state.clone(), app_state_tx));
+			let app_state_to_ws = app_state_rx.map(Ok).forward(write);
+
+			let read_handler = read.for_each(|msg| handle_message(pool.clone(), app_state, msg));
+
+			pin_mut!(app_state_to_ws, read_handler);
+			future::select(app_state_to_ws, read_handler).await;
+
+			log::warn!("Lost connection to websocket");
+		}
+		Err(err) => {
+			log::warn!("Failed to connect to websocket: {}", err,);
+		}
+	}
+	Ok(())
+}
+
+async fn read_app_state(
+	app_state: Addr<AppState>,
+	tx: futures_channel::mpsc::UnboundedSender<Message>,
+) -> Result<(), EmgauwaError> {
+	let notifier = &*app_state_get_relay_notifier(&app_state).await?;
+	loop {
+		notifier.notified().await;
+		log::debug!("Relay change detected");
+		let this = app_state_get_this(&app_state).await?;
+		let relay_states = this.get_relay_states();
+		let ws_action = ControllerWsAction::RelayStates((this.c.uid, relay_states));
+
+		let ws_action_json = serde_json::to_string(&ws_action)?;
+		tx.unbounded_send(Message::text(ws_action_json))
+			.map_err(|_| {
+				EmgauwaError::Other(String::from(
+					"Failed to forward message from app state to websocket",
+				))
+			})?;
+	}
+}
+
+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, &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>,
+	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)?,
+			);
+		}
+
+		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
+}