//! Property data: the row-major quantized feature matrix plus the side tables
//! (addresses, postcodes, renovation/price history, POI metrics) built from the
//! properties + postcode parquet files.
//!
//! Split by concern:
//! - [`loading`]: parquet ingestion, validation, spatial sort and matrix build
//! - [`stats`]: histograms, percentiles and slider-bound computation
//! - [`quant`]: u16 quantization encode/decode
//! - [`poi_metrics`]: postcode-level POI metric side table
//! - [`address_search`]: address tokenization, indexing and ranked search
//! - [`h3`]: H3 cell precomputation

mod address_search;
mod h3;
mod loading;
mod poi_metrics;
mod quant;
mod stats;

pub use h3::precompute_h3;
pub use poi_metrics::PostcodePoiMetrics;
pub use quant::QuantRef;
pub use stats::{FeatureStats, Histogram};

use rustc_hash::FxHashMap;
use serde::Serialize;

use crate::consts::NAN_U16;
use crate::data::spill::SpillVec;

#[derive(Serialize, Clone)]
pub struct RenovationEvent {
    pub year: i32,
    pub event: String,
}

#[derive(Serialize, Clone)]
pub struct HistoricalPrice {
    pub year: i32,
    pub month: u8,
    pub price: i64,
    /// Whether this specific sale was a new-build transfer (PPD `old_new == "Y"`).
    /// Defaults to `false` for older parquet that predates the per-sale flag.
    pub is_new: bool,
}

/// A point on the property's tenure timeline: the year a certificate first
/// recorded a new occupancy `status` ("Owner-occupied" / "Rented (private)" /
/// "Rented (social)"). Derived per-EPC and reduced to change points by the
/// pipeline, so this surfaces *when* a home was let out vs lived in.
#[derive(Serialize, Clone)]
pub struct TenureEvent {
    pub year: i32,
    pub status: String,
}

pub struct PropertyData {
    pub lat: Vec<f32>,
    pub lon: Vec<f32>,
    pub feature_names: Vec<String>,
    pub num_features: usize,
    /// Number of numeric features (enum features start at this index).
    pub num_numeric: usize,
    /// Row-major flat array: feature_data[row * num_features + feat_idx].
    /// Quantized to u16. NaN sentinel = u16::MAX (65535).
    /// Numeric features: encoded via (val - min) / range * 65534.
    /// Enum features: stored directly as u16 cast of the f32 index.
    /// Heap-resident in prod; mmap-backed under `--spill-dir` (the largest array,
    /// so the biggest dev-memory win). Derefs to `[u16]`, so reads are unchanged.
    pub feature_data: SpillVec<u16>,
    /// Per-feature: range / QUANT_SCALE for fast decode.
    dequant_a: Vec<f32>,
    /// Per-feature: minimum value (offset for dequantization).
    quant_min: Vec<f32>,
    /// Per-feature: max - min (for encoding filter bounds).
    quant_range: Vec<f32>,
    pub feature_stats: Vec<FeatureStats>,
    pub poi_metrics: PostcodePoiMetrics,
    /// Unquantized last sale price used by the price-history chart.
    last_known_price_raw: Vec<f32>,
    /// Contiguous buffer holding all address strings end-to-end (valid UTF-8 by
    /// construction; spillable, so stored as bytes rather than `String`).
    address_buffer: SpillVec<u8>,
    /// Byte offset into `address_buffer` where each row's address starts.
    address_offsets: SpillVec<u32>,
    /// Length in bytes of each row's address.
    address_lengths: SpillVec<u16>,
    /// Interned postcodes: reader is thread-safe, keys index into it.
    postcode_interner: lasso::RodeoReader,
    postcode_keys: SpillVec<lasso::Spur>,
    /// Rows for each postcode, keyed by the interned postcode key.
    postcode_row_index: FxHashMap<lasso::Spur, Vec<u32>>,
    /// Postcode keys for each postcode sector (e.g. "E14 2"). Stores interned
    /// keys, not rows, so the wider-area price-history charts can enumerate a
    /// sector's properties (via `postcode_row_index`) for ~1/13th the memory of
    /// duplicating the row ids.
    sector_postcode_index: FxHashMap<String, Vec<lasso::Spur>>,
    /// Postcode keys for each outward code (e.g. "E14"). Same rationale as
    /// `sector_postcode_index`.
    outcode_postcode_index: FxHashMap<String, Vec<lasso::Spur>>,
    /// Inverted index from address tokens to property rows.
    address_token_index: FxHashMap<String, Vec<u32>>,
    /// Prefix lookup from typed address-token prefix to indexed full address tokens.
    address_prefix_index: FxHashMap<String, Vec<String>>,
    /// Interned normalized address-search tokens used for per-row scoring.
    /// Resolve-only (no string->key reverse map): scoring only resolves keys.
    address_search_interner: lasso::RodeoResolver,
    /// Flat per-row normalized address-search token keys.
    address_search_token_keys: SpillVec<lasso::Spur>,
    /// Offset into `address_search_token_keys` for each row.
    address_search_token_offsets: SpillVec<u32>,
    /// Number of normalized address-search token keys for each row.
    address_search_token_lengths: SpillVec<u16>,
    /// For enum features: maps feature index to list of possible string values.
    /// Index in values list corresponds to the u16 value stored in feature_data.
    pub enum_values: rustc_hash::FxHashMap<usize, Vec<String>>,
    /// For enum features: maps feature index to per-value global counts (same order as enum_values).
    pub enum_counts: rustc_hash::FxHashMap<usize, Vec<u64>>,
    /// Per-row flag: true = construction date is approximate (from EPC band),
    /// false = exact (from new-build transaction date).
    /// Bit-packed: byte `row / 8`, bit `row % 8`. 8x smaller than Vec<bool>.
    approx_build_date_bits: Vec<u8>,
    /// Per-row renovation events. Keyed by (permuted) row index.
    /// Only rows with events are present in the map.
    renovation_history: FxHashMap<u32, Vec<RenovationEvent>>,
    /// Per-row historical sale transactions (Land Registry price-paid).
    /// Keyed by (permuted) row index. Only rows with prices are present.
    historical_prices: FxHashMap<u32, Vec<HistoricalPrice>>,
    /// Per-row tenure timeline (owner-occupied <-> rented change points from
    /// EPC). Keyed by (permuted) row index. Only rows with changes are present.
    tenure_history: FxHashMap<u32, Vec<TenureEvent>>,
    property_sub_type: FxHashMap<u32, String>,
    price_qualifier: FxHashMap<u32, String>,
}

