diff --git a/Cargo.lock b/Cargo.lock index e25cacb..9cb9a7a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -561,6 +561,21 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + [[package]] name = "crossterm" version = "0.28.1" @@ -598,6 +613,12 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" + [[package]] name = "derive_arbitrary" version = "1.4.2" @@ -867,6 +888,7 @@ dependencies = [ "tokio-util", "tower-http", "tracing", + "tracing-appender", "tracing-subscriber", ] @@ -1474,6 +1496,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "num-integer" version = "0.1.47" @@ -1598,6 +1626,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -2088,6 +2122,12 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "2.0.119" @@ -2204,6 +2244,36 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + +[[package]] +name = "time-macros" +version = "0.2.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e689342a48d2ea927c87ea50cabf8594854bf940e9310208848d680d668ed85" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tiny-keccak" version = "2.0.2" @@ -2366,6 +2436,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" diff --git a/Cargo.toml b/Cargo.toml index 75fb93c..2ca4e5f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ serde_json = "1" thiserror = "2" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +tracing-appender = "0.2" anyhow = "1" duckdb = { version = "1.10505", features = ["bundled"] } axum = "0.8" diff --git a/src/api.rs b/src/api.rs index 1265698..5af2cdb 100644 --- a/src/api.rs +++ b/src/api.rs @@ -1,6 +1,6 @@ use crate::db; use crate::types::{ - Config, LandingDto, MonthlyAggregate, PriceTrend, SpeciesDto, SummaryDto, TopSpecies, + Config, LandingDto, LookupDto, MonthlyAggregate, PriceTrend, SpeciesDto, SummaryDto, TopSpecies, }; use axum::{ Json, Router, @@ -394,50 +394,51 @@ async fn get_summary( async fn export_parquet(State(state): State) -> ApiResult { let conn = state.conn.clone(); - let (file_path, file) = - tokio::task::spawn_blocking(move || -> ApiResult<(String, std::fs::File)> { - let conn = conn.blocking_lock(); + let file_path = tokio::task::spawn_blocking(move || -> ApiResult { + let conn = conn.blocking_lock(); - // Use builder to ensure consistent naming for cleanup - let tmp = tempfile::Builder::new() - .prefix("hagfish-") - .suffix(".parquet") - .tempfile() - .map_err(|e| ApiError { - status: StatusCode::INTERNAL_SERVER_ERROR, - message: format!("Failed to create temp file: {e}"), - })?; - - let (_kept_file, path) = tmp.keep().map_err(|e| ApiError { + let tmp = tempfile::Builder::new() + .prefix("hagfish-") + .suffix(".parquet") + .tempfile() + .map_err(|e| ApiError { status: StatusCode::INTERNAL_SERVER_ERROR, - message: format!("Failed to persist temp file: {e}"), + message: format!("Failed to create temp file: {e}"), })?; - let path_str = path - .to_str() - .ok_or_else(|| ApiError { - status: StatusCode::INTERNAL_SERVER_ERROR, - message: "Temp path contains invalid UTF-8".to_string(), - })? - .to_string(); + let (_kept_file, path) = tmp.keep().map_err(|e| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("Failed to persist temp file: {e}"), + })?; - db::export_parquet(&conn, &path_str)?; - - let file = std::fs::File::open(&path_str).map_err(|e| ApiError { + let path_str = path + .to_str() + .ok_or_else(|| ApiError { status: StatusCode::INTERNAL_SERVER_ERROR, - message: format!("Failed to open exported file: {e}"), - })?; + message: "Temp path contains invalid UTF-8".to_string(), + })? + .to_string(); - Ok((path_str, file)) - }) + db::export_parquet(&conn, &path_str)?; + + drop(conn); + + Ok(path_str) + }) + .await + .map_err(|e| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("Task join error: {e}"), + })??; + + let file = tokio::fs::File::open(&file_path) .await .map_err(|e| ApiError { status: StatusCode::INTERNAL_SERVER_ERROR, - message: format!("Task join error: {e}"), - })??; + message: format!("Failed to open exported file: {e}"), + })?; - let tokio_file = tokio::fs::File::from(file); - let stream = tokio_util::io::ReaderStream::new(tokio_file); + let stream = tokio_util::io::ReaderStream::new(file); let body = Body::from_stream(stream); let cleanup_path = file_path.clone(); @@ -448,7 +449,7 @@ async fn export_parquet(State(state): State) -> ApiResult { } }); - Ok(Response::builder() + Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") .header( @@ -456,7 +457,10 @@ async fn export_parquet(State(state): State) -> ApiResult { "attachment; filename=\"landings.parquet\"", ) .body(body) - .unwrap()) + .map_err(|e| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("Failed to build response: {e}"), + }) } async fn static_handler(uri: axum::http::Uri) -> Response { @@ -498,13 +502,13 @@ async fn static_handler(uri: axum::http::Uri) -> Response { } } -async fn get_zones(State(state): State) -> ApiResult>> { +async fn get_zones(State(state): State) -> ApiResult>> { let conn = state.conn.clone(); - let rows = tokio::task::spawn_blocking(move || -> db::Result> { + let rows = tokio::task::spawn_blocking(move || -> db::Result> { let conn = conn.blocking_lock(); let mut stmt = conn.prepare("SELECT code, label FROM zone ORDER BY code")?; let rows = stmt.query_map([], |row| { - Ok(SpeciesDto { + Ok(LookupDto { code: row.get(0)?, label: row.get(1)?, }) @@ -520,13 +524,13 @@ async fn get_zones(State(state): State) -> ApiResult) -> ApiResult>> { +async fn get_gear(State(state): State) -> ApiResult>> { let conn = state.conn.clone(); - let rows = tokio::task::spawn_blocking(move || -> db::Result> { + let rows = tokio::task::spawn_blocking(move || -> db::Result> { let conn = conn.blocking_lock(); let mut stmt = conn.prepare("SELECT code, label FROM gear ORDER BY code")?; let rows = stmt.query_map([], |row| { - Ok(SpeciesDto { + Ok(LookupDto { code: row.get(0)?, label: row.get(1)?, }) @@ -962,7 +966,7 @@ mod tests { #[tokio::test] async fn test_concurrent_export_requests() { let state = test_state(); - let app = build_router(state.clone()); + let app = build_router(state); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -970,12 +974,11 @@ mod tests { axum::serve(listener, app).await.unwrap(); }); - let resp1 = reqwest::get(format!("http://{addr}/api/export.parquet")) - .await - .unwrap(); - let resp2 = reqwest::get(format!("http://{addr}/api/export.parquet")) - .await - .unwrap(); + let url = format!("http://{addr}/api/export.parquet"); + let (resp1, resp2) = tokio::join!(reqwest::get(&url), reqwest::get(&url)); + + let resp1 = resp1.unwrap(); + let resp2 = resp2.unwrap(); assert_eq!(resp1.status(), StatusCode::OK); assert_eq!(resp2.status(), StatusCode::OK); @@ -985,8 +988,8 @@ mod tests { assert_eq!(&bytes1[..4], b"PAR1"); assert_eq!(&bytes2[..4], b"PAR1"); - assert_ne!(bytes1.len(), 0); - assert_ne!(bytes2.len(), 0); + assert!(!bytes1.is_empty()); + assert!(!bytes2.is_empty()); } #[tokio::test] diff --git a/src/cli.rs b/src/cli.rs index 6c8f7bd..2185fd8 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -1,13 +1,11 @@ use clap::{Parser, Subcommand}; use std::path::PathBuf; -/// Faroese fisheries data pipeline and dashboard #[derive(Parser)] #[command(name = "hagfish", version, about)] pub struct Cli { - /// Path to configuration file - #[arg(short, long, value_name = "FILE", default_value = "config.json")] - pub config: PathBuf, + #[arg(short, long, value_name = "FILE")] + pub config: Option, #[command(subcommand)] pub command: Command, @@ -15,17 +13,12 @@ pub struct Cli { #[derive(Subcommand)] pub enum Command { - /// Fetch data from Hagstova API and store in DuckDB Ingest { - /// Force full backfill instead of incremental #[arg(long)] full: bool, }, - /// Start the HTTP server Serve, - /// Export landings data to Parquet Export { - /// Output file path #[arg(short, long, value_name = "FILE")] out: PathBuf, }, diff --git a/src/db.rs b/src/db.rs index 266f64e..ac07a1e 100644 --- a/src/db.rs +++ b/src/db.rs @@ -17,7 +17,7 @@ const LOOKUP_TABLES: &[(&str, &str)] = &[ pub enum DbError { #[error("DuckDB error: {0}")] Duckdb(#[from] duckdb::Error), - #[error("invalid export path: contains single quote: {0}")] + #[error("invalid export path: {0}")] InvalidPath(String), } @@ -156,6 +156,16 @@ pub fn export_parquet(conn: &Connection, path: &str) -> Result<()> { if path.contains('\'') || path.contains('\0') { return Err(DbError::InvalidPath(path.to_string())); } + + if !path + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '/' | '-' | '_' | ' ')) + { + return Err(DbError::InvalidPath(format!( + "Path contains disallowed characters: {path}" + ))); + } + conn.execute( &format!( "COPY (SELECT * FROM landings) TO '{}' (FORMAT PARQUET)", diff --git a/src/ingest.rs b/src/ingest.rs index 2169706..42b0273 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -12,12 +12,10 @@ const DIM_PRESERVATION: &str = "Preservation (EUMOFAPreservation)"; const DIM_SHIPSIZE: &str = "Shipsize"; const DIM_MEASURE: &str = "measure"; -const SENTINEL_VALUES: [f64; 1] = [-1.0]; +const SENTINEL_VALUE: f64 = -1.0; fn is_sentinel(v: f64) -> bool { - SENTINEL_VALUES - .iter() - .any(|s| v.total_cmp(s) == std::cmp::Ordering::Equal) + v.total_cmp(&SENTINEL_VALUE) == std::cmp::Ordering::Equal } pub async fn fetch_metadata(client: &Client, url: &str) -> Result<(LookupMap, MetadataResponse)> { @@ -83,13 +81,14 @@ pub fn chunk_months(months: &[String], batch_size: usize) -> Vec> { .collect() } -pub fn build_query(all_months: &[String]) -> Query { - assert!( - !all_months.is_empty(), - "build_query requires at least one month" - ); +pub fn build_query(all_months: &[String]) -> Result { + if all_months.is_empty() { + return Err(IngestError::InvalidValueCode( + "build_query requires at least one month".to_string(), + )); + } - Query { + Ok(Query { query: vec![ QueryItem { code: DIM_MONTH.to_string(), @@ -149,7 +148,7 @@ pub fn build_query(all_months: &[String]) -> Query { }, ], response: QueryResponse::default(), - } + }) } pub async fn fetch_data(client: &Client, url: &str, query: &Query) -> Result { diff --git a/src/main.rs b/src/main.rs index 3b643bf..4099ccc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,16 +13,45 @@ use clap::Parser; use tokio::net::TcpListener; use tracing_subscriber::EnvFilter; -fn init_logging() { +struct LoggerGuard { + _guard: Option, +} + +fn init_logging(log_file_path: Option<&str>) -> LoggerGuard { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); - if std::env::var("HAGFISH_LOG_FORMAT").as_deref() == Ok("json") { - tracing_subscriber::fmt() - .with_env_filter(filter) - .json() - .init(); - } else { - tracing_subscriber::fmt().with_env_filter(filter).init(); + let is_json = std::env::var("HAGFISH_LOG_FORMAT").as_deref() == Ok("json"); + + match log_file_path { + Some(path) => { + let file = std::fs::File::create(path).expect("Failed to create log file"); + let (non_blocking, guard) = tracing_appender::non_blocking(file); + + let subscriber = tracing_subscriber::fmt() + .with_env_filter(filter.clone()) + .with_writer(non_blocking); + + if is_json { + subscriber.json().init(); + } else { + subscriber.init(); + } + + LoggerGuard { + _guard: Some(guard), + } + } + None => { + let subscriber = tracing_subscriber::fmt().with_env_filter(filter.clone()); + + if is_json { + subscriber.json().init(); + } else { + subscriber.init(); + } + + LoggerGuard { _guard: None } + } } } @@ -51,10 +80,23 @@ fn cleanup_stale_parquet_files() { #[tokio::main] async fn main() -> anyhow::Result<()> { - init_logging(); - let cli = cli::Cli::parse(); - let config = types::Config::load(&cli.config)?; + + let config = match &cli.config { + Some(path) => types::Config::load(path)?, + None => { + let default_path = std::path::Path::new("config.json"); + if default_path.exists() { + types::Config::load(default_path)? + } else { + tracing::warn!("No config file found, using defaults"); + types::Config::default() + } + } + }; + + // Initialize logging AFTER config is loaded so we have log_file_path + let _logger_guard = init_logging(config.log_file_path.as_deref()); match cli.command { cli::Command::Serve => run_serve(&config).await, @@ -142,7 +184,6 @@ async fn run_ingest(config: &types::Config, full: bool) -> anyhow::Result<()> { ); let mut total_rows = 0; - let mut total_skipped = 0; let mut errors = 0; for (i, batch) in batches.iter().enumerate() { @@ -156,7 +197,7 @@ async fn run_ingest(config: &types::Config, full: bool) -> anyhow::Result<()> { "fetching batch" ); - let query = ingest::build_query(batch); + let query = ingest::build_query(batch)?; let data = match ingest::fetch_data(&client, &config.data_source_url, &query).await { Ok(d) => d, @@ -178,11 +219,7 @@ async fn run_ingest(config: &types::Config, full: bool) -> anyhow::Result<()> { for j in 0..data.dataset.value.len() { match ingest::parse_row(j, &data.dataset, &lookup_maps) { Ok(row) => { - if row.value.is_some() { - landings.push(ingest::data_row_to_landing(&row)); - } else { - total_skipped += 1; - } + landings.push(ingest::data_row_to_landing(&row)); } Err(e) => { tracing::warn!( @@ -218,11 +255,15 @@ async fn run_ingest(config: &types::Config, full: bool) -> anyhow::Result<()> { batch = i + 1, total_batches = batches.len(), rows_inserted = count, - null_rows_skipped = total_skipped, parse_errors = parse_errors, elapsed_ms = batch_elapsed.as_millis() as u64, "batch complete" ); + + // Backoff between batches to avoid rate limiting + if i < batches.len() - 1 { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } } let elapsed = run_start.elapsed(); @@ -232,7 +273,6 @@ async fn run_ingest(config: &types::Config, full: bool) -> anyhow::Result<()> { months_requested = months_to_fetch.len(), batches = batches.len(), total_rows = total_rows, - null_rows_skipped = total_skipped, errors = errors, elapsed_ms = elapsed.as_millis() as u64, "ingestion run complete" diff --git a/src/types.rs b/src/types.rs index fad5978..54d47dc 100644 --- a/src/types.rs +++ b/src/types.rs @@ -31,15 +31,12 @@ impl Default for Config { impl Config { pub fn load(path: &Path) -> anyhow::Result { - if path.exists() { - let contents = std::fs::read_to_string(path)?; - let config: Config = serde_json::from_str(&contents)?; - tracing::info!("Loaded config from {}", path.display()); - Ok(config) - } else { - tracing::warn!("Config file {} not found, using defaults", path.display()); - Ok(Self::default()) - } + let contents = std::fs::read_to_string(path) + .map_err(|e| anyhow::anyhow!("Failed to read config file {}: {e}", path.display()))?; + let config: Config = serde_json::from_str(&contents) + .map_err(|e| anyhow::anyhow!("Failed to parse config file {}: {e}", path.display()))?; + tracing::info!("Loaded config from {}", path.display()); + Ok(config) } } @@ -179,6 +176,12 @@ pub struct SpeciesDto { pub label: String, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LookupDto { + pub code: String, + pub label: String, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct LandingDto { pub month: String,