prop/rust/prop_grid_rs/src/profiles_file.rs

//! On-disk profile store in MessagePack + gzip.
//!
//! Rust-side writer for the f00 ProfilesFile. Replaces the Elixir ETF
//! format (`:erlang.term_to_binary/2`, `.etf.gz`) so Rust can produce
//! the files without porting the Erlang serializer. Elixir's
//! `ProfilesFile` module learns to read `.mp.gz` alongside the legacy
//! `.etf.gz` during the cutover.
//!
//! Wire layout (top-level msgpack map):
//!
//! ```text
//! {
//!   "v":          1,
//!   "valid_time": "2026-04-19T14:00:00Z",
//!   "cells": [
//!     {
//!       "lat": 32.9, "lon": -97.0,
//!       "profile": {
//!         "native_min_gradient": -150.5,
//!         "best_duct_freq_ghz":  24.0,
//!         "duct_count":          1,
//!         "ducts": [
//!           {"base_m": 100, "top_m": 200, "thickness_m": 100,
//!            "m_deficit": 12.3, "min_freq_ghz": 15.2}
//!         ],
//!         "surface_temp_c":     22.5,
//!         "surface_dewpoint_c": 18.1,
//!         "surface_pressure_mb": 1013.2,
//!         "hpbl_m":  320.0,
//!         "pwat_mm": 25.3,
//!         "profile": [
//!           {"pres_mb":1000,"hght_m":100,"tmpc":20.0,"dwpc":18.0},
//!           ...
//!         ]
//!       }
//!     }
//!   ]
//! }
//! ```
//!
//! All keys are strings — MessagePack has no atom type, and Elixir's
//! reader converts known keys back to atoms via a whitelist.

use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};

use chrono::{DateTime, Utc};
use flate2::write::GzEncoder;
use flate2::Compression;
use serde::Serialize;

/// Bump whenever the on-disk layout changes in a way that isn't a
/// transparent superset. Elixir reader gates on this.
pub const FORMAT_VERSION: u32 = 1;

/// One grid cell to persist. `profile` is an arbitrary map of
/// string-keyed values — sized at write time. Kept as a
/// `rmp_serde::Value` so the writer stays agnostic about what the
/// f00 pipeline decides to include.
#[derive(Debug, Serialize)]
pub struct CellEntry {
    pub lat: f64,
    pub lon: f64,
    pub profile: rmpv::Value,
}

#[derive(Debug, Serialize)]
struct FileBody<'a> {
    v: u32,
    valid_time: String,
    cells: &'a [CellEntry],
}

#[derive(Debug, thiserror::Error)]
pub enum WriteError {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("encode: {0}")]
    Encode(#[from] rmp_serde::encode::Error),
}

/// Absolute path the writer lands at for `valid_time`.
/// Sibling of the Elixir `.etf.gz` file; the Elixir reader prefers
/// `.mp.gz` when both exist.
pub fn path_for(scores_dir: &Path, valid_time: DateTime<Utc>) -> PathBuf {
    let iso = valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string();
    scores_dir.join("profiles").join(format!("{iso}.mp.gz"))
}

/// Write `cells` to `<scores_dir>/profiles/<iso>.mp.gz` atomically.
/// Pattern mirrors `scores_file::write_atomic`: write to a tmp sibling,
/// rename into place. The NFS rename is atomic, so a concurrent reader
/// sees either the old file, the new file, or nothing — never a torn
/// write.
pub fn write_atomic(
    scores_dir: &Path,
    valid_time: DateTime<Utc>,
    cells: &[CellEntry],
) -> Result<PathBuf, WriteError> {
    let path = path_for(scores_dir, valid_time);
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let pid = std::process::id();
    let mut tmp = path.clone().into_os_string();
    tmp.push(format!(".tmp.{nanos}.{pid}"));
    let tmp = PathBuf::from(tmp);

    {
        let file = File::create(&tmp)?;
        // ProfilesFile is a write-once / read-many blob — each hourly
        // cycle writes it ~once but every LiveView mount, point_detail
        // click, and forecast scrub pays the read cost (up to ~3 MB
        // uncompressed per file × ~5 cached hours per pod). Level 9 is
        // ~30% slower to encode than the default (6) but 5–10% smaller
        // on semi-structured MessagePack, and our write path already
        // off-loads to `spawn_blocking` so encode latency isn't on the
        // user path. Readers are level-agnostic — same gzip stream.
        let mut gz = GzEncoder::new(BufWriter::new(file), Compression::best());
        let body = FileBody {
            v: FORMAT_VERSION,
            valid_time: valid_time.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
            cells,
        };
        let buf = rmp_serde::to_vec_named(&body)?;
        gz.write_all(&buf)?;
        gz.finish()?;
    }

    std::fs::rename(&tmp, &path)?;
    Ok(path)
}

