Add fmt options and move handlers

emgauwa-core/Cargo.toml

@@ -21,5 +21,7 @@ serde = "1.0"
 serde_json = "1.0"
 serde_derive = "1.0"
 
+sqlx = { version = "0.7", features = ["sqlite", "runtime-tokio", "macros", "chrono"] }
+
 futures = "0.3.29"
 libc = "0.2"

emgauwa-core/src/handlers/errors.rs

@@ -0,0 +1,83 @@
+use std::fmt::{Display, Formatter};
+
+use actix_web::http::StatusCode;
+use actix_web::HttpResponse;
+use emgauwa_lib::db::errors::DatabaseError;
+use serde::ser::SerializeStruct;
+use serde::{Serialize, Serializer};
+
+#[derive(Debug)]
+pub enum ApiError {
+	BadUid,
+	ProtectedSchedule,
+	DatabaseError(DatabaseError),
+	InternalError(String),
+}
+
+impl ApiError {
+	fn get_code(&self) -> StatusCode {
+		match self {
+			ApiError::BadUid => StatusCode::BAD_REQUEST,
+			ApiError::ProtectedSchedule => StatusCode::FORBIDDEN,
+			ApiError::DatabaseError(db_error) => db_error.get_code(),
+			ApiError::InternalError(_) => StatusCode::INTERNAL_SERVER_ERROR,
+		}
+	}
+}
+
+impl Serialize for ApiError {
+	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+	where
+		S: Serializer,
+	{
+		let mut s = serializer.serialize_struct("error", 2)?;
+		s.serialize_field("code", &self.get_code().as_u16())?;
+		s.serialize_field("description", &String::from(self))?;
+		s.end()
+	}
+}
+
+impl From<&ApiError> for String {
+	fn from(err: &ApiError) -> Self {
+		match err {
+			ApiError::BadUid => String::from("the uid is in a bad format"),
+			ApiError::ProtectedSchedule => String::from("the targeted schedule is protected"),
+			ApiError::DatabaseError(db_err) => String::from(db_err),
+			ApiError::InternalError(msg) => msg.clone(),
+		}
+	}
+}
+
+impl From<&ApiError> for HttpResponse {
+	fn from(err: &ApiError) -> Self {
+		HttpResponse::build(err.get_code()).json(err)
+	}
+}
+
+impl Display for ApiError {
+	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+		write!(f, "{}: {}", self.get_code(), String::from(self))
+	}
+}
+
+impl actix_web::error::ResponseError for ApiError {
+	fn status_code(&self) -> StatusCode {
+		self.get_code()
+	}
+
+	fn error_response(&self) -> HttpResponse {
+		HttpResponse::from(self)
+	}
+}
+
+impl From<sqlx::Error> for ApiError {
+	fn from(err: sqlx::Error) -> Self {
+		ApiError::DatabaseError(DatabaseError::from(err))
+	}
+}
+
+impl From<DatabaseError> for ApiError {
+	fn from(err: DatabaseError) -> Self {
+		ApiError::DatabaseError(err)
+	}
+}

emgauwa-core/src/handlers/mod.rs

@@ -0,0 +1,38 @@
+use actix_web::{error, Error, HttpRequest, HttpResponse};
+use serde::ser::SerializeStruct;
+use serde::{Serialize, Serializer};
+
+pub(crate) mod errors;
+pub mod v1;
+
+enum EmgauwaJsonPayLoadError {
+	Error(error::JsonPayloadError),
+}
+
+impl Serialize for EmgauwaJsonPayLoadError {
+	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+	where
+		S: Serializer,
+	{
+		let mut s = serializer.serialize_struct("error", 3)?;
+		s.serialize_field("type", "json-payload-error")?;
+		s.serialize_field("code", &400)?;
+		s.serialize_field(
+			"description",
+			&match self {
+				EmgauwaJsonPayLoadError::Error(err) => format!("{}", err),
+			},
+		)?;
+		s.end()
+	}
+}
+
+pub fn json_error_handler(err: error::JsonPayloadError, _: &HttpRequest) -> Error {
+	error::InternalError::from_response(
+		"",
+		HttpResponse::BadRequest()
+			.content_type("application/json")
+			.json(EmgauwaJsonPayLoadError::Error(err)),
+	)
+	.into()
+}

