//! Database tests against a real PostgreSQL. Each test runs in its own
//! throwaway schema. Set TEST_DATABASE_URL to run them, e.g.
//! `postgres://postgres:ping@localhost:55432/ping`; without it they are
//! skipped (reported as passed with a note on stderr).

use crate::dams::DamRecord;
use crate::db;
use crate::hii::{RainRecord, WaterLevelRecord};
use crate::rain::RainRow;
use crate::rid::{self, Reading, RidSource};
use crate::stations::Stations;
use anyhow::Result;
use chrono::{Duration, NaiveDate, NaiveDateTime, NaiveTime};
use sqlx::PgPool;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Mutex;

pub(crate) struct TestDb {
    pub(crate) pool: PgPool,
    admin: PgPool,
    schema: String,
}

impl TestDb {
    pub(crate) async fn new() -> Option<Self> {
        let Ok(url) = std::env::var("TEST_DATABASE_URL") else {
            eprintln!("TEST_DATABASE_URL not set: skipping database test");
            return None;
        };
        // the counter: tests run in parallel and the clock is coarse on Windows
        static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
        let schema = format!(
            "t_{}_{}_{}",
            std::process::id(),
            chrono::Utc::now().timestamp_nanos_opt().unwrap(),
            N.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        );
        let admin = PgPool::connect(&url).await.expect("connect");
        sqlx::query(&format!("CREATE SCHEMA {schema}"))
            .execute(&admin)
            .await
            .unwrap();
        let opts = PgConnectOptions::from_str(&url)
            .unwrap()
            .options([("search_path", schema.as_str())]);
        let pool = PgPoolOptions::new()
            .max_connections(2)
            .connect_with(opts)
            .await
            .unwrap();
        db::ensure_schema(&pool).await.expect("schema");
        Some(TestDb {
            pool,
            admin,
            schema,
        })
    }

    pub(crate) async fn drop(self) {
        self.pool.close().await;
        sqlx::query(&format!("DROP SCHEMA {} CASCADE", self.schema))
            .execute(&self.admin)
            .await
            .unwrap();
    }
}

fn d(y: i32, m: u32, day: u32) -> NaiveDate {
    NaiveDate::from_ymd_opt(y, m, day).unwrap()
}
fn t(y: i32, m: u32, day: u32, h: u32, min: u32) -> NaiveDateTime {
    d(y, m, day).and_hms_opt(h, min, 0).unwrap()
}

/// Canned RID: date -> last published hour (1..=24); P.1 (column 8) and P.20
/// (column 1). Records which reports were requested.
struct FakeRid {
    published: HashMap<NaiveDate, i64>,
    calls: Mutex<Vec<NaiveDate>>,
}

impl FakeRid {
    fn new(published: &[(NaiveDate, i64)]) -> Self {
        FakeRid {
            published: published.iter().copied().collect(),
            calls: Mutex::new(Vec::new()),
        }
    }
    fn calls(&self) -> Vec<NaiveDate> {
        self.calls.lock().unwrap().clone()
    }
}

impl RidSource for FakeRid {
    async fn fetch_report(&self, date: NaiveDate) -> Result<Vec<Reading>> {
        self.calls.lock().unwrap().push(date);
        let last = self.published.get(&date).copied().unwrap_or(0);
        Ok((1..=last)
            .flat_map(|h| {
                let ts = date.and_time(NaiveTime::MIN) + Duration::hours(h);
                [8u32, 1].map(|n| Reading {
                    timestamp: ts,
                    station_num: n,
                    water_level: 1.0 + h as f64 / 100.0,
                    discharge: Some(50.0),
                    discharge_percent: Some(12.5),
                })
            })
            .collect())
    }
}

async fn stored_hours(pool: &PgPool, day: NaiveDate) -> Vec<u32> {
    let rec = db::recorded_hours_by_day(pool, day, day).await.unwrap();
    let mut h: Vec<u32> = rec
        .get(&day)
        .cloned()
        .unwrap_or_default()
        .into_iter()
        .collect();
    h.sort();
    h
}

