in gateway tracker, update location_changed_at for new records being … (#1055)

* in gateway tracker, update location_changed_at for new records being inserted

* Add migration to fix null location_changed_at

* Add csv import

* Fix API return

* Add location check

* PR comments fixes

---------

Co-authored-by: Macpie <[email protected]>

Author: bbalser

Cargo.lock

@@ -15,8 +15,10 @@ bs58 = { workspace = true }
 chrono = { workspace = true }
 clap = { workspace = true }
 config = { workspace = true }
+csv = { workspace = true }
 futures = { workspace = true }
 futures-util = { workspace = true }
+h3o = { workspace = true }
 helium-crypto = { workspace = true, features = ["sqlx-postgres"] }
 helium-proto = { workspace = true }
 hextree = { workspace = true }
@@ -45,7 +47,6 @@ tower-http = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 triggered = { workspace = true }
-csv = { workspace = true }
 
 coverage-map = { path = "../coverage_map" }
 custom-tracing = { path = "../custom_tracing", features = ["grpc"] }

mobile_config/Cargo.toml

@@ -0,0 +1,102 @@
+use crate::{
+    gateway::db::{Gateway, LocationChangedAtUpdate},
+    settings::Settings,
+};
+use chrono::{DateTime, Utc};
+use h3o::{error::InvalidCellIndex, CellIndex};
+use helium_crypto::PublicKey;
+use serde::Deserialize;
+use std::{
+    fs::File,
+    path::{Path, PathBuf},
+    str::FromStr,
+    time::Instant,
+};
+
+#[derive(Debug, clap::Parser)]
+pub struct Import {
+    #[clap(short = 'f')]
+    file: PathBuf,
+
+    #[clap(subcommand)]
+    cmd: Cmd,
+}
+
+#[derive(Debug, clap::Subcommand)]
+pub enum Cmd {
+    HotspotsAssertions,
+}
+
+impl Import {
+    pub async fn run(self, settings: &Settings) -> anyhow::Result<()> {
+        match self.cmd {
+            Cmd::HotspotsAssertions => {
+                custom_tracing::init(settings.log.clone(), settings.custom_tracing.clone()).await?;
+
+                tracing::info!("started");
+
+                let pool = settings.database.connect("mobile-config-store").await?;
+
+                let start = Instant::now();
+
+                let updates = read_csv(self.file)?
+                    .into_iter()
+                    .filter_map(|row| row.try_into().ok())
+                    .collect::<Vec<LocationChangedAtUpdate>>();
+
+                tracing::info!("file read, updating {} records", updates.len());
+
+                let updated = Gateway::update_bulk_location_changed_at(&pool, &updates).await?;
+
+                let elapsed = start.elapsed();
+                tracing::info!(?elapsed, updated, "finished");
+
+                Ok(())
+            }
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+struct CsvRow {
+    public_key: PublicKey,
+    // serialnumber: String,
+    time: DateTime<Utc>,
+    // latitude: f64,
+    // longitude: f64,
+    h3: String,
+    // assertion_type: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum CsvRowError {
+    #[error("H3 index parse error: {0}")]
+    H3IndexParseError(#[from] InvalidCellIndex),
+}
+
+impl TryFrom<CsvRow> for LocationChangedAtUpdate {
+    type Error = CsvRowError;
+
+    fn try_from(row: CsvRow) -> Result<Self, Self::Error> {
+        let cell = CellIndex::from_str(&row.h3)?;
+
+        Ok(Self {
+            address: row.public_key.into(),
+            location_changed_at: row.time,
+            location: cell.into(),
+        })
+    }
+}
+
+fn read_csv<P: AsRef<Path>>(path: P) -> anyhow::Result<Vec<CsvRow>> {
+    let file = File::open(path)?;
+    let mut rdr = csv::Reader::from_reader(file);
+    let mut rows = Vec::new();
+
+    for result in rdr.deserialize() {
+        let record: CsvRow = result?;
+        rows.push(record);
+    }
+
+    Ok(rows)
+}

mobile_config/src/cli/import.rs

@@ -1,5 +1,5 @@
 use crate::{
-    cli::{api::Api, server::Server},
+    cli::{api::Api, import::Import, server::Server},
     settings::Settings,
 };
 use base64::{engine::general_purpose, Engine};
@@ -9,6 +9,7 @@ use std::io::Write;
 use std::{fs::File, path::PathBuf};
 
 pub mod api;
+pub mod import;
 pub mod server;
 
 #[derive(Debug, clap::Parser)]
@@ -28,9 +29,9 @@ pub struct Cli {
 impl Cli {
     pub async fn run(self) -> anyhow::Result<()> {
         match self.cmd {
-            Cmd::Server(server) => {
+            Cmd::Api(api) => {
                 let settings = Settings::new(self.config)?;
-                server.run(&settings).await
+                api.run(&settings).await
             }
             Cmd::GenerateKey => {
                 let kp = Keypair::generate(KeyTag::default(), &mut OsRng);
@@ -54,17 +55,22 @@ impl Cli {
 
                 Ok(())
             }
-            Cmd::Api(api) => {
+            Cmd::Import(import) => {
                 let settings = Settings::new(self.config)?;
-                api.run(&settings).await
+                import.run(&settings).await
+            }
+            Cmd::Server(server) => {
+                let settings = Settings::new(self.config)?;
+                server.run(&settings).await
             }
         }
     }
 }
 
 #[derive(Debug, clap::Subcommand)]
 pub enum Cmd {
-    Server(Server),
-    GenerateKey,
     Api(Api),
+    GenerateKey,
+    Import(Import),
+    Server(Server),
 }

mobile_config/src/cli/mod.rs

@@ -96,6 +96,12 @@ pub struct Gateway {
     pub location_changed_at: Option<DateTime<Utc>>,
     pub location_asserts: Option<u32>,
 }
+#[derive(Debug)]
+pub struct LocationChangedAtUpdate {
+    pub address: PublicKeyBinary,
+    pub location_changed_at: DateTime<Utc>,
+    pub location: u64,
+}
 
 impl Gateway {
     pub async fn insert_bulk(pool: &PgPool, rows: &[Gateway]) -> anyhow::Result<u64> {
@@ -335,6 +341,48 @@ impl Gateway {
         .map_err(anyhow::Error::from)
         .filter_map(|res| async move { res.ok() })
     }
+
+    pub async fn update_bulk_location_changed_at(
+        pool: &PgPool,
+        updates: &[LocationChangedAtUpdate],
+    ) -> anyhow::Result<u64> {
+        if updates.is_empty() {
+            return Ok(0);
+        }
+
+        const MAX_ROWS: usize = 20000;
+        let mut total = 0;
+
+        for chunk in updates.chunks(MAX_ROWS) {
+            let mut qb = QueryBuilder::<Postgres>::new(
+                r#"
+                    UPDATE gateways AS g
+                    SET location_changed_at = v.location_changed_at
+                    FROM ( 
+                "#,
+            );
+
+            qb.push_values(chunk, |mut b, update| {
+                b.push_bind(update.address.as_ref())
+                    .push_bind(update.location_changed_at)
+                    .push_bind(update.location as i64);
+            });
+
+            qb.push(
+                r#"
+                    ) AS v(address, location_changed_at, location)
+                    WHERE g.address = v.address
+                        AND g.location_changed_at IS NULL
+                        AND g.location = v.location
+                "#,
+            );
+
+            let res = qb.build().execute(pool).await?;
+            total += res.rows_affected();
+        }
+
+        Ok(total)
+    }
 }
 
 impl FromRow<'_, PgRow> for Gateway {

mobile_config/src/gateway/db.rs

@@ -132,8 +132,12 @@ impl MobileHotspotInfo {
             elevation,
             azimuth,
             location,
-            // Updated via SQL query see Gateway::insert
-            location_changed_at: None,
+            // Set to refreshed_at when hotspot has a location, None otherwise
+            location_changed_at: if location.is_some() {
+                Some(self.refreshed_at.unwrap_or_else(Utc::now))
+            } else {
+                None
+            },
             location_asserts: self.num_location_asserts.map(|n| n as u32),
         }))
     }

mobile_config/src/gateway/metadata_db.rs

@@ -252,7 +252,8 @@ impl From<Gateway> for GatewayInfo {
             metadata,
             device_type: gateway.gateway_type.into(),
             created_at: Some(gateway.created_at),
-            updated_at: Some(gateway.updated_at),
+            // because updated_at refers to the last time the data was actually changed.
+            updated_at: Some(gateway.last_changed_at),
             refreshed_at: Some(gateway.refreshed_at),
         }
     }

mobile_config/src/gateway/service/info.rs

@@ -46,7 +46,7 @@ async fn execute_test(pool: PgPool) -> anyhow::Result<()> {
     assert_eq!(gateway1.elevation, None);
     assert_eq!(gateway1.azimuth, None);
     assert_eq!(gateway1.location, Some(hex1 as u64));
-    assert_eq!(gateway1.location_changed_at, None);
+    assert_eq!(gateway1.location_changed_at, Some(now));
     assert_eq!(gateway1.location_asserts, Some(1));
 
     Ok(())