use std::path::Path;

use anyhow::{Context, Result};
use polars::lazy::frame::LazyFrame;
use polars::prelude::*;
use serde::Serialize;
use tracing::info;

use crate::consts::{NAN_U16, QUANT_SCALE};
use crate::data::{
    combine_nearest_distances, combined_station_feature_name,
    combined_station_source_feature_names, PropertyData, QuantRef,
};
use crate::utils::{normalize_postcode, GridIndex, InternedColumn};

const GRID_CELL_SIZE: f32 = 0.01;

/// One observation on a listing's accruing asking-price timeline, recovered from
/// the scraper's forward-only store (finder/price_history.py). `reason` is one of
/// "listed" | "reduced" | "increased"; `date` is `YYYY-MM-DD`.
#[derive(Serialize, Clone)]
pub struct PriceHistoryPoint {
    pub date: String,
    pub price: i64,
    pub reason: String,
}

#[derive(Serialize, Clone)]
pub struct ActualListing {
    pub lat: f32,
    pub lon: f32,
    pub postcode: String,
    pub address: Option<String>,
    pub property_type: Option<String>,
    pub property_sub_type: Option<String>,
    pub leasehold_freehold: Option<String>,
    pub price_qualifier: Option<String>,
    pub bedrooms: Option<i32>,
    pub bathrooms: Option<i32>,
    pub rooms_total: Option<i32>,
    pub floor_area_sqm: Option<f32>,
    pub asking_price: Option<i64>,
    pub asking_price_per_sqm: Option<f32>,
    pub listing_url: String,
    pub listing_status: Option<String>,
    pub listing_date_iso: Option<String>,
    pub features: Vec<String>,
    /// Accrued asking-price observations, oldest -> newest. Empty until the
    /// scraper's forward-only store has recorded at least one run for this id.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub price_history: Vec<PriceHistoryPoint>,
}

pub struct ActualListingData {
    pub lat: Vec<f32>,
    pub lon: Vec<f32>,
    /// Normalized (uppercase, canonical spacing) postcode per row.
    pub postcode: Vec<String>,
    pub address: Vec<Option<String>>,
    pub property_type: InternedColumn,
    pub property_sub_type: InternedColumn,
    pub leasehold_freehold: InternedColumn,
    pub price_qualifier: InternedColumn,
    pub bedrooms: Vec<Option<i32>>,
    pub bathrooms: Vec<Option<i32>>,
    pub rooms_total: Vec<Option<i32>>,
    pub floor_area_sqm: Vec<Option<f32>>,
    pub asking_price: Vec<Option<i64>>,
    pub asking_price_per_sqm: Vec<Option<f32>>,
    pub listing_url: Vec<String>,
    pub listing_status: InternedColumn,
    pub listing_date_iso: Vec<Option<String>>,
    pub features: Vec<Vec<String>>,
    /// Per-row accrued asking-price series (empty when absent from the parquet).
    pub price_history: Vec<Vec<PriceHistoryPoint>>,
    /// Row-major feature matrix aligned with PropertyData::feature_names.
    ///
    /// Rows start from a best-effort address/postcode join to the historical property
    /// dataset, then live listing fields such as asking price and property type are
    /// overlaid where available. This lets the listings endpoint use the same filter
    /// execution path as the property endpoints.
    pub filter_feature_data: Vec<u16>,
    /// Row-major dynamic postcode POI metrics aligned with
    /// PropertyData::poi_metrics.feature_names.
    pub poi_filter_feature_data: Vec<u16>,
    pub grid: GridIndex,
}

impl ActualListingData {
    pub fn load(parquet_path: &Path, property_data: &PropertyData) -> Result<Self> {
        super::run_polars_io(|| Self::load_inner(parquet_path, Some(property_data)))
    }