#[tokio::test]
async fn late_report_is_stored_and_reported_behind() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let today = d(2026, 10, 17);
    let yesterday = today - Duration::days(1);
    let stations = Stations::bundled();
    // yesterday complete already
    let full = FakeRid::new(&[(yesterday, 24)]);
    rid::fill_gaps(&full, &db.pool, &stations, Some(1), t(2026, 10, 17, 0, 30))
        .await
        .unwrap();

    // 21:30 and RID has published up to 20:00
    let src = FakeRid::new(&[(today, 20)]);
    let o = rid::run_cycle(&src, &db.pool, &stations, t(2026, 10, 17, 21, 30)).await;
    assert!(!o.current);
    assert_eq!(o.latest, Some(t(2026, 10, 17, 20, 0)));
    assert_eq!(o.saved, 40);
    assert_eq!(src.calls(), vec![today]);
    assert_eq!(
        stored_hours(&db.pool, today).await,
        (0..=20).collect::<Vec<_>>()
    );

    // the 21:00 reading lands; the next poll is current
    let src = FakeRid::new(&[(today, 21)]);
    let o = rid::run_cycle(&src, &db.pool, &stations, t(2026, 10, 17, 21, 40)).await;
    assert!(o.current);
    db.drop().await;
}

#[tokio::test]
async fn yesterdays_late_hours_are_recovered_after_midnight() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let today = d(2026, 10, 17);
    let yesterday = today - Duration::days(1);
    let stations = Stations::bundled();
    // stored up to 21:00 yesterday (the evening RID published late)
    let partial = FakeRid::new(&[(yesterday, 21)]);
    rid::run_cycle(&partial, &db.pool, &stations, t(2026, 10, 16, 21, 5)).await;
    assert_eq!(
        stored_hours(&db.pool, yesterday).await,
        (1..=21).collect::<Vec<_>>()
    );

    // 02:10: RID now has all of yesterday and today up to 02:00
    let src = FakeRid::new(&[(yesterday, 24), (today, 2)]);
    let o = rid::run_cycle(&src, &db.pool, &stations, t(2026, 10, 17, 2, 10)).await;
    assert!(o.current);
    assert_eq!(src.calls(), vec![yesterday, today]);
    assert_eq!(
        stored_hours(&db.pool, yesterday).await,
        (1..=23).collect::<Vec<_>>()
    );
    assert_eq!(stored_hours(&db.pool, today).await, vec![0, 1, 2]);

    // complete now: the next cycle fetches only today's report
    let src = FakeRid::new(&[(yesterday, 24), (today, 3)]);
    rid::run_cycle(&src, &db.pool, &stations, t(2026, 10, 17, 3, 5)).await;
    assert_eq!(src.calls(), vec![today]);
    db.drop().await;
}

#[tokio::test]
async fn rid_upsert_revises_values_and_rounds_like_numeric() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let stations = Stations::bundled();
    let r = |level: f64| Reading {
        timestamp: t(2026, 10, 17, 9, 0),
        station_num: 8,
        water_level: level,
        discharge: None,
        discharge_percent: None,
    };
    db::save_rid(&db.pool, &[r(1.0700000524520874)], &stations)
        .await
        .unwrap();
    db::save_rid(&db.pool, &[r(1.0900000333786)], &stations)
        .await
        .unwrap();
    let rows: Vec<(String, f64, Option<f64>, String)> = sqlx::query_as(
        "SELECT s.station_code, w.water_level::float8, w.discharge::float8, w.status
         FROM water_measurements w JOIN stations s ON s.id = w.station_id",
    )
    .fetch_all(&db.pool)
    .await
    .unwrap();
    assert_eq!(rows, vec![("P.1".into(), 1.09, None, "active".into())]);
    db.drop().await;
}