emgauwa-core/src/handlers/v1/controllers.rs

@@ -0,0 +1,17 @@
+use actix_web::{get, web, HttpResponse};
+use emgauwa_lib::db::DbController;
+use emgauwa_lib::models::{convert_db_list, Controller};
+use sqlx::{Pool, Sqlite};
+
+use crate::handlers::errors::ApiError;
+
+#[get("/api/v1/controllers")]
+pub async fn index(pool: web::Data<Pool<Sqlite>>) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let db_controllers = DbController::get_all(&mut pool_conn).await?;
+
+	let controllers: Vec<Controller> = convert_db_list(&mut pool_conn, db_controllers)?;
+
+	Ok(HttpResponse::Ok().json(controllers))
+}

emgauwa-core/src/handlers/v1/mod.rs

@@ -0,0 +1,4 @@
+pub mod controllers;
+pub mod relays;
+pub mod schedules;
+pub mod ws;

emgauwa-core/src/handlers/v1/relays.rs

@@ -0,0 +1,24 @@
+use actix_web::{get, web, HttpResponse};
+use emgauwa_lib::db::DbRelay;
+use emgauwa_lib::models::{convert_db_list, Relay};
+use serde::{Deserialize, Serialize};
+use sqlx::{Pool, Sqlite};
+
+use crate::handlers::errors::ApiError;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RequestRelay {
+	name: String,
+	tags: Vec<String>,
+}
+
+#[get("/api/v1/relays")]
+pub async fn index(pool: web::Data<Pool<Sqlite>>) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let db_relays = DbRelay::get_all(&mut pool_conn).await?;
+
+	let relays: Vec<Relay> = convert_db_list(&mut pool_conn, db_relays)?;
+
+	Ok(HttpResponse::Ok().json(relays))
+}

emgauwa-core/src/handlers/v1/schedules.rs