    fn load_inner(parquet_path: &Path, property_data: Option<&PropertyData>) -> Result<Self> {
        info!("Loading actual listings from {:?}", parquet_path);
        let pl_path = PlRefPath::try_from_path(parquet_path)
            .context("Failed to normalize actual listings parquet path")?;
        let df = LazyFrame::scan_parquet(pl_path, Default::default())
            .context("Failed to scan actual listings parquet")?
            .collect()
            .context("Failed to read actual listings parquet")?;

        let row_count = df.height();
        info!(rows = row_count, "Actual listings parquet read");

        let lat = extract_f32(&df, "lat")?;
        let lon = extract_f32(&df, "lon")?;
        let postcode_raw = extract_str(&df, "Postcode")?;
        let address = extract_opt_str(&df, "Address per Property Register")?;
        let property_type_raw = extract_opt_str(&df, "Property type")?;
        let property_sub_type_raw = extract_opt_str(&df, "Property sub-type")?;
        let leasehold_freehold_raw = extract_opt_str(&df, "Leasehold/Freehold")?;
        let price_qualifier_raw = extract_opt_str(&df, "Price qualifier")?;
        let bedrooms = extract_opt_i32(&df, "Bedrooms")?;
        let bathrooms = extract_opt_i32(&df, "Bathrooms")?;
        let rooms_total = extract_opt_i32(&df, "Number of bedrooms & living rooms")?;
        let floor_area_sqm = extract_opt_f32(&df, "Total floor area (sqm)")?;
        let asking_price = extract_opt_i64(&df, "Asking price")?;
        let asking_price_per_sqm = extract_opt_f32(&df, "Asking price per sqm")?;
        let listing_url = extract_str(&df, "Listing URL")?;
        let listing_status_raw = extract_opt_str(&df, "Listing status")?;
        let listing_date_iso = extract_opt_datetime_iso(&df, "Listing date")?;
        let features = extract_str_list(&df, "Listing features")?;
        let price_history = extract_price_history(&df)?;

        let postcode: Vec<String> = postcode_raw.iter().map(|s| normalize_postcode(s)).collect();

        let property_type = InternedColumn::build(&opt_to_string(&property_type_raw));
        let property_sub_type = InternedColumn::build(&opt_to_string(&property_sub_type_raw));
        let leasehold_freehold = InternedColumn::build(&opt_to_string(&leasehold_freehold_raw));
        let price_qualifier = InternedColumn::build(&opt_to_string(&price_qualifier_raw));
        let listing_status = InternedColumn::build(&opt_to_string(&listing_status_raw));

        let filter_feature_data = build_filter_feature_data(
            &df,
            property_data,
            &property_type_raw,
            &leasehold_freehold_raw,
            &rooms_total,
            &floor_area_sqm,
            &asking_price,
            &asking_price_per_sqm,
        )?;
        let poi_filter_feature_data = build_poi_filter_feature_data(&df, property_data)?;

        let grid = GridIndex::build(&lat, &lon, GRID_CELL_SIZE);

        info!(rows = row_count, "Actual listings loaded");

        Ok(Self {
            lat,
            lon,
            postcode,
            address,
            property_type,
            property_sub_type,
            leasehold_freehold,
            price_qualifier,
            bedrooms,
            bathrooms,
            rooms_total,
            floor_area_sqm,
            asking_price,
            asking_price_per_sqm,
            listing_url,
            listing_status,
            listing_date_iso,
            features,
            price_history,
            filter_feature_data,
            poi_filter_feature_data,
            grid,
        })
    }

    pub fn listing_at(&self, row: usize) -> ActualListing {
        ActualListing {
            lat: self.lat[row],
            lon: self.lon[row],
            postcode: self.postcode[row].clone(),
            address: self.address[row].clone(),
            property_type: opt_from_interned(&self.property_type, row),
            property_sub_type: opt_from_interned(&self.property_sub_type, row),
            leasehold_freehold: opt_from_interned(&self.leasehold_freehold, row),
            price_qualifier: opt_from_interned(&self.price_qualifier, row),
            bedrooms: self.bedrooms[row],
            bathrooms: self.bathrooms[row],
            rooms_total: self.rooms_total[row],
            floor_area_sqm: self.floor_area_sqm[row],
            asking_price: self.asking_price[row],
            asking_price_per_sqm: self.asking_price_per_sqm[row],
            listing_url: self.listing_url[row].clone(),
            listing_status: opt_from_interned(&self.listing_status, row),
            listing_date_iso: self.listing_date_iso[row].clone(),
            features: self.features[row].clone(),
            price_history: self.price_history[row].clone(),
        }
    }
}