#[tokio::test]
async fn hii_dams_and_rain_round_trip() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let now = t(2026, 10, 17, 21, 5);
    let rain = RainRecord {
        station_id: 11,
        oldcode: Some("CHM001".into()),
        name_th: None,
        name_en: Some("Chiang Mai 1".into()),
        latitude: Some(18.79),
        longitude: Some(99.0),
        sub_basin_id: Some("601".into()),
        agency: Some("HII".into()),
        timestamp: t(2026, 10, 17, 21, 0),
        rain_1h: Some(1.234),
        rain_24h: Some(12.0),
    };
    assert_eq!(
        db::save_hii_rain(&db.pool, &[rain.clone(), rain.clone()], now)
            .await
            .unwrap(),
        1
    );
    let (r1h,): (f64,) = sqlx::query_as("SELECT rain_1h::float8 FROM hii_rainfall")
        .fetch_one(&db.pool)
        .await
        .unwrap();
    assert_eq!(r1h, 1.23);

    let wl = WaterLevelRecord {
        station_id: 5,
        oldcode: Some("ridhydro_P.1".into()),
        rid_code: Some("P.1".into()),
        name_th: Some("สะพานนวรัฐ".into()),
        name_en: None,
        latitude: None,
        longitude: None,
        agency: None,
        river_name: None,
        offset_msl: Some(300.0),
        ground_level_msl: None,
        min_bank_msl: None,
        critical_level_msl: None,
        critical_level_m: None,
        qmax: None,
        is_key_station: true,
        timestamp: t(2026, 10, 17, 21, 0),
        wl_msl: Some(301.2),
        wl_m: None,
        discharge: None,
        flow_rate: None,
        storage_percent: None,
        situation_level: Some(2),
        diff_wl_bank: None,
    };
    db::save_hii_waterlevel(&db.pool, &[wl], now).await.unwrap();
    let (code, key, sit): (String, bool, i32) = sqlx::query_as(
        "SELECT s.rid_code, s.is_key_station, w.situation_level
         FROM hii_waterlevel w JOIN hii_wl_stations s ON s.id = w.station_id",
    )
    .fetch_one(&db.pool)
    .await
    .unwrap();
    assert_eq!((code.as_str(), key, sit), ("P.1", true, 2));

    let dam = |name: Option<&str>, level: Option<f64>| DamRecord {
        dam_id: "200103".into(),
        region: Some("north".into()),
        name_th: name.map(str::to_string),
        latitude: Some(19.17),
        longitude: None,
        capacity_max_mcm: Some(265.0),
        capacity_normal_mcm: None,
        date: d(2026, 10, 17),
        storage_mcm: Some(200.0),
        storage_pct: Some(75.5),
        inflow_mcm: None,
        outflow_mcm: Some(1.2),
        level_msl: level,
    };
    db::save_dams(&db.pool, &[dam(Some("แม่งัด"), Some(395.1))], now)
        .await
        .unwrap();
    // a later snapshot without name / level must not blank them
    db::save_dams(&db.pool, &[dam(None, None)], now)
        .await
        .unwrap();
    let (name, level): (Option<String>, Option<f64>) = sqlx::query_as(
        "SELECT d.name_th, r.level_msl::float8 FROM rid_reservoir_daily r JOIN rid_dams d USING (dam_id)",
    )
    .fetch_one(&db.pool)
    .await
    .unwrap();
    assert_eq!(name.as_deref(), Some("แม่งัด"));
    assert_eq!(level, Some(395.1));

    let rows = vec![RainRow {
        timestamp: t(2026, 10, 18, 3, 0),
        points: [Some(1.0), None, Some(2.0), None, Some(0.333)],
        catchment_mean: Some(1.111),
    }];
    db::save_rain(&db.pool, &rows).await.unwrap();
    let (mean, missing): (f64, Option<f64>) =
        sqlx::query_as("SELECT catchment_mean::float8, mae_taeng::float8 FROM openmeteo_rain")
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert_eq!((mean, missing), (1.11, None));
    db.drop().await;
}

fn forecast_row(code: &str, h: u32, p: f64) -> ping_forecast::Forecast {
    ping_forecast::Forecast {
        station_code: code.into(),
        horizon_hours: h,
        p_warning: p,
        p_danger: p / 2.0,
        predicted_max_level: 3.123456,
        current_level: 1.23,
        as_of: t(2026, 10, 18, 13, 0),
        model_version: "hgb-v3+test".into(),
        trained_at: Some("2026-10-01T03:30:00".into()),
        source: "model".into(),
        threshold_warning: 3.7,
        threshold_danger: 4.2,
        stages: None,
    }
}

