From 45bcc1378860175f1d6262444344dd960736608e Mon Sep 17 00:00:00 2001 From: Jonas Rabenstein Date: Wed, 12 Aug 2026 17:49:27 +0200 Subject: [PATCH] state --- scripts/import-vg250.sh | 97 +++++++-- src/intervals.rs | 431 +++++++++++++++++++++++++++++++--------- src/main.rs | 316 +++++++++++++++++++++++------ 3 files changed, 674 insertions(+), 170 deletions(-) diff --git a/scripts/import-vg250.sh b/scripts/import-vg250.sh index 5ea96ca..406f33d 100755 --- a/scripts/import-vg250.sh +++ b/scripts/import-vg250.sh @@ -16,24 +16,55 @@ fi : "${DATABASE_URL:?DATABASE_URL must be set}" echo "Available layers:" -ogrinfo "$INPUT" | sed -n '1,80p' +ogrinfo "$INPUT" | sed -n '1,120p' echo -echo "Importing vg250_krs..." +echo "Inspecting vg250_krs..." +ogrinfo "$INPUT" vg250_krs -so +echo +echo "Importing only vg250_krs..." + +# Remove an old failed import. +psql "$DATABASE_URL" <<'SQL' +DROP TABLE IF EXISTS vg250_krs_import CASCADE; +SQL + +# IMPORTANT: +# The layer name is explicitly specified here. ogr2ogr \ -f PostgreSQL \ "$DATABASE_URL" \ "$INPUT" \ + vg250_krs \ -nln vg250_krs_import \ -nlt PROMOTE_TO_MULTI \ -t_srs EPSG:4326 \ + -lco GEOMETRY_NAME=wkb_geometry \ -overwrite -echo "Import completed." +echo +echo "Source import completed." + +echo +echo "Imported source columns:" +psql "$DATABASE_URL" <<'SQL' +SELECT + column_name, + data_type +FROM information_schema.columns +WHERE table_name = 'vg250_krs_import' +ORDER BY ordinal_position; +SQL + +echo +echo "Creating county boundaries..." psql "$DATABASE_URL" <<'SQL' -TRUNCATE county_boundaries RESTART IDENTITY; +BEGIN; + +TRUNCATE county_crossings; +TRUNCATE county_boundaries RESTART IDENTITY CASCADE; INSERT INTO county_boundaries ( name, @@ -41,9 +72,9 @@ INSERT INTO county_boundaries ( ) SELECT COALESCE( - NULLIF("GEN", ''), - NULLIF("BEZ", ''), - NULLIF("ARS", ''), + NULLIF(TRIM(gen), ''), + NULLIF(TRIM(bez), ''), + NULLIF(TRIM(ars), ''), 'unknown' ) AS name, @@ -52,13 +83,29 @@ SELECT ST_MakeValid(wkb_geometry), 3 ) - )::geometry(MultiPolygon,4326) + )::geometry(MultiPolygon, 4326) AS geometry FROM vg250_krs_import -WHERE wkb_geometry IS NOT NULL; +WHERE + wkb_geometry IS NOT NULL + AND NOT ST_IsEmpty(wkb_geometry); +COMMIT; +SQL + +echo +echo "Removing temporary import table..." + +psql "$DATABASE_URL" <<'SQL' DROP TABLE vg250_krs_import; +SQL + +echo +echo "Recreating spatial index..." + +psql "$DATABASE_URL" <<'SQL' +DROP INDEX IF EXISTS county_boundaries_geometry_idx; CREATE INDEX county_boundaries_geometry_idx ON county_boundaries @@ -68,6 +115,32 @@ ANALYZE county_boundaries; SQL echo -echo "Imported counties:" -psql "$DATABASE_URL" \ - -c "SELECT id, name FROM county_boundaries ORDER BY name;" +echo "==========================================" +echo " Imported counties" +echo "==========================================" + +psql "$DATABASE_URL" <<'SQL' +SELECT + COUNT(*) AS total, + COUNT(*) FILTER (WHERE geometry IS NOT NULL) AS with_geometry, + COUNT(*) FILTER (WHERE ST_IsValid(geometry)) AS valid_geometry +FROM county_boundaries; +SQL + +echo +echo "First 20 counties:" +echo + +psql "$DATABASE_URL" <<'SQL' +SELECT + id, + name, + ST_GeometryType(geometry) AS geometry_type, + ST_SRID(geometry) AS srid +FROM county_boundaries +ORDER BY name +LIMIT 20; +SQL + +echo +echo "Import completed successfully." diff --git a/src/intervals.rs b/src/intervals.rs index e6048eb..44083c4 100644 --- a/src/intervals.rs +++ b/src/intervals.rs @@ -1,4 +1,4 @@ -use anyhow::{Context, Result, anyhow}; +use anyhow::{anyhow, Context, Result}; use chrono::{DateTime, NaiveDateTime, Utc}; use reqwest::Client; use serde_json::Value; @@ -26,7 +26,10 @@ impl IntervalsClient { /// Create a client using an explicitly supplied personal API key. /// /// Primarily used by /dev/sync/{api_key}/{athlete_id}. - pub fn with_api_key(config: Config, api_key: impl Into) -> Self { + pub fn with_api_key( + config: Config, + api_key: impl Into, + ) -> Self { Self { client: Client::new(), config, @@ -35,13 +38,17 @@ impl IntervalsClient { } fn base_url(&self) -> &str { - self.config.intervals_base_url.trim_end_matches('/') + self.config + .intervals_base_url + .trim_end_matches('/') } fn api_key(&self) -> Result<&str> { self.api_key .as_deref() - .context("Intervals.icu API key is not configured") + .context( + "Intervals.icu API key is not configured", + ) } fn api_url(&self, path: &str) -> String { @@ -52,30 +59,54 @@ impl IntervalsClient { ) } - fn authenticated(&self, request: reqwest::RequestBuilder) -> Result { - Ok(request.basic_auth("API_KEY", Some(self.api_key()?))) + fn authenticated( + &self, + request: reqwest::RequestBuilder, + ) -> Result { + Ok( + request.basic_auth( + "API_KEY", + Some(self.api_key()?), + ) + ) } - async fn get_json(&self, url: String) -> Result { + async fn get_json( + &self, + url: String, + ) -> Result { let response = self .authenticated(self.client.get(&url))? .send() .await - .context("Intervals.icu request failed")?; + .context( + "Intervals.icu request failed", + )?; let status = response.status(); let body = response .text() .await - .context("cannot read Intervals.icu response")?; + .context( + "cannot read Intervals.icu response", + )?; if !status.is_success() { - return Err(anyhow!("Intervals.icu returned HTTP {}: {}", status, body)); + return Err(anyhow!( + "Intervals.icu returned HTTP {}: {}", + status, + body + )); } serde_json::from_str(&body) - .with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body)) + .with_context(|| { + format!( + "invalid JSON returned by Intervals.icu: {}", + body + ) + }) } /// Get the athlete belonging to the configured API key. @@ -83,12 +114,18 @@ impl IntervalsClient { /// GET /api/v1/athlete/0 pub async fn owner(&self) -> Result { let url = self.api_url("athlete/0"); + self.get_json(url).await } /// Get a specific athlete. - pub async fn athlete(&self, athlete_id: &str) -> Result { - let url = self.api_url(&format!("athlete/{athlete_id}")); + pub async fn athlete( + &self, + athlete_id: &str, + ) -> Result { + let url = self.api_url( + &format!("athlete/{athlete_id}"), + ); self.get_json(url).await } @@ -96,8 +133,20 @@ impl IntervalsClient { /// Get a single activity. /// /// GET /api/v1/activity/{id} - pub async fn activity(&self, activity_id: &str) -> Result { - let url = self.api_url(&format!("activity/{activity_id}?intervals=true")); + pub async fn activity( + &self, + activity_id: &str, + ) -> Result { + /* + * Ask Intervals.icu to include interval information, + * which is later used to associate county crossings + * with intervals. + */ + let url = self.api_url( + &format!( + "activity/{activity_id}?intervals=true" + ), + ); self.get_json(url).await } @@ -106,78 +155,180 @@ impl IntervalsClient { /// /// GET /api/v1/activity/{id}/streams.json /// - /// We explicitly request the streams needed for GPS processing. - pub async fn streams(&self, activity_id: &str) -> Result { - let url = self.api_url(&format!( - "activity/{activity_id}/streams.json?types=time,latlng" - )); + /// We explicitly request the streams needed for GPS + /// processing. + pub async fn streams( + &self, + activity_id: &str, + ) -> Result { + let url = self.api_url( + &format!( + "activity/{activity_id}/streams.json?types=time,latlng" + ), + ); - self.get_json(url).await + tracing::info!( + activity_id = %activity_id, + "requesting Intervals.icu streams" + ); + + let value = + self.get_json(url).await?; + + let stream_count = value + .as_array() + .map(|streams| streams.len()) + .unwrap_or(0); + + let time_points = + Self::stream_values( + &value, + "time", + ) + .map(|values| values.len()) + .unwrap_or(0); + + let (latitude_points, longitude_points) = + Self::latlng_values(&value) + .map(|(lat, lon)| { + (lat.len(), lon.len()) + }) + .unwrap_or((0, 0)); + + tracing::info!( + activity_id = %activity_id, + stream_count, + time_points, + latitude_points, + longitude_points, + "received Intervals.icu streams" + ); + + Ok(value) } /// Find a stream object by type. - pub fn find_stream<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Value> { - // Normal API response: - // - // [ - // { - // "type": "time", - // "data": [...] - // }, - // { - // "type": "latlng", - // "data": [...], - // "data2": [...] - // } - // ] - if let Some(array) = streams.as_array() { - return array - .iter() - .find(|stream| stream.get("type").and_then(Value::as_str) == Some(stream_type)); + /// + /// Normal Intervals.icu response: + /// + /// [ + /// { + /// "type": "time", + /// "data": [...] + /// }, + /// { + /// "type": "latlng", + /// "data": [...], + /// "data2": [...] + /// } + /// ] + /// + /// Also tolerates a dictionary-like response. + pub fn find_stream<'a>( + streams: &'a Value, + stream_type: &str, + ) -> Option<&'a Value> { + if let Some(array) = + streams.as_array() + { + return array.iter().find( + |stream| { + stream + .get("type") + .and_then(Value::as_str) + == Some(stream_type) + }, + ); } - // Also tolerate: - // - // { - // "time": {...}, - // "latlng": {...} - // } streams.get(stream_type) } /// Get the `data` array of a normal stream. - pub fn stream_values<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Vec> { - let stream = Self::find_stream(streams, stream_type)?; + /// + /// For example: + /// + /// { + /// "type": "time", + /// "data": [0, 1, 2, 3] + /// } + pub fn stream_values<'a>( + streams: &'a Value, + stream_type: &str, + ) -> Option<&'a Vec> { + let stream = + Self::find_stream( + streams, + stream_type, + )?; - stream.get("data").and_then(Value::as_array) + stream + .get("data") + .and_then(Value::as_array) } - /// Get the latitude and longitude arrays from the + /// Get latitude and longitude arrays from the /// Intervals.icu `latlng` stream. /// /// Intervals.icu represents latlng as: /// + /// { + /// "type": "latlng", + /// "data": [latitude, latitude, ...], + /// "data2": [longitude, longitude, ...] + /// } + /// + /// Therefore: + /// /// data = latitude /// data2 = longitude /// - pub fn latlng_values<'a>(streams: &'a Value) -> Option<(&'a Vec, &'a Vec)> { - let stream = Self::find_stream(streams, "latlng")?; + /// It is NOT: + /// + /// data = [[lat, lon], [lat, lon], ...] + pub fn latlng_values<'a>( + streams: &'a Value, + ) -> Option<( + &'a Vec, + &'a Vec, + )> { + let stream = + Self::find_stream( + streams, + "latlng", + )?; - let lat = stream.get("data")?.as_array()?; + let latitudes = stream + .get("data") + .and_then(Value::as_array)?; - let lon = stream.get("data2")?.as_array()?; + let longitudes = stream + .get("data2") + .and_then(Value::as_array)?; - Some((lat, lon)) + Some(( + latitudes, + longitudes, + )) } /// Fetch an athlete's activities. /// /// `athlete_id = "0"` means the athlete belonging to /// the API key. - pub async fn activities(&self, athlete_id: &str, oldest: &str, newest: &str) -> Result { + pub async fn activities( + &self, + athlete_id: &str, + oldest: &str, + newest: &str, + ) -> Result { let url = format!( "{}?oldest={}&newest={}", - self.api_url(&format!("athlete/{athlete_id}/activities")), + self.api_url( + &format!( + "athlete/{athlete_id}/activities" + ), + ), urlencoding::encode(oldest), urlencoding::encode(newest), ); @@ -186,39 +337,68 @@ impl IntervalsClient { } /// Extract the activity start timestamp. - pub fn activity_start(activity: &Value) -> Result> { + pub fn activity_start( + activity: &Value, + ) -> Result> { let value = activity .get("start_date") - .or_else(|| activity.get("start_date_local")) - .or_else(|| activity.get("start_time")) - .ok_or_else(|| anyhow!("activity has no start timestamp"))?; + .or_else(|| { + activity.get("start_date_local") + }) + .or_else(|| { + activity.get("start_time") + }) + .ok_or_else(|| { + anyhow!( + "activity has no start timestamp" + ) + })?; let string = value .as_str() - .ok_or_else(|| anyhow!("activity start timestamp is not a string"))?; + .ok_or_else(|| { + anyhow!( + "activity start timestamp is not a string" + ) + })?; parse_datetime(string) } /// Try to find an Intervals.icu interval corresponding /// to a GPS stream index. - pub fn matching_interval(activity: &Value, track_index: usize) -> Option { - let intervals = activity.get("icu_intervals")?; + pub fn matching_interval( + activity: &Value, + track_index: usize, + ) -> Option { + let intervals = + activity.get("icu_intervals")?; - let array = intervals.as_array()?; + let array = + intervals.as_array()?; for interval in array { let start = interval .get("start_index") - .or_else(|| interval.get("start")); + .or_else(|| { + interval.get("start") + }); - let end = interval.get("end_index").or_else(|| interval.get("end")); + let end = interval + .get("end_index") + .or_else(|| { + interval.get("end") + }); - let start = start.and_then(Value::as_u64)?; + let start = + start.and_then(Value::as_u64)?; - let end = end.and_then(Value::as_u64)?; + let end = + end.and_then(Value::as_u64)?; - if (start as usize) <= track_index && track_index <= end as usize { + if (start as usize) <= track_index + && track_index <= end as usize + { return Some(interval.clone()); } } @@ -227,18 +407,25 @@ impl IntervalsClient { } /// OAuth methods retained for the normal multi-user flow. - pub fn oauth_authorize_url(&self, state: &str) -> Result { + pub fn oauth_authorize_url( + &self, + state: &str, + ) -> Result { let client_id = self .config .intervals_client_id .as_deref() - .context("INTERVALS_CLIENT_ID is not configured")?; + .context( + "INTERVALS_CLIENT_ID is not configured", + )?; let redirect_uri = self .config .intervals_redirect_uri .as_deref() - .context("INTERVALS_REDIRECT_URI is not configured")?; + .context( + "INTERVALS_REDIRECT_URI is not configured", + )?; Ok(format!( "{}/oauth/authorize?client_id={}&redirect_uri={}&scope={}&state={}", @@ -250,22 +437,32 @@ impl IntervalsClient { )) } - pub async fn exchange_code(&self, code: &str) -> Result { + pub async fn exchange_code( + &self, + code: &str, + ) -> Result { let client_id = self .config .intervals_client_id .as_deref() - .context("INTERVALS_CLIENT_ID is not configured")?; + .context( + "INTERVALS_CLIENT_ID is not configured", + )?; let client_secret = self .config .intervals_client_secret .as_deref() - .context("INTERVALS_CLIENT_SECRET is not configured")?; + .context( + "INTERVALS_CLIENT_SECRET is not configured", + )?; let response = self .client - .post(format!("{}/api/oauth/token", self.base_url())) + .post(format!( + "{}/api/oauth/token", + self.base_url() + )) .form(&[ ("client_id", client_id), ("client_secret", client_secret), @@ -273,11 +470,15 @@ impl IntervalsClient { ]) .send() .await - .context("OAuth token request failed")?; + .context( + "OAuth token request failed", + )?; let status = response.status(); - let body = response.text().await?; + let body = response + .text() + .await?; if !status.is_success() { return Err(anyhow!( @@ -290,11 +491,20 @@ impl IntervalsClient { Ok(serde_json::from_str(&body)?) } - pub fn token_data(response: &Value) -> Result<(String, String, String, String)> { + pub fn token_data( + response: &Value, + ) -> Result<( + String, + String, + String, + String, + )> { let access_token = response .get("access_token") .and_then(Value::as_str) - .context("OAuth response has no access_token")? + .context( + "OAuth response has no access_token", + )? .to_string(); let scope = response @@ -305,7 +515,9 @@ impl IntervalsClient { let athlete = response .get("athlete") - .context("OAuth response has no athlete")?; + .context( + "OAuth response has no athlete", + )?; let athlete_id = athlete .get("id") @@ -321,27 +533,62 @@ impl IntervalsClient { let display_name = athlete .get("name") - .or_else(|| athlete.get("display_name")) + .or_else(|| { + athlete.get("display_name") + }) .and_then(Value::as_str) - .unwrap_or("Intervals.icu athlete") + .unwrap_or( + "Intervals.icu athlete", + ) .to_string(); - Ok((access_token, scope, athlete_id, display_name)) + Ok(( + access_token, + scope, + athlete_id, + display_name, + )) } } -fn parse_datetime(value: &str) -> Result> { - if let Ok(value) = DateTime::parse_from_rfc3339(value) { - return Ok(value.with_timezone(&Utc)); +fn parse_datetime( + value: &str, +) -> Result> { + if let Ok(value) = + DateTime::parse_from_rfc3339(value) + { + return Ok( + value.with_timezone(&Utc) + ); } - if let Ok(value) = DateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S%z") { - return Ok(value.with_timezone(&Utc)); + if let Ok(value) = + DateTime::parse_from_str( + value, + "%Y-%m-%dT%H:%M:%S%z", + ) + { + return Ok( + value.with_timezone(&Utc) + ); } - if let Ok(value) = NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S") { - return Ok(DateTime::::from_naive_utc_and_offset(value, Utc)); + if let Ok(value) = + NaiveDateTime::parse_from_str( + value, + "%Y-%m-%dT%H:%M:%S", + ) + { + return Ok( + DateTime::::from_naive_utc_and_offset( + value, + Utc, + ) + ); } - Err(anyhow!("cannot parse activity timestamp: {}", value)) + Err(anyhow!( + "cannot parse activity timestamp: {}", + value + )) } diff --git a/src/main.rs b/src/main.rs index ea09997..434639b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,20 +8,24 @@ mod model; mod web; mod webhook; -use anyhow::{Context, Result, anyhow}; +use anyhow::{anyhow, Context, Result}; use axum::{ - Router, extract::{Path, State}, response::Json, routing::{get, post}, + Router, }; use chrono::{DateTime, Duration, Utc}; -use serde_json::{Value, json}; +use serde_json::{json, Value}; use sqlx::PgPool; use tower_http::trace::TraceLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; -use crate::{config::Config, intervals::IntervalsClient, model::TrackPoint}; +use crate::{ + config::Config, + intervals::IntervalsClient, + model::TrackPoint, +}; #[derive(Clone)] pub struct AppState { @@ -115,7 +119,6 @@ async fn dev_sync( ) -> Result, axum::http::StatusCode> { match dev_sync_inner(state, api_key, athlete_id).await { Ok(result) => Ok(Json(result)), - Err(error) => { tracing::error!( error = %error, @@ -129,18 +132,17 @@ async fn dev_sync( /// One-segment variant: /// -/// /dev/sync/{api_key} +/// GET /dev/sync/{api_key} /// /// Equivalent to: /// -/// /dev/sync/{api_key}/0 +/// GET /dev/sync/{api_key}/0 async fn dev_sync_one( State(state): State, Path(api_key): Path, ) -> Result, axum::http::StatusCode> { match dev_sync_inner(state, api_key, "0".to_string()).await { Ok(result) => Ok(Json(result)), - Err(error) => { tracing::error!( error = %error, @@ -152,7 +154,11 @@ async fn dev_sync_one( } } -async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> Result { +async fn dev_sync_inner( + state: AppState, + api_key: String, + athlete_id: String, +) -> Result { if api_key.trim().is_empty() { return Err(anyhow!("API key is empty")); } @@ -167,7 +173,8 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> * This client is explicitly authenticated using the API key * supplied to the development endpoint. */ - let client = IntervalsClient::with_api_key(state.config.clone(), api_key.clone()); + let client = + IntervalsClient::with_api_key(state.config.clone(), api_key.clone()); /* * Resolve athlete 0 through Intervals.icu. @@ -197,11 +204,9 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> }; let newest = Utc::now().date_naive(); - let oldest = newest - Duration::days(state.config.dev_sync_days); let oldest_string = oldest.format("%Y-%m-%d").to_string(); - let newest_string = newest.format("%Y-%m-%d").to_string(); tracing::info!( @@ -212,13 +217,14 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> ); /* - * IMPORTANT: - * - * Use the resolved athlete ID here, not the original - * "0" value. + * Use the resolved athlete ID here, not the original "0". */ let activities = client - .activities(&resolved_athlete_id, &oldest_string, &newest_string) + .activities( + &resolved_athlete_id, + &oldest_string, + &newest_string, + ) .await .context("cannot fetch development activities")?; @@ -235,14 +241,21 @@ async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> /* * Make sure there is a corresponding local athlete row. * - * We store the development API key as access_token because - * process_activity() uses the local athlete record and the - * existing activity-processing pipeline expects that field. + * The development API key is stored as access_token so that + * process_activity() can use exactly the same credential. */ - ensure_dev_athlete(&state, &resolved_athlete_id, &client, &api_key).await?; + ensure_dev_athlete( + &state, + &resolved_athlete_id, + &client, + &api_key, + ) + .await?; for activity in activities_array { - let Some(activity_id) = activity.get("id").and_then(Value::as_str) else { + let Some(activity_id) = + activity.get("id").and_then(Value::as_str) + else { skipped += 1; results.push(json!({ @@ -339,9 +352,6 @@ async fn ensure_dev_athlete( /* * If this athlete already exists, update the development * credential. - * - * This is particularly useful if the row was originally - * created by an earlier development sync with an empty token. */ let existing = sqlx::query( r#" @@ -401,11 +411,18 @@ async fn ensure_dev_athlete( Ok(()) } +/// Process one activity. +/// +/// The API credential is retrieved from the local athlete record. +/// This is important for development sync because the API key +/// supplied to /dev/sync/{api_key} is stored in access_token. pub async fn process_activity( state: AppState, intervals_athlete_id: String, activity_id: String, ) -> Result<()> { + use sqlx::Row; + let athlete = sqlx::query( r#" SELECT @@ -419,8 +436,6 @@ pub async fn process_activity( .fetch_optional(&state.db) .await?; - use sqlx::Row; - let Some(athlete) = athlete else { return Err(anyhow!( "no local athlete for Intervals.icu athlete {}", @@ -430,14 +445,21 @@ pub async fn process_activity( let athlete_db_id: i64 = athlete.try_get("id")?; + let access_token: String = athlete + .try_get("access_token") + .context("local athlete has no access token")?; + /* - * The normal activity-processing path uses the configured - * Intervals client. + * IMPORTANT: * - * The development endpoint has already ensured that the - * athlete exists locally and that its credential is stored. + * Use the athlete's stored credential rather than creating + * IntervalsClient::new(), which would use only the globally + * configured INTERVALS_API_KEY. */ - let client = IntervalsClient::new(state.config.clone()); + let client = IntervalsClient::with_api_key( + state.config.clone(), + access_token, + ); /* * Fetch activity metadata. @@ -445,19 +467,27 @@ pub async fn process_activity( let activity = client .activity(&activity_id) .await - .with_context(|| format!("cannot fetch Intervals.icu activity {}", activity_id))?; + .with_context(|| { + format!( + "cannot fetch Intervals.icu activity {}", + activity_id + ) + })?; let start_time = IntervalsClient::activity_start(&activity)?; /* * Fetch GPS/time streams. */ - let streams = client.streams(&activity_id).await.with_context(|| { - format!( - "cannot fetch streams for Intervals.icu activity {}", - activity_id - ) - })?; + let streams = client + .streams(&activity_id) + .await + .with_context(|| { + format!( + "cannot fetch streams for Intervals.icu activity {}", + activity_id + ) + })?; let points = parse_track(&streams, start_time)?; @@ -485,19 +515,46 @@ pub async fn process_activity( let mut tx = state.db.begin().await?; let activity_db_id = - db::ensure_activity(&mut tx, athlete_db_id, &activity_id, start_time, &activity).await?; + db::ensure_activity( + &mut tx, + athlete_db_id, + &activity_id, + start_time, + &activity, + ) + .await?; - db::delete_activity_crossings(&mut tx, activity_db_id).await?; + db::delete_activity_crossings( + &mut tx, + activity_db_id, + ) + .await?; let mut crossing_count = 0usize; for pair in points.windows(2) { - let crossings = geo::crossings_between(&state.db, &pair[0], &pair[1]).await?; + let crossings = + geo::crossings_between( + &state.db, + &pair[0], + &pair[1], + ) + .await?; for crossing in crossings { - let interval = IntervalsClient::matching_interval(&activity, crossing.track_index); + let interval = + IntervalsClient::matching_interval( + &activity, + crossing.track_index, + ); - db::save_crossing(&mut tx, activity_db_id, &crossing, interval.as_ref()).await?; + db::save_crossing( + &mut tx, + activity_db_id, + &crossing, + interval.as_ref(), + ) + .await?; crossing_count += 1; @@ -512,12 +569,17 @@ pub async fn process_activity( } } - db::mark_activity_processed(&mut tx, activity_db_id).await?; + db::mark_activity_processed( + &mut tx, + activity_db_id, + ) + .await?; tx.commit().await?; tracing::info!( activity_id = %activity_id, + points = points.len(), crossings = crossing_count, "activity processed" ); @@ -525,14 +587,65 @@ pub async fn process_activity( Ok(()) } -fn parse_track(streams: &Value, start_time: DateTime) -> Result> { - let times = IntervalsClient::stream_values(streams, "time") - .ok_or_else(|| anyhow!("Intervals.icu response has no time stream"))?; +/// Parse Intervals.icu GPS streams. +/// +/// Intervals.icu returns the streams as an array similar to: +/// +/// [ +/// { +/// "type": "time", +/// "data": [0, 1, 2, 3] +/// }, +/// { +/// "type": "latlng", +/// "data": [51.5, 51.5001, 51.5002, 51.5003], +/// "data2": [-0.1, -0.1001, -0.1002, -0.1003] +/// } +/// ] +/// +/// IMPORTANT: +/// +/// `latlng.data` is latitude. +/// +/// `latlng.data2` is longitude. +/// +/// It is NOT: +/// +/// `latlng.data = [[lat, lon], [lat, lon], ...]` +fn parse_track( + streams: &Value, + start_time: DateTime, +) -> Result> { + let times = IntervalsClient::stream_values( + streams, + "time", + ) + .ok_or_else(|| { + anyhow!( + "Intervals.icu response has no time stream" + ) + })?; - let latlng = IntervalsClient::stream_values(streams, "latlng") - .ok_or_else(|| anyhow!("Intervals.icu response has no latlng stream"))?; + let (latitudes, longitudes) = + IntervalsClient::latlng_values(streams) + .ok_or_else(|| { + anyhow!( + "Intervals.icu response has no usable latlng stream" + ) + })?; - let length = times.len().min(latlng.len()); + let length = times + .len() + .min(latitudes.len()) + .min(longitudes.len()); + + tracing::info!( + time_points = times.len(), + latitude_points = latitudes.len(), + longitude_points = longitudes.len(), + usable_points = length, + "parsing Intervals.icu GPS streams" + ); let mut result = Vec::with_capacity(length); @@ -541,26 +654,25 @@ fn parse_track(streams: &Value, start_time: DateTime) -> Result) -> Result