#[allow(clippy::too_many_arguments)]
fn build_filter_feature_data(
    df: &DataFrame,
    property_data: Option<&PropertyData>,
    property_type: &[Option<String>],
    leasehold_freehold: &[Option<String>],
    rooms_total: &[Option<i32>],
    floor_area_sqm: &[Option<f32>],
    asking_price: &[Option<i64>],
    asking_price_per_sqm: &[Option<f32>],
) -> Result<Vec<u16>> {
    let Some(property_data) = property_data else {
        return Ok(Vec::new());
    };

    let num_features = property_data.num_features;
    let row_count = df.height();
    let mut feature_data = vec![NAN_U16; row_count * num_features];
    let quant = property_data.quant_ref();
    let mut encoded_columns = 0usize;

    for (feat_idx, name) in property_data.feature_names.iter().enumerate() {
        if feat_idx < property_data.num_numeric {
            if let Some(values) = extract_optional_feature_f32(df, name)? {
                encode_numeric_feature(&mut feature_data, property_data, &quant, feat_idx, values);
                encoded_columns += 1;
            }
        } else if let Some(values) = extract_optional_feature_str(df, name)? {
            encode_enum_feature(&mut feature_data, property_data, feat_idx, values);
            encoded_columns += 1;
        }
    }

    overlay_numeric_feature(
        &mut feature_data,
        property_data,
        &quant,
        "Total floor area (sqm)",
        floor_area_sqm.iter().copied(),
        false,
    );
    overlay_numeric_feature(
        &mut feature_data,
        property_data,
        &quant,
        "Number of bedrooms & living rooms",
        rooms_total.iter().map(|value| value.map(|v| v as f32)),
        false,
    );
    overlay_numeric_feature(
        &mut feature_data,
        property_data,
        &quant,
        "Estimated current price",
        asking_price.iter().map(|value| value.map(|v| v as f32)),
        true,
    );
    overlay_numeric_feature(
        &mut feature_data,
        property_data,
        &quant,
        "Last known price",
        asking_price.iter().map(|value| value.map(|v| v as f32)),
        true,
    );
    overlay_numeric_feature(
        &mut feature_data,
        property_data,
        &quant,
        "Est. price per sqm",
        asking_price_per_sqm.iter().copied(),
        true,
    );
    overlay_numeric_feature(
        &mut feature_data,
        property_data,
        &quant,
        "Price per sqm",
        asking_price_per_sqm.iter().copied(),
        true,
    );
    overlay_enum_feature(
        &mut feature_data,
        property_data,
        "Property type",
        property_type.iter().map(Option::as_deref),
        false,
    );
    overlay_enum_feature(
        &mut feature_data,
        property_data,
        "Leasehold/Freehold",
        leasehold_freehold.iter().map(Option::as_deref),
        false,
    );

    info!(
        rows = row_count,
        encoded_columns, "Actual listings feature matrix read from enriched parquet"
    );

    Ok(feature_data)
}

fn build_poi_filter_feature_data(
    df: &DataFrame,
    property_data: Option<&PropertyData>,
) -> Result<Vec<u16>> {
    let Some(property_data) = property_data else {
        return Ok(Vec::new());
    };
    let poi_metrics = &property_data.poi_metrics;
    let num_features = poi_metrics.num_features();
    if num_features == 0 {
        return Ok(Vec::new());
    }

    let row_count = df.height();
    let mut feature_data = vec![NAN_U16; row_count * num_features];
    let quant = poi_metrics.quant_ref();
    let mut encoded_columns = 0usize;

    for (metric_idx, name) in poi_metrics.feature_names.iter().enumerate() {
        let values = match extract_optional_feature_f32(df, name)? {
            Some(values) => values,
            // Some POI columns (the combined-station distance) are synthesized
            // in memory by PostcodePoiMetrics and never written to the listings
            // parquet. Reconstruct them from their source columns so listing POI
            // filters match the map exactly instead of silently rejecting every
            // row on an all-NaN column.
            None => match synthesize_missing_poi_column(df, name, row_count)? {
                Some(values) => values,
                None => continue,
            },
        };
        for (row, value) in values.into_iter().enumerate() {
            let dst = row * num_features + metric_idx;
            feature_data[dst] = value
                .map(|value| encode_numeric_value(&quant, metric_idx, value))
                .unwrap_or(NAN_U16);
        }
        encoded_columns += 1;
    }

    info!(
        rows = row_count,
        encoded_columns, "Actual listings POI metrics read from enriched parquet"
    );

    Ok(feature_data)
}

