fix: add ASOS startup backfill + ingest audit notes

- daemon.rs: add get_asos_locations(), check_asos_staleness(), pub backfill_asos_station()
- main.rs: check ASOS staleness on startup; backfill stale/empty stations (30 days)
  mirrors existing USGS + CWMS startup backfill pattern
- TODO.md: document accurate ingest status and 7 known gaps found in audit

Author: treeherder

TODO.md

@@ -5,9 +5,9 @@
 ## ✅ Completed
 
 - Zone-based HTTP API (`/zones`, `/zone/{id}`, `/status`, `/backwater`, `/health`)
-- USGS NWIS ingest (real-time IV + 87 years DV historical backfill)
-- USACE CWMS ingest with runtime catalog discovery
-- IEM/ASOS precipitation ingest with basin assignment
+- USGS NWIS ingest (real-time IV every 15min + startup backfill up to 120 days IV)
+- USACE CWMS ingest with runtime catalog discovery (pool, tailwater, stage)
+- IEM/ASOS precipitation ingest with basin assignment (hourly, all weather fields)
 - 7-zone hydrological model (Illinois River reference implementation)
 - Flood type classification (top-down, bottom-up, local tributary, compound)
 - Backwater detection (Grafton stage + LaGrange pool/tailwater differential)
@@ -21,9 +21,19 @@
 - Integration test suites (ASOS, USGS, CWMS, daemon lifecycle, peak flow)
 - Dockerfile healthcheck (curl-based)
 
+## Ingest Bugs & Known Gaps
+
+- **ASOS: no startup backfill** — `backfill_asos_station()` is defined in `daemon.rs` but never called from `main.rs`; ASOS data only accumulates forward from first run; fix: add ASOS staleness check + backfill call in `main.rs` startup sequence alongside USGS/CWMS
+- **USGS: DV history not fetched** — `backfill_days: 120` (default) only fetches IV data; to get years of daily history, increase `backfill_days > 120` or add a one-time historical ingest CLI subcommand; the "87 years" history claim requires a manual config change or separate tool
+- **CWMS staleness check uses `location_id`** — `check_cwms_staleness()` queries all timeseries for a location in aggregate; if pool has fresh data but tailwater is empty, tailwater won't get backfilled on startup
+- **`asos_precip_summary` table is never populated** — schema defines 6h/12h/24h/48h pre-computed rolling totals, but no code ever inserts into it; `/zone/{id}` now correctly sums raw `asos_observations` directly instead
+- **`IemAsosMinute` (1-min precip) struct unused** — the 1-minute ASOS data structure exists in `ingest/iem.rs` but `fetch_recent_precip` only fetches hourly data; no 1-min poll path exists in the daemon
+- **ASOS poll ignores priority intervals** — all ASOS stations (PRIMARY through EXTENDED) are polled every 15min; the `poll_interval_minutes` field from priority is stored in DB but the daemon doesn't respect it
+- **USGS only fetches 2 parameters** — hardcoded `["00060", "00065"]` (discharge + stage); stations may also report temperature (00010), conductance (00095), pH (00400) — not currently ingested
+
 ---
 
-## Generic Multi-Waterway Architecture
+
 
 - `waterway.toml` — top-level deployment config: `waterway_id`, display name, bounding box, datum, timezone
 - Dynamic zone deserialization — replace hardcoded `ZoneCollection { zone_0…zone_6 }` with `HashMap<String, Zone>`

flomon_service/src/daemon.rs

@@ -210,6 +210,35 @@ impl Daemon {
         &self.cwms_locations
     }
 
+    /// Get reference to loaded ASOS locations
+    pub fn get_asos_locations(&self) -> &[AsosLocation] {
+        &self.asos_locations
+    }
+    
+    /// Check staleness of ASOS data for a specific station
+    pub fn check_asos_staleness(&mut self, station_id: &str) -> Result<Option<Duration>, Box<dyn Error>> {
+        let client = self.client.as_mut()
+            .ok_or("Daemon not initialized")?;
+        
+        let rows = client.query(
+            "SELECT MAX(observation_time) as latest 
+             FROM asos_observations 
+             WHERE station_id = $1",
+            &[&station_id]
+        )?;
+        
+        if rows.is_empty() {
+            return Ok(None);
+        }
+        
+        let latest: Option<DateTime<Utc>> = rows[0].get(0);
+        
+        match latest {
+            Some(dt) => Ok(Some(Utc::now() - dt)),
+            None => Ok(None),
+        }
+    }
+    
     /// Check staleness of data for a specific station
     pub fn check_staleness(&mut self, site_code: &str) -> Result<Option<Duration>, Box<dyn Error>> {
         let client = self.client.as_mut()
@@ -683,7 +712,7 @@ impl Daemon {
     }
 
     /// Backfill ASOS historical data for a station
-    fn backfill_asos_station(&mut self, station_id: &str, days: i64) -> Result<usize, Box<dyn Error>> {
+    pub fn backfill_asos_station(&mut self, station_id: &str, days: i64) -> Result<usize, Box<dyn Error>> {
         let http_client = reqwest::blocking::Client::builder()
             .timeout(std::time::Duration::from_secs(30))
             .build()?;

flomon_service/src/main.rs

@@ -186,6 +186,50 @@ fn main() {
         println!();
     }
 
+    // Check ASOS stations for stale data
+    println!("📋 Checking ASOS data freshness...");
+    let asos_locations: Vec<_> = daemon.get_asos_locations().to_vec();
+    let mut asos_backfill_needed = Vec::new();
+    
+    for location in &asos_locations {
+        // Strip leading "K" to match IEM API station codes
+        let station_id = if location.station_id.starts_with('K') && location.station_id.len() == 4 {
+            &location.station_id[1..]
+        } else {
+            &location.station_id[..]
+        };
+        
+        match daemon.check_asos_staleness(station_id) {
+            Ok(None) => {
+                println!("   {} - No data found (needs backfill)", location.station_id);
+                asos_backfill_needed.push(station_id.to_string());
+            }
+            Ok(Some(staleness)) => {
+                let hours = staleness.num_hours();
+                if hours > 2 {
+                    println!("   {} - Data is {} hours old (stale)", location.station_id, hours);
+                    asos_backfill_needed.push(station_id.to_string());
+                } else {
+                    println!("   {} - Data is fresh ({} min old)", location.station_id, staleness.num_minutes());
+                }
+            }
+            Err(e) => {
+                eprintln!("   {} - Error checking staleness: {}", location.station_id, e);
+            }
+        }
+    }
+    
+    if !asos_backfill_needed.is_empty() {
+        println!("\n📥 Backfilling {} ASOS stations (last 30 days)...", asos_backfill_needed.len());
+        for station_id in &asos_backfill_needed {
+            match daemon.backfill_asos_station(station_id, 30) {
+                Ok(count) => println!("   ✓ {} - Inserted {} observations", station_id, count),
+                Err(e) => eprintln!("   ✗ {} - Backfill failed: {}", station_id, e),
+            }
+        }
+        println!();
+    }
+    
     // Start HTTP endpoint if requested (in background thread)
     if let Some(port) = endpoint_port {
         println!("🚀 Starting HTTP endpoint server...");