prop/rust/prop_grid_rs/src/commercial.rs

//! Commercial-link degradation lookup.
//!
//! Port of `Microwaveprop.Commercial.build_link_lookup/2` +
//! `link_degradation_from_lookup/3`. The Elixir context stays intact for
//! LiveView sensors and SNMP poller; this Rust side is a read-only
//! consumer for the f00 chain step (Phase 3 Stream A).
//!
//! Commercial LOS microwave links act as an inverse tropo sensor: when
//! rx_power drops significantly below its 7-day baseline with the link
//! still in state=1, the refractivity structure hurting the Fresnel
//! zone is usually the same structure *helping* long beyond-LOS
//! amateur propagation. We report baseline/current/delta per grid cell
//! and let the scorer fold it in.

use sqlx::PgPool;
use uuid::Uuid;

const DEFAULT_RADIUS_KM: f64 = 75.0;
const DEFAULT_BASELINE_DAYS: i64 = 7;
const DEFAULT_CURRENT_WINDOW_SECS: i64 = 900;
const MIN_BASELINE_SAMPLES: i64 = 5;

#[derive(Debug, Clone)]
pub struct LinkLookupEntry {
    pub link_id: Uuid,
    pub endpoint: (f64, f64),
    pub baseline_dbm: f64,
    pub current_dbm: f64,
}

/// Aggregate degradation at one grid cell, matching the shape produced
/// by `Commercial.link_degradation_from_lookup/3`.
#[derive(Debug, Clone, Copy)]
pub struct Degradation {
    pub degradation_db: f64,
    pub baseline_dbm: f64,
    pub current_dbm: f64,
    pub n_links: u32,
}

