Connect old APIs to new DB

macpie · macpie · commit 28eefd7bef02 · 2025-10-29T11:04:55.000-07:00
diff --git a/iot_config/migrations/20251027000000_gateways.sql b/iot_config/migrations/20251027000000_gateways.sql
@@ -12,4 +12,10 @@ CREATE TABLE IF NOT EXISTS gateways (
     location_changed_at TIMESTAMPTZ,
     refreshed_at TIMESTAMPTZ NOT NULL,
     updated_at TIMESTAMPTZ NOT NULL
-);
+);
+
+CREATE INDEX IF NOT EXISTS gateways_last_changed_idx ON gateways (last_changed_at DESC);
+
+CREATE INDEX IF NOT EXISTS gateways_location_changed_idx ON gateways (location_changed_at DESC)
+WHERE
+    location IS NOT NULL;
diff --git a/iot_config/src/cli/daemon.rs b/iot_config/src/cli/daemon.rs
@@ -36,7 +36,7 @@ impl Daemon {
 
         let gateway_svc = GatewayService::new(
             signing_keypair.clone(),
-            metadata_pool.clone(),
+            pool.clone(),
             region_map.clone(),
             auth_cache.clone(),
             delegate_key_cache,
diff --git a/iot_config/src/gateway/db.rs b/iot_config/src/gateway/db.rs
@@ -1,4 +1,5 @@
 use chrono::{DateTime, Utc};
+use futures::{Stream, StreamExt, TryStreamExt};
 use helium_crypto::PublicKeyBinary;
 use sqlx::{postgres::PgRow, FromRow, PgExecutor, PgPool, Postgres, QueryBuilder, Row};
 
@@ -122,6 +123,42 @@ impl Gateway {
 
         Ok(gateway)
     }
+
+    pub fn stream<'a>(
+        db: impl PgExecutor<'a> + 'a,
+        min_last_changed_at: DateTime<Utc>,
+        min_location_changed_at: Option<DateTime<Utc>>,
+    ) -> impl Stream<Item = Self> + 'a {
+        sqlx::query_as::<_, Self>(
+            r#"
+                SELECT
+                    address,
+                    created_at,
+                    elevation,
+                    gain,
+                    hash,
+                    is_active,
+                    is_full_hotspot,
+                    last_changed_at,
+                    location,
+                    location_asserts,
+                    location_changed_at,
+                    refreshed_at,
+                    updated_at
+                FROM gateways
+                WHERE last_changed_at >= $1
+                AND (
+                    $2::timestamptz IS NULL
+                    OR (location IS NOT NULL AND location_changed_at >= $2)
+                )
+            "#,
+        )
+        .bind(min_last_changed_at)
+        .bind(min_location_changed_at)
+        .fetch(db)
+        .map_err(anyhow::Error::from)
+        .filter_map(|res| async move { res.ok() })
+    }
 }
 
 impl FromRow<'_, PgRow> for Gateway {
diff --git a/iot_config/src/gateway/service/info.rs b/iot_config/src/gateway/service/info.rs
@@ -1,16 +1,40 @@
-use crate::region_map;
+use crate::{gateway::db::Gateway, region_map};
 use anyhow::anyhow;
-use futures::stream::BoxStream;
+use chrono::{DateTime, Utc};
+use futures::{stream::BoxStream, Stream, StreamExt};
 use helium_crypto::PublicKeyBinary;
 use helium_proto::{
     services::iot_config::{
         GatewayInfo as GatewayInfoProto, GatewayMetadata as GatewayMetadataProto,
     },
     Region,
 };
+use sqlx::PgExecutor;
 
 pub type GatewayInfoStream = BoxStream<'static, GatewayInfo>;
 