@@ -0,0 +1,160 @@
+use actix_web::{delete, get, post, put, web, HttpResponse};
+use emgauwa_lib::db::errors::DatabaseError;
+use emgauwa_lib::db::{DbPeriods, DbSchedule, DbTag};
+use emgauwa_lib::models::{convert_db_list, FromDbModel, Schedule};
+use emgauwa_lib::types::ScheduleUid;
+use serde::{Deserialize, Serialize};
+use sqlx::pool::PoolConnection;
+use sqlx::{Pool, Sqlite};
+
+use crate::handlers::errors::ApiError;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RequestSchedule {
+	name: String,
+	periods: DbPeriods,
+	tags: Vec<String>,
+}
+
+#[get("/api/v1/schedules")]
+pub async fn index(pool: web::Data<Pool<Sqlite>>) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let db_schedules = DbSchedule::get_all(&mut pool_conn).await?;
+
+	let schedules: Vec<Schedule> = convert_db_list(&mut pool_conn, db_schedules)?;
+
+	Ok(HttpResponse::Ok().json(schedules))
+}
+
+#[get("/api/v1/schedules/tag/{tag}")]
+pub async fn tagged(
+	pool: web::Data<Pool<Sqlite>>,
+	path: web::Path<(String,)>,
+) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let (tag,) = path.into_inner();
+	let tag_db = DbTag::get_by_tag(&mut pool_conn, &tag)
+		.await?
+		.ok_or(DatabaseError::NotFound)?;
+
+	let db_schedules = DbSchedule::get_by_tag(&mut pool_conn, &tag_db).await?;
+
+	let schedules: Vec<Schedule> = convert_db_list(&mut pool_conn, db_schedules)?;
+
+	Ok(HttpResponse::Ok().json(schedules))
+}
+
+#[get("/api/v1/schedules/{schedule_id}")]
+pub async fn show(
+	pool: web::Data<Pool<Sqlite>>,
+	path: web::Path<(String,)>,
+) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let (schedule_uid,) = path.into_inner();
+	let uid = ScheduleUid::try_from(schedule_uid.as_str()).or(Err(ApiError::BadUid))?;
+
+	let schedule = DbSchedule::get_by_uid(&mut pool_conn, &uid)
+		.await?
+		.ok_or(DatabaseError::NotFound)?;
+
+	let return_schedule = Schedule::from_db_model(&mut pool_conn, schedule);
+	Ok(HttpResponse::Ok().json(return_schedule))
+}
+
+#[post("/api/v1/schedules")]
+pub async fn add(
+	pool: web::Data<Pool<Sqlite>>,
+	data: web::Json<RequestSchedule>,
+) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let new_schedule = DbSchedule::create(&mut pool_conn, &data.name, &data.periods).await?;
+
+	new_schedule
+		.set_tags(&mut pool_conn, data.tags.as_slice())
+		.await?;
+
+	let return_schedule = Schedule::from_db_model(&mut pool_conn, new_schedule);
+	Ok(HttpResponse::Created().json(return_schedule))
+}
+
+async fn add_list_single(
+	conn: &mut PoolConnection<Sqlite>,
+	request_schedule: &RequestSchedule,
+) -> Result<DbSchedule, DatabaseError> {
+	let new_schedule =
+		DbSchedule::create(conn, &request_schedule.name, &request_schedule.periods).await?;
+
+	new_schedule
+		.set_tags(conn, request_schedule.tags.as_slice())
+		.await?;
+
+	Ok(new_schedule)
+}
+
+#[post("/api/v1/schedules/list")]
+pub async fn add_list(
+	pool: web::Data<Pool<Sqlite>>,
+	data: web::Json<Vec<RequestSchedule>>,
+) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let mut db_schedules: Vec<DbSchedule> = Vec::new();
+	for s in data.iter() {
+		let new_s = futures::executor::block_on(add_list_single(&mut pool_conn, s))?;
+		db_schedules.push(new_s);
+	}
+
+	let schedules: Vec<Schedule> = convert_db_list(&mut pool_conn, db_schedules)?;
+	Ok(HttpResponse::Created().json(schedules))
+}
+
+#[put("/api/v1/schedules/{schedule_id}")]
+pub async fn update(
+	pool: web::Data<Pool<Sqlite>>,
+	path: web::Path<(String,)>,
+	data: web::Json<RequestSchedule>,
+) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let (schedule_uid,) = path.into_inner();
+	let uid = ScheduleUid::try_from(schedule_uid.as_str()).or(Err(ApiError::BadUid))?;
+
+	let schedule = DbSchedule::get_by_uid(&mut pool_conn, &uid)
+		.await?
+		.ok_or(DatabaseError::NotFound)?;
+
+	let schedule = schedule
+		.update(&mut pool_conn, data.name.as_str(), &data.periods)
+		.await?;
+
+	schedule
+		.set_tags(&mut pool_conn, data.tags.as_slice())
+		.await?;
+
+	let return_schedule = Schedule::from_db_model(&mut pool_conn, schedule);
+	Ok(HttpResponse::Ok().json(return_schedule))
+}
+
+#[delete("/api/v1/schedules/{schedule_id}")]
+pub async fn delete(
+	pool: web::Data<Pool<Sqlite>>,
+	path: web::Path<(String,)>,
+) -> Result<HttpResponse, ApiError> {
+	let mut pool_conn = pool.acquire().await?;
+
+	let (schedule_uid,) = path.into_inner();
+	let uid = ScheduleUid::try_from(schedule_uid.as_str()).or(Err(ApiError::BadUid))?;
+
+	match uid {
+		ScheduleUid::Off => Err(ApiError::ProtectedSchedule),
+		ScheduleUid::On => Err(ApiError::ProtectedSchedule),
+		ScheduleUid::Any(_) => {
+			DbSchedule::delete_by_uid(&mut pool_conn, uid).await?;
+			Ok(HttpResponse::Ok().json("schedule got deleted"))
+		}
+	}
+}

