Merge pull request #8 from FxQnLr/eta

Eta

11 files changed, 109 insertions, 21 deletions

diff --git a/.sqlx/query-f179f38584f97842cd41159dd2293fa19e5f1975ea8810526924e77683ba92c4.json b/.sqlx/query-1dc5f44967ffdee882f4cef32262fd643b452aacca373ee527c978e816115de6.json
index 322db91..33d524d 100644
--- a/.sqlx/query-f179f38584f97842cd41159dd2293fa19e5f1975ea8810526924e77683ba92c4.json
+++ b/.sqlx/query-1dc5f44967ffdee882f4cef32262fd643b452aacca373ee527c978e816115de6.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            UPDATE devices\n            SET mac = $1, broadcast_addr = $2, ip = $3 WHERE id = $4\n            RETURNING id, mac, broadcast_addr, ip;\n            ",
+  "query": "\n            UPDATE devices\n            SET mac = $1, broadcast_addr = $2, ip = $3 WHERE id = $4\n            RETURNING id, mac, broadcast_addr, ip, times;\n            ",
   "describe": {
     "columns": [
       {
@@ -22,6 +22,11 @@
         "ordinal": 3,
         "name": "ip",
         "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "times",
+        "type_info": "Int8Array"
       }
     ],
     "parameters": {
@@ -36,8 +41,9 @@
       false,
       false,
       false,
-      false
+      false,
+      true
     ]
   },
-  "hash": "f179f38584f97842cd41159dd2293fa19e5f1975ea8810526924e77683ba92c4"
+  "hash": "1dc5f44967ffdee882f4cef32262fd643b452aacca373ee527c978e816115de6"
 }
diff --git a/.sqlx/query-25ab5caa92ff80ef3f85ddce1174021ce0fb896a818a6deee291a1eb85ae2a8a.json b/.sqlx/query-25ab5caa92ff80ef3f85ddce1174021ce0fb896a818a6deee291a1eb85ae2a8a.json
new file mode 100644
index 0000000..71438e3
--- /dev/null
+++ b/.sqlx/query-25ab5caa92ff80ef3f85ddce1174021ce0fb896a818a6deee291a1eb85ae2a8a.json
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            UPDATE devices\n            SET times = array_append(times, $1)\n            WHERE id = $2;\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "25ab5caa92ff80ef3f85ddce1174021ce0fb896a818a6deee291a1eb85ae2a8a"
+}
diff --git a/.sqlx/query-5ac2b9a76338dd1342938cc09e37b14c828c4ee44b696d83f87f739022be6948.json b/.sqlx/query-5ac2b9a76338dd1342938cc09e37b14c828c4ee44b696d83f87f739022be6948.json
new file mode 100644
index 0000000..c385101
--- /dev/null
+++ b/.sqlx/query-5ac2b9a76338dd1342938cc09e37b14c828c4ee44b696d83f87f739022be6948.json
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT times FROM devices;",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "times",
+        "type_info": "Int8Array"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "5ac2b9a76338dd1342938cc09e37b14c828c4ee44b696d83f87f739022be6948"
+}
diff --git a/.sqlx/query-82c11b5a47389884e4ed945b87dfcee9067933c4df892609700c0766c20fc5c5.json b/.sqlx/query-62c84231c7e9c85dc91d71f6b4f7ee6dae2130c2109fb6f1e47e0990ec395744.json
index c0a933d..5ec47e3 100644
--- a/.sqlx/query-82c11b5a47389884e4ed945b87dfcee9067933c4df892609700c0766c20fc5c5.json
+++ b/.sqlx/query-62c84231c7e9c85dc91d71f6b4f7ee6dae2130c2109fb6f1e47e0990ec395744.json
@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            SELECT id, mac, broadcast_addr, ip\n            FROM devices\n            WHERE id = $1;\n            ",
+  "query": "\n            SELECT id, mac, broadcast_addr, ip, times\n            FROM devices\n            WHERE id = $1;\n            ",
   "describe": {
     "columns": [
       {
@@ -22,6 +22,11 @@
         "ordinal": 3,
         "name": "ip",
         "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "times",
+        "type_info": "Int8Array"
       }
     ],
     "parameters": {
@@ -33,8 +38,9 @@
       false,
       false,
       false,
-      false
+      false,
+      true
     ]
   },
-  "hash": "82c11b5a47389884e4ed945b87dfcee9067933c4df892609700c0766c20fc5c5"
+  "hash": "62c84231c7e9c85dc91d71f6b4f7ee6dae2130c2109fb6f1e47e0990ec395744"
 }
diff --git a/Cargo.lock b/Cargo.lock
index 9873a4c..8f62060 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2187,7 +2187,7 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
 
 [[package]]
 name = "webol"
-version = "0.2.1"
+version = "0.3.0"
 dependencies = [
  "axum",
  "axum-macros",
diff --git a/Cargo.toml b/Cargo.toml
index 5279abf..3207453 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "webol"
-version = "0.2.1"
+version = "0.3.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
diff --git a/migrations/20231009123228_devices.sql b/migrations/20231009123228_devices.sql
index b911b19..d36946c 100644
--- a/migrations/20231009123228_devices.sql
+++ b/migrations/20231009123228_devices.sql
@@ -4,5 +4,6 @@ CREATE TABLE IF NOT EXISTS "devices"
     "id"                VARCHAR(255) PRIMARY KEY NOT NULL,
     "mac"               VARCHAR(17) NOT NULL,
     "broadcast_addr"    VARCHAR(39) NOT NULL,
-    "ip"                VARCHAR(39) NOT NULL
+    "ip"                VARCHAR(39) NOT NULL,
+    "times"             BIGINT[]
 )
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");