prop/rust/prop_grid_rs/src/hrdps_fetcher.rs

//! HRDPS (Canadian) fetcher. Sibling of `fetcher.rs` (HRRR) but with
//! ECCC's MSC Datamart shape:
//!
//!   * Date-prefixed URL structure
//!     (`https://dd.weather.gc.ca/{YYYYMMDD}/WXO-DD/model_hrdps/...`)
//!   * **One variable per GRIB2 file** (~3.5 MB each, ~14 files total).
//!     No idx / no byte-range partials.
//!   * Concurrent per-variable fetch + byte-concatenate into a single
//!     multi-record blob — wgrib2 reads multi-record files natively, so
//!     the downstream `decoder::extract_grid` path is unchanged from HRRR.
//!   * Rotated lat/lon projection handled transparently by wgrib2.
//!
//! 1:1 port of `lib/microwaveprop/weather/hrdps_client.ex`.

use std::time::Duration;

use chrono::{DateTime, Timelike, Utc};
use reqwest::Client;
use tokio::task::JoinSet;

pub const DATAMART_BASE_DEFAULT: &str = "https://dd.weather.gc.ca";
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
pub const MAX_PARALLEL_FETCHES: usize = 4;

/// Pressure levels matching `lib/microwaveprop/weather/hrdps_client.ex`'s
/// `@grid_pressure_levels`. Mirrors HRRR's grid set (1000-700 mb) so
/// SoundingParams.derive sees a comparable refractivity-gradient profile.
pub const HRDPS_GRID_PRESSURE_LEVELS: &[u16] = &[1000, 950, 900, 850, 800, 750, 700];

/// Surface variables we fetch per cycle/forecast hour. Each entry is
/// (internal_atom, msc_var, msc_level) — the MSC level slug becomes part
/// of the filename. Mirrors `HrdpsClient.@surface_vars` in Elixir.
pub const HRDPS_SURFACE_VARS: &[(&str, &str, &str)] = &[
    ("tmp_2m", "TMP", "AGL-2m"),
    ("depr_2m", "DEPR", "AGL-2m"),
    ("dpt_2m", "DPT", "AGL-2m"),
    ("pres_sfc", "PRES", "Sfc"),
    ("hpbl_sfc", "HPBL", "Sfc"),
    ("ugrd_10m", "UGRD", "AGL-10m"),
    ("vgrd_10m", "VGRD", "AGL-10m"),
    ("tcdc_sfc", "TCDC", "Sfc"),
];