emgauwa-core/src/handlers/v1/ws/controllers.rs

@@ -0,0 +1,141 @@
+use std::time::Instant;
+
+use actix::{Actor, ActorContext, AsyncContext, StreamHandler};
+use actix_web_actors::ws;
+use actix_web_actors::ws::ProtocolError;
+use emgauwa_lib::constants::{HEARTBEAT_INTERVAL, HEARTBEAT_TIMEOUT};
+use emgauwa_lib::db::errors::DatabaseError;
+use emgauwa_lib::db::{DbController, DbRelay};
+use emgauwa_lib::models::{Controller, FromDbModel};
+use emgauwa_lib::types::{ConnectedControllersType, ControllerUid, ControllerWsAction};
+use sqlx::pool::PoolConnection;
+use sqlx::{Pool, Sqlite};
+use ws::Message;
+
+pub struct ControllerWs {
+	pub pool: Pool<Sqlite>,
+	pub controller_uid: Option<ControllerUid>,
+	pub connected_controllers: ConnectedControllersType,
+	pub hb: Instant,
+}
+
+impl Actor for ControllerWs {
+	type Context = ws::WebsocketContext<Self>;
+
+	fn started(&mut self, ctx: &mut Self::Context) {
+		self.hb(ctx);
+	}
+
+	fn stopped(&mut self, _ctx: &mut Self::Context) {
+		if let Some(controller_uid) = &self.controller_uid {
+			let mut pool_conn = futures::executor::block_on(self.pool.acquire()).unwrap();
+
+			let mut data = self.connected_controllers.lock().unwrap();
+			if let Some(controller) = data.remove(controller_uid) {
+				futures::executor::block_on(controller.c.update_active(&mut pool_conn, false))
+					.unwrap();
+			}
+		}
+	}
+}
+
+impl ControllerWs {
+	pub fn handle_action(
+		&mut self,
+		conn: &mut PoolConnection<Sqlite>,
+		action: ControllerWsAction,
+	) -> Result<(), DatabaseError> {
+		match action {
+			ControllerWsAction::Register(controller) => {
+				log::info!("Registering controller: {:?}", controller);
+				let c = &controller.c;
+				let controller_db = futures::executor::block_on(
+					DbController::get_by_uid_or_create(conn, &c.uid, &c.name, c.relay_count),
+				)?;
+				futures::executor::block_on(controller_db.update_active(conn, true))?;
+
+				for relay in &controller.relays {
+					let r = &relay.r;
+					futures::executor::block_on(DbRelay::get_by_controller_and_num_or_create(
+						conn,
+						&controller_db,
+						r.number,
+						&r.name,
+					))?;
+				}
+
+				let controller = Controller::from_db_model(conn, controller_db)?;
+
+				let controller_uid = &controller.c.uid;
+				self.controller_uid = Some(controller_uid.clone());
+
+				let mut data = self.connected_controllers.lock().unwrap();
+				data.insert(controller_uid.clone(), controller);
+
+				Ok(())
+			}
+		}
+	}
+
+	/// helper method that sends ping to client every 5 seconds (HEARTBEAT_INTERVAL).
+	///
+	/// also this method checks heartbeats from client
+	fn hb(&self, ctx: &mut ws::WebsocketContext<Self>) {
+		ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
+			// check client heartbeats
+			if Instant::now().duration_since(act.hb) > HEARTBEAT_TIMEOUT {
+				log::warn!("Websocket Controller heartbeat failed, disconnecting!");
+				ctx.stop();
+				// don't try to send a ping
+				return;
+			}
+
+			ctx.ping(&[]);
+		});
+	}
+}
+
+impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs {
+	fn handle(&mut self, msg: Result<Message, ProtocolError>, ctx: &mut Self::Context) {
+		let mut pool_conn = futures::executor::block_on(self.pool.acquire()).unwrap();
+
+		let msg = match msg {
+			Err(_) => {
+				ctx.stop();
+				return;
+			}
+			Ok(msg) => msg,
+		};
+
+		match msg {
+			Message::Ping(msg) => {
+				self.hb = Instant::now();
+				ctx.pong(&msg)
+			}
+			Message::Pong(_) => {
+				self.hb = Instant::now();
+			}
+			Message::Text(text) => {
+				let action: ControllerWsAction = serde_json::from_str(&text).unwrap();
+				let action_res = self.handle_action(&mut pool_conn, action);
+				if let Err(e) = action_res {
+					log::error!("Error handling action: {:?}", e);
+					ctx.text(serde_json::to_string(&e).unwrap());
+				}
+			}
+			Message::Binary(_) => log::warn!("Received unexpected binary in controller ws"),
+			Message::Close(reason) => {
+				ctx.close(reason);
+				ctx.stop();
+			}
+			Message::Continuation(_) => {
+				ctx.stop();
+			}
+			Message::Nop => (),
+		}
+
+		//let schedules = futures::executor::block_on(DbSchedule::get_all(&mut pool_conn)).unwrap();
+		//let schedules_json = serde_json::to_string(&schedules).unwrap();
+		//ctx.text(schedules_json);
+	}
+}