+// Hotspot gain default; dbi * 10
+const DEFAULT_GAIN: u32 = 12;
+// Hotspot elevation default; meters above sea level
+const DEFAULT_ELEVATION: u32 = 0;
+
+pub async fn get(
+    db: impl PgExecutor<'_>,
+    address: &PublicKeyBinary,
+) -> anyhow::Result<Option<IotMetadata>> {
+    let gateway = Gateway::get_by_address(db, address).await?;
+    Ok(gateway.map(IotMetadata::from))
+}
+
+pub fn stream<'a>(
+    db: impl PgExecutor<'a> + 'a,
+    min_last_changed_at: DateTime<Utc>,
+    min_location_changed_at: Option<DateTime<Utc>>,
+) -> impl Stream<Item = IotMetadata> + 'a {
+    let stream = Gateway::stream(db, min_last_changed_at, min_location_changed_at);
+    stream.map(IotMetadata::from).boxed()
+}
+
 #[derive(Clone, Debug)]
 pub struct GatewayMetadata {
     pub location: u64,
@@ -26,9 +50,17 @@ pub struct GatewayInfo {
     pub is_full_hotspot: bool,
 }
 
+pub struct IotMetadata {
+    pub address: PublicKeyBinary,
+    pub location: Option<u64>,
+    pub elevation: i32,
+    pub gain: i32,
+    pub is_full_hotspot: bool,
+}
+
 impl GatewayInfo {
     pub fn chain_metadata_to_info(
-        meta: db::IotMetadata,
+        meta: IotMetadata,
         region_map: &region_map::RegionMapReader,
     ) -> Self {
         let metadata = if let Some(location) = meta.location {
@@ -112,69 +144,14 @@ impl TryFrom<GatewayInfo> for GatewayInfoProto {
     }
 }
 
-pub(crate) mod db {
-    use futures::stream::{Stream, StreamExt};
-    use helium_crypto::PublicKeyBinary;
-    use sqlx::{PgExecutor, Row};
-    use std::str::FromStr;
-
-    // Hotspot gain default; dbi * 10
-    const DEFAULT_GAIN: i32 = 12;
-    // Hotspot elevation default; meters above sea level
-    const DEFAULT_ELEVATION: i32 = 0;
-
-    pub struct IotMetadata {
-        pub address: PublicKeyBinary,
-        pub location: Option<u64>,
-        pub elevation: i32,
-        pub gain: i32,
-        pub is_full_hotspot: bool,
-    }
-
-    const GET_METADATA_SQL: &str = r#"
-            select kta.entity_key, infos.location::bigint, CAST(infos.elevation AS integer), CAST(infos.gain as integer), infos.is_full_hotspot
-            from iot_hotspot_infos infos
-            join key_to_assets kta on infos.asset = kta.asset
-        "#;
-
-    pub async fn get_info(
-        db: impl PgExecutor<'_>,
-        address: &PublicKeyBinary,
-    ) -> anyhow::Result<Option<IotMetadata>> {
-        let entity_key = bs58::decode(address.to_string()).into_vec()?;
-        let mut query: sqlx::QueryBuilder<sqlx::Postgres> =
-            sqlx::QueryBuilder::new(GET_METADATA_SQL);
-        query.push(" where kta.entity_key = $1 ");
-        Ok(query
-            .build_query_as::<IotMetadata>()
-            .bind(entity_key)
-            .fetch_optional(db)
-            .await?)
-    }
-
-    pub fn all_info_stream<'a>(
-        db: impl PgExecutor<'a> + 'a,
-    ) -> impl Stream<Item = IotMetadata> + 'a {
-        sqlx::query_as::<_, IotMetadata>(GET_METADATA_SQL)
-            .fetch(db)
-            .filter_map(|metadata| async move { metadata.ok() })
-            .boxed()
-    }
-
-    impl sqlx::FromRow<'_, sqlx::postgres::PgRow> for IotMetadata {
-        fn from_row(row: &sqlx::postgres::PgRow) -> sqlx::Result<Self> {
-            Ok(Self {
-                address: PublicKeyBinary::from_str(
-                    &bs58::encode(row.get::<&[u8], &str>("entity_key")).into_string(),
-                )
-                .map_err(|err| sqlx::Error::Decode(Box::new(err)))?,
-                location: row.get::<Option<i64>, &str>("location").map(|v| v as u64),
-                elevation: row
-                    .get::<Option<i32>, &str>("elevation")
-                    .unwrap_or(DEFAULT_ELEVATION),
-                gain: row.get::<Option<i32>, &str>("gain").unwrap_or(DEFAULT_GAIN),
-                is_full_hotspot: row.get("is_full_hotspot"),
-            })
+impl From<Gateway> for IotMetadata {
+    fn from(gateway: Gateway) -> Self {
+        Self {
+            address: gateway.address,
+            location: gateway.location,
+            elevation: gateway.elevation.unwrap_or(DEFAULT_ELEVATION) as i32,
+            gain: gateway.gain.unwrap_or(DEFAULT_GAIN) as i32,
+            is_full_hotspot: gateway.is_full_hotspot.unwrap_or(false),
         }
     }
 }
diff --git a/iot_config/src/gateway/service/mod.rs b/iot_config/src/gateway/service/mod.rs
@@ -3,7 +3,7 @@ use crate::{
     telemetry, verify_public_key, GrpcResult, GrpcStreamResult,
 };
 use anyhow::Result;
-use chrono::Utc;
+use chrono::{DateTime, Utc};
 use file_store::traits::{MsgVerify, TimestampEncode};
 use futures::stream::StreamExt;
 use helium_crypto::{Keypair, PublicKey, PublicKeyBinary, Sign};
@@ -30,7 +30,7 @@ const CACHE_TTL: Duration = Duration::from_secs(60 * 60 * 3);
 pub struct GatewayService {
     auth_cache: AuthCache,
     gateway_cache: Arc<Cache<PublicKeyBinary, GatewayInfo>>,
-    metadata_pool: Pool<Postgres>,
+    pool: Pool<Postgres>,
     region_map: RegionMapReader,
     signing_key: Arc<Keypair>,
     delegate_cache: watch::Receiver<org::DelegateCache>,
@@ -39,7 +39,7 @@ pub struct GatewayService {
 impl GatewayService {
     pub fn new(
         signing_key: Arc<Keypair>,
-        metadata_pool: Pool<Postgres>,
+        pool: Pool<Postgres>,
         region_map: RegionMapReader,
         auth_cache: AuthCache,
         delegate_cache: watch::Receiver<org::DelegateCache>,
@@ -51,7 +51,7 @@ impl GatewayService {
         Ok(Self {
             auth_cache,
             gateway_cache,
-            metadata_pool,
+            pool,
             region_map,
             signing_key,
             delegate_cache,
@@ -102,7 +102,7 @@ impl GatewayService {
             Some(gateway) => Ok(gateway.value().clone()),
             None => {
                 let metadata = tokio::select! {
-                    query_result = info::db::get_info(&self.metadata_pool, pubkey) => {
+                    query_result = info::get(&self.pool, pubkey) => {
                         query_result.map_err(|_| Status::internal("error fetching gateway info"))?
                             .ok_or_else(|| {
                                 telemetry::count_gateway_info_lookup("not-found");
@@ -284,7 +284,7 @@ impl iot_config::Gateway for GatewayService {
 
         tracing::debug!("fetching all gateways' info");
 
-        let pool = self.metadata_pool.clone();
+        let pool = self.pool.clone();
         let signing_key = self.signing_key.clone();
         let batch_size = request.batch_size;
         let region_map = self.region_map.clone();
@@ -316,7 +316,10 @@ async fn stream_all_gateways_info(
 ) -> anyhow::Result<()> {
     let timestamp = Utc::now().encode_timestamp();
     let signer: Vec<u8> = signing_key.public_key().into();
-    let mut stream = info::db::all_info_stream(pool).chunks(batch_size as usize);
+
+    let epoch: DateTime<Utc> = "1970-01-01T00:00:00Z".parse().unwrap();
+
+    let mut stream = info::stream(pool, epoch, None).chunks(batch_size as usize);
     while let Some(infos) = stream.next().await {
         let gateway_infos = infos
             .into_iter()
diff --git a/iot_config/tests/integrations/gateway_service.rs b/iot_config/tests/integrations/gateway_service.rs
@@ -1,7 +1,4 @@
-use crate::common::{
-    gateway_metadata_db::{self, create_tables},
-    make_keypair, spawn_gateway_service,
-};
+use crate::common::{make_keypair, spawn_gateway_service};
 use chrono::{DateTime, Utc};
 use futures::StreamExt;
 use h3o::{LatLng, Resolution};
@@ -74,8 +71,6 @@ async fn gateway_location_v1(pool: PgPool) -> anyhow::Result<()> {
     let pub_key = make_keypair().public_key().clone();
     let now = Utc::now();
 
-    create_tables(&pool).await?;
-
     let gateway = insert_gateway(&pool, now, pub_key.clone().into()).await?;
 
     let (addr, _) = spawn_gateway_service(pool.clone(), admin_key.public_key().clone()).await?;
@@ -97,8 +92,6 @@ async fn gateway_region_params_v1(pool: PgPool) -> anyhow::Result<()> {
     let pub_key = keypair.public_key().clone();
     let now = Utc::now();
 
-    create_tables(&pool).await?;
-
     let gateway = insert_gateway(&pool, now, pub_key.clone().into()).await?;
 
     let (addr, _) = spawn_gateway_service(pool.clone(), admin_key.public_key().clone()).await?;
@@ -125,8 +118,6 @@ async fn gateway_info_v1(pool: PgPool) -> anyhow::Result<()> {
     let pub_key = make_keypair().public_key().clone();
     let now = Utc::now();
 
-    create_tables(&pool).await?;
-
     let gateway = insert_gateway(&pool, now, pub_key.clone().into()).await?;
 
     let (addr, _) = spawn_gateway_service(pool.clone(), admin_key.public_key().clone()).await?;
@@ -158,8 +149,6 @@ async fn gateway_info_stream_v1(pool: PgPool) -> anyhow::Result<()> {
     let pub_key = make_keypair().public_key().clone();
     let now = Utc::now();
 
-    create_tables(&pool).await?;
-
     let gateway = insert_gateway(&pool, now, pub_key.clone().into()).await?;
 
     let (addr, _) = spawn_gateway_service(pool.clone(), admin_key.public_key().clone()).await?;
@@ -295,24 +284,7 @@ async fn insert_gateway(
         updated_at: now,
     };
 
-    // Insert test data into iot_hotspot_infos
-    gateway_metadata_db::insert_gateway(
-        pool,
-        "address1",                                 // address (PRIMARY KEY)
-        "asset1",                                   // asset
-        Some(gateway.location.unwrap() as i64),     // location
-        gateway.elevation.map(|v| v as i64),        // elevation
-        gateway.gain.map(|v| v as i64),             // gain
-        gateway.is_full_hotspot,                    // is_full_hotspot
-        gateway.location_asserts.map(|v| v as i32), // num_location_asserts
-        gateway.is_active,                          // is_active
-        Some(0),                                    // dc_onboarding_fee_paid
-        gateway.created_at,                         // created_at
-        gateway.refreshed_at,                       // refreshed_at
-        Some(0),                                    // last_block
-        gateway.address.clone(),                    // key (PublicKeyBinary)
-    )
-    .await?;
+    Gateway::insert_bulk(pool, std::slice::from_ref(&gateway)).await?;
 
     let loc_bytes = h3_index.to_le_bytes();
     let mut encoder = Encoder::new(Vec::new())?;
