adds initial user defined panels

server/src/grpc.rs

@@ -86,6 +86,7 @@ impl TelemetryService for CoreTelemetryService {
 }
 
 impl CoreTelemetryService {
+    #[allow(clippy::result_large_err)]
     fn handle_new_tlm_item(
         tlm_management: &Arc<TelemetryManagementService>,
         tlm_item: &TelemetryItem,

server/src/http/api/mod.rs

@@ -1,85 +1,15 @@
-use crate::http::error::HttpServerResultError;
-use crate::telemetry::management_service::TelemetryManagementService;
-use actix_web::{get, web, Responder};
-use chrono::{DateTime, TimeDelta, Utc};
-use log::trace;
-use serde::Deserialize;
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::time::timeout;
+mod panels;
+mod tlm;
 
-#[get("/tlm/info/{name:[\\w\\d/_-]+}")]
-async fn get_tlm_definition(
-    data: web::Data<Arc<TelemetryManagementService>>,
-    name: web::Path<String>,
-) -> Result<impl Responder, HttpServerResultError> {
-    let string = name.to_string();
-    trace!("get_tlm_definition {}", string);
-    let Some(data) = data.get_by_name(&string) else {
-        return Err(HttpServerResultError::TlmNameNotFound { tlm: string });
-    };
-    Ok(web::Json(data.definition.clone()))
-}
-
-#[get("/tlm/info")]
-async fn get_all_tlm_definitions(
-    data: web::Data<Arc<TelemetryManagementService>>,
-) -> Result<impl Responder, HttpServerResultError> {
-    trace!("get_all_tlm_definitions");
-    Ok(web::Json(data.get_all_definitions()))
-}
-
-#[derive(Deserialize)]
-struct HistoryQuery {
-    from: String,
-    to: String,
-    resolution: i64,
-}
-
-#[get("/tlm/history/{uuid:[0-9a-f]+}")]
-async fn get_tlm_history(
-    data_arc: web::Data<Arc<TelemetryManagementService>>,
-    uuid: web::Path<String>,
-    info: web::Query<HistoryQuery>,
-) -> Result<impl Responder, HttpServerResultError> {
-    let uuid = uuid.to_string();
-    trace!(
-        "get_tlm_history {} from {} to {} resolution {}",
-        uuid,
-        info.from,
-        info.to,
-        info.resolution
-    );
-
-    let Ok(from) = info.from.parse::<DateTime<Utc>>() else {
-        return Err(HttpServerResultError::InvalidDateTime {
-            date_time: info.from.clone(),
-        });
-    };
-    let Ok(to) = info.to.parse::<DateTime<Utc>>() else {
-        return Err(HttpServerResultError::InvalidDateTime {
-            date_time: info.to.clone(),
-        });
-    };
-    let maximum_resolution = TimeDelta::milliseconds(info.resolution);
-
-    let history_service = data_arc.history_service();
-    let data = data_arc.pin();
-
-    match data.get_by_uuid(&uuid) {
-        None => Err(HttpServerResultError::TlmUuidNotFound { uuid }),
-        Some(tlm) => timeout(
-            Duration::from_secs(10),
-            tlm.get(from, to, maximum_resolution, &history_service),
-        )
-        .await
-        .map(|result| Ok(web::Json(result)))
-        .unwrap_or_else(|_| Err(HttpServerResultError::Timeout)),
-    }
-}
+use actix_web::web;
 
 pub fn setup_api(cfg: &mut web::ServiceConfig) {
-    cfg.service(get_all_tlm_definitions)
-        .service(get_tlm_definition)
-        .service(get_tlm_history);
+    cfg.service(tlm::get_all_tlm_definitions)
+        .service(tlm::get_tlm_definition)
+        .service(tlm::get_tlm_history)
+        .service(panels::new)
+        .service(panels::get_all)
+        .service(panels::get_one)
+        .service(panels::set)
+        .service(panels::delete);
 }

server/src/http/api/panels.rs

@@ -0,0 +1,67 @@
+use crate::http::error::HttpServerResultError;
+use crate::panels::panel::PanelUpdate;
+use crate::panels::PanelService;
+use actix_web::{delete, get, post, put, web, Responder};
+use serde::Deserialize;
+use std::sync::Arc;
+
+#[derive(Deserialize)]
+struct CreateParam {
+    name: String,
+    data: String,
+}
+
+#[derive(Deserialize)]
+struct IdParam {
+    id: String,
+}
+
+#[post("/panel")]
+pub(super) async fn new(
+    panels: web::Data<Arc<PanelService>>,
+    data: web::Json<CreateParam>,
+) -> Result<impl Responder, HttpServerResultError> {
+    let uuid = panels.create(&data.name, &data.data).await?;
+    Ok(web::Json(uuid.value))
+}
+
+#[get("/panel")]
+pub(super) async fn get_all(
+    panels: web::Data<Arc<PanelService>>,
+) -> Result<impl Responder, HttpServerResultError> {
+    let result = panels.read_all().await?;
+    Ok(web::Json(result))
+}
+
+#[get("/panel/{id}")]
+pub(super) async fn get_one(
+    panels: web::Data<Arc<PanelService>>,
+    path: web::Path<IdParam>,
+) -> Result<impl Responder, HttpServerResultError> {
+    let result = panels.read(path.id.clone().into()).await?;
+    match result {
+        Some(result) => Ok(web::Json(result)),
+        None => Err(HttpServerResultError::PanelUuidNotFound {
+            uuid: path.id.clone(),
+        }),
+    }
+}
+
+#[put("/panel/{id}")]
+pub(super) async fn set(
+    panels: web::Data<Arc<PanelService>>,
+    path: web::Path<IdParam>,
+    data: web::Json<PanelUpdate>,
+) -> Result<impl Responder, HttpServerResultError> {
+    panels.update(path.id.clone().into(), data.0).await?;
+    Ok(web::Json(()))
+}
+
+#[delete("/panel/{id}")]
+pub(super) async fn delete(
+    panels: web::Data<Arc<PanelService>>,
+    path: web::Path<IdParam>,
+) -> Result<impl Responder, HttpServerResultError> {
+    panels.delete(path.id.clone().into()).await?;
+    Ok(web::Json(()))
+}

server/src/http/api/tlm.rs

@@ -0,0 +1,79 @@
+use crate::http::error::HttpServerResultError;
+use crate::telemetry::management_service::TelemetryManagementService;
+use actix_web::{get, web, Responder};
+use chrono::{DateTime, TimeDelta, Utc};
+use log::trace;
+use serde::Deserialize;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::timeout;
+
+#[get("/tlm/info/{name:[\\w\\d/_-]+}")]
+pub(super) async fn get_tlm_definition(
+    data: web::Data<Arc<TelemetryManagementService>>,
+    name: web::Path<String>,
+) -> Result<impl Responder, HttpServerResultError> {
+    let string = name.to_string();
+    trace!("get_tlm_definition {}", string);
+    let Some(data) = data.get_by_name(&string) else {
+        return Err(HttpServerResultError::TlmNameNotFound { tlm: string });
+    };
+    Ok(web::Json(data.definition.clone()))
+}
+
+#[get("/tlm/info")]
+pub(super) async fn get_all_tlm_definitions(
+    data: web::Data<Arc<TelemetryManagementService>>,
+) -> Result<impl Responder, HttpServerResultError> {
+    trace!("get_all_tlm_definitions");
+    Ok(web::Json(data.get_all_definitions()))
+}
+
+#[derive(Deserialize)]
+struct HistoryQuery {
+    from: String,
+    to: String,
+    resolution: i64,
+}
+
+#[get("/tlm/history/{uuid:[0-9a-f]+}")]
+pub(super) async fn get_tlm_history(
+    data_arc: web::Data<Arc<TelemetryManagementService>>,
+    uuid: web::Path<String>,
+    info: web::Query<HistoryQuery>,
+) -> Result<impl Responder, HttpServerResultError> {
+    let uuid = uuid.to_string();
+    trace!(
+        "get_tlm_history {} from {} to {} resolution {}",
+        uuid,
+        info.from,
+        info.to,
+        info.resolution
+    );
+
+    let Ok(from) = info.from.parse::<DateTime<Utc>>() else {
+        return Err(HttpServerResultError::InvalidDateTime {
+            date_time: info.from.clone(),
+        });
+    };
+    let Ok(to) = info.to.parse::<DateTime<Utc>>() else {
+        return Err(HttpServerResultError::InvalidDateTime {
+            date_time: info.to.clone(),
+        });
+    };
+    let maximum_resolution = TimeDelta::milliseconds(info.resolution);
+
+    let history_service = data_arc.history_service();
+    let data = data_arc.pin();
+
+    match data.get_by_uuid(&uuid) {
+        None => Err(HttpServerResultError::TlmUuidNotFound { uuid }),
+        Some(tlm) => timeout(
+            Duration::from_secs(10),
+            tlm.get(from, to, maximum_resolution, &history_service),
+        )
+        .await
+        .map(|result| Ok(web::Json(result)))
+        .unwrap_or_else(|_| Err(HttpServerResultError::Timeout)),
+    }
+}

server/src/http/error.rs

@@ -2,11 +2,9 @@ use actix_web::error::ResponseError;
 use actix_web::http::header::ContentType;
 use actix_web::http::StatusCode;
 use actix_web::HttpResponse;
+use anyhow::Error;
 use thiserror::Error;
 
-#[derive(Error, Debug)]
-pub enum WebsocketResponseError {}
-
 #[derive(Error, Debug)]
 pub enum HttpServerResultError {
     #[error("Telemetry Name Not Found: {tlm}")]
@@ -17,6 +15,10 @@ pub enum HttpServerResultError {
     InvalidDateTime { date_time: String },
     #[error("Timed out")]
     Timeout,
+    #[error("Internal Error")]
+    InternalError(anyhow::Error),
+    #[error("Panel Uuid Not Found: {uuid}")]
+    PanelUuidNotFound { uuid: String },
 }
 
 impl ResponseError for HttpServerResultError {
@@ -25,7 +27,9 @@ impl ResponseError for HttpServerResultError {
             HttpServerResultError::TlmNameNotFound { .. } => StatusCode::NOT_FOUND,
             HttpServerResultError::TlmUuidNotFound { .. } => StatusCode::NOT_FOUND,
             HttpServerResultError::InvalidDateTime { .. } => StatusCode::BAD_REQUEST,
-            HttpServerResultError::Timeout { .. } => StatusCode::GATEWAY_TIMEOUT,
+            HttpServerResultError::Timeout => StatusCode::GATEWAY_TIMEOUT,
+            HttpServerResultError::InternalError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            HttpServerResultError::PanelUuidNotFound { .. } => StatusCode::NOT_FOUND,
         }
     }
     fn error_response(&self) -> HttpResponse {
@@ -34,3 +38,9 @@ impl ResponseError for HttpServerResultError {
             .body(self.to_string())
     }
 }
+
+impl From<anyhow::Error> for HttpServerResultError {
+    fn from(value: Error) -> Self {
+        Self::InternalError(value)
+    }
+}

server/src/http/mod.rs

@@ -4,7 +4,9 @@ mod websocket;
 
 use crate::http::api::setup_api;
 use crate::http::websocket::setup_websocket;
+use crate::panels::PanelService;
 use crate::telemetry::management_service::TelemetryManagementService;
+use actix_web::middleware::Logger;
 use actix_web::{web, App, HttpServer};
 use log::info;
 use std::sync::Arc;
@@ -13,17 +15,21 @@ use tokio_util::sync::CancellationToken;
 pub async fn setup(
     cancellation_token: CancellationToken,
     telemetry_definitions: Arc<TelemetryManagementService>,
+    panel_service: PanelService,
 ) -> anyhow::Result<()> {
     let data = web::Data::new(telemetry_definitions);
     let cancel_token = web::Data::new(cancellation_token);
+    let panel_service = web::Data::new(Arc::new(panel_service));
 
     info!("Starting HTTP Server");
     HttpServer::new(move || {
         App::new()
             .app_data(data.clone())
             .app_data(cancel_token.clone())
+            .app_data(panel_service.clone())
             .service(web::scope("/ws").configure(setup_websocket))
             .service(web::scope("/api").configure(setup_api))
+            .wrap(Logger::default())
     })
     .bind("localhost:8080")?
     .run()

server/src/lib.rs

@@ -1,5 +1,6 @@
 mod grpc;
 mod http;
+mod panels;
 mod serialization;
 mod telemetry;
 mod uuid;
@@ -8,9 +9,13 @@ pub mod core {
     tonic::include_proto!("core");
 }
 
+use crate::panels::PanelService;
 use crate::telemetry::history::TelemetryHistoryService;
 use crate::telemetry::management_service::TelemetryManagementService;
-use log::error;
+use log::{error, info};
+use sqlx::sqlite::SqliteConnectOptions;
+use sqlx::SqlitePool;
+use std::path;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::sleep;
@@ -26,18 +31,36 @@ pub async fn setup() -> anyhow::Result<()> {
         });
     }
 
+    let data_folder = path::absolute("data")?;
+    let telemetry_folder = data_folder.join("telemetry");
+
+    let database_url = data_folder.join("database.db");
+
+    info!("Opening Database: {database_url:?}");
+    let sqlite = SqlitePool::connect_with(
+        SqliteConnectOptions::new()
+            .filename(database_url)
+            .create_if_missing(true),
+    )
+    .await?;
+    sqlx::migrate!().run(&sqlite).await?;
+
     let tlm = Arc::new(TelemetryManagementService::new(
-        TelemetryHistoryService::new()?,
+        TelemetryHistoryService::new(telemetry_folder)?,
     )?);
 
     let grpc_server = grpc::setup(cancellation_token.clone(), tlm.clone())?;
 
-    let result = http::setup(cancellation_token.clone(), tlm.clone()).await;
+    let panel_service = PanelService::new(sqlite.clone());
+
+    let result = http::setup(cancellation_token.clone(), tlm.clone(), panel_service).await;
     cancellation_token.cancel();
     result?; // result is dropped
     grpc_server.await?; //grpc server is dropped
     drop(cancellation_token); // All cancellation tokens are now dropped
 
+    sqlite.close().await;
+
     // Perform cleanup functions - at this point all servers have stopped and we can be sure that cleaning things up is safe
     for _ in 0..15 {
         if Arc::strong_count(&tlm) != 1 {

server/src/main.rs

@@ -21,6 +21,7 @@ async fn main() -> anyhow::Result<()> {
         })
         .level(log::LevelFilter::Warn)
         .level_for("server", log_level)
+        .level_for("actix_web::middleware::logger", log_level)
         .chain(std::io::stdout());
     if let Ok(log_file) = log_file {
         log_config = log_config.chain(fern::log_file(log_file)?)

server/src/panels/mod.rs

@@ -0,0 +1,130 @@
+pub mod panel;
+
+use crate::core::Uuid;
+use crate::panels::panel::{PanelRequired, PanelUpdate};
+use panel::Panel;
+use sqlx::SqlitePool;
+
+pub struct PanelService {
+    pool: SqlitePool,
+}
+
+impl PanelService {
+    pub fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+
+    pub async fn create(&self, name: &str, data: &str) -> anyhow::Result<Uuid> {
+        let id = Uuid::random();
+
+        let mut transaction = self.pool.begin().await?;
+
+        let _ = sqlx::query!(
+            r#"
+                INSERT INTO PANELS (id, name, data, deleted)
+                VALUES ($1, $2, $3, FALSE);
+            "#,
+            id.value,
+            name,
+            data
+        )
+        .execute(&mut *transaction)
+        .await?;
+
+        transaction.commit().await?;
+
+        Ok(id)
+    }
+
+    pub async fn read_all(&self) -> anyhow::Result<Vec<PanelRequired>> {
+        let mut transaction = self.pool.begin().await?;
+
+        let panels = sqlx::query_as!(
+            PanelRequired,
+            r#"
+                    SELECT id, name
+                    FROM PANELS
+                    WHERE deleted = FALSE
+                "#,
+        )
+        .fetch_all(&mut *transaction)
+        .await?;
+
+        transaction.commit().await?;
+
+        Ok(panels)
+    }
+
+    pub async fn read(&self, id: Uuid) -> anyhow::Result<Option<Panel>> {
+        let mut transaction = self.pool.begin().await?;
+
+        let panel = sqlx::query_as(
+            r#"
+                    SELECT id, name, data
+                    FROM PANELS
+                    WHERE id = $1 AND deleted = FALSE
+                "#,
+        )
+        .bind(id.value)
+        .fetch_optional(&mut *transaction)
+        .await?;
+
+        transaction.commit().await?;
+
+        Ok(panel)
+    }
+
+    pub async fn update(&self, id: Uuid, data: PanelUpdate) -> anyhow::Result<()> {
+        let mut transaction = self.pool.begin().await?;
+
+        if let Some(name) = data.name {
+            let _ = sqlx::query!(
+                r#"
+                        UPDATE PANELS
+                        SET name = $2
+                        WHERE id = $1;
+                    "#,
+                id.value,
+                name
+            )
+            .execute(&mut *transaction)
+            .await?;
+        }
+        if let Some(data) = data.data {
+            let _ = sqlx::query!(
+                r#"
+                        UPDATE PANELS
+                        SET data = $2
+                        WHERE id = $1;
+                    "#,
+                id.value,
+                data
+            )
+            .execute(&mut *transaction)
+            .await?;
+        }
+
+        transaction.commit().await?;
+
+        Ok(())
+    }
+
+    pub async fn delete(&self, id: Uuid) -> anyhow::Result<()> {
+        let mut transaction = self.pool.begin().await?;
+
+        let _ = sqlx::query!(
+            r#"
+                    UPDATE PANELS
+                    SET deleted = TRUE
+                    WHERE id = $1;
+                "#,
+            id.value,
+        )
+        .execute(&mut *transaction)
+        .await?;
+
+        transaction.commit().await?;
+
+        Ok(())
+    }
+}

server/src/panels/panel.rs

@@ -0,0 +1,22 @@
+use serde::{Deserialize, Serialize};
+use sqlx::FromRow;
+
+#[derive(Clone, FromRow, Serialize, Deserialize)]
+pub struct PanelRequired {
+    pub id: String,
+    pub name: String,
+}
+
+#[derive(Clone, FromRow, Serialize, Deserialize)]
+pub struct Panel {
+    #[sqlx(flatten)]
+    #[serde(flatten)]
+    pub header: PanelRequired,
+    pub data: String,
+}
+
+#[derive(Default, Clone, Serialize, Deserialize)]
+pub struct PanelUpdate {
+    pub name: Option<String>,
+    pub data: Option<String>,
+}

server/src/telemetry/history.rs

@@ -9,11 +9,11 @@ use chrono::{DateTime, DurationRound, SecondsFormat, TimeDelta, Utc};
 use log::{error, info};
 use std::cmp::min;
 use std::collections::VecDeque;
+use std::fs;
 use std::fs::File;
 use std::io::{BufReader, BufWriter, Seek, SeekFrom, Write};
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
-use std::{fs, path};
 use tokio::task::{spawn_blocking, JoinHandle};
 
 const FOLDER_DURATION: TimeDelta = TimeDelta::hours(1);
@@ -484,7 +484,7 @@ impl TelemetryHistory {
             drop(segments);
             let mut segments = self.segments.write().await;
 
-            if segments.len() == 0 {
+            if segments.is_empty() {
                 let start_time = timestamp.duration_trunc(service.segment_width).unwrap();
                 segments.push_back(
                     self.create_ram_segment(start_time, service, self.data.definition.data_type)
@@ -636,11 +636,11 @@ pub struct TelemetryHistoryService {
 }
 
 impl TelemetryHistoryService {
-    pub fn new() -> anyhow::Result<Self> {
+    pub fn new(data_folder: PathBuf) -> anyhow::Result<Self> {
         let result = Self {
             segment_width: TimeDelta::minutes(1),
             max_segments: 5,
-            data_root_folder: path::absolute("telemetry")?,
+            data_root_folder: data_folder,
         };
 
         fs::create_dir_all(&result.data_root_folder)?;
@@ -654,8 +654,6 @@ impl TelemetryHistoryService {
     }
 
     pub fn get_metadata_file(&self) -> PathBuf {
-        let mut result = self.data_root_folder.clone();
-        result.push("metadata.json");
-        result
+        self.data_root_folder.join("metadata.json")
     }
 }

server/src/telemetry/management_service.rs

@@ -154,7 +154,7 @@ impl TelemetryManagementService {
             .collect()
     }
 
-    pub fn pin(&self) -> TelemetryManagementServicePin {
+    pub fn pin(&self) -> TelemetryManagementServicePin<'_> {
         TelemetryManagementServicePin {
             tlm_data: self.tlm_data.pin(),
         }

server/src/uuid.rs

@@ -10,3 +10,9 @@ impl Uuid {
         }
     }
 }
+
+impl From<String> for Uuid {
+    fn from(value: String) -> Self {
+        Self { value }
+    }
+}