fix(heartbeats): Don't fetch heartbeats outside of the defined window range

Author: tizoc

tools/heartbeats-processor/src/local_db.rs

@@ -245,9 +245,11 @@ pub async fn process_heartbeats(
     db: &FirestoreDb,
     pool: &SqlitePool,
     config: &Config,
-) -> Result<()> {
+) -> Result<usize> {
     let last_processed_time = get_last_processed_time(pool, Some(config)).await?;
     let now = Utc::now();
+    // Don't fetch heartbeats beyond window range end
+    let end_time = config.window_range_end.min(now);
 
     let mut total_heartbeats = 0;
     let mut latest_time = last_processed_time;
@@ -266,7 +268,8 @@ pub async fn process_heartbeats(
     };
 
     loop {
-        let heartbeats = crate::remote_db::fetch_heartbeat_chunk(db, &mut chunk_state, now).await?;
+        let heartbeats =
+            crate::remote_db::fetch_heartbeat_chunk(db, &mut chunk_state, end_time).await?;
         if heartbeats.is_empty() {
             break;
         }
@@ -430,7 +433,7 @@ pub async fn process_heartbeats(
         update_last_processed_time(pool, latest_time).await?;
     }
 
-    Ok(())
+    Ok(total_heartbeats)
 }
 
 pub async fn create_tables_from_file(pool: &SqlitePool) -> Result<()> {

tools/heartbeats-processor/src/main.rs

@@ -112,10 +112,12 @@ async fn run_process_loop(
 
     loop {
         println!("Processing heartbeats...");
-        local_db::process_heartbeats(db, pool, config).await?;
+        let count = local_db::process_heartbeats(db, pool, config).await?;
 
-        println!("Posting scores...");
-        post_scores_to_firestore(pool, db, config).await?;
+        if count > 0 {
+            println!("Posting scores...");
+            post_scores_to_firestore(pool, db, config).await?;
+        }
 
         println!("Sleeping for {} seconds...", interval_seconds);
         tokio::time::sleep(interval).await;

tools/heartbeats-processor/src/remote_db.rs

@@ -171,10 +171,14 @@ pub async fn fetch_heartbeat_chunk(
     let chunk_end = (state.chunk_start + chunk_duration).min(end_time);
 
     if state.chunk_start >= end_time {
+        println!("Reached end of testing window: {}", end_time);
         return Ok(Vec::new());
     }
 
-    println!("Fetching heartbeat chunk... {}", state.chunk_start);
+    println!(
+        "Fetching heartbeat chunk... {} to {}",
+        state.chunk_start, chunk_end
+    );
 
     let query = db
         .fluent()