#[derive(Debug, thiserror::Error)]
pub enum CommercialError {
    #[error(transparent)]
    Sqlx(#[from] sqlx::Error),
}

/// Precompute every enabled link's degradation once. Callers fold this
/// lookup across the ~92k CONUS cells. Links whose current/baseline
/// samples don't pass the quality gate (`MIN_BASELINE_SAMPLES`, both
/// sides non-null) are dropped so the per-cell path stays simple.
pub async fn build_link_lookup(
    pool: &PgPool,
    valid_time: chrono::DateTime<chrono::Utc>,
) -> Result<Vec<LinkLookupEntry>, CommercialError> {
    let baseline_cutoff = valid_time - chrono::Duration::days(DEFAULT_BASELINE_DAYS);
    let current_cutoff = valid_time - chrono::Duration::seconds(DEFAULT_CURRENT_WINDOW_SECS);

    // Pull enabled links with a usable endpoint_a. The JSONB → (lat, lon)
    // dereference mirrors `link_endpoint/1` in Elixir: the schema
    // convention is that endpoint_a holds a map with "lat" and "lon"
    // keys. Failures of that assumption are dropped rather than
    // surfaced — a malformed endpoint is a data-quality issue, not a
    // chain-blocker.
    let link_rows = sqlx::query_as::<_, (Uuid, serde_json::Value)>(
        r#"
        SELECT id, endpoint_a
        FROM commercial_links
        WHERE enabled = true
        "#,
    )
    .fetch_all(pool)
    .await?;

    let mut out = Vec::with_capacity(link_rows.len());
    for (link_id, endpoint_json) in link_rows {
        let Some(endpoint) = extract_endpoint(&endpoint_json) else {
            continue;
        };
        if let Some(deg) =
            fetch_per_link_degradation(pool, link_id, baseline_cutoff, current_cutoff, valid_time)
                .await?
        {
            out.push(LinkLookupEntry {
                link_id,
                endpoint,
                baseline_dbm: deg.0,
                current_dbm: deg.1,
            });
        }
    }
    Ok(out)
}

fn extract_endpoint(v: &serde_json::Value) -> Option<(f64, f64)> {
    let lat = v.get("lat")?.as_f64()?;
    let lon = v.get("lon")?.as_f64()?;
    Some((lat, lon))
}

async fn fetch_per_link_degradation(
    pool: &PgPool,
    link_id: Uuid,
    baseline_cutoff: chrono::DateTime<chrono::Utc>,
    current_cutoff: chrono::DateTime<chrono::Utc>,
    valid_time: chrono::DateTime<chrono::Utc>,
) -> Result<Option<(f64, f64)>, CommercialError> {
    // Baseline: average rx_power_0 over the 7-day window ending at
    // current_cutoff. sqlx decodes AVG(numeric) as Option<BigDecimal>
    // by default; reading it as f64 keeps the downstream math simple.
    let baseline: Option<f64> = sqlx::query_scalar(
        r#"
        SELECT AVG(rx_power_0)::float8
        FROM commercial_samples
        WHERE link_id = $1
          AND sampled_at >= $2
          AND sampled_at <  $3
          AND link_state = 1
          AND rx_power_0 IS NOT NULL
        "#,
    )
    .bind(link_id)
    .bind(baseline_cutoff.naive_utc())
    .bind(current_cutoff.naive_utc())
    .fetch_one(pool)
    .await?;

    let current: Option<f64> = sqlx::query_scalar(
        r#"
        SELECT rx_power_0::float8
        FROM commercial_samples
        WHERE link_id = $1
          AND sampled_at >= $2
          AND sampled_at <= $3
          AND link_state = 1
          AND rx_power_0 IS NOT NULL
        ORDER BY sampled_at DESC
        LIMIT 1
        "#,
    )
    .bind(link_id)
    .bind(current_cutoff.naive_utc())
    .bind(valid_time.naive_utc())
    .fetch_optional(pool)
    .await?
    .flatten();

    let baseline_n: i64 = sqlx::query_scalar(
        r#"
        SELECT COUNT(id)
        FROM commercial_samples
        WHERE link_id = $1
          AND sampled_at < $2
          AND link_state = 1
          AND rx_power_0 IS NOT NULL
        "#,
    )
    .bind(link_id)
    .bind(current_cutoff.naive_utc())
    .fetch_one(pool)
    .await?;

    match (baseline, current) {
        (Some(b), Some(c)) if baseline_n >= MIN_BASELINE_SAMPLES => Ok(Some((b, c))),
        _ => Ok(None),
    }
}

/// Aggregate degradation at a single grid cell given a precomputed
/// link lookup. `None` when no link is in range.
pub fn degradation_at(cell: (f64, f64), lookup: &[LinkLookupEntry]) -> Option<Degradation> {
    let in_range: Vec<&LinkLookupEntry> = lookup
        .iter()
        .filter(|e| haversine_km(cell.0, cell.1, e.endpoint.0, e.endpoint.1) <= DEFAULT_RADIUS_KM)
        .collect();
    if in_range.is_empty() {
        return None;
    }
    let n = in_range.len() as f64;
    let baseline_avg: f64 = in_range.iter().map(|e| e.baseline_dbm).sum::<f64>() / n;
    let current_avg: f64 = in_range.iter().map(|e| e.current_dbm).sum::<f64>() / n;
    Some(Degradation {
        degradation_db: round2(baseline_avg - current_avg),
        baseline_dbm: round2(baseline_avg),
        current_dbm: round2(current_avg),
        n_links: in_range.len() as u32,
    })
}

fn haversine_km(lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 {
    let r = 6371.0_f64;
    let rad = std::f64::consts::PI / 180.0;
    let phi1 = lat1 * rad;
    let phi2 = lat2 * rad;
    let dphi = (lat2 - lat1) * rad;
    let dlambda = (lon2 - lon1) * rad;
    let a = (dphi / 2.0).sin().powi(2) + phi1.cos() * phi2.cos() * (dlambda / 2.0).sin().powi(2);
    2.0 * r * a.sqrt().atan2((1.0 - a).sqrt())
}

fn round2(v: f64) -> f64 {
    (v * 100.0).round() / 100.0
}

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

    fn entry(n: u8, lat: f64, lon: f64, baseline: f64, current: f64) -> LinkLookupEntry {
        // Tests don't care about the actual UUID value, only that each entry
        // has a distinct identity — a fixed-byte-0 pattern is plenty.
        let mut bytes = [0u8; 16];
        bytes[0] = n;
        LinkLookupEntry {
            link_id: Uuid::from_bytes(bytes),
            endpoint: (lat, lon),
            baseline_dbm: baseline,
            current_dbm: current,
        }
    }

    #[test]
    fn haversine_known_distances() {
        // DFW (32.9°N) → Austin (30.3°N): ~300 km (Wikipedia lists ~304 km
        // road, great-circle is ~297 km).
        let d = haversine_km(32.8998, -97.0403, 30.2672, -97.7431);
        assert!((d - 300.0).abs() < 10.0, "got {d}");
        // Same point → 0
        assert!(haversine_km(32.0, -97.0, 32.0, -97.0).abs() < 1e-9);
    }

    #[test]
    fn no_links_in_range_returns_none() {
        let lookup = vec![entry(1, 32.9, -97.0, -50.0, -52.0)];
        // Austin cell is ~290 km from a DFW link → out of 75 km radius.
        let d = degradation_at((30.27, -97.74), &lookup);
        assert!(d.is_none());
    }

    #[test]
    fn aggregates_within_radius() {
        let lookup = vec![
            entry(1, 32.9, -97.0, -50.0, -54.0), // 4 dB drop
            entry(2, 33.0, -97.1, -50.0, -52.0), // 2 dB drop
        ];
        let d = degradation_at((32.95, -97.05), &lookup).unwrap();
        assert_eq!(d.n_links, 2);
        assert!((d.baseline_dbm - -50.0).abs() < 1e-9);
        assert!((d.current_dbm - -53.0).abs() < 1e-9);
        assert!((d.degradation_db - 3.0).abs() < 1e-9);
    }

    #[test]
    fn degradation_rounds_to_two_decimals() {
        let lookup = vec![entry(1, 32.9, -97.0, -50.12345, -52.67890)];
        let d = degradation_at((32.9, -97.0), &lookup).unwrap();
        assert_eq!(d.baseline_dbm, -50.12);
        assert_eq!(d.current_dbm, -52.68);
        assert_eq!(d.degradation_db, 2.56);
    }

    #[test]
    fn extract_endpoint_handles_missing_keys() {
        assert!(extract_endpoint(&serde_json::json!({})).is_none());
        assert!(extract_endpoint(&serde_json::json!({"lat": 32.0})).is_none());
        assert!(extract_endpoint(&serde_json::json!({"lon": -97.0})).is_none());
        assert_eq!(
            extract_endpoint(&serde_json::json!({"lat": 32.0, "lon": -97.0})),
            Some((32.0, -97.0))
        );
    }
}