impl PropertyData {
    /// Get the address string for a given row.
    pub fn address(&self, row: usize) -> &str {
        let offset = self.address_offsets[row] as usize;
        let length = self.address_lengths[row] as usize;
        let bytes = &self.address_buffer[offset..offset + length];
        // SAFETY: `address_buffer` is built by concatenating `&str` slices on UTF-8
        // char boundaries, so every recorded `[offset, offset + length)` range is
        // valid UTF-8.
        unsafe { std::str::from_utf8_unchecked(bytes) }
    }

    /// Get the postcode string for a given row.
    pub fn postcode(&self, row: usize) -> &str {
        self.postcode_interner.resolve(&self.postcode_keys[row])
    }

    /// Get postcode components for field-level borrowing (avoids conflicting borrows with feature_data).
    pub fn postcode_parts(&self) -> (&lasso::RodeoReader, &[lasso::Spur]) {
        (&self.postcode_interner, self.postcode_keys.as_slice())
    }

    /// Property rows for a given postcode string, or empty if unknown.
    pub fn rows_for_postcode(&self, postcode: &str) -> &[u32] {
        self.postcode_interner
            .get(postcode)
            .and_then(|key| self.postcode_row_index.get(&key))
            .map(Vec::as_slice)
            .unwrap_or(&[])
    }

    /// All property rows in a postcode sector (e.g. "E14 2"), gathered across the
    /// sector's postcodes. Empty if the sector is unknown. Materializes a new Vec
    /// since the rows live in per-postcode buckets; callers downsample it.
    pub fn rows_for_sector(&self, sector: &str) -> Vec<u32> {
        self.rows_for_area(self.sector_postcode_index.get(sector))
    }

    /// All property rows in an outward code (e.g. "E14"). See `rows_for_sector`.
    pub fn rows_for_outcode(&self, outcode: &str) -> Vec<u32> {
        self.rows_for_area(self.outcode_postcode_index.get(outcode))
    }

    fn rows_for_area(&self, postcode_keys: Option<&Vec<lasso::Spur>>) -> Vec<u32> {
        let Some(keys) = postcode_keys else {
            return Vec::new();
        };
        let mut rows = Vec::new();
        for key in keys {
            if let Some(bucket) = self.postcode_row_index.get(key) {
                rows.extend_from_slice(bucket);
            }
        }
        rows
    }