#[derive(Debug, thiserror::Error)]
pub enum HrdpsFetchError {
    #[error("http: {0}")]
    Http(#[from] reqwest::Error),
    #[error("HRDPS file fetch HTTP {status} for {url}")]
    Status { status: u16, url: String },
    #[error("HRDPS fetch returned empty body for {url}")]
    EmptyBody { url: String },
}

/// Build the MSC Datamart URL for one HRDPS variable file. ECCC reorganized
/// to a date-prefixed structure as of 2026 — the old `/model_hrdps/...`
/// flat root 404s. Filename pattern verified against the live datamart on
/// 2026-04-29.
pub fn hrdps_url(
    base: &str,
    cycle: DateTime<Utc>,
    forecast_hour: u8,
    msc_var: &str,
    msc_level: &str,
) -> String {
    let date = cycle.date_naive();
    let date_str = date.format("%Y%m%d").to_string();
    let hour_str = format!("{:02}", cycle.hour());
    let fff_str = format!("{:03}", forecast_hour);
    let trimmed = base.trim_end_matches('/');

    format!(
        "{trimmed}/{date_str}/WXO-DD/model_hrdps/continental/2.5km/{hour_str}/{fff_str}/\
         {date_str}T{hour_str}Z_MSC_HRDPS_{msc_var}_{msc_level}_RLatLon0.0225_PT{fff_str}H.grib2"
    )
}

/// Build the variable manifest the fetcher pulls per cycle/forecast hour.
/// Returns `[(msc_var, msc_level)]` in stable order so cargo-test fixtures
/// match production output byte-for-byte. Surface variables come first,
/// then pressure-level (TMP / DEPR / HGT × levels).
pub fn variable_manifest() -> Vec<(&'static str, String)> {
    let mut out: Vec<(&'static str, String)> = HRDPS_SURFACE_VARS
        .iter()
        .map(|(_, var, level)| (*var, level.to_string()))
        .collect();

    for &level in HRDPS_GRID_PRESSURE_LEVELS {
        let level_slug = format!("ISBL_{:04}", level);
        out.push(("TMP", level_slug.clone()));
        out.push(("DEPR", level_slug.clone()));
        out.push(("HGT", level_slug));
    }

    out
}

#[derive(Debug, Clone)]
pub struct HrdpsClient {
    http: Client,
    base: String,
}

impl HrdpsClient {
    pub fn new(base: impl Into<String>) -> Result<Self, HrdpsFetchError> {
        let http = Client::builder()
            .timeout(REQUEST_TIMEOUT)
            .pool_max_idle_per_host(MAX_PARALLEL_FETCHES)
            .build()?;

        Ok(Self {
            http,
            base: base.into(),
        })
    }

    pub fn default_base() -> Result<Self, HrdpsFetchError> {
        Self::new(DATAMART_BASE_DEFAULT)
    }

    pub fn base(&self) -> &str {
        &self.base
    }

    /// Fetch every variable file for `(cycle, forecast_hour)` concurrently,
    /// then byte-concatenate into a single multi-record GRIB2 blob. wgrib2
    /// reads multi-record files natively, so the returned blob plugs into
    /// `decoder::extract_grid` exactly like an HRRR `wrfsfcf`/`wrfprsf`
    /// blob would — same `match` pattern works because HRDPS uses the same
    /// NCEP-style level strings (`TMP:2 m above ground`, etc.) at the
    /// wgrib2 inventory layer regardless of the MSC filename.
    pub async fn fetch_combined_blob(
        &self,
        cycle: DateTime<Utc>,
        forecast_hour: u8,
    ) -> Result<Vec<u8>, HrdpsFetchError> {
        let manifest = variable_manifest();
        let mut futs: JoinSet<Result<(usize, Vec<u8>), HrdpsFetchError>> = JoinSet::new();

        // Cap concurrent fetches at MAX_PARALLEL_FETCHES (datamart isn't
        // aggressively rate-limited but ECCC has explicit guidance to
        // limit hammer-rate). Index preserves variable order so the
        // concatenated output is deterministic.
        let manifest_len = manifest.len();
        for (idx, (msc_var, msc_level)) in manifest.into_iter().enumerate() {
            let client = self.http.clone();
            let url = hrdps_url(&self.base, cycle, forecast_hour, msc_var, &msc_level);
            futs.spawn(async move {
                let bytes = retry_get(&client, &url).await?;
                Ok((idx, bytes))
            });
        }

        let mut results: Vec<(usize, Vec<u8>)> = Vec::with_capacity(manifest_len);
        while let Some(joined) = futs.join_next().await {
            // Per CLAUDE.md: never silently swallow a task failure. A
            // single missing variable here means a Canadian dead-zone
            // we won't notice for hours.
            results.push(joined.expect("hrdps fetch task panicked")?);
        }

        // Stable ordering by manifest index. wgrib2 doesn't care about
        // record order but deterministic concatenation makes byte-level
        // golden tests possible.
        results.sort_by_key(|(idx, _)| *idx);

        let total: usize = results.iter().map(|(_, b)| b.len()).sum();
        let mut out = Vec::with_capacity(total);
        for (_, bytes) in results {
            out.extend_from_slice(&bytes);
        }
        Ok(out)
    }
}

async fn retry_get(client: &Client, url: &str) -> Result<Vec<u8>, HrdpsFetchError> {
    for attempt in 0..5 {
        match client.get(url).send().await {
            Ok(resp) => {
                let status = resp.status().as_u16();
                if status == 200 {
                    let bytes = resp.bytes().await?;
                    if bytes.is_empty() {
                        return Err(HrdpsFetchError::EmptyBody {
                            url: url.to_string(),
                        });
                    }
                    return Ok(bytes.to_vec());
                }
                if !is_retryable_status(status) {
                    return Err(HrdpsFetchError::Status {
                        status,
                        url: url.to_string(),
                    });
                }
                tracing::warn!(status, attempt, url, "hrdps retryable status");
            }
            Err(e) if e.is_connect() || e.is_timeout() => {
                tracing::warn!(error = %e, attempt, url, "hrdps transient error");
            }
            Err(e) => return Err(HrdpsFetchError::Http(e)),
        }
        tokio::time::sleep(retry_backoff(attempt)).await;
    }
    Err(HrdpsFetchError::Status {
        status: 0,
        url: url.to_string(),
    })
}

fn is_retryable_status(status: u16) -> bool {
    matches!(status, 429 | 500 | 502 | 503 | 504)
}

fn retry_backoff(attempt: u32) -> Duration {
    let base_ms = 1000_u64 << attempt;
    let jitter = rand_jitter_ms();
    Duration::from_millis(base_ms + jitter)
}

// Tiny stdlib-only jitter — avoids pulling in `rand` for one call site.
// Returns 0..=999 ms.
fn rand_jitter_ms() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.subsec_nanos() as u64)
        .unwrap_or(0);
    nanos % 1_000
}

