[Mobile config] Add v2 endpoints to GatewayInfoV2. Change min_refreshed_at to min_updated_at

Cargo.toml

@@ -70,10 +70,10 @@ helium-lib = { git = "https://github.com/helium/helium-wallet-rs.git", branch =
 hextree = { git = "https://github.com/jaykickliter/HexTree", branch = "main", features = [
   "disktree",
 ] }
-helium-proto = { git = "https://github.com/helium/proto", branch = "master", features = [
+helium-proto = { git = "https://github.com/helium/proto", branch = "min_update_at", features = [
   "services",
 ] }
-beacon = { git = "https://github.com/helium/proto", branch = "master" }
+beacon = { git = "https://github.com/helium/proto", branch = "min_update_at" }
 solana-client = "1.18"
 solana-sdk = "1.18"
 solana-program = "1.18"

mobile_config/src/gateway_info.rs

@@ -360,7 +360,7 @@ pub(crate) mod db {
     use futures::stream::{Stream, StreamExt};
     use helium_crypto::PublicKeyBinary;
     use sqlx::{types::Json, PgExecutor, Row};
-    use std::str::FromStr;
+    use std::{collections::HashSet, str::FromStr};
 
     const GET_METADATA_SQL: &str = r#"
             select kta.entity_key, infos.location::bigint, infos.device_type,
@@ -369,15 +369,34 @@ pub(crate) mod db {
             join key_to_assets kta on infos.asset = kta.asset
         "#;
     const BATCH_SQL_WHERE_SNIPPET: &str = " where kta.entity_key = any($1::bytea[]) ";
-    const DEVICE_TYPES_AND_SNIPPET: &str = " and device_type::text = any($2) ";
+    const DEVICE_TYPES_WHERE_SNIPPET: &str = " where device_type::text = any($1) ";
+
+    const GET_UPDATED_RADIOS: &str =
+        "SELECT entity_key FROM mobile_radio_tracker WHERE last_changed_at >= $1";
 
     lazy_static::lazy_static! {
         static ref BATCH_METADATA_SQL: String = format!("{GET_METADATA_SQL} {BATCH_SQL_WHERE_SNIPPET}");
-        static ref GET_METADATA_SQL_REFRESHED_AT: String = format!(r#"{GET_METADATA_SQL}
-            where ( infos.refreshed_at >= $1 OR (infos.refreshed_at IS NULL AND infos.created_at >= $1) ) "#);
+        static ref DEVICE_TYPES_METADATA_SQL: String = format!("{GET_METADATA_SQL} {DEVICE_TYPES_WHERE_SNIPPET}");
+    }
 
-        static ref DEVICE_TYPES_METADATA_SQL: String = format!("{} {}", *GET_METADATA_SQL_REFRESHED_AT, DEVICE_TYPES_AND_SNIPPET);
+    pub async fn get_updated_radios(
+        db: impl PgExecutor<'_>,
+        min_updated_at: DateTime<Utc>,
+    ) -> anyhow::Result<HashSet<PublicKeyBinary>> {
+        let rows: Vec<Vec<u8>> = sqlx::query_scalar(GET_UPDATED_RADIOS)
+            .bind(min_updated_at)
+            .fetch_all(db)
+            .await?;
+        let mut radios = HashSet::new();
+
+        for row in rows {
+            let entity_key_b: &[u8] = &row;
+            let entity_key = bs58::encode(entity_key_b).into_string();
+            let pk = PublicKeyBinary::from_str(&entity_key)?;
+            radios.insert(pk);
+        }
 
+        Ok(radios)
     }
 
     pub async fn get_info(
@@ -413,16 +432,13 @@ pub(crate) mod db {
     pub fn all_info_stream<'a>(
         db: impl PgExecutor<'a> + 'a,
         device_types: &'a [DeviceType],
-        min_refreshed_at: DateTime<Utc>,
     ) -> impl Stream<Item = GatewayInfo> + 'a {
         match device_types.is_empty() {
-            true => sqlx::query_as::<_, GatewayInfo>(&GET_METADATA_SQL_REFRESHED_AT)
-                .bind(min_refreshed_at)
+            true => sqlx::query_as::<_, GatewayInfo>(GET_METADATA_SQL)
                 .fetch(db)
                 .filter_map(|metadata| async move { metadata.ok() })
                 .boxed(),
             false => sqlx::query_as::<_, GatewayInfo>(&DEVICE_TYPES_METADATA_SQL)
-                .bind(min_refreshed_at)
                 .bind(
                     device_types
                         .iter()

mobile_config/src/gateway_service.rs

@@ -1,11 +1,12 @@
 use crate::{
-    gateway_info::{self, DeviceType, GatewayInfo},
+    gateway_info::{self, db::get_updated_radios, DeviceType, GatewayInfo},
     key_cache::KeyCache,
     telemetry, verify_public_key, GrpcResult, GrpcStreamResult,
 };
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::{TimeZone, Utc};
 use file_store::traits::{MsgVerify, TimestampEncode};
 use futures::{
+    future,
     stream::{Stream, StreamExt, TryStreamExt},
     TryFutureExt,
 };
@@ -23,16 +24,23 @@ use tonic::{Request, Response, Status};
 
 pub struct GatewayService {
     key_cache: KeyCache,
+    mobile_config_db_pool: Pool<Postgres>,
     metadata_pool: Pool<Postgres>,
     signing_key: Arc<Keypair>,
 }
 
 impl GatewayService {
-    pub fn new(key_cache: KeyCache, metadata_pool: Pool<Postgres>, signing_key: Keypair) -> Self {
+    pub fn new(
+        key_cache: KeyCache,
+        metadata_pool: Pool<Postgres>,
+        signing_key: Keypair,
+        mobile_config_db_pool: Pool<Postgres>,
+    ) -> Self {
         Self {
             key_cache,
             metadata_pool,
             signing_key: Arc::new(signing_key),
+            mobile_config_db_pool,
         }
     }
 
@@ -170,8 +178,7 @@ impl mobile_config::Gateway for GatewayService {
         );
 
         tokio::spawn(async move {
-            let stream =
-                gateway_info::db::all_info_stream(&pool, &device_types, DateTime::UNIX_EPOCH);
+            let stream = gateway_info::db::all_info_stream(&pool, &device_types);
             stream_multi_gateways_info(stream, tx.clone(), signing_key.clone(), batch_size).await
         });
 
@@ -191,27 +198,39 @@ impl mobile_config::Gateway for GatewayService {
         self.verify_request_signature(&signer, &request)?;
 
         let pool = self.metadata_pool.clone();
+        let mc_pool = self.mobile_config_db_pool.clone();
         let signing_key = self.signing_key.clone();
         let batch_size = request.batch_size;
 
         let (tx, rx) = tokio::sync::mpsc::channel(100);
 
         let device_types: Vec<DeviceType> = request.device_types().map(|v| v.into()).collect();
-        let min_refreshed_at = Utc
-            .timestamp_opt(request.min_refreshed_at as i64, 0)
-            .single()
-            .ok_or(Status::invalid_argument(
-                "Invalid min_refreshed_at argument",
-            ))?;
 
         tracing::debug!(
             "fetching all gateways' info (v2). Device types: {:?} ",
             device_types
         );
 
         tokio::spawn(async move {
-            let stream = gateway_info::db::all_info_stream(&pool, &device_types, min_refreshed_at);
-            stream_multi_gateways_info(stream, tx.clone(), signing_key.clone(), batch_size).await
+            let stream = gateway_info::db::all_info_stream(&pool, &device_types);
+            if request.min_updated_at > 0 {
+                let min_updated_at = Utc
+                    .timestamp_opt(request.min_updated_at as i64, 0)
+                    .single()
+                    .ok_or(Status::invalid_argument(
+                        "Invalid min_refreshed_at argument",
+                    ))?;
+
+                let updated_redios = get_updated_radios(&mc_pool, min_updated_at).await?;
+                let stream = stream
+                    .filter(|v| future::ready(updated_redios.contains(&v.address)))
+                    .boxed();
+                stream_multi_gateways_info(stream, tx.clone(), signing_key.clone(), batch_size)
+                    .await
+            } else {
+                stream_multi_gateways_info(stream, tx.clone(), signing_key.clone(), batch_size)
+                    .await
+            }
         });
 
         Ok(Response::new(GrpcStreamResult::new(rx)))

mobile_config/src/main.rs

@@ -75,6 +75,7 @@ impl Daemon {
             key_cache.clone(),
             metadata_pool.clone(),
             settings.signing_keypair()?,
+            pool.clone(),
         );
         let auth_svc = AuthorizationService::new(key_cache.clone(), settings.signing_keypair()?);
         let entity_svc = EntityService::new(