diff --git a/src/config.rs b/src/config.rs index 7a95062..4489293 100644 --- a/src/config.rs +++ b/src/config.rs @@ -16,6 +16,8 @@ pub struct Config { pub intervals_webhook_secret: Option, pub cookie_secure: bool, + + pub dev_sync_days: i64, } impl Config { @@ -23,6 +25,7 @@ impl Config { 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()); @@ -47,6 +50,15 @@ impl Config { .parse::() .context("COOKIE_SECURE must be true or false")?; + let dev_sync_days = env::var("DEV_SYNC_DAYS") + .unwrap_or_else(|_| "30".to_string()) + .parse::() + .context("DEV_SYNC_DAYS must be an integer")?; + + if dev_sync_days <= 0 { + anyhow::bail!("DEV_SYNC_DAYS must be greater than zero"); + } + Ok(Self { database_url, bind_address, @@ -61,6 +73,8 @@ impl Config { intervals_webhook_secret, cookie_secure, + + dev_sync_days, }) } diff --git a/src/db.rs b/src/db.rs index 9620735..d1723ab 100644 --- a/src/db.rs +++ b/src/db.rs @@ -8,7 +8,9 @@ use crate::model::DetectedCrossing; pub fn hash_token(token: &str) -> Vec { let mut hasher = Sha256::new(); + hasher.update(token.as_bytes()); + hasher.finalize().to_vec() } @@ -54,6 +56,7 @@ pub async fn athlete_for_session(db: &PgPool, token: &str) -> Result, } impl IntervalsClient { pub fn new(config: Config) -> Self { + let api_key = config.intervals_api_key.clone(); + Self { client: Client::new(), config, + api_key, + } + } + + /// 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 { + Self { + client: Client::new(), + config, + api_key: Some(api_key.into()), } } @@ -24,7 +39,9 @@ impl IntervalsClient { } fn api_key(&self) -> Result<&str> { - self.config.require_api_key() + self.api_key + .as_deref() + .context("Intervals.icu API key is not configured") } fn api_url(&self, path: &str) -> String { @@ -41,12 +58,13 @@ impl IntervalsClient { async fn get_json(&self, url: String) -> Result { let response = self - .authenticated(self.client.get(url))? + .authenticated(self.client.get(&url))? .send() .await .context("Intervals.icu request failed")?; let status = response.status(); + let body = response .text() .await @@ -60,40 +78,103 @@ impl IntervalsClient { .with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body)) } + /// Get the athlete belonging to the configured API key. + /// + /// 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}")); + + self.get_json(url).await + } + /// 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}")); + let url = self.api_url(&format!("activity/{activity_id}?intervals=true")); 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":[...]}, - /// ... - /// ] + /// 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")); + let url = self.api_url(&format!( + "activity/{activity_id}/streams.json?types=time,latlng" + )); 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; + /// 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)); + } + // 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)?; + + stream.get("data").and_then(Value::as_array) + } + + /// Get the latitude and longitude arrays from the + /// Intervals.icu `latlng` stream. + /// + /// Intervals.icu represents latlng as: + /// + /// data = latitude + /// data2 = longitude + /// + 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 lon = stream.get("data2")?.as_array()?; + + Some((lat, lon)) + } + + /// 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 { let url = format!( "{}?oldest={}&newest={}", self.api_url(&format!("athlete/{athlete_id}/activities")), @@ -104,27 +185,6 @@ impl IntervalsClient { self.get_json(url).await } - /// 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 stream in array { - if stream.get("type").and_then(Value::as_str) == Some(stream_type) { - return stream.get("data").and_then(Value::as_array); - } - } - } - - // Also support a dictionary-like response: - // - // { - // "time": [...], - // "latlng": [...] - // } - streams.get(stream_type).and_then(Value::as_array) - } - /// Extract the activity start timestamp. pub fn activity_start(activity: &Value) -> Result> { let value = activity @@ -142,9 +202,6 @@ impl IntervalsClient { /// 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")?; @@ -158,6 +215,7 @@ impl IntervalsClient { 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 { @@ -168,8 +226,7 @@ impl IntervalsClient { None } - /// OAuth methods are intentionally retained so we can - /// re-enable multi-user OAuth later. + /// OAuth methods retained for the normal multi-user flow. pub fn oauth_authorize_url(&self, state: &str) -> Result { let client_id = self .config @@ -219,6 +276,7 @@ impl IntervalsClient { .context("OAuth token request failed")?; let status = response.status(); + let body = response.text().await?; if !status.is_success() { @@ -252,9 +310,14 @@ impl IntervalsClient { let athlete_id = athlete .get("id") .and_then(Value::as_str) - .or_else(|| athlete.get("id").and_then(Value::as_i64).map(|_| "")) - .unwrap_or("") - .to_string(); + .map(str::to_string) + .or_else(|| { + athlete + .get("id") + .and_then(Value::as_i64) + .map(|id| id.to_string()) + }) + .unwrap_or_default(); let display_name = athlete .get("name") diff --git a/src/main.rs b/src/main.rs index e997940..ea09997 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,11 +12,12 @@ use anyhow::{Context, Result, anyhow}; use axum::{ Router, extract::{Path, State}, + response::Json, routing::{get, post}, }; use chrono::{DateTime, Duration, Utc}; -use serde_json::Value; -use sqlx::{PgPool, Row}; +use serde_json::{Value, json}; +use sqlx::PgPool; use tower_http::trace::TraceLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; @@ -64,27 +65,16 @@ async fn main() -> Result<()> { .route("/logout", get(auth::logout)) .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)) + // Development-only activity synchronisation. + // + // /dev/sync/{api_key} + // /dev/sync/{api_key}/{athlete_id} + .route("/dev/sync/{api_key}", get(dev_sync_one)) + .route("/dev/sync/{api_key}/{athlete_id}", get(dev_sync)) .layer(TraceLayer::new_for_http()) .with_state(state); - let listener = tokio::net::TcpListener::bind(&config.bind_address) - .await - .with_context(|| format!("cannot bind to {}", config.bind_address))?; + let listener = tokio::net::TcpListener::bind(&config.bind_address).await?; tracing::info!( address = %config.bind_address, @@ -100,64 +90,327 @@ async fn health() -> &'static str { "ok" } -/// Development endpoint for processing one real Intervals.icu -/// activity using INTERVALS_API_KEY. +fn redact_database_url(url: &str) -> String { + if let Some((prefix, _)) = url.split_once('@') { + if prefix.contains("://") { + return format!("{prefix}@***"); + } + } + + url.to_string() +} + +/// Development endpoint for importing/analyzing activities. /// -/// This intentionally bypasses OAuth and webhooks. -async fn dev_process_activity( +/// Examples: +/// +/// GET /dev/sync/MY_API_KEY +/// GET /dev/sync/MY_API_KEY/0 +/// GET /dev/sync/MY_API_KEY/i123456 +/// +/// `0` means the owner of the supplied API key. +async fn dev_sync( 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) + Path((api_key, athlete_id)): Path<(String, String)>, +) -> 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, + "development sync failed" + ); + + Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR) + } + } } -fn configured_athlete_id() -> String { - std::env::var("INTERVALS_ATHLETE_ID").unwrap_or_else(|_| "0".to_string()) +/// One-segment variant: +/// +/// /dev/sync/{api_key} +/// +/// Equivalent to: +/// +/// /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, + "development sync failed" + ); + + Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR) + } + } } -fn internal_error(error: anyhow::Error) -> (axum::http::StatusCode, String) { - tracing::error!( - %error, - "request failed" +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")); + } + + let athlete_id = if athlete_id.trim().is_empty() { + "0".to_string() + } else { + athlete_id + }; + + /* + * 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()); + + /* + * Resolve athlete 0 through Intervals.icu. + * + * Intervals.icu supports athlete/0 for personal API keys; + * it refers to the athlete belonging to that key. + */ + let resolved_athlete_id = if athlete_id == "0" { + let owner = client + .owner() + .await + .context("cannot resolve API-key owner")?; + + owner + .get("id") + .and_then(Value::as_str) + .map(str::to_string) + .or_else(|| { + owner + .get("id") + .and_then(Value::as_i64) + .map(|id| id.to_string()) + }) + .context("Intervals.icu owner response has no athlete id")? + } else { + athlete_id.clone() + }; + + 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!( + athlete_id = %resolved_athlete_id, + oldest = %oldest_string, + newest = %newest_string, + "starting development activity sync" ); - ( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - error.to_string(), - ) + /* + * IMPORTANT: + * + * Use the resolved athlete ID here, not the original + * "0" value. + */ + let activities = client + .activities(&resolved_athlete_id, &oldest_string, &newest_string) + .await + .context("cannot fetch development activities")?; + + let activities_array = activities + .as_array() + .context("Intervals.icu activities response is not an array")?; + + let mut processed = 0usize; + let mut skipped = 0usize; + let mut failed = 0usize; + + let mut results = Vec::new(); + + /* + * 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. + */ + 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 { + skipped += 1; + + results.push(json!({ + "status": "skipped", + "reason": "activity has no id" + })); + + continue; + }; + + match process_activity( + state.clone(), + resolved_athlete_id.clone(), + activity_id.to_string(), + ) + .await + { + Ok(()) => { + processed += 1; + + results.push(json!({ + "id": activity_id, + "status": "processed" + })); + } + + Err(error) => { + failed += 1; + + tracing::error!( + activity_id = %activity_id, + error = %error, + "development activity processing failed" + ); + + results.push(json!({ + "id": activity_id, + "status": "failed", + "error": error.to_string() + })); + } + } + } + + tracing::info!( + athlete_id = %resolved_athlete_id, + found = activities_array.len(), + processed, + skipped, + failed, + "development activity sync finished" + ); + + Ok(json!({ + "athlete_id": resolved_athlete_id, + "oldest": oldest_string, + "newest": newest_string, + "found": activities_array.len(), + "processed": processed, + "skipped": skipped, + "failed": failed, + "activities": results + })) } -/// Process one Intervals.icu activity. +/// Make sure the athlete used by /dev/sync exists locally. /// -/// In development mode the Intervals.icu personal API key from -/// INTERVALS_API_KEY is used by IntervalsClient. +/// If the athlete already exists, update its development API +/// credential so process_activity() can use it. /// -/// 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 +/// We deliberately do not print the API key. +async fn ensure_dev_athlete( + state: &AppState, + intervals_athlete_id: &str, + client: &IntervalsClient, + api_key: &str, +) -> Result<()> { + let athlete = client + .athlete(intervals_athlete_id) + .await + .with_context(|| { + format!( + "cannot fetch Intervals.icu athlete {}", + intervals_athlete_id + ) + })?; + + let display_name = athlete + .get("name") + .or_else(|| athlete.get("display_name")) + .and_then(Value::as_str) + .unwrap_or("Intervals.icu 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#" + SELECT id + FROM athletes + WHERE intervals_athlete_id = $1 + "#, + ) + .bind(intervals_athlete_id) + .fetch_optional(&state.db) + .await?; + + if existing.is_some() { + sqlx::query( + r#" + UPDATE athletes + SET + access_token = $2, + display_name = $3, + updated_at = now() + WHERE intervals_athlete_id = $1 + "#, + ) + .bind(intervals_athlete_id) + .bind(api_key) + .bind(display_name) + .execute(&state.db) + .await + .context("cannot update local dev athlete")?; + + return Ok(()); + } + + /* + * `scopes` is NOT NULL in the initial migration, so it must + * be supplied here. + */ + sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + display_name, + access_token, + scopes + ) + VALUES ($1, $2, $3, $4) + "#, + ) + .bind(intervals_athlete_id) + .bind(display_name) + .bind(api_key) + .bind("dev") + .execute(&state.db) + .await + .context("cannot create local dev athlete")?; + + Ok(()) +} + 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 + SELECT + id, + access_token FROM athletes WHERE intervals_athlete_id = $1 "#, @@ -166,73 +419,45 @@ pub async fn process_activity( .fetch_optional(&state.db) .await?; - let athlete_db_id: i64 = match athlete { - Some(row) => row.try_get("id")?, + use sqlx::Row; - 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 Some(athlete) = athlete else { + return Err(anyhow!( + "no local athlete for Intervals.icu athlete {}", + intervals_athlete_id + )); }; + let athlete_db_id: i64 = athlete.try_get("id")?; + + /* + * The normal activity-processing path uses the configured + * Intervals client. + * + * The development endpoint has already ensured that the + * athlete exists locally and that its credential is stored. + */ let client = IntervalsClient::new(state.config.clone()); /* - * First fetch the activity metadata. - * - * Among other things this may contain Intervals.icu's - * automatically detected intervals. + * Fetch activity metadata. */ - let activity = client.activity(&activity_id).await?; + let activity = client + .activity(&activity_id) + .await + .with_context(|| format!("cannot fetch Intervals.icu activity {}", activity_id))?; let start_time = IntervalsClient::activity_start(&activity)?; /* - * 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. + * Fetch GPS/time streams. */ - let streams = client.streams(&activity_id).await?; + 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)?; @@ -246,22 +471,16 @@ pub async fn process_activity( return Ok(()); } - 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: + * Re-processing is idempotent: * - * old crossings are deleted - * new crossings are inserted - * activity is marked processed + * - update activity + * - remove existing crossings + * - calculate crossings + * - insert crossings + * - mark processed * - * all atomically. + * All changes happen inside one transaction. */ let mut tx = state.db.begin().await?; @@ -276,14 +495,6 @@ pub async fn process_activity( let crossings = geo::crossings_between(&state.db, &pair[0], &pair[1]).await?; for crossing in crossings { - /* - * 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?; @@ -296,7 +507,6 @@ 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" ); } @@ -351,15 +561,6 @@ fn parse_track(streams: &Value, start_time: DateTime) -> Result) -> Result 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};