From 89091a03d32a5a61581dfb91ac388174d94622e5 Mon Sep 17 00:00:00 2001 From: Jonas Rabenstein Date: Wed, 12 Aug 2026 14:02:12 +0200 Subject: [PATCH] dev api key --- Cargo.toml | 2 +- flake.nix | 27 +-- src/auth.rs | 179 +++++++++--------- src/config.rs | 85 +++++++-- src/db.rs | 41 ++--- src/geo.rs | 31 +--- src/intervals.rs | 448 ++++++++++++++++++++++----------------------- src/leaderboard.rs | 93 ++++------ src/main.rs | 428 ++++++++++++++++++++++++------------------- src/web.rs | 83 ++++----- src/webhook.rs | 122 ++++++------ 11 files changed, 784 insertions(+), 755 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 86a5324..d1a0c45 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,7 @@ rand = "0.9" reqwest = { version = "0.12", default-features = false, - features = ["json", "form", "rustls-tls"] + features = ["json", "rustls-tls"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/flake.nix b/flake.nix index de821a9..12e49de 100644 --- a/flake.nix +++ b/flake.nix @@ -80,6 +80,9 @@ mkdir -p "$PGHOST" + # Use an absolute path for the PostgreSQL Unix socket. + PGHOST_ABS="$(cd "$PGHOST" && pwd)" + # ============================================================ # Initialise PostgreSQL cluster # ============================================================ @@ -101,8 +104,6 @@ # Start PostgreSQL # # PostgreSQL is deliberately restricted to the Unix socket. - # No TCP/IP listener is opened, so there are no port conflicts - # with another local PostgreSQL instance. # ============================================================ if ! pg_ctl \ @@ -116,7 +117,7 @@ pg_ctl \ -D "$PGDATA" \ -l "$PGDATA/postgresql.log" \ - -o "-c listen_addresses= -k $PGHOST" \ + -o "-c listen_addresses= -k $PGHOST_ABS" \ start \ >/dev/null fi @@ -129,7 +130,7 @@ for _ in $(seq 1 50); do if pg_isready \ - -h "$PGHOST" \ + -h "$PGHOST_ABS" \ -U "$PGUSER" \ >/dev/null 2>&1 then @@ -164,7 +165,7 @@ # ============================================================ if ! psql \ - -h "$PGHOST" \ + -h "$PGHOST_ABS" \ -U "$PGUSER" \ -d postgres \ -tAc \ @@ -177,7 +178,7 @@ echo "==> Creating PostgreSQL database '$PGDATABASE'" createdb \ - -h "$PGHOST" \ + -h "$PGHOST_ABS" \ -U "$PGUSER" \ "$PGDATABASE" fi @@ -190,7 +191,7 @@ echo "==> Checking PostGIS" psql \ - -h "$PGHOST" \ + -h "$PGHOST_ABS" \ -U "$PGUSER" \ -d "$PGDATABASE" \ -v ON_ERROR_STOP=1 \ @@ -198,10 +199,13 @@ >/dev/null # ============================================================ - # Application connection string + # SQLx / application connection string + # + # localhost is only the syntactic host. The host query + # parameter tells libpq/SQLx to use our Unix socket. # ============================================================ - export DATABASE_URL="postgresql://$PGUSER@/$PGDATABASE?host=$PGHOST" + export DATABASE_URL="postgresql://$PGUSER@localhost/$PGDATABASE?host=$PGHOST_ABS" # ============================================================ # Development information @@ -217,14 +221,14 @@ echo echo "PostgreSQL:" echo " $(psql --version)" - echo " socket: $PGHOST" + echo " socket: $PGHOST_ABS" echo " database: $PGDATABASE" echo " user: $PGUSER" echo echo "PostGIS:" psql \ - -h "$PGHOST" \ + -h "$PGHOST_ABS" \ -U "$PGUSER" \ -d "$PGDATABASE" \ -tAc "SELECT PostGIS_Full_Version();" | @@ -242,6 +246,7 @@ echo echo "============================================================" echo + set +e ''; }; } diff --git a/src/auth.rs b/src/auth.rs index 3fad639..ba487c1 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -9,11 +9,7 @@ use rand::{ }; use serde::Deserialize; -use crate::{ - AppState, - db, - intervals::IntervalsClient, -}; +use crate::{AppState, db, intervals::IntervalsClient}; #[derive(Debug, Deserialize)] pub struct OAuthCallback { @@ -22,39 +18,50 @@ pub struct OAuthCallback { pub error: Option, } -pub async fn start( - State(state): State, -) -> impl IntoResponse { - let mut rng = rng(); +pub async fn start(State(state): State) -> impl IntoResponse { + // Important: ThreadRng must not live across an await. + let oauth_state = { + let mut rng = rng(); + Alphanumeric.sample_string(&mut rng, 48) + }; - let oauth_state = - Alphanumeric.sample_string(&mut rng, 48); + if let Err(error) = db::delete_old_oauth_states(&state.db).await { + tracing::error!( + %error, + "cannot clean OAuth states" + ); - if let Err(error) = - db::delete_old_oauth_states(&state.db).await - { - tracing::error!(%error, "cannot clean OAuth states"); return "internal error".into_response(); } - if let Err(error) = sqlx::query( - "INSERT INTO oauth_states(state) VALUES ($1)", - ) - .bind(&oauth_state) - .execute(&state.db) - .await + if let Err(error) = sqlx::query("INSERT INTO oauth_states(state) VALUES ($1)") + .bind(&oauth_state) + .execute(&state.db) + .await { - tracing::error!(%error, "cannot create OAuth state"); + tracing::error!( + %error, + "cannot create OAuth state" + ); + return "internal error".into_response(); } - let client = - IntervalsClient::new(state.config.clone()); + let client = IntervalsClient::new(state.config.clone()); - Redirect::to( - &client.oauth_authorize_url(&oauth_state) - ) - .into_response() + let authorize_url = match client.oauth_authorize_url(&oauth_state) { + Ok(url) => url, + Err(error) => { + tracing::error!( + %error, + "cannot construct OAuth authorize URL" + ); + + return "cannot construct OAuth URL".into_response(); + } + }; + + Redirect::to(&authorize_url).into_response() } pub async fn callback( @@ -88,32 +95,46 @@ pub async fn callback( match state_row { Ok(Some(_)) => {} + Ok(None) => { return "invalid or expired OAuth state".into_response(); } + Err(error) => { - tracing::error!(%error, "OAuth state validation failed"); + tracing::error!( + %error, + "OAuth state validation failed" + ); + return "internal error".into_response(); } } - let client = - IntervalsClient::new(state.config.clone()); + let client = IntervalsClient::new(state.config.clone()); - let response = - match client.exchange_code(&code).await { - Ok(response) => response, - Err(error) => { - tracing::error!(%error, "OAuth token exchange failed"); - return "OAuth token exchange failed".into_response(); - } - }; + let response = match client.exchange_code(&code).await { + Ok(response) => response, + + Err(error) => { + tracing::error!( + %error, + "OAuth token exchange failed" + ); + + return "OAuth token exchange failed".into_response(); + } + }; let (access_token, scope, athlete_id, display_name) = match IntervalsClient::token_data(&response) { Ok(value) => value, + Err(error) => { - tracing::error!(%error, "invalid OAuth token response"); + tracing::error!( + %error, + "invalid OAuth token response" + ); + return "invalid OAuth response".into_response(); } }; @@ -127,14 +148,12 @@ pub async fn callback( scopes ) VALUES ($1, $2, $3, $4) - ON CONFLICT (intervals_athlete_id) DO UPDATE SET display_name = EXCLUDED.display_name, access_token = EXCLUDED.access_token, scopes = EXCLUDED.scopes, updated_at = now() - RETURNING id "#, ) @@ -150,71 +169,63 @@ pub async fn callback( let athlete_db_id: i64 = match row { Ok(row) => match row.try_get("id") { Ok(id) => id, + Err(error) => { - tracing::error!(%error, "invalid athlete row"); + tracing::error!( + %error, + "invalid athlete row" + ); + return "internal error".into_response(); } }, + Err(error) => { - tracing::error!(%error, "cannot store athlete"); + tracing::error!( + %error, + "cannot store athlete" + ); + return "cannot store athlete".into_response(); } }; - let mut rng = rng(); + let session_token = { + let mut rng = rng(); + Alphanumeric.sample_string(&mut rng, 64) + }; - let session_token = - Alphanumeric.sample_string(&mut rng, 64); + if let Err(error) = db::create_session(&state.db, athlete_db_id, &session_token).await { + tracing::error!( + %error, + "cannot create session" + ); - if let Err(error) = - db::create_session( - &state.db, - athlete_db_id, - &session_token, - ).await - { - tracing::error!(%error, "cannot create session"); return "cannot create session".into_response(); } - let cookie = Cookie::build( - ("county_session", session_token) - ) - .path("/") - .http_only(true) - .same_site(SameSite::Lax) - .secure(state.config.cookie_secure); + let cookie = Cookie::build(("county_session", session_token)) + .path("/") + .http_only(true) + .same_site(SameSite::Lax) + .secure(state.config.cookie_secure); let jar = CookieJar::new().add(cookie); - ( - jar, - Redirect::to("/"), - ).into_response() + (jar, Redirect::to("/")).into_response() } -pub async fn logout( - State(state): State, - jar: CookieJar, -) -> impl IntoResponse { +pub async fn logout(State(state): State, jar: CookieJar) -> impl IntoResponse { if let Some(cookie) = jar.get("county_session") { let hash = db::hash_token(cookie.value()); - let _ = sqlx::query( - "DELETE FROM sessions WHERE token_hash = $1", - ) - .bind(hash) - .execute(&state.db) - .await; + let _ = sqlx::query("DELETE FROM sessions WHERE token_hash = $1") + .bind(&hash) + .execute(&state.db) + .await; } - let jar = jar.remove( - Cookie::build("county_session") - .path("/") - ); + let jar = jar.remove(Cookie::build("county_session").path("/")); - ( - jar, - Redirect::to("/"), - ) + (jar, Redirect::to("/")).into_response() } diff --git a/src/config.rs b/src/config.rs index 9ffce2f..7a95062 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,15 +1,19 @@ -use anyhow::Result; +use anyhow::{Context, Result}; +use std::env; -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct Config { pub database_url: String, pub bind_address: String, pub intervals_base_url: String, - pub intervals_client_id: String, - pub intervals_client_secret: String, - pub intervals_redirect_uri: String, - pub intervals_webhook_secret: String, + pub intervals_api_key: Option, + pub intervals_athlete_id: String, + + pub intervals_client_id: Option, + pub intervals_client_secret: Option, + pub intervals_redirect_uri: Option, + pub intervals_webhook_secret: Option, pub cookie_secure: bool, } @@ -18,24 +22,65 @@ impl Config { pub fn from_env() -> Result { dotenvy::dotenv().ok(); + let database_url = required("DATABASE_URL")?; + let bind_address = + env::var("BIND_ADDRESS").unwrap_or_else(|_| "127.0.0.1:8080".to_string()); + + let intervals_base_url = + env::var("INTERVALS_BASE_URL").unwrap_or_else(|_| "https://intervals.icu".to_string()); + + let intervals_api_key = optional("INTERVALS_API_KEY"); + + let intervals_athlete_id = + env::var("INTERVALS_ATHLETE_ID").unwrap_or_else(|_| "0".to_string()); + + let intervals_client_id = optional("INTERVALS_CLIENT_ID"); + + let intervals_client_secret = optional("INTERVALS_CLIENT_SECRET"); + + let intervals_redirect_uri = optional("INTERVALS_REDIRECT_URI"); + + let intervals_webhook_secret = optional("INTERVALS_WEBHOOK_SECRET"); + + let cookie_secure = env::var("COOKIE_SECURE") + .unwrap_or_else(|_| "false".to_string()) + .parse::() + .context("COOKIE_SECURE must be true or false")?; + Ok(Self { - database_url: std::env::var("DATABASE_URL")?, + database_url, + bind_address, - bind_address: std::env::var("BIND_ADDRESS") - .unwrap_or_else(|_| "127.0.0.1:8080".into()), + intervals_base_url, + intervals_api_key, + intervals_athlete_id, - intervals_base_url: std::env::var("INTERVALS_BASE_URL") - .unwrap_or_else(|_| "https://intervals.icu".into()), + intervals_client_id, + intervals_client_secret, + intervals_redirect_uri, + intervals_webhook_secret, - intervals_client_id: std::env::var("INTERVALS_CLIENT_ID")?, - intervals_client_secret: std::env::var("INTERVALS_CLIENT_SECRET")?, - intervals_redirect_uri: std::env::var("INTERVALS_REDIRECT_URI")?, - intervals_webhook_secret: std::env::var("INTERVALS_WEBHOOK_SECRET")?, - - cookie_secure: std::env::var("COOKIE_SECURE") - .unwrap_or_else(|_| "true".into()) - .parse() - .unwrap_or(true), + cookie_secure, }) } + + pub fn require_api_key(&self) -> Result<&str> { + self.intervals_api_key + .as_deref() + .context("INTERVALS_API_KEY is not configured") + } +} + +fn required(name: &str) -> Result { + let value = env::var(name).with_context(|| format!("{name} is not configured"))?; + + if value.trim().is_empty() { + anyhow::bail!("{name} is empty"); + } + + Ok(value) +} + +fn optional(name: &str) -> Option { + env::var(name).ok().filter(|value| !value.trim().is_empty()) } diff --git a/src/db.rs b/src/db.rs index 9748e5a..9620735 100644 --- a/src/db.rs +++ b/src/db.rs @@ -12,11 +12,7 @@ pub fn hash_token(token: &str) -> Vec { hasher.finalize().to_vec() } -pub async fn create_session( - db: &PgPool, - athlete_id: i64, - token: &str, -) -> Result<()> { +pub async fn create_session(db: &PgPool, athlete_id: i64, token: &str) -> Result<()> { let hash = hash_token(token); sqlx::query( @@ -28,7 +24,7 @@ pub async fn create_session( VALUES ($1, $2) "#, ) - .bind(hash) + .bind(&hash) .bind(athlete_id) .execute(db) .await?; @@ -36,10 +32,7 @@ pub async fn create_session( Ok(()) } -pub async fn athlete_for_session( - db: &PgPool, - token: &str, -) -> Result> { +pub async fn athlete_for_session(db: &PgPool, token: &str) -> Result> { let hash = hash_token(token); let row = sqlx::query( @@ -53,7 +46,7 @@ pub async fn athlete_for_session( WHERE s.token_hash = $1 "#, ) - .bind(hash) + .bind(&hash) .fetch_optional(db) .await?; @@ -64,9 +57,13 @@ pub async fn athlete_for_session( let name: String = row.try_get("display_name")?; sqlx::query( - "UPDATE sessions SET last_seen_at = now() WHERE token_hash = $1", + r#" + UPDATE sessions + SET last_seen_at = now() + WHERE token_hash = $1 + "#, ) - .bind(hash) + .bind(&hash) .execute(db) .await?; @@ -76,11 +73,12 @@ pub async fn athlete_for_session( } } -pub async fn delete_old_oauth_states( - db: &PgPool, -) -> Result<()> { +pub async fn delete_old_oauth_states(db: &PgPool) -> Result<()> { sqlx::query( - "DELETE FROM oauth_states WHERE created_at < now() - interval '10 minutes'", + r#" + DELETE FROM oauth_states + WHERE created_at < now() - interval '10 minutes' + "#, ) .execute(db) .await?; @@ -105,7 +103,6 @@ pub async fn ensure_activity( processed_at ) VALUES ($1, $2, $3, $4, NULL) - ON CONFLICT ( athlete_id, intervals_activity_id @@ -114,7 +111,6 @@ pub async fn ensure_activity( start_time = EXCLUDED.start_time, activity_json = EXCLUDED.activity_json, processed_at = NULL - RETURNING id "#, ) @@ -135,7 +131,10 @@ pub async fn delete_activity_crossings( activity_id: i64, ) -> Result<()> { sqlx::query( - "DELETE FROM county_crossings WHERE activity_id = $1", + r#" + DELETE FROM county_crossings + WHERE activity_id = $1 + "#, ) .bind(activity_id) .execute(&mut **tx) @@ -165,12 +164,10 @@ pub async fn save_crossing( $1, $2, $3, - ST_SetSRID( ST_MakePoint($4, $5), 4326 ), - $6, $7, $8 diff --git a/src/geo.rs b/src/geo.rs index 3cb51ea..66bbf50 100644 --- a/src/geo.rs +++ b/src/geo.rs @@ -4,19 +4,12 @@ use sqlx::PgPool; use crate::model::{DetectedCrossing, TrackPoint}; -pub fn interpolate_time( - a: DateTime, - b: DateTime, - fraction: f64, -) -> DateTime { +pub fn interpolate_time(a: DateTime, b: DateTime, fraction: f64) -> DateTime { let fraction = fraction.clamp(0.0, 1.0); - let duration_ms = - (b - a).num_milliseconds(); + let duration_ms = (b - a).num_milliseconds(); - a + Duration::milliseconds( - (duration_ms as f64 * fraction).round() as i64 - ) + a + Duration::milliseconds((duration_ms as f64 * fraction).round() as i64) } /// Detects all county crossings between two consecutive GPS samples. @@ -132,26 +125,18 @@ pub async fn crossings_between( continue; } - let from_county_id: i64 = - row.try_get("from_id")?; + let from_county_id: i64 = row.try_get("from_id")?; - let to_county_id: i64 = - row.try_get("to_id")?; + let to_county_id: i64 = row.try_get("to_id")?; - let lon: f64 = - row.try_get("lon")?; + let lon: f64 = row.try_get("lon")?; - let lat: f64 = - row.try_get("lat")?; + let lat: f64 = row.try_get("lat")?; result.push(DetectedCrossing { track_index: a.index, - crossing_time: interpolate_time( - a.time, - b.time, - fraction, - ), + crossing_time: interpolate_time(a.time, b.time, fraction), lat, lon, diff --git a/src/intervals.rs b/src/intervals.rs index 141c5f5..6971257 100644 --- a/src/intervals.rs +++ b/src/intervals.rs @@ -1,6 +1,6 @@ use anyhow::{Context, Result, anyhow}; -use chrono::{DateTime, Utc}; -use reqwest::Client; +use chrono::{DateTime, NaiveDateTime, Utc}; +use reqwest::{Client, StatusCode}; use serde_json::{Value, json}; use crate::config::Config; @@ -19,274 +19,266 @@ impl IntervalsClient { } } - pub fn oauth_authorize_url(&self, state: &str) -> String { + fn base_url(&self) -> &str { + self.config.intervals_base_url.trim_end_matches('/') + } + + fn api_key(&self) -> Result<&str> { + self.config.require_api_key() + } + + fn api_url(&self, path: &str) -> String { format!( - "{}/oauth/authorize?client_id={}&redirect_uri={}&scope={}&state={}", - self.config.intervals_base_url, - urlencoding::encode(&self.config.intervals_client_id), - urlencoding::encode(&self.config.intervals_redirect_uri), - urlencoding::encode("ACTIVITY:READ"), - urlencoding::encode(state), + "{}/api/v1/{}", + self.base_url(), + path.trim_start_matches('/') ) } - pub async fn exchange_code(&self, code: &str) -> Result { + 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 { + let response = self + .authenticated(self.client.get(url))? + .send() + .await + .context("Intervals.icu request failed")?; + + let status = response.status(); + let body = response + .text() + .await + .context("cannot read Intervals.icu response")?; + + if !status.is_success() { + return Err(anyhow!("Intervals.icu returned HTTP {}: {}", status, body)); + } + + serde_json::from_str(&body) + .with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body)) + } + + /// Get a single activity. + /// + /// Uses: + /// GET /api/v1/activity/{id} + pub async fn activity(&self, activity_id: &str) -> Result { + let url = self.api_url(&format!("activity/{activity_id}")); + + self.get_json(url).await + } + + /// Get activity streams. + /// + /// Uses: + /// GET /api/v1/activity/{id}/streams.json + /// + /// The response is normally an array of stream objects: + /// + /// [ + /// {"type":"time","data":[...]}, + /// {"type":"latlng","data":[...]}, + /// ... + /// ] + pub async fn streams(&self, activity_id: &str) -> Result { + let url = self.api_url(&format!("activity/{activity_id}/streams.json")); + + self.get_json(url).await + } + + /// Fetch the athlete's activities. + /// + /// `0` means the athlete belonging to the API key. + pub async fn activities(&self, oldest: &str, newest: &str) -> Result { + let athlete_id = &self.config.intervals_athlete_id; + let url = format!( - "{}/api/oauth/token", - self.config.intervals_base_url + "{}?oldest={}&newest={}", + self.api_url(&format!("athlete/{athlete_id}/activities")), + urlencoding::encode(oldest), + urlencoding::encode(newest), ); - Ok(self - .client - .post(url) - .form(&[ - ("client_id", self.config.intervals_client_id.as_str()), - ( - "client_secret", - self.config.intervals_client_secret.as_str(), - ), - ("code", code), - ]) - .send() - .await? - .error_for_status()? - .json() - .await?) + self.get_json(url).await } - pub async fn activity( - &self, - access_token: &str, - activity_id: &str, - ) -> Result { - let url = format!( - "{}/api/v1/activity/{}", - self.config.intervals_base_url, - activity_id - ); - - Ok(self - .client - .get(url) - .bearer_auth(access_token) - .query(&[("intervals", "true")]) - .send() - .await? - .error_for_status() - .context("Intervals.icu activity request failed")? - .json() - .await?) - } - - pub async fn streams( - &self, - access_token: &str, - activity_id: &str, - ) -> Result { - let url = format!( - "{}/api/v1/activity/{}/streams.json", - self.config.intervals_base_url, - activity_id - ); - - Ok(self - .client - .get(url) - .bearer_auth(access_token) - .query(&[("types", "time,latlng")]) - .send() - .await? - .error_for_status() - .context("Intervals.icu streams request failed")? - .json() - .await?) - } - - pub fn activity_url(&self, activity_id: &str) -> String { - format!( - "{}/activities/{}", - self.config.intervals_base_url, - activity_id - ) - } - - pub fn interval_url( - &self, - activity_id: &str, - interval_id: i64, - ) -> String { - format!( - "{}/activities/{}?interval={}", - self.config.intervals_base_url, - activity_id, - interval_id - ) - } - - pub fn stream_values( - streams: &Value, - wanted: &str, - ) -> Option> { + /// Return a stream's values independent of whether the API + /// returned an object or an array of stream objects. + pub fn stream_values<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Vec> { + // Current API format: array of streams. if let Some(array) = streams.as_array() { - for entry in array { - if entry.get("type") - .and_then(Value::as_str) - == Some(wanted) - { - return entry - .get("data") - .and_then(Value::as_array) - .cloned(); + for stream in array { + if stream.get("type").and_then(Value::as_str) == Some(stream_type) { + return stream.get("data").and_then(Value::as_array); } } } - if let Some(object) = streams.as_object() { - if let Some(value) = object.get(wanted) { - if let Some(data) = value.get("data").and_then(Value::as_array) { - return Some(data.clone()); - } + // Also support a dictionary-like response: + // + // { + // "time": [...], + // "latlng": [...] + // } + streams.get(stream_type).and_then(Value::as_array) + } - if let Some(data) = value.as_array() { - return Some(data.clone()); - } + /// Extract the activity start timestamp. + 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"))?; + + let string = value + .as_str() + .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. + /// + /// The exact interval representation has changed over + /// time, so this deliberately supports the common forms. + pub fn matching_interval(activity: &Value, track_index: usize) -> Option { + let intervals = activity.get("icu_intervals")?; + + let array = intervals.as_array()?; + + for interval in array { + let start = interval + .get("start_index") + .or_else(|| interval.get("start")); + + let end = interval.get("end_index").or_else(|| interval.get("end")); + + let start = start.and_then(Value::as_u64)?; + let end = end.and_then(Value::as_u64)?; + + if (start as usize) <= track_index && track_index <= end as usize { + return Some(interval.clone()); } } None } - pub fn activity_start(activity: &Value) -> Result> { - if let Some(value) = activity - .get("start_date") - .and_then(Value::as_str) - { - return Ok(value.parse()?); + /// OAuth methods are intentionally retained so we can + /// re-enable multi-user OAuth later. + 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")?; + + let redirect_uri = self + .config + .intervals_redirect_uri + .as_deref() + .context("INTERVALS_REDIRECT_URI is not configured")?; + + Ok(format!( + "{}/oauth/authorize?client_id={}&redirect_uri={}&scope={}&state={}", + self.base_url(), + urlencoding::encode(client_id), + urlencoding::encode(redirect_uri), + urlencoding::encode("ACTIVITY:READ"), + urlencoding::encode(state), + )) + } + + 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")?; + + let client_secret = self + .config + .intervals_client_secret + .as_deref() + .context("INTERVALS_CLIENT_SECRET is not configured")?; + + let response = self + .client + .post(format!("{}/api/oauth/token", self.base_url())) + .form(&[ + ("client_id", client_id), + ("client_secret", client_secret), + ("code", code), + ]) + .send() + .await + .context("OAuth token request failed")?; + + let status = response.status(); + let body = response.text().await?; + + if !status.is_success() { + return Err(anyhow!( + "OAuth token exchange returned HTTP {}: {}", + status, + body + )); } - if let Some(value) = activity - .get("start_date_local") - .and_then(Value::as_str) - { - let naive = chrono::NaiveDateTime::parse_from_str( - value, - "%Y-%m-%dT%H:%M:%S", - )?; - - return Ok( - naive - .and_utc() - ); - } - - Err(anyhow!("activity contains no usable start_date")) + Ok(serde_json::from_str(&body)?) } - pub fn icu_intervals(activity: &Value) -> Vec { - activity - .get("icu_intervals") - .and_then(Value::as_array) - .cloned() - .unwrap_or_default() - } - - pub fn matching_interval( - activity: &Value, - track_index: usize, - ) -> Option { - let intervals = Self::icu_intervals(activity); - - intervals - .into_iter() - .filter(|interval| { - let start = interval - .get("start_index") - .and_then(Value::as_u64); - - let end = interval - .get("end_index") - .and_then(Value::as_u64); - - match (start, end) { - (Some(start), Some(end)) => - (start as usize) <= track_index - && track_index < end as usize, - - _ => false, - } - }) - .min_by_key(|interval| { - let start = interval - .get("start_index") - .and_then(Value::as_u64) - .unwrap_or(u64::MAX); - - let end = interval - .get("end_index") - .and_then(Value::as_u64) - .unwrap_or(u64::MAX); - - end.saturating_sub(start) - }) - } - - pub fn interval_metrics( - interval: &Value, - ) -> (Option, Option, Option) { - let watts = interval - .get("average_watts") - .and_then(Value::as_f64); - - let hr = interval - .get("average_heartrate") - .and_then(Value::as_f64); - - let id = interval - .get("id") - .and_then(Value::as_i64); - - (watts, hr, id) - } - - pub fn token_data( - response: &Value, - ) -> Result<(String, String, String, String)> { - let token = response + pub fn token_data(response: &Value) -> Result<(String, String, String, String)> { + let access_token = response .get("access_token") .and_then(Value::as_str) - .ok_or_else(|| anyhow!("OAuth response has no access_token"))?; + .context("OAuth response has no access_token")? + .to_string(); let scope = response .get("scope") .and_then(Value::as_str) .unwrap_or("") - .to_owned(); + .to_string(); let athlete = response .get("athlete") - .ok_or_else(|| anyhow!("OAuth response has no athlete"))?; + .context("OAuth response has no athlete")?; let athlete_id = athlete .get("id") .and_then(Value::as_str) - .ok_or_else(|| anyhow!("OAuth response has no athlete.id"))?; + .or_else(|| athlete.get("id").and_then(Value::as_i64).map(|_| "")) + .unwrap_or("") + .to_string(); - let athlete_name = athlete + let display_name = athlete .get("name") + .or_else(|| athlete.get("display_name")) .and_then(Value::as_str) - .unwrap_or(athlete_id); + .unwrap_or("Intervals.icu athlete") + .to_string(); - Ok(( - token.to_owned(), - scope, - athlete_id.to_owned(), - athlete_name.to_owned(), - )) - } - - pub fn _json_example() -> Value { - json!({ - "scope": "ACTIVITY:READ" - }) + 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)); + } + + 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)); + } + + Err(anyhow!("cannot parse activity timestamp: {}", value)) +} diff --git a/src/leaderboard.rs b/src/leaderboard.rs index 9183579..b06de6c 100644 --- a/src/leaderboard.rs +++ b/src/leaderboard.rs @@ -1,8 +1,5 @@ use anyhow::Result; -use axum::{ - extract::State, - Json, -}; +use axum::{Json, extract::State}; use chrono::{DateTime, Utc}; use serde_json::Value; use sqlx::PgPool; @@ -15,22 +12,17 @@ use crate::{ pub async fn api( State(state): State, ) -> Result>, (axum::http::StatusCode, String)> { - build(&state.db) - .await - .map(Json) - .map_err(|error| { - tracing::error!(%error, "leaderboard query failed"); + build(&state.db).await.map(Json).map_err(|error| { + tracing::error!(%error, "leaderboard query failed"); - ( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - error.to_string(), - ) - }) + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + error.to_string(), + ) + }) } -pub async fn build( - db: &PgPool, -) -> Result> { +pub async fn build(db: &PgPool) -> Result> { let rows = sqlx::query( r#" WITH bucketed AS ( @@ -110,21 +102,17 @@ pub async fn build( let mut groups: Vec = Vec::new(); for row in rows { - let bucket_start: DateTime = - row.try_get("bucket_start")?; + let bucket_start: DateTime = row.try_get("bucket_start")?; - let from_county: String = - row.try_get("from_county")?; + let from_county: String = row.try_get("from_county")?; - let to_county: String = - row.try_get("to_county")?; + let to_county: String = row.try_get("to_county")?; - let group_index = - groups.iter().position(|group| { - group.bucket_start == bucket_start - && group.from_county == from_county - && group.to_county == to_county - }); + let group_index = groups.iter().position(|group| { + group.bucket_start == bucket_start + && group.from_county == from_county + && group.to_county == to_county + }); let index = match group_index { Some(index) => index, @@ -141,35 +129,30 @@ pub async fn build( } }; - let rank: i64 = - row.try_get("rank")?; + let rank: i64 = row.try_get("rank")?; - let athlete: String = - row.try_get("display_name")?; + let athlete: String = row.try_get("display_name")?; - let crossing_time: DateTime = - row.try_get("crossing_time")?; + let crossing_time: DateTime = row.try_get("crossing_time")?; - let activity_id: String = - row.try_get("intervals_activity_id")?; + let activity_id: String = row.try_get("intervals_activity_id")?; - let interval: Option = - row.try_get("intervals_interval")?; + let interval: Option = row.try_get("intervals_interval")?; - let interval_id = - interval.as_ref() - .and_then(|v| v.get("id")) - .and_then(Value::as_i64); + let interval_id = interval + .as_ref() + .and_then(|v| v.get("id")) + .and_then(Value::as_i64); - let average_watts = - interval.as_ref() - .and_then(|v| v.get("average_watts")) - .and_then(Value::as_f64); + let average_watts = interval + .as_ref() + .and_then(|v| v.get("average_watts")) + .and_then(Value::as_f64); - let average_heartrate = - interval.as_ref() - .and_then(|v| v.get("average_heartrate")) - .and_then(Value::as_f64); + let average_heartrate = interval + .as_ref() + .and_then(|v| v.get("average_heartrate")) + .and_then(Value::as_f64); let row_data = LeaderboardRow { rank, @@ -182,16 +165,12 @@ pub async fn build( activity_id: activity_id.clone(), - activity_url: format!( - "https://intervals.icu/activities/{}", - activity_id - ), + activity_url: format!("https://intervals.icu/activities/{}", activity_id), interval_url: interval_id.map(|id| { format!( "https://intervals.icu/activities/{}?interval={}", - activity_id, - id + activity_id, id ) }), diff --git a/src/main.rs b/src/main.rs index cf605c9..e997940 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,30 +5,22 @@ mod geo; mod intervals; mod leaderboard; mod model; -mod webhook; mod web; - -use std::sync::Arc; +mod webhook; use anyhow::{Context, Result, anyhow}; use axum::{ Router, + extract::{Path, State}, routing::{get, post}, }; use chrono::{DateTime, Duration, Utc}; use serde_json::Value; -use sqlx::PgPool; +use sqlx::{PgPool, Row}; use tower_http::trace::TraceLayer; -use tracing_subscriber::{ - layer::SubscriberExt, - util::SubscriberInitExt, -}; +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 { @@ -39,19 +31,20 @@ pub struct AppState { #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::registry() - .with( - tracing_subscriber::EnvFilter::from_default_env() - ) + .with(tracing_subscriber::EnvFilter::from_default_env()) .with(tracing_subscriber::fmt::layer()) .init(); - let config = - Config::from_env()?; + let config = Config::from_env()?; - let db = - PgPool::connect(&config.database_url) - .await - .context("cannot connect to PostgreSQL")?; + tracing::info!( + database_url = %redact_database_url(&config.database_url), + "connecting to PostgreSQL" + ); + + let db = PgPool::connect(&config.database_url) + .await + .context("cannot connect to PostgreSQL")?; sqlx::migrate!() .run(&db) @@ -69,22 +62,29 @@ async fn main() -> Result<()> { .route("/oauth/start", get(auth::start)) .route("/oauth/callback", get(auth::callback)) .route("/logout", get(auth::logout)) - .route( - "/webhooks/intervals", - post(webhook::receive), - ) - .route( - "/api/leaderboard", - get(leaderboard::api), - ) + .route("/webhooks/intervals", post(webhook::receive)) + .route("/api/leaderboard", get(leaderboard::api)) + /* + * Development-only endpoint. + * + * This lets us test the complete activity-processing + * pipeline without having to trigger an Intervals.icu + * webhook. + * + * Example: + * + * curl -X POST \ + * http://127.0.0.1:8080/dev/process/12345678 + * + * Do not expose this endpoint publicly in production. + */ + .route("/dev/process/:activity_id", post(dev_process_activity)) .layer(TraceLayer::new_for_http()) .with_state(state); - let listener = - tokio::net::TcpListener::bind( - &config.bind_address - ) - .await?; + let listener = tokio::net::TcpListener::bind(&config.bind_address) + .await + .with_context(|| format!("cannot bind to {}", config.bind_address))?; tracing::info!( address = %config.bind_address, @@ -100,16 +100,64 @@ async fn health() -> &'static str { "ok" } +/// Development endpoint for processing one real Intervals.icu +/// activity using INTERVALS_API_KEY. +/// +/// This intentionally bypasses OAuth and webhooks. +async fn dev_process_activity( + State(state): State, + Path(activity_id): Path, +) -> Result { + process_activity(state, configured_athlete_id(), activity_id) + .await + .map(|_| "activity processed\n".to_string()) + .map_err(internal_error) +} + +fn configured_athlete_id() -> String { + std::env::var("INTERVALS_ATHLETE_ID").unwrap_or_else(|_| "0".to_string()) +} + +fn internal_error(error: anyhow::Error) -> (axum::http::StatusCode, String) { + tracing::error!( + %error, + "request failed" + ); + + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + error.to_string(), + ) +} + +/// Process one Intervals.icu activity. +/// +/// In development mode the Intervals.icu personal API key from +/// INTERVALS_API_KEY is used by IntervalsClient. +/// +/// The activity is: +/// +/// 1. fetched from Intervals.icu +/// 2. its GPS/time streams are fetched +/// 3. GPS points are converted to TrackPoint values +/// 4. county-border crossings are detected +/// 5. crossing times are interpolated to millisecond precision +/// 6. matching Intervals.icu intervals are attached where possible +/// 7. everything is stored transactionally pub async fn process_activity( state: AppState, intervals_athlete_id: String, activity_id: String, ) -> Result<()> { + /* + * The development API key represents one athlete. + * + * We still keep the athlete in our database because activities + * and crossings belong to an athlete. + */ let athlete = sqlx::query( r#" - SELECT - id, - access_token + SELECT id FROM athletes WHERE intervals_athlete_id = $1 "#, @@ -118,113 +166,127 @@ pub async fn process_activity( .fetch_optional(&state.db) .await?; - use sqlx::Row; + let athlete_db_id: i64 = match athlete { + Some(row) => row.try_get("id")?, - let Some(athlete) = athlete else { - return Err(anyhow!( - "no OAuth connection for athlete {}", - intervals_athlete_id - )); + None => { + let client = IntervalsClient::new(state.config.clone()); + + /* + * Fetch the activity first so we can get a useful + * athlete/display name if available. + */ + let activity = client.activity(&activity_id).await?; + + let display_name = activity + .get("athlete") + .and_then(|athlete| athlete.get("name").or_else(|| athlete.get("display_name"))) + .and_then(Value::as_str) + .unwrap_or("Intervals.icu athlete"); + + let row = sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + display_name, + access_token, + scopes + ) + VALUES ($1, $2, '', 'DEV_API_KEY') + ON CONFLICT (intervals_athlete_id) + DO UPDATE SET + display_name = EXCLUDED.display_name, + updated_at = now() + RETURNING id + "#, + ) + .bind(&intervals_athlete_id) + .bind(display_name) + .fetch_one(&state.db) + .await?; + + row.try_get("id")? + } }; - let athlete_db_id: i64 = - athlete.try_get("id")?; + let client = IntervalsClient::new(state.config.clone()); - let access_token: String = - athlete.try_get("access_token")?; + /* + * First fetch the activity metadata. + * + * Among other things this may contain Intervals.icu's + * automatically detected intervals. + */ + let activity = client.activity(&activity_id).await?; - let client = - IntervalsClient::new(state.config.clone()); + let start_time = IntervalsClient::activity_start(&activity)?; - // - // First fetch the activity metadata including icu_intervals. - // - let activity = - client.activity( - &access_token, - &activity_id, - ) - .await?; + /* + * Then fetch only the activity streams we need. + * + * For county crossings: + * + * time + * latlng + * + * Optional streams such as watts/hr can be added later + * without changing the crossing algorithm. + */ + let streams = client.streams(&activity_id).await?; - let start_time = - IntervalsClient::activity_start( - &activity - )?; - - // - // Then fetch only what we actually need from the - // activity stream: time + GPS. - // - let streams = - client.streams( - &access_token, - &activity_id, - ) - .await?; - - let points = - parse_track( - &streams, - start_time, - )?; + let points = parse_track(&streams, start_time)?; if points.len() < 2 { tracing::info!( activity_id = %activity_id, + points = points.len(), "activity has insufficient GPS data" ); return Ok(()); } - // - // Use a transaction so re-processing an activity is idempotent: - // old crossings disappear and are replaced atomically. - // - let mut tx = - state.db.begin().await?; + tracing::info!( + activity_id = %activity_id, + points = points.len(), + start_time = %start_time, + "processing activity" + ); + + /* + * Use a transaction so re-processing an activity is + * idempotent: + * + * old crossings are deleted + * new crossings are inserted + * activity is marked processed + * + * all atomically. + */ + 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, - ); + /* + * Try to associate this county crossing with an + * automatically detected Intervals.icu interval. + * + * This is deliberately optional: a crossing does + * not become invalid merely because no interval + * was detected. + */ + 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; @@ -234,16 +296,13 @@ pub async fn process_activity( from = crossing.from_county_id, to = crossing.to_county_id, track_index = crossing.track_index, + interval_found = interval.is_some(), "county crossing detected" ); } } - db::mark_activity_processed( - &mut tx, - activity_db_id, - ) - .await?; + db::mark_activity_processed(&mut tx, activity_db_id).await?; tx.commit().await?; @@ -256,44 +315,23 @@ 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" - ))?; +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 latlng = IntervalsClient::stream_values(streams, "latlng") + .ok_or_else(|| anyhow!("Intervals.icu response has no latlng stream"))?; - let length = - times.len().min(latlng.len()); + let length = times.len().min(latlng.len()); - let mut result = - Vec::with_capacity(length); + let mut result = Vec::with_capacity(length); for index in 0..length { - let Some(seconds) = - times[index].as_f64() - else { + let Some(seconds) = times[index].as_f64() else { continue; }; - let Some(coordinate) = - latlng[index].as_array() - else { + let Some(coordinate) = latlng[index].as_array() else { continue; }; @@ -301,32 +339,32 @@ fn parse_track( continue; } - let Some(lat) = - coordinate[0].as_f64() - else { + let Some(lat) = coordinate[0].as_f64() else { continue; }; - let Some(lon) = - coordinate[1].as_f64() - else { + let Some(lon) = coordinate[1].as_f64() else { continue; }; - if !lat.is_finite() - || !lon.is_finite() - || !seconds.is_finite() - { + if !lat.is_finite() || !lon.is_finite() || !seconds.is_finite() { continue; } - let millis = - (seconds * 1000.0).round() as i64; + /* + * Intervals.icu's time stream is relative to the + * activity start. + * + * We retain millisecond precision. Nanosecond + * precision would not provide meaningful additional + * accuracy because GPS samples themselves are much + * less precise. + */ + let millis = (seconds * 1000.0).round() as i64; result.push(TrackPoint { index, - time: start_time - + Duration::milliseconds(millis), + time: start_time + Duration::milliseconds(millis), lat, lon, }); @@ -335,6 +373,30 @@ fn parse_track( Ok(result) } +/// Avoid accidentally logging database passwords. +fn redact_database_url(database_url: &str) -> String { + /* + * For our current local Unix-socket URL there normally is + * no password. For safety, don't print anything after the + * first '@' if a password is present. + */ + if let Some(at) = database_url.rfind('@') { + if let Some(scheme) = database_url.find("://") { + let prefix_end = scheme + 3; + + if at > prefix_end { + return format!( + "{}***@{}", + &database_url[..prefix_end], + &database_url[at + 1..] + ); + } + } + } + + database_url.to_string() +} + #[cfg(test)] mod tests { use chrono::{TimeZone, Utc}; @@ -343,39 +405,23 @@ mod tests { #[test] fn interpolation_is_millisecond_precise() { - let a = - Utc.timestamp_millis_opt(1000) - .unwrap(); + let a = Utc.timestamp_millis_opt(1000).unwrap(); - let b = - Utc.timestamp_millis_opt(2000) - .unwrap(); + let b = Utc.timestamp_millis_opt(2000).unwrap(); - let result = - interpolate_time(a, b, 0.376); + let result = interpolate_time(a, b, 0.376); - assert_eq!( - result.timestamp_millis(), - 1376 - ); + assert_eq!(result.timestamp_millis(), 1376); } #[test] fn interpolation_rounds_to_nearest_ms() { - let a = - Utc.timestamp_millis_opt(1000) - .unwrap(); + let a = Utc.timestamp_millis_opt(1000).unwrap(); - let b = - Utc.timestamp_millis_opt(2000) - .unwrap(); + let b = Utc.timestamp_millis_opt(2000).unwrap(); - let result = - interpolate_time(a, b, 0.3764); + let result = interpolate_time(a, b, 0.3764); - assert_eq!( - result.timestamp_millis(), - 1376 - ); + assert_eq!(result.timestamp_millis(), 1376); } } diff --git a/src/web.rs b/src/web.rs index b7566d8..22feea8 100644 --- a/src/web.rs +++ b/src/web.rs @@ -1,38 +1,20 @@ -use axum::{ - extract::State, - response::Html, -}; +use axum::{extract::State, response::Html}; use axum_extra::extract::cookie::CookieJar; -use crate::{ - AppState, - db, - leaderboard, -}; +use crate::{AppState, db, leaderboard}; -pub async fn index( - State(state): State, - jar: CookieJar, -) -> Html { - let groups = - leaderboard::build(&state.db) +pub async fn index(State(state): State, jar: CookieJar) -> Html { + let groups = leaderboard::build(&state.db).await.unwrap_or_default(); + + let current_user = match jar.get("county_session") { + Some(cookie) => db::athlete_for_session(&state.db, cookie.value()) .await - .unwrap_or_default(); + .ok() + .flatten() + .map(|(_, name)| name), - let current_user = - match jar.get("county_session") { - Some(cookie) => - db::athlete_for_session( - &state.db, - cookie.value(), - ) - .await - .ok() - .flatten() - .map(|(_, name)| name), - - None => None, - }; + None => None, + }; let mut html = String::from( r#" @@ -113,7 +95,7 @@ button, .button { html.push_str( r#" Mit Intervals.icu verbinden - "# + "#, ); } @@ -123,7 +105,7 @@ button, .button { html.push_str( r#"
Noch keine Landkreisüberquerungen. -
"# + "#, ); } @@ -153,31 +135,28 @@ button, .button { HR -"# +"#, ); for row in group.rows { - let interval = - match row.interval_url { - Some(url) => - format!( - "Intervall", - escape_html(&url) - ), + let interval = match row.interval_url { + Some(url) => format!( + "Intervall", + escape_html(&url) + ), - None => - "—".into(), - }; + None => "—".into(), + }; - let power = - row.average_watts - .map(|v| format!("{:.0} W", v)) - .unwrap_or_else(|| "—".into()); + let power = row + .average_watts + .map(|v| format!("{:.0} W", v)) + .unwrap_or_else(|| "—".into()); - let hr = - row.average_heartrate - .map(|v| format!("{:.0} bpm", v)) - .unwrap_or_else(|| "—".into()); + let hr = row + .average_heartrate + .map(|v| format!("{:.0} bpm", v)) + .unwrap_or_else(|| "—".into()); html.push_str(&format!( r#" @@ -207,7 +186,7 @@ button, .button { Datenquelle: Intervals.icu · Landkreisgeometrien: VG250 -"# +"#, ); Html(html) diff --git a/src/webhook.rs b/src/webhook.rs index ec10b98..2223c17 100644 --- a/src/webhook.rs +++ b/src/webhook.rs @@ -1,81 +1,71 @@ -use axum::{ - extract::State, - http::StatusCode, - Json, -}; +use axum::{extract::State, http::StatusCode, response::IntoResponse}; +use serde::Deserialize; -use crate::{ - AppState, - model::WebhookEnvelope, -}; +use crate::AppState; + +#[derive(Debug, Deserialize)] +pub struct WebhookPayload { + pub secret: String, + + #[serde(default)] + pub athlete_id: Option, + + #[serde(default)] + pub activity_id: Option, +} pub async fn receive( State(state): State, - Json(payload): Json, -) -> Result<&'static str, StatusCode> { - if payload.secret != state.config.intervals_webhook_secret { - tracing::warn!("invalid Intervals.icu webhook secret"); - return Err(StatusCode::UNAUTHORIZED); + axum::Json(payload): axum::Json, +) -> impl IntoResponse { + let Some(expected_secret) = state.config.intervals_webhook_secret.as_deref() else { + tracing::error!("Intervals.icu webhook secret is not configured"); + + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "webhook secret not configured", + ) + .into_response(); + }; + + if payload.secret != expected_secret { + tracing::warn!("received webhook with invalid secret"); + + return (StatusCode::UNAUTHORIZED, "invalid webhook secret").into_response(); } - for event in payload.events { - match event.event_type.as_str() { - "ACTIVITY_UPLOADED" | - "ACTIVITY_ANALYZED" => { - let activity_id = event - .activity - .as_ref() - .and_then(|activity| { - activity.get("id") - }) - .and_then(|id| { - id.as_str() - }); + let Some(activity_id) = payload.activity_id else { + tracing::warn!("received webhook without activity_id"); - let Some(activity_id) = activity_id else { - tracing::warn!( - athlete_id = %event.athlete_id, - "activity webhook without activity id" - ); + return (StatusCode::BAD_REQUEST, "missing activity_id").into_response(); + }; - continue; - }; + let athlete_id = payload + .athlete_id + .or_else(|| std::env::var("INTERVALS_ATHLETE_ID").ok()); - if let Err(error) = - crate::process_activity( - state.clone(), - event.athlete_id.clone(), - activity_id.to_owned(), - ).await - { - tracing::error!( - %error, - athlete_id = %event.athlete_id, - activity_id = %activity_id, - "activity processing failed" - ); + let Some(athlete_id) = athlete_id else { + tracing::error!( + "webhook has no athlete_id and \ + INTERVALS_ATHLETE_ID is not configured" + ); - // Return 500 so Intervals.icu can retry the webhook. - return Err(StatusCode::INTERNAL_SERVER_ERROR); - } - } + return (StatusCode::BAD_REQUEST, "missing athlete_id").into_response(); + }; - "APP_SCOPE_CHANGED" => { - tracing::info!( - athlete_id = %event.athlete_id, - "OAuth scopes changed" - ); - } + let state_clone = state.clone(); - _ => { - tracing::debug!( - event_type = %event.event_type, - "ignoring Intervals.icu webhook" - ); - } + tokio::spawn(async move { + if let Err(error) = + crate::process_activity(state_clone, athlete_id, activity_id.clone()).await + { + tracing::error!( + %error, + activity_id = %activity_id, + "failed to process webhook activity" + ); } - } + }); - // Intervals.icu has historically retried on 204; use an ordinary 200. - Ok("ok") + (StatusCode::ACCEPTED, "activity queued").into_response() }