Modeled Coverage - Phase 1

mobile_verifier/migrations/15_modeled_coverage.sql

@@ -12,13 +12,14 @@ CREATE TABLE hex_coverage (
        cbsd_id TEXT NOT NULL,
        signal_level signal_level NOT NULL,
        coverage_claim_time TIMESTAMPTZ NOT NULL,
-       inserted_at TIMESTAMPTZ NOT NULL,
+       inserted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
        PRIMARY KEY (uuid, hex)
 );
 
 CREATE TABLE seniority (
        cbsd_id TEXT NOT NULL,
        seniority_ts TIMESTAMPTZ NOT NULL,
+       inserted_at TIMESTAMPTZ NOT NULL,
        last_heartbeat TIMESTAMPTZ NOT NULL,
        uuid UUID NOT NULL,
        PRIMARY KEY (cbsd_id, seniority_ts)

mobile_verifier/src/coverage.rs

@@ -176,9 +176,9 @@ impl CoverageObject {
             sqlx::query(
                 r#"
                 INSERT INTO hex_coverage
-                  (uuid, hex, indoor, cbsd_id, signal_level, coverage_claim_time, inserted_at)
+                  (uuid, hex, indoor, cbsd_id, signal_level, coverage_claim_time)
                 VALUES
-                  ($1, $2, $3, $4, $5, $6, $7)
+                  ($1, $2, $3, $4, $5, $6)
                 "#,
             )
             .bind(self.coverage_obj.uuid)
@@ -187,7 +187,6 @@ impl CoverageObject {
             .bind(&self.coverage_obj.cbsd_id)
             .bind(SignalLevel::from(hex.signal_level))
             .bind(self.coverage_obj.coverage_claim_time)
-            .bind(Utc::now())
             .execute(&mut *transaction)
             .await?;
         }
@@ -445,37 +444,41 @@ impl CoverageClaimTimeCache {
         cbsd_id: &str,
         coverage_object: &Option<Uuid>,
         exec: &mut Transaction<'_, Postgres>,
-    ) -> Result<DateTime<Utc>, sqlx::Error> {
+    ) -> Result<Option<DateTime<Utc>>, sqlx::Error> {
         let key = (cbsd_id.to_string(), *coverage_object);
         if let Some(coverage_claim_time) = self.cache.get(&key).await {
-            Ok(*coverage_claim_time)
+            Ok(Some(*coverage_claim_time))
         } else {
-            let coverage_claim_time: DateTime<Utc> = sqlx::query_scalar(
+            let coverage_claim_time: Option<DateTime<Utc>> = sqlx::query_scalar(
                 r#"
                 SELECT coverage_claim_time FROM hex_coverage WHERE cbsd_id = $1 AND uuid = $2 LIMIT 1
                 "#,
             )
             .bind(cbsd_id)
             .bind(coverage_object)
-            .fetch_one(&mut *exec)
+            .fetch_optional(&mut *exec)
             .await?;
-            self.cache
-                .insert(
-                    key,
-                    coverage_claim_time,
-                    std::time::Duration::from_secs(60 * 60 * 24),
-                )
-                .await;
+            if let Some(coverage_claim_time) = coverage_claim_time {
+                self.cache
+                    .insert(
+                        key,
+                        coverage_claim_time,
+                        std::time::Duration::from_secs(60 * 60 * 24),
+                    )
+                    .await;
+            }
             Ok(coverage_claim_time)
         }
     }
 }
 
+#[allow(dead_code)]
 pub struct CoveredHexCache {
     pool: Pool<Postgres>,
     covered_hexes: Arc<Cache<Uuid, CachedCoverage>>,
 }
 
+#[allow(dead_code)]
 impl CoveredHexCache {
     pub fn new(pool: &Pool<Postgres>) -> Self {
         let cache = Arc::new(Cache::new());
@@ -524,11 +527,13 @@ impl CoveredHexCache {
 }
 
 #[derive(Clone)]
+#[allow(dead_code)]
 pub struct CachedCoverage {
     pub cbsd_id: String,
     coverage: Vec<CellIndex>,
 }
 
+#[allow(dead_code)]
 impl CachedCoverage {
     pub fn max_distance_km(&self, latlng: LatLng) -> f64 {
         self.coverage.iter().fold(0.0, |curr_max, curr_cov| {

mobile_verifier/src/heartbeats.rs

@@ -13,7 +13,7 @@ use futures::{
     stream::{Stream, StreamExt, TryStreamExt},
     TryFutureExt,
 };
-use h3o::LatLng;
+// use h3o::LatLng;
 use helium_crypto::PublicKeyBinary;
 use helium_proto::services::poc_mobile as proto;
 use mobile_config::{gateway_info::GatewayInfoResolver, GatewayClient};
@@ -132,16 +132,22 @@ impl HeartbeatDaemon {
 
         while let Some(heartbeat) = validated_heartbeats.next().await.transpose()? {
             if heartbeat.is_valid() && heartbeat.coverage_object.is_some() {
-                let coverage_claim_time = coverage_claim_time_cache
+                if let Some(coverage_claim_time) = coverage_claim_time_cache
                     .fetch_coverage_claim_time(
                         &heartbeat.cbsd_id,
                         &heartbeat.coverage_object,
                         &mut transaction,
                     )
-                    .await?;
-                heartbeat
-                    .update_seniority(coverage_claim_time, &self.seniority_sink, &mut transaction)
-                    .await?;
+                    .await?
+                {
+                    heartbeat
+                        .update_seniority(
+                            coverage_claim_time,
+                            &self.seniority_sink,
+                            &mut transaction,
+                        )
+                        .await?;
+                }
             }
 
             heartbeat.write(&self.heartbeat_sink).await?;
@@ -393,7 +399,7 @@ impl Heartbeat {
                 INSERT INTO heartbeats (cbsd_id, hotspot_key, cell_type, latest_timestamp, truncated_timestamp, coverage_object)
                 VALUES ($1, $2, $3, $4, $5, $6)
                 ON CONFLICT (cbsd_id, truncated_timestamp) DO UPDATE SET
-                latest_timestamp = EXCLUDED.latest_timestamp
+                latest_timestamp = EXCLUDED.latest_timestamp,
                 coverage_object = EXCLUDED.coverage_object
                 RETURNING (xmax = 0) as inserted
                 "#
@@ -414,9 +420,9 @@ impl Heartbeat {
 async fn validate_heartbeat(
     heartbeat: &CellHeartbeatIngestReport,
     gateway_client: &GatewayClient,
-    coverage_cache: &CoveredHexCache,
+    _coverage_cache: &CoveredHexCache,
     epoch: &Range<DateTime<Utc>>,
-    max_distance: f64,
+    _max_distance: f64,
 ) -> anyhow::Result<(Option<CellType>, proto::HeartbeatValidity)> {
     let cell_type = match CellType::from_cbsd_id(&heartbeat.report.cbsd_id) {
         Some(ty) => Some(ty),
@@ -439,9 +445,9 @@ async fn validate_heartbeat(
         return Ok((cell_type, proto::HeartbeatValidity::GatewayOwnerNotFound));
     }
 
+    /*
     let Some(coverage_object) = heartbeat.report.coverage_object() else {
-        return Ok((cell_type, proto::HeartbeatValidity::Valid));
-        // return Ok((cell_type, proto::HeartbeatValidity::BadCoverageObject));
+        return Ok((cell_type, proto::HeartbeatValidity::BadCoverageObject));
     };
 
     let Some(coverage) = coverage_cache.fetch_coverage(&coverage_object).await? else {
@@ -459,6 +465,7 @@ async fn validate_heartbeat(
     if coverage.max_distance_km(latlng) > max_distance {
         return Ok((cell_type, proto::HeartbeatValidity::TooFarFromCoverage));
     }
+    */
 
     Ok((cell_type, proto::HeartbeatValidity::Valid))
 }

mobile_verifier/src/settings.rs

@@ -27,12 +27,6 @@ pub struct Settings {
     pub start_after: u64,
     #[serde(default = "default_max_heartbeat_distance_from_coverage_km")]
     pub max_heartbeat_distance_from_coverage_km: f64,
-    #[serde(default = "default_disable_discovery_loc_rewards_to_s3")]
-    pub disable_discovery_loc_rewards_to_s3: bool,
-}
-
-pub fn default_disable_discovery_loc_rewards_to_s3() -> bool {
-    true
 }
 
 pub fn default_log() -> String {