/// Reconstruct a POI metric column that `PostcodePoiMetrics` synthesizes in
/// memory and therefore is absent from the listings parquet. Currently only the
/// combined-station distance, derived as the elementwise nearest of its per-mode
/// source columns (the same rule the postcode side table uses). Returns `None`
/// if `name` is not a synthesized column, or if none of its sources are present.
fn synthesize_missing_poi_column(
    df: &DataFrame,
    name: &str,
    row_count: usize,
) -> Result<Option<Vec<Option<f32>>>> {
    if name != combined_station_feature_name() {
        return Ok(None);
    }

    let mut sources: Vec<Vec<f32>> = Vec::new();
    for source_name in combined_station_source_feature_names() {
        if let Some(values) = extract_optional_feature_f32(df, &source_name)? {
            sources.push(
                values
                    .into_iter()
                    .map(|value| value.unwrap_or(f32::NAN))
                    .collect(),
            );
        }
    }
    if sources.is_empty() {
        return Ok(None);
    }

    let source_refs: Vec<&[f32]> = sources.iter().map(Vec::as_slice).collect();
    let combined = combine_nearest_distances(&source_refs, row_count);
    Ok(Some(
        combined
            .into_iter()
            .map(|value| value.is_finite().then_some(value))
            .collect(),
    ))
}

fn feature_index(property_data: &PropertyData, name: &str) -> Option<usize> {
    property_data
        .feature_names
        .iter()
        .position(|candidate| candidate == name)
}

fn overlay_numeric_feature<I>(
    feature_data: &mut [u16],
    property_data: &PropertyData,
    quant: &QuantRef<'_>,
    name: &str,
    values: I,
    clear_missing: bool,
) where
    I: IntoIterator<Item = Option<f32>>,
{
    let Some(feat_idx) = feature_index(property_data, name) else {
        return;
    };
    if feat_idx >= property_data.num_numeric {
        return;
    }

    let num_features = property_data.num_features;
    for (row, value) in values.into_iter().enumerate() {
        let dst = row * num_features + feat_idx;
        match value {
            Some(value) => feature_data[dst] = encode_numeric_value(quant, feat_idx, value),
            None if clear_missing => feature_data[dst] = NAN_U16,
            None => {}
        }
    }
}

fn encode_numeric_feature<I>(
    feature_data: &mut [u16],
    property_data: &PropertyData,
    quant: &QuantRef<'_>,
    feat_idx: usize,
    values: I,
) where
    I: IntoIterator<Item = Option<f32>>,
{
    let num_features = property_data.num_features;
    for (row, value) in values.into_iter().enumerate() {
        let dst = row * num_features + feat_idx;
        feature_data[dst] = value
            .map(|value| encode_numeric_value(quant, feat_idx, value))
            .unwrap_or(NAN_U16);
    }
}

fn extract_optional_feature_f32(df: &DataFrame, name: &str) -> Result<Option<Vec<Option<f32>>>> {
    let Ok(column) = df.column(name) else {
        return Ok(None);
    };

    if matches!(column.dtype(), DataType::Datetime(_, _) | DataType::Date) {
        let projected = df
            .clone()
            .lazy()
            .select([(col(name).dt().year().cast(DataType::Float32)
                + (col(name).dt().month().cast(DataType::Float32) - lit(1.0f32)) / lit(12.0f32))
            .alias("__feature")])
            .collect()
            .with_context(|| format!("Failed to convert datetime feature '{name}'"))?;
        return Ok(Some(extract_opt_f32(&projected, "__feature")?));
    }

    let cast = column
        .cast(&DataType::Float32)
        .with_context(|| format!("Failed to cast feature '{name}' to Float32"))?;
    let values = cast
        .f32()
        .with_context(|| format!("Feature '{name}' is not Float32"))?
        .into_iter()
        .map(|value| value.filter(|v| v.is_finite()))
        .collect();
    Ok(Some(values))
}