emgauwa-core/src/handlers/v1/ws/mod.rs

@@ -0,0 +1,32 @@
+use std::time::Instant;
+
+use actix_web::{get, web, HttpRequest, HttpResponse};
+use actix_web_actors::ws;
+use emgauwa_lib::types::ConnectedControllersType;
+use sqlx::{Pool, Sqlite};
+
+use crate::handlers::errors::ApiError;
+use crate::handlers::v1::ws::controllers::ControllerWs;
+
+pub mod controllers;
+
+#[get("/api/v1/ws/controllers")]
+pub async fn ws_controllers(
+	pool: web::Data<Pool<Sqlite>>,
+	connected_controllers: web::Data<ConnectedControllersType>,
+	req: HttpRequest,
+	stream: web::Payload,
+) -> Result<HttpResponse, ApiError> {
+	let resp = ws::start(
+		ControllerWs {
+			pool: pool.get_ref().clone(),
+			controller_uid: None,
+			connected_controllers: connected_controllers.get_ref().clone(),
+			hb: Instant::now(),
+		},
+		&req,
+		stream,
+	)
+	.map_err(|_| ApiError::InternalError(String::from("error starting websocket")));
+	resp
+}

emgauwa-core/src/main.rs

@@ -1,17 +1,17 @@
-use actix_cors::Cors;
 use std::collections::HashMap;
 use std::net::TcpListener;
-use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 
-use crate::utils::drop_privileges;
+use actix_cors::Cors;
 use actix_web::middleware::TrailingSlash;
 use actix_web::{middleware, web, App, HttpServer};
 use emgauwa_lib::db::DbController;
-use emgauwa_lib::handlers;
 use emgauwa_lib::types::ConnectedControllersType;
 use emgauwa_lib::utils::init_logging;
 
+use crate::utils::drop_privileges;
+
+mod handlers;
 mod settings;
 mod utils;
 

emgauwa-core/src/utils.rs

@@ -1,8 +1,8 @@
-use crate::settings::Settings;
-use log::log;
 use std::ffi::CString;
 use std::io::{Error, ErrorKind};
 
+use crate::settings::Settings;
+
 // https://blog.lxsang.me/post/id/28.0
 pub fn drop_privileges(settings: &Settings) -> Result<(), Error> {
 	log::info!(