Merge pull request #8 from FxQnLr/eta

Eta

4 files changed, 52 insertions, 12 deletions

diff --git a/src/db.rs b/src/db.rs
index c012b47..8a6b16e 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -13,7 +13,8 @@ pub struct Device {
     pub id: String,
     pub mac: String,
     pub broadcast_addr: String,
-    pub ip: String
+    pub ip: String,
+    pub times: Option<Vec<i64>>
 }
 
 pub async fn init_db_pool() -> PgPool {
diff --git a/src/routes/device.rs b/src/routes/device.rs
index 1eeff0b..678d117 100644
--- a/src/routes/device.rs
+++ b/src/routes/device.rs
@@ -16,7 +16,7 @@ pub async fn get_device(State(state): State<Arc<crate::AppState>>, headers: Head
         let device = sqlx::query_as!(
             Device,
             r#"
-            SELECT id, mac, broadcast_addr, ip
+            SELECT id, mac, broadcast_addr, ip, times
             FROM devices
             WHERE id = $1;
             "#,
@@ -79,7 +79,7 @@ pub async fn post_device(State(state): State<Arc<crate::AppState>>, headers: Hea
             r#"
             UPDATE devices
             SET mac = $1, broadcast_addr = $2, ip = $3 WHERE id = $4
-            RETURNING id, mac, broadcast_addr, ip;
+            RETURNING id, mac, broadcast_addr, ip, times;
             "#,
             payload.mac,
             payload.broadcast_addr,
diff --git a/src/routes/start.rs b/src/routes/start.rs
index 271f924..1555db3 100644
--- a/src/routes/start.rs
+++ b/src/routes/start.rs
@@ -22,7 +22,7 @@ pub async fn start(State(state): State<Arc<crate::AppState>>, headers: HeaderMap
         let device = sqlx::query_as!(
             Device,
             r#"
-            SELECT id, mac, broadcast_addr, ip
+            SELECT id, mac, broadcast_addr, ip, times
             FROM devices
             WHERE id = $1;
             "#,
@@ -40,19 +40,31 @@ pub async fn start(State(state): State<Arc<crate::AppState>>, headers: HeaderMap
             &device.broadcast_addr.parse().map_err(WebolError::IpParse)?,
             create_buffer(&device.mac)?
         )?;
-
+        let dev_id = device.id.clone();
         let uuid = if payload.ping.is_some_and(|ping| ping) {
-            let uuid_gen = Uuid::new_v4().to_string();
+            let mut uuid: Option<String> = None;
+            for (key, value) in state.ping_map.clone() {
+                if value.ip == device.ip {
+                    debug!("service already exists");
+                    uuid = Some(key);
+                    break;
+                }
+            };
+            let uuid_gen = match uuid {
+                Some(u) => u,
+                None => Uuid::new_v4().to_string(),
+            };
             let uuid_genc = uuid_gen.clone();
+
             tokio::spawn(async move {
                 debug!("init ping service");
                 state.ping_map.insert(uuid_gen.clone(), PingValue { ip: device.ip.clone(), online: false });
 
-                crate::services::ping::spawn(state.ping_send.clone(), device.ip, uuid_gen.clone(), &state.ping_map).await
+                crate::services::ping::spawn(state.ping_send.clone(), device, uuid_gen.clone(), &state.ping_map, &state.db).await
             });
             Some(uuid_genc)
         } else { None };
-        Ok(Json(json!(StartResponse { id: device.id, boot: true, uuid })))
+        Ok(Json(json!(StartResponse { id: dev_id, boot: true, uuid })))
     } else {
         Err(WebolError::Generic)
     }
diff --git a/src/services/ping.rs b/src/services/ping.rs
index 67acc1c..c3bdced 100644
--- a/src/services/ping.rs
+++ b/src/services/ping.rs
@@ -5,11 +5,13 @@ use std::sync::Arc;
 use axum::extract::{ws::WebSocket};
 use axum::extract::ws::Message;
 use dashmap::DashMap;
+use sqlx::PgPool;
 use time::{Duration, Instant};
 use tokio::sync::broadcast::{Sender};
 use tracing::{debug, error, trace};
 use crate::AppState;
 use crate::config::SETTINGS;
+use crate::db::Device;
 
 pub type PingMap = DashMap<String, PingValue>;
 
@@ -19,11 +21,11 @@ pub struct PingValue {
     pub online: bool
 }
 
-pub async fn spawn(tx: Sender<BroadcastCommands>, ip: String, uuid: String, ping_map: &PingMap) {
+pub async fn spawn(tx: Sender<BroadcastCommands>, device: Device, uuid: String, ping_map: &PingMap, db: &PgPool) {
     let timer = Instant::now();
     let payload = [0; 8];
 
-    let ping_ip = IpAddr::from_str(&ip).expect("bad ip");
+    let ping_ip = IpAddr::from_str(&device.ip).expect("bad ip");
 
     let mut msg: Option<BroadcastCommands> = None;
     while msg.is_none() {
@@ -52,7 +54,16 @@ pub async fn spawn(tx: Sender<BroadcastCommands>, ip: String, uuid: String, ping
 
     let _ = tx.send(msg.clone());
     if let BroadcastCommands::Success(..) = msg {
-        ping_map.insert(uuid.clone(), PingValue { ip: ip.clone(), online: true });
+        sqlx::query!(
+            r#"
+            UPDATE devices
+            SET times = array_append(times, $1)
+            WHERE id = $2;
+            "#,
+            timer.elapsed().whole_seconds(),
+            device.id
+        ).execute(db).await.unwrap();
+        ping_map.insert(uuid.clone(), PingValue { ip: device.ip.clone(), online: true });
         tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
     }
     trace!("remove {} from ping_map", uuid);
@@ -71,7 +82,10 @@ pub async fn status_websocket(mut socket: WebSocket, state: Arc<AppState>) {
     let msg = socket.recv().await;
     let uuid = msg.unwrap().unwrap().into_text().unwrap();
 
-    trace!("Search for uuid: {:?}", uuid);
+    trace!("Search for uuid: {}", uuid);
+
+    let eta = get_eta(&state.db).await;
+    let _ = socket.send(Message::Text(format!("eta_{}_{}", eta, uuid))).await;
 
     let device_exists = state.ping_map.contains_key(&uuid);
     match device_exists {
@@ -87,6 +101,19 @@ pub async fn status_websocket(mut socket: WebSocket, state: Arc<AppState>) {
     let _ = socket.close().await;
 }
 
+async fn get_eta(db: &PgPool) -> i64 {
+    let query = sqlx::query!(
+        r#"SELECT times FROM devices;"#
+    ).fetch_one(db).await.unwrap();
+
+    let times = match query.times {
+        None => { vec![0] },
+        Some(t) => t,
+    };
+    times.iter().sum::<i64>() / times.len() as i64
+
+}
+
 async fn process_device(state: Arc<AppState>, uuid: String) -> Message {
     let pm = state.ping_map.clone().into_read_only();
     let device = pm.get(&uuid).expect("fatal error");