fn overlay_enum_feature<'a, I>(
    feature_data: &mut [u16],
    property_data: &PropertyData,
    name: &str,
    values: I,
    clear_missing: bool,
) where
    I: IntoIterator<Item = Option<&'a str>>,
{
    let Some(feat_idx) = feature_index(property_data, name) else {
        return;
    };
    let Some(enum_values) = property_data.enum_values.get(&feat_idx) else {
        return;
    };

    let num_features = property_data.num_features;
    for (row, value) in values.into_iter().enumerate() {
        let dst = row * num_features + feat_idx;
        let encoded = value
            .map(str::trim)
            .filter(|text| !text.is_empty())
            .and_then(|text| enum_values.iter().position(|candidate| candidate == text))
            .map(|position| position as u16);
        match encoded {
            Some(value) => feature_data[dst] = value,
            None if clear_missing => feature_data[dst] = NAN_U16,
            None => {}
        }
    }
}

fn encode_enum_feature(
    feature_data: &mut [u16],
    property_data: &PropertyData,
    feat_idx: usize,
    values: Vec<Option<String>>,
) {
    let Some(enum_values) = property_data.enum_values.get(&feat_idx) else {
        return;
    };
    let num_features = property_data.num_features;
    for (row, value) in values.into_iter().enumerate() {
        let dst = row * num_features + feat_idx;
        feature_data[dst] = value
            .as_deref()
            .map(str::trim)
            .filter(|text| !text.is_empty())
            .and_then(|text| enum_values.iter().position(|candidate| candidate == text))
            .map(|position| position as u16)
            .unwrap_or(NAN_U16);
    }
}

fn extract_optional_feature_str(df: &DataFrame, name: &str) -> Result<Option<Vec<Option<String>>>> {
    let Ok(column) = df.column(name) else {
        return Ok(None);
    };
    let cast = column
        .cast(&DataType::String)
        .with_context(|| format!("Failed to cast feature '{name}' to String"))?;
    let strings = cast
        .str()
        .with_context(|| format!("Feature '{name}' is not a string column"))?;
    Ok(Some(
        strings
            .into_iter()
            .map(|value| value.and_then(|text| (!text.trim().is_empty()).then(|| text.to_string())))
            .collect(),
    ))
}

fn encode_numeric_value(quant: &QuantRef<'_>, feat_idx: usize, value: f32) -> u16 {
    if !value.is_finite() {
        return NAN_U16;
    }
    let range = quant.quant_range[feat_idx];
    if range <= 0.0 {
        return 0;
    }
    let normalized = (value - quant.quant_min[feat_idx]) / range;
    (normalized * QUANT_SCALE).round().clamp(0.0, QUANT_SCALE) as u16
}

fn opt_to_string(values: &[Option<String>]) -> Vec<String> {
    values
        .iter()
        .map(|value| value.clone().unwrap_or_default())
        .collect()
}

fn opt_from_interned(column: &InternedColumn, row: usize) -> Option<String> {
    let value = column.get(row);
    if value.is_empty() {
        None
    } else {
        Some(value.to_string())
    }
}

fn extract_f32(df: &DataFrame, name: &str) -> Result<Vec<f32>> {
    let cast = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?
        .cast(&DataType::Float32)
        .with_context(|| format!("Failed to cast '{name}' to Float32"))?;
    let column = cast
        .f32()
        .with_context(|| format!("Column '{name}' is not Float32"))?;
    column
        .into_iter()
        .enumerate()
        .map(|(row, value)| value.with_context(|| format!("Column '{name}' has null at row {row}")))
        .collect()
}

fn extract_str(df: &DataFrame, name: &str) -> Result<Vec<String>> {
    let column = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?;
    let strings = column
        .str()
        .with_context(|| format!("Column '{name}' is not a string column"))?;
    strings
        .into_iter()
        .enumerate()
        .map(|(row, value)| {
            value
                .map(ToString::to_string)
                .with_context(|| format!("Column '{name}' has null at row {row}"))
        })
        .collect()
}

fn extract_opt_str(df: &DataFrame, name: &str) -> Result<Vec<Option<String>>> {
    let column = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?;
    let strings = column
        .str()
        .with_context(|| format!("Column '{name}' is not a string column"))?;
    Ok(strings
        .into_iter()
        .map(|value| value.and_then(|text| (!text.is_empty()).then(|| text.to_string())))
        .collect())
}