    /// Get the is_approx_build_date flag for a given row (bit-packed).
    pub fn is_approx_build_date(&self, row: usize) -> bool {
        let byte = self.approx_build_date_bits[row / 8];
        byte & (1 << (row % 8)) != 0
    }

    /// Get renovation events for a given row (empty slice if none).
    pub fn renovation_history(&self, row: usize) -> &[RenovationEvent] {
        self.renovation_history
            .get(&(row as u32))
            .map(|v| v.as_slice())
            .unwrap_or(&[])
    }

    /// Get historical sale transactions for a given row (empty slice if none).
    pub fn historical_prices(&self, row: usize) -> &[HistoricalPrice] {
        self.historical_prices
            .get(&(row as u32))
            .map(|v| v.as_slice())
            .unwrap_or(&[])
    }

    /// Get tenure timeline events for a given row (empty slice if none).
    pub fn tenure_history(&self, row: usize) -> &[TenureEvent] {
        self.tenure_history
            .get(&(row as u32))
            .map(|v| v.as_slice())
            .unwrap_or(&[])
    }

    /// Get property sub-type for a given row.
    pub fn property_sub_type(&self, row: usize) -> Option<&str> {
        self.property_sub_type
            .get(&(row as u32))
            .map(String::as_str)
    }

    /// Get price qualifier for a given row.
    pub fn price_qualifier(&self, row: usize) -> Option<&str> {
        self.price_qualifier.get(&(row as u32)).map(String::as_str)
    }

    /// Get the unquantized last sale price for charting.
    #[inline]
    pub fn last_known_price_raw(&self, row: usize) -> f32 {
        self.last_known_price_raw[row]
    }

    /// Decode a single feature value from quantized u16 storage.
    #[inline]
    pub fn get_feature(&self, row: usize, feat_idx: usize) -> f32 {
        let raw = self.feature_data[row * self.num_features + feat_idx];
        if raw == NAN_U16 {
            return f32::NAN;
        }
        if feat_idx >= self.num_numeric {
            raw as f32
        } else {
            raw as f32 * self.dequant_a[feat_idx] + self.quant_min[feat_idx]
        }
    }

    /// Get a QuantRef for passing to aggregation/filter functions.
    pub fn quant_ref(&self) -> QuantRef<'_> {
        QuantRef {
            dequant_a: &self.dequant_a,
            quant_min: &self.quant_min,
            quant_range: &self.quant_range,
            num_numeric: self.num_numeric,
        }
    }
}

#[cfg(test)]
impl PropertyData {
    /// Minimal empty instance for integration tests that need an `AppState`
    /// but never touch property data (e.g. checkout/webhook/invite flows).
    pub(crate) fn empty_for_tests() -> Self {
        PropertyData {
            lat: Vec::new(),
            lon: Vec::new(),
            feature_names: Vec::new(),
            num_features: 0,
            num_numeric: 0,
            feature_data: SpillVec::owned(Vec::new()),
            dequant_a: Vec::new(),
            quant_min: Vec::new(),
            quant_range: Vec::new(),
            feature_stats: Vec::new(),
            poi_metrics: PostcodePoiMetrics::empty(0),
            last_known_price_raw: Vec::new(),
            address_buffer: SpillVec::owned(Vec::new()),
            address_offsets: SpillVec::owned(Vec::new()),
            address_lengths: SpillVec::owned(Vec::new()),
            postcode_interner: lasso::Rodeo::default().into_reader(),
            postcode_keys: SpillVec::owned(Vec::new()),
            postcode_row_index: FxHashMap::default(),
            sector_postcode_index: FxHashMap::default(),
            outcode_postcode_index: FxHashMap::default(),
            address_token_index: FxHashMap::default(),
            address_prefix_index: FxHashMap::default(),
            address_search_interner: lasso::Rodeo::default().into_resolver(),
            address_search_token_keys: SpillVec::owned(Vec::new()),
            address_search_token_offsets: SpillVec::owned(Vec::new()),
            address_search_token_lengths: SpillVec::owned(Vec::new()),
            enum_values: rustc_hash::FxHashMap::default(),
            enum_counts: rustc_hash::FxHashMap::default(),
            approx_build_date_bits: Vec::new(),
            renovation_history: FxHashMap::default(),
            historical_prices: FxHashMap::default(),
            tenure_history: FxHashMap::default(),
            property_sub_type: FxHashMap::default(),
            price_qualifier: FxHashMap::default(),
        }
    }
}