diff --git a/server-rs/src/data.rs b/server-rs/src/data.rs index 15db5f5..7a7c181 100644 --- a/server-rs/src/data.rs +++ b/server-rs/src/data.rs @@ -1,5 +1,7 @@ mod poi; +mod postcodes; mod property; pub use poi::{POICategoryGroup, POIData}; +pub use postcodes::PostcodeData; pub use property::{precompute_h3, Histogram, PropertyData}; diff --git a/server-rs/src/main.rs b/server-rs/src/main.rs index ca57999..8271d6d 100644 --- a/server-rs/src/main.rs +++ b/server-rs/src/main.rs @@ -1,6 +1,7 @@ mod consts; mod data; mod features; +mod metrics; mod og_middleware; pub mod parsing; mod routes; @@ -35,6 +36,14 @@ struct Cli { #[arg(long)] pois: PathBuf, + /// Path to the postcode boundaries directory + #[arg(long)] + postcodes: PathBuf, + + /// Path to the PMTiles file for map tiles + #[arg(long)] + tiles: PathBuf, + /// Path to the frontend dist directory #[arg(long)] dist: Option, @@ -61,6 +70,10 @@ async fn main() -> anyhow::Result<()> { .with_ansi(true) .init(); + // Initialize Prometheus metrics + let metrics_handle = metrics::init_metrics(); + info!("Prometheus metrics initialized"); + let cli = Cli::parse(); let parquet_path = &cli.data; @@ -107,6 +120,33 @@ async fn main() -> anyhow::Result<()> { let poi_grid = utils::GridIndex::build(&poi_data.lat, &poi_data.lng, consts::GRID_CELL_SIZE); + // Load postcode boundaries + let postcodes_path = &cli.postcodes; + if !postcodes_path.exists() { + bail!( + "Postcode boundaries not found: {}", + postcodes_path.display() + ); + } + info!( + "Loading postcode boundaries from {}", + postcodes_path.display() + ); + let postcode_data = data::PostcodeData::load(postcodes_path)?; + info!( + postcodes = postcode_data.postcodes.len(), + "Postcode boundaries loaded" + ); + + // Initialize tile reader + let tiles_path = &cli.tiles; + if !tiles_path.exists() { + bail!("PMTiles file not found: {}", tiles_path.display()); + } + info!("Loading PMTiles from {}", tiles_path.display()); + let tile_reader = Arc::new(routes::init_tile_reader(tiles_path).await?); + info!("PMTiles loaded successfully"); + let min_keys: Vec = property_data .feature_names .iter() @@ -165,12 +205,20 @@ async fn main() -> anyhow::Result<()> { "Precomputed features response" ); + // Record data loading metrics + metrics::record_data_stats( + property_data.lat.len(), + poi_data.lat.len(), + postcode_data.postcodes.len(), + ); + let state = Arc::new(AppState { data: property_data, grid, h3_cells, poi_data, poi_grid, + postcode_data, min_keys, max_keys, poi_category_groups, @@ -188,6 +236,7 @@ async fn main() -> anyhow::Result<()> { let state_features = state.clone(); let state_hexagons = state.clone(); + let state_postcodes = state.clone(); let state_pois = state.clone(); let state_poi_categories = state.clone(); let state_hexagon_properties = state.clone(); @@ -204,6 +253,10 @@ async fn main() -> anyhow::Result<()> { "/api/hexagons", get(move |query| routes::get_hexagons(state_hexagons.clone(), query)), ) + .route( + "/api/postcodes", + get(move |query| routes::get_postcodes(state_postcodes.clone(), query)), + ) .route( "/api/pois", get(move |query| routes::get_pois(state_pois.clone(), query)), @@ -227,6 +280,22 @@ async fn main() -> anyhow::Result<()> { get(move |query| routes::get_og_image(state_og_image.clone(), query)), ); + // Add tile routes + let reader_tile = tile_reader.clone(); + let reader_style = tile_reader.clone(); + let api = api + .route( + "/api/tiles/{z}/{x}/{y}", + get(move |path| routes::get_tile(axum::extract::State(reader_tile.clone()), path)), + ) + .route( + "/api/tiles/style.json", + get(move |headers, query| { + routes::get_style(axum::extract::State(reader_style.clone()), headers, query) + }), + ) + .route("/metrics", get(move || metrics::metrics_handler(metrics_handle.clone()))); + let app = if frontend_dist.exists() { api.fallback_service(ServeDir::new(&frontend_dist)) } else { @@ -234,6 +303,7 @@ async fn main() -> anyhow::Result<()> { }; let app = app + .layer(middleware::from_fn(metrics::track_metrics)) .layer(middleware::from_fn( move |req: axum::extract::Request, next: middleware::Next| { let st = state_crawler.clone(); diff --git a/server-rs/src/routes.rs b/server-rs/src/routes.rs index 3ee7d9f..129a473 100644 --- a/server-rs/src/routes.rs +++ b/server-rs/src/routes.rs @@ -3,11 +3,15 @@ mod hexagon_stats; pub(crate) mod hexagons; mod og_image; mod pois; +mod postcodes; pub(crate) mod properties; +mod tiles; pub use features::{build_features_response, get_features, FeaturesResponse}; pub use hexagon_stats::get_hexagon_stats; pub use hexagons::get_hexagons; pub use og_image::get_og_image; pub use pois::{get_poi_categories, get_pois}; +pub use postcodes::get_postcodes; pub use properties::get_hexagon_properties; +pub use tiles::{get_style, get_tile, init_tile_reader}; diff --git a/server-rs/src/routes/postcodes.rs b/server-rs/src/routes/postcodes.rs new file mode 100644 index 0000000..50b9980 --- /dev/null +++ b/server-rs/src/routes/postcodes.rs @@ -0,0 +1,246 @@ +use std::sync::Arc; + +use axum::extract::Query; +use axum::http::StatusCode; +use axum::response::Json; +use rustc_hash::FxHashMap; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use tracing::info; + +use crate::parsing::{parse_bounds, parse_filters, row_passes_filters}; +use crate::state::AppState; + +#[derive(Serialize)] +pub struct PostcodesResponse { + features: Vec>, +} + +#[derive(Deserialize)] +pub struct PostcodeParams { + bounds: Option, + /// Comma-separated filters: `name:min:max,...` + filters: Option, + /// Comma-separated feature names to include in min/max aggregation. + fields: Option, +} + +/// Per-postcode accumulator for aggregating features. +struct PostcodeAgg { + count: u32, + mins: Box<[f32]>, + maxs: Box<[f32]>, +} + +impl PostcodeAgg { + fn new(num_features: usize) -> Self { + PostcodeAgg { + count: 0, + mins: vec![f32::INFINITY; num_features].into_boxed_slice(), + maxs: vec![f32::NEG_INFINITY; num_features].into_boxed_slice(), + } + } + + #[inline] + fn add_row(&mut self, feature_data: &[f32], row: usize, num_features: usize) { + self.count += 1; + let base = row * num_features; + let row_slice = &feature_data[base..base + num_features]; + for (feat_index, &value) in row_slice.iter().enumerate() { + if value.is_finite() { + if value < self.mins[feat_index] { + self.mins[feat_index] = value; + } + if value > self.maxs[feat_index] { + self.maxs[feat_index] = value; + } + } + } + } + + #[inline] + fn add_row_selective( + &mut self, + feature_data: &[f32], + row: usize, + num_features: usize, + indices: &[usize], + ) { + self.count += 1; + let base = row * num_features; + for &feat_index in indices { + let value = feature_data[base + feat_index]; + if value.is_finite() { + if value < self.mins[feat_index] { + self.mins[feat_index] = value; + } + if value > self.maxs[feat_index] { + self.maxs[feat_index] = value; + } + } + } + } +} + +pub async fn get_postcodes( + state: Arc, + Query(params): Query, +) -> Result, (StatusCode, String)> { + let bounds_str = params.bounds.ok_or(( + StatusCode::BAD_REQUEST, + "bounds parameter is required".into(), + ))?; + + let (south, west, north, east) = parse_bounds(&bounds_str)?; + + let filters_str = params.filters.clone(); + let (parsed_filters, parsed_enum_filters) = parse_filters( + params.filters.as_deref(), + &state.data.feature_names, + &state.data.enum_values, + ); + let num_filters = parsed_filters.len() + parsed_enum_filters.len(); + + // Parse optional `fields` param into feature indices + let field_indices: Option> = params.fields.as_ref().map(|fields_str| { + if fields_str.is_empty() { + return Vec::new(); + } + fields_str + .split(',') + .filter_map(|name| { + let name = name.trim(); + if name.is_empty() { + return None; + } + state + .data + .feature_names + .iter() + .position(|feat| feat == name) + }) + .collect() + }); + + let response = tokio::task::spawn_blocking(move || -> Result { + let postcode_data = &state.postcode_data; + let t0 = std::time::Instant::now(); + + let num_features = state.data.num_features; + let feature_data = &state.data.feature_data; + let min_keys = &state.min_keys; + let max_keys = &state.max_keys; + + let has_selective = field_indices.is_some(); + let sel_indices = field_indices.as_deref().unwrap_or(&[]); + + // Step 1: Find postcodes within bounds using spatial grid on centroids + let postcode_indices: Vec = postcode_data.grid.query(south, west, north, east); + + // Step 2: For each postcode, aggregate properties + let mut postcode_aggs: FxHashMap = FxHashMap::default(); + + // Build postcode -> rows mapping by iterating properties in bounds + // and grouping by their postcode + let mut postcode_rows: FxHashMap> = FxHashMap::default(); + + state + .grid + .for_each_in_bounds(south, west, north, east, |row_idx| { + let row = row_idx as usize; + if !row_passes_filters( + row, + &parsed_filters, + &parsed_enum_filters, + feature_data, + num_features, + ) { + return; + } + + // Get postcode for this property + let postcode = state.data.postcode(row); + if let Some(&pc_idx) = postcode_data.postcode_to_idx.get(postcode) { + postcode_rows.entry(pc_idx).or_default().push(row); + } + }); + + // Now aggregate for each postcode that's in bounds and has properties + for &pc_idx in &postcode_indices { + let idx = pc_idx as usize; + if let Some(rows) = postcode_rows.get(&idx) { + let agg = postcode_aggs + .entry(idx) + .or_insert_with(|| PostcodeAgg::new(num_features)); + for &row in rows { + if has_selective { + agg.add_row_selective(feature_data, row, num_features, sel_indices); + } else { + agg.add_row(feature_data, row, num_features); + } + } + } + } + + // Build response + let mut features = Vec::with_capacity(postcode_aggs.len()); + + for (pc_idx, aggregation) in postcode_aggs { + if aggregation.count == 0 { + continue; + } + + let mut map = Map::new(); + map.insert( + "postcode".into(), + Value::String(postcode_data.postcodes[pc_idx].clone()), + ); + map.insert("count".into(), Value::Number(aggregation.count.into())); + + // Add vertices as array of [lon, lat] pairs + let vertices_array: Vec = postcode_data.vertices[pc_idx] + .iter() + .map(|[lon, lat]| Value::Array(vec![Value::from(*lon as f64), Value::from(*lat as f64)])) + .collect(); + map.insert("vertices".into(), Value::Array(vertices_array)); + + let iter: Box> = if let Some(idx) = field_indices.as_ref() { + Box::new(idx.iter().copied()) + } else { + Box::new(0..num_features) + }; + + for feat_index in iter { + if aggregation.mins[feat_index].is_finite() + && aggregation.maxs[feat_index].is_finite() + { + if let (Some(min_num), Some(max_num)) = ( + serde_json::Number::from_f64(aggregation.mins[feat_index] as f64), + serde_json::Number::from_f64(aggregation.maxs[feat_index] as f64), + ) { + map.insert(min_keys[feat_index].clone(), Value::Number(min_num)); + map.insert(max_keys[feat_index].clone(), Value::Number(max_num)); + } + } + } + + features.push(map); + } + + let t_total = t0.elapsed(); + info!( + postcodes = features.len(), + filters = num_filters, + filters_raw = filters_str.as_deref().unwrap_or("-"), + total_ms = format_args!("{:.1}", t_total.as_secs_f64() * 1000.0), + "GET /api/postcodes" + ); + + Ok(PostcodesResponse { features }) + }) + .await + .map_err(|error| (StatusCode::INTERNAL_SERVER_ERROR, error.to_string()))? + .map_err(|error| (StatusCode::INTERNAL_SERVER_ERROR, error))?; + + Ok(Json(response)) +} diff --git a/server-rs/src/state.rs b/server-rs/src/state.rs index 5b52ec4..043b109 100644 --- a/server-rs/src/state.rs +++ b/server-rs/src/state.rs @@ -1,4 +1,4 @@ -use crate::data::{POICategoryGroup, POIData, PropertyData}; +use crate::data::{POICategoryGroup, POIData, PostcodeData, PropertyData}; use crate::routes::FeaturesResponse; use crate::utils::GridIndex; @@ -10,6 +10,8 @@ pub struct AppState { pub h3_cells: Vec, pub poi_data: POIData, pub poi_grid: GridIndex, + /// Postcode boundary data for high-zoom rendering + pub postcode_data: PostcodeData, /// Precomputed JSON key names: "min_{feature_name}" for each feature pub min_keys: Vec, /// Precomputed JSON key names: "max_{feature_name}" for each feature