fn extract_opt_i32(df: &DataFrame, name: &str) -> Result<Vec<Option<i32>>> {
    let cast = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?
        .cast(&DataType::Int32)
        .with_context(|| format!("Failed to cast '{name}' to Int32"))?;
    let column = cast
        .i32()
        .with_context(|| format!("Column '{name}' is not Int32"))?;
    Ok(column.into_iter().collect())
}

fn extract_opt_i64(df: &DataFrame, name: &str) -> Result<Vec<Option<i64>>> {
    let cast = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?
        .cast(&DataType::Int64)
        .with_context(|| format!("Failed to cast '{name}' to Int64"))?;
    let column = cast
        .i64()
        .with_context(|| format!("Column '{name}' is not Int64"))?;
    Ok(column.into_iter().collect())
}

fn extract_opt_f32(df: &DataFrame, name: &str) -> Result<Vec<Option<f32>>> {
    let cast = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?
        .cast(&DataType::Float32)
        .with_context(|| format!("Failed to cast '{name}' to Float32"))?;
    let column = cast
        .f32()
        .with_context(|| format!("Column '{name}' is not Float32"))?;
    Ok(column
        .into_iter()
        .map(|value| value.filter(|v| v.is_finite()))
        .collect())
}

fn extract_opt_datetime_iso(df: &DataFrame, name: &str) -> Result<Vec<Option<String>>> {
    let column = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?;
    let cast = column
        .cast(&DataType::Datetime(TimeUnit::Microseconds, None))
        .with_context(|| format!("Failed to cast '{name}' to Datetime(us)"))?;
    let datetime = cast
        .datetime()
        .with_context(|| format!("Column '{name}' is not a Datetime column"))?;
    Ok(datetime
        .as_datetime_iter()
        .map(|value| value.map(|date| date.format("%Y-%m-%dT%H:%M:%SZ").to_string()))
        .collect())
}

fn extract_str_list(df: &DataFrame, name: &str) -> Result<Vec<Vec<String>>> {
    let column = df
        .column(name)
        .with_context(|| format!("Missing column '{name}'"))?;
    let list = column
        .list()
        .with_context(|| format!("Column '{name}' is not a list column"))?;
    let mut out = Vec::with_capacity(list.len());
    for series_opt in list.into_iter() {
        let entries = match series_opt {
            Some(series) => {
                let strings = series.str().with_context(|| {
                    format!("Column '{name}' list inner is not a string column")
                })?;
                strings
                    .into_iter()
                    .filter_map(|value| value.map(ToString::to_string))
                    .collect()
            }
            None => Vec::new(),
        };
        out.push(entries);
    }
    Ok(out)
}