/// Snap a lat/lon to the propagation grid step. Mirrors Elixir's
/// `Microwaveprop.Propagation.ProfilesFile.snap/2`:
///
/// ```elixir
/// Float.round(Float.round(coord / step) * step, 3)
/// ```
///
/// Two implementations of "snap" — Rust's old plain 3-decimal round
/// and Elixir's step-aware round — disagreed on every off-grid input,
/// so a coordinate written by Elixir was unreachable from Rust and
/// vice versa.
const GRID_STEP: f64 = 0.125;

pub fn snap_coords(lat: f64, lon: f64) -> (f64, f64) {
    (snap_step(lat), snap_step(lon))
}

fn snap_step(coord: f64) -> f64 {
    let snapped = (coord / GRID_STEP).round() * GRID_STEP;
    (snapped * 1000.0).round() / 1000.0
}

/// Convenience helper: build a `rmpv::Value` map from an
/// iterator of `(name, value)` pairs.
pub fn value_map(entries: impl IntoIterator<Item = (&'static str, rmpv::Value)>) -> rmpv::Value {
    let pairs: Vec<(rmpv::Value, rmpv::Value)> = entries
        .into_iter()
        .map(|(k, v)| (rmpv::Value::String(k.into()), v))
        .collect();
    rmpv::Value::Map(pairs)
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::TimeZone;
    use flate2::read::GzDecoder;
    use std::io::Read;

    #[test]
    fn writes_and_reads_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap();
        let cells = vec![CellEntry {
            lat: 32.9,
            lon: -97.0,
            profile: value_map([
                ("surface_temp_c", rmpv::Value::F64(22.5)),
                ("surface_pressure_mb", rmpv::Value::F64(1013.2)),
                ("duct_count", rmpv::Value::Integer(1.into())),
            ]),
        }];

        let path = write_atomic(dir.path(), vt, &cells).unwrap();
        assert!(path.exists());
        assert!(path.to_string_lossy().ends_with(".mp.gz"));

        // Decode and confirm the top-level shape.
        let raw = std::fs::read(&path).unwrap();
        let mut gz = GzDecoder::new(&raw[..]);
        let mut buf = Vec::new();
        gz.read_to_end(&mut buf).unwrap();
        let decoded: rmpv::Value = rmp_serde::from_slice(&buf).unwrap();
        let map = decoded.as_map().unwrap();
        assert!(map
            .iter()
            .any(|(k, v)| k.as_str() == Some("v") && v.as_u64() == Some(1)));
        assert!(map
            .iter()
            .any(|(k, v)| k.as_str() == Some("valid_time")
                && v.as_str() == Some("2026-04-19T14:00:00Z")));
        let cells_val = &map
            .iter()
            .find(|(k, _)| k.as_str() == Some("cells"))
            .unwrap()
            .1;
        let cells_arr = cells_val.as_array().unwrap();
        assert_eq!(cells_arr.len(), 1);
    }

    #[test]
    fn atomic_rename_replaces_existing() {
        let dir = tempfile::tempdir().unwrap();
        let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap();
        let first = vec![CellEntry {
            lat: 32.0,
            lon: -97.0,
            profile: value_map([("v", rmpv::Value::Integer(1.into()))]),
        }];
        let second = vec![CellEntry {
            lat: 32.0,
            lon: -97.0,
            profile: value_map([("v", rmpv::Value::Integer(2.into()))]),
        }];
        let p1 = write_atomic(dir.path(), vt, &first).unwrap();
        let p2 = write_atomic(dir.path(), vt, &second).unwrap();
        assert_eq!(p1, p2);
        // Only one file in the profiles dir — the tmp was renamed away.
        let entries: Vec<_> = std::fs::read_dir(dir.path().join("profiles"))
            .unwrap()
            .map(|e| e.unwrap().file_name().into_string().unwrap())
            .collect();
        assert_eq!(entries.len(), 1);
    }

    #[test]
    fn path_for_layout() {
        let vt = Utc.with_ymd_and_hms(2026, 4, 19, 14, 0, 0).unwrap();
        let p = path_for(std::path::Path::new("/data/scores"), vt);
        assert_eq!(
            p.to_str().unwrap(),
            "/data/scores/profiles/2026-04-19T14:00:00Z.mp.gz"
        );
    }

    #[test]
    fn snap_coords_matches_elixir_step_snap() {
        // The Elixir reader writes the profiles map with keys snapped via
        // `Float.round(Float.round(coord / 0.125) * 0.125, 3)`. This Rust
        // helper has to produce the same keys or `read_point` looks up a
        // coordinate that was never written and the UI shows "no data".
        assert_eq!(snap_coords(32.8998, -97.04031), (32.875, -97.0));
        assert_eq!(snap_coords(32.07, -97.07), (32.125, -97.125));
        // Round-trip — already-snapped coordinates stay put.
        assert_eq!(snap_coords(32.875, -97.0), (32.875, -97.0));
    }
}