/// Snap a UTC timestamp to the start of the most recent HRDPS cycle hour
/// (00/06/12/18Z). Always rounds DOWN — never points at a future cycle
/// that may not have published yet.
pub fn nearest_hrdps_cycle(dt: DateTime<Utc>) -> DateTime<Utc> {
    let cycle_hour = (dt.hour() / 6) * 6;
    dt.with_hour(cycle_hour)
        .and_then(|d| d.with_minute(0))
        .and_then(|d| d.with_second(0))
        .and_then(|d| d.with_nanosecond(0))
        .unwrap_or(dt)
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::TimeZone;

    #[test]
    fn url_matches_live_datamart_format() {
        let cycle = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
        let url = hrdps_url(DATAMART_BASE_DEFAULT, cycle, 0, "TMP", "AGL-2m");
        assert_eq!(
            url,
            "https://dd.weather.gc.ca/20260429/WXO-DD/model_hrdps/continental/2.5km/12/000/\
             20260429T12Z_MSC_HRDPS_TMP_AGL-2m_RLatLon0.0225_PT000H.grib2"
        );
    }

    #[test]
    fn url_pads_forecast_hour_to_three_digits() {
        let cycle = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
        let url = hrdps_url(DATAMART_BASE_DEFAULT, cycle, 24, "TMP", "AGL-2m");
        assert!(url.contains("/024/"));
        assert!(url.contains("PT024H.grib2"));
    }

    #[test]
    fn url_pads_cycle_hour_to_two_digits() {
        let cycle = Utc.with_ymd_and_hms(2026, 4, 29, 6, 0, 0).unwrap();
        let url = hrdps_url(DATAMART_BASE_DEFAULT, cycle, 0, "TMP", "AGL-2m");
        assert!(url.contains("/06/000/"));
        assert!(url.contains("T06Z_"));
    }

    #[test]
    fn pressure_level_url_uses_isbl_padding() {
        let cycle = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
        let url = hrdps_url(DATAMART_BASE_DEFAULT, cycle, 0, "TMP", "ISBL_0850");
        assert!(url.contains("MSC_HRDPS_TMP_ISBL_0850_"));
    }

    #[test]
    fn manifest_covers_surface_and_pressure_set() {
        let manifest = variable_manifest();
        // 8 surface + 7 pressure levels × 3 vars (TMP/DEPR/HGT) = 29
        assert_eq!(manifest.len(), 29);
        assert!(manifest.iter().any(|(v, l)| *v == "TMP" && l == "AGL-2m"));
        assert!(manifest.iter().any(|(v, l)| *v == "TMP" && l == "ISBL_0850"));
        assert!(manifest.iter().any(|(v, l)| *v == "HGT" && l == "ISBL_1000"));
    }

    #[test]
    fn nearest_cycle_snaps_to_six_hour_boundary() {
        let dt = Utc.with_ymd_and_hms(2026, 4, 29, 13, 45, 0).unwrap();
        let cycle = nearest_hrdps_cycle(dt);
        assert_eq!(cycle, Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap());
    }

    #[test]
    fn nearest_cycle_rounds_down_at_boundary() {
        // Just inside the 06Z window; mustn't jump forward to 12Z.
        let dt = Utc.with_ymd_and_hms(2026, 4, 29, 11, 59, 0).unwrap();
        let cycle = nearest_hrdps_cycle(dt);
        assert_eq!(cycle, Utc.with_ymd_and_hms(2026, 4, 29, 6, 0, 0).unwrap());
    }

    #[test]
    fn nearest_cycle_idempotent_on_boundary() {
        let dt = Utc.with_ymd_and_hms(2026, 4, 29, 12, 0, 0).unwrap();
        assert_eq!(nearest_hrdps_cycle(dt), dt);
    }
}