/// Extract `price_history: List<Struct{date: Utf8, price: Int64, reason: Utf8}>`.
///
/// Schema-gated: a parquet written before the column existed simply yields an
/// empty series per row (the column is dropped from serialized listings). A
/// null/empty list is likewise an empty series.
fn extract_price_history(df: &DataFrame) -> Result<Vec<Vec<PriceHistoryPoint>>> {
    let row_count = df.height();
    let Ok(column) = df.column("price_history") else {
        return Ok(vec![Vec::new(); row_count]);
    };
    let list_ca = column
        .list()
        .context("price_history is not a list column")?;

    let mut out: Vec<Vec<PriceHistoryPoint>> = Vec::with_capacity(row_count);
    for row in 0..row_count {
        let mut points = Vec::new();
        if let Some(inner) = list_ca.get_as_series(row) {
            if !inner.is_empty() {
                let structs = inner
                    .struct_()
                    .context("price_history inner is not a struct")?;
                let dates_s = structs
                    .field_by_name("date")
                    .context("Missing 'date' field in price_history struct")?;
                let date_ca = dates_s.str().context("price_history.date is not a string")?;
                let prices_s = structs
                    .field_by_name("price")
                    .context("Missing 'price' field in price_history struct")?;
                let prices_i64 = prices_s
                    .cast(&DataType::Int64)
                    .context("price_history.price is not castable to Int64")?;
                let price_ca = prices_i64.i64().context("price_history.price is not Int64")?;
                let reasons_s = structs
                    .field_by_name("reason")
                    .context("Missing 'reason' field in price_history struct")?;
                let reason_ca = reasons_s
                    .str()
                    .context("price_history.reason is not a string")?;
                for idx in 0..inner.len() {
                    let (Some(date), Some(price), Some(reason)) =
                        (date_ca.get(idx), price_ca.get(idx), reason_ca.get(idx))
                    else {
                        continue;
                    };
                    points.push(PriceHistoryPoint {
                        date: date.to_string(),
                        price,
                        reason: reason.to_string(),
                    });
                }
            }
        }
        out.push(points);
    }
    Ok(out)
}

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

    fn sample_path() -> Option<PathBuf> {
        [
            "../finder/data/online_listings_buy_enriched.parquet",
            "../finder/data/online_listings_buy.parquet",
        ]
        .into_iter()
        .map(PathBuf::from)
        .find(|path| path.exists())
    }

    #[test]
    fn loads_sample_listings_when_available() {
        let Some(path) = sample_path() else {
            eprintln!("sample parquet not present; skipping");
            return;
        };

        let data = ActualListingData::load_inner(&path, None).expect("listings load");
        assert!(!data.lat.is_empty());
        assert_eq!(data.lat.len(), data.lon.len());
        assert_eq!(data.lat.len(), data.postcode.len());
        assert_eq!(data.lat.len(), data.listing_url.len());
        assert_eq!(data.lat.len(), data.features.len());

        let any_listing = data.listing_at(0);
        assert!(any_listing.lat.is_finite());
        assert!(any_listing.lon.is_finite());
        assert!(!any_listing.listing_url.is_empty());
    }

    #[test]
    fn synthesizes_combined_station_column_from_source_columns() {
        // Mirrors the postcode side table: the combined-station distance is the
        // finite minimum of the per-mode source columns present in the parquet.
        let df = df![
            "Distance to nearest amenity (Rail station) (km)" => [2.0f32, f32::NAN, 5.0],
            "Distance to nearest amenity (Tube station) (km)" => [1.0f32, f32::NAN, f32::NAN],
            "Distance to nearest amenity (DLR station) (km)" => [3.0f32, 4.0, f32::NAN],
        ]
        .unwrap();

        let combined = synthesize_missing_poi_column(&df, &combined_station_feature_name(), 3)
            .unwrap()
            .expect("combined-station column should be synthesized");

        assert_eq!(combined[0], Some(1.0)); // min(2, 1, 3)
        assert_eq!(combined[1], Some(4.0)); // only DLR present
        assert_eq!(combined[2], Some(5.0)); // only Rail present
    }

    #[test]
    fn does_not_synthesize_non_combined_or_sourceless_columns() {
        let df = df![
            "Distance to nearest amenity (Rail station) (km)" => [1.0f32],
        ]
        .unwrap();

        // A real per-mode column is read from the parquet, never synthesized.
        assert!(synthesize_missing_poi_column(
            &df,
            "Distance to nearest amenity (Rail station) (km)",
            1
        )
        .unwrap()
        .is_none());

        // The combined column cannot be built when no source columns exist.
        let empty = df!["Postcode" => ["AB1 2CD"]].unwrap();
        assert!(
            synthesize_missing_poi_column(&empty, &combined_station_feature_name(), 1)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn extracts_price_history_from_parquet() {
        let path = PathBuf::from("src/data/testdata/listings_price_history.parquet");
        if !path.exists() {
            eprintln!("price_history fixture not present; skipping");
            return;
        }
        let data = ActualListingData::load_inner(&path, None).expect("listings load");
        assert_eq!(data.price_history.len(), data.lat.len());

        // Row 0: two points, listed -> reduced, oldest first.
        let r0 = data.listing_at(0);
        assert_eq!(r0.price_history.len(), 2);
        assert_eq!(r0.price_history[0].date, "2026-07-01");
        assert_eq!(r0.price_history[0].price, 500_000);
        assert_eq!(r0.price_history[0].reason, "listed");
        assert_eq!(r0.price_history[1].price, 480_000);
        assert_eq!(r0.price_history[1].reason, "reduced");

        // Row 1: a single "listed" point.
        assert_eq!(data.listing_at(1).price_history.len(), 1);
        // Row 2 (empty list) and row 3 (null) carry no points.
        assert!(data.listing_at(2).price_history.is_empty());
        assert!(data.listing_at(3).price_history.is_empty());
    }
}