#[tokio::test]
async fn forecast_outputs_round_trip() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    // 0.123455 + 1 ulp: Python's repr gives 17 digits and NUMERIC(7,5)
    // rounds that up; a float8 -> numeric cast (15 digits) would round down
    let tricky = f64::from_bits(0.123455f64.to_bits() + 1);
    let rows = [
        forecast_row("P.1", 24, tricky),
        forecast_row("P.1", 24, 0.2),
        forecast_row("P.20", 6, 1e-7),
    ];
    assert_eq!(db::save_forecast_history(&db.pool, &rows).await.unwrap(), 2);
    db::save_forecast_history(&db.pool, &rows[..1])
        .await
        .unwrap();
    let got: Vec<(String, i32, f64, f64, f64)> = sqlx::query_as(
        "SELECT station_code, horizon_hours, p_warning::float8, predicted_max_level::float8,
                current_level::float8 FROM forecast_history ORDER BY station_code",
    )
    .fetch_all(&db.pool)
    .await
    .unwrap();
    assert_eq!(
        got,
        vec![
            ("P.1".into(), 24, 0.12346, 3.123, 1.23),
            ("P.20".into(), 6, 0.0, 3.123, 1.23)
        ]
    );

    let now = t(2026, 10, 18, 13, 7);
    db::save_forecast_current(&db.pool, now, &rows[1..])
        .await
        .unwrap();
    db::save_forecast_current(&db.pool, now, &rows[..1])
        .await
        .unwrap();
    let (at, versions, payload): (NaiveDateTime, String, String) =
        sqlx::query_as("SELECT computed_at, model_versions, payload FROM forecast_current")
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert_eq!((at, versions.as_str()), (now, "hgb-v3+test"));
    let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
    assert_eq!(v[0]["as_of"], "2026-10-18T13:00:00");
    assert_eq!(v[0]["p_warning"].as_f64(), Some(tricky));
    assert!(v[0].get("stages").is_none());
    db.drop().await;
}

#[tokio::test]
async fn heuristic_never_replaces_a_model_forecast() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    // overnight the run re-issues the 21:00 reading, after 6 h as heuristic
    let model = forecast_row("P.1", 24, 0.2);
    let heuristic = ping_forecast::Forecast {
        p_warning: 0.0,
        predicted_max_level: 1.23,
        model_version: "heuristic-v1".into(),
        source: "heuristic".into(),
        ..forecast_row("P.1", 24, 0.0)
    };
    let stored = async || -> (String, f64) {
        sqlx::query_as("SELECT source::text, predicted_max_level::float8 FROM forecast_history")
            .fetch_one(&db.pool)
            .await
            .unwrap()
    };
    db::save_forecast_history(&db.pool, std::slice::from_ref(&model))
        .await
        .unwrap();
    db::save_forecast_history(&db.pool, std::slice::from_ref(&heuristic))
        .await
        .unwrap();
    assert_eq!(stored().await, ("model".into(), 3.123));
    // a model row does replace a heuristic one (and a newer model row)
    sqlx::query("DELETE FROM forecast_history")
        .execute(&db.pool)
        .await
        .unwrap();
    db::save_forecast_history(&db.pool, &[heuristic])
        .await
        .unwrap();
    assert_eq!(stored().await, ("heuristic".into(), 1.23));
    db::save_forecast_history(&db.pool, &[model]).await.unwrap();
    assert_eq!(stored().await, ("model".into(), 3.123));
    db.drop().await;
}

#[tokio::test]
async fn hours_without_a_model_forecast() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let stations = Stations::bundled();
    let r = |h: u32, num: u32| Reading {
        timestamp: t(2026, 10, 18, h, 0),
        station_num: num, // 8 = P.1, 1 = P.20
        water_level: 1.1,
        discharge: None,
        discharge_percent: None,
    };
    // RID's morning batch: 11..13 arrive at once; only 13:00 (the newest)
    // and P.20's 11:00 were forecast, and 12:00 only by the heuristic
    db::save_rid(
        &db.pool,
        &[r(11, 8), r(12, 8), r(13, 8), r(11, 1), r(12, 1), r(13, 1)],
        &stations,
    )
    .await
    .unwrap();
    let issued = |code: &str, h: u32, source: &str| ping_forecast::Forecast {
        as_of: t(2026, 10, 18, h, 0),
        source: source.into(),
        ..forecast_row(code, 24, 0.1)
    };
    db::save_forecast_history(
        &db.pool,
        &[
            issued("P.1", 13, "model"),
            issued("P.20", 13, "model"),
            issued("P.20", 11, "model"),
            issued("P.1", 12, "heuristic"),
        ],
    )
    .await
    .unwrap();
    let codes = vec!["P.1".to_string(), "P.20".to_string()];
    let missing = db::hours_without_model_forecast(
        &db.pool,
        &codes,
        t(2026, 10, 18, 0, 0),
        t(2026, 10, 18, 14, 0),
    )
    .await
    .unwrap();
    let at = |h: u32, code: &str| (t(2026, 10, 18, h, 0), code.to_string());
    assert_eq!(missing, vec![at(11, "P.1"), at(12, "P.1"), at(12, "P.20")]);
    // stations without a model are not asked about
    let only_p20 = db::hours_without_model_forecast(
        &db.pool,
        &codes[1..],
        t(2026, 10, 18, 0, 0),
        t(2026, 10, 18, 14, 0),
    )
    .await
    .unwrap();
    assert_eq!(only_p20, vec![at(12, "P.20")]);
    db.drop().await;
}

#[tokio::test]
async fn notification_state_round_trip() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let now = t(2026, 10, 18, 2, 0);
    db::set_notification_state(&db.pool, "level:P.1", "warning", Some(3.7512), now)
        .await
        .unwrap();
    db::set_notification_state(&db.pool, "feed", "stale", None, now)
        .await
        .unwrap();
    db::set_notification_state(&db.pool, "level:P.1", "clear", Some(3.5), now)
        .await
        .unwrap();
    let states = db::notification_states(&db.pool).await.unwrap();
    assert_eq!(states.len(), 2);
    assert_eq!(states["level:P.1"], "clear");
    assert_eq!(states["feed"], "stale");
    db.drop().await;
}

#[tokio::test]
async fn forecast_inputs_load_like_python() {
    let Some(db) = TestDb::new().await else {
        return;
    };
    let stations = Stations::bundled();
    let r = |h: u32, level: f64| Reading {
        timestamp: t(2026, 10, 18, h, 0),
        station_num: 8, // P.1
        water_level: level,
        discharge: Some(51.25),
        discharge_percent: Some(40.5),
    };
    db::save_rid(&db.pool, &[r(10, 1.1), r(11, 1.2), r(12, 1.3)], &stations)
        .await
        .unwrap();
    let rows = db::load_rid_window(&db.pool, t(2026, 10, 18, 11, 0), t(2026, 10, 18, 12, 30))
        .await
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(
        (rows[0].station_code.as_str(), rows[0].water_level),
        ("P.1", 1.2)
    );
    assert_eq!(rows[1].discharge, 51.25);

    let latest = db::latest_rid_readings(&db.pool).await.unwrap();
    assert_eq!(latest.len(), 1);
    assert_eq!(
        (
            latest[0].timestamp,
            latest[0].water_level,
            latest[0].discharge_percent
        ),
        (t(2026, 10, 18, 12, 0), Some(1.3), Some(40.5))
    );

    let wl = |minute: u32, v: f64| WaterLevelRecord {
        station_id: 5,
        oldcode: None,
        rid_code: Some("P.1".into()),
        name_th: None,
        name_en: None,
        latitude: None,
        longitude: None,
        agency: None,
        river_name: None,
        offset_msl: None,
        ground_level_msl: None,
        min_bank_msl: None,
        critical_level_msl: None,
        critical_level_m: None,
        qmax: None,
        is_key_station: true,
        timestamp: t(2026, 10, 18, 13, minute),
        wl_msl: Some(v),
        wl_m: None,
        discharge: Some(52.0),
        flow_rate: None,
        storage_percent: None,
        situation_level: None,
        diff_wl_bank: None,
    };
    db::save_hii_waterlevel(
        &db.pool,
        &[wl(0, 301.4), wl(40, 301.45)],
        t(2026, 10, 18, 13, 45),
    )
    .await
    .unwrap();
    let codes = vec!["P.1".to_string(), "P.103".to_string()];
    let hii = db::load_hii_window(
        &db.pool,
        &codes,
        t(2026, 10, 18, 0, 0),
        t(2026, 10, 18, 14, 0),
    )
    .await
    .unwrap();
    assert_eq!(hii.len(), 2);
    assert_eq!(ping_forecast::hii::prepare(&hii)[0].wl_msl, 301.45);
    assert_eq!(
        db::latest_hii_timestamp(&db.pool, &codes).await.unwrap(),
        Some(t(2026, 10, 18, 13, 40))
    );
    assert!(
        db::load_dam_days(&db.pool, "200103", d(2026, 10, 1))
            .await
            .unwrap()
            .is_empty()
    );
    db.drop().await;
}