From 859603399a8f5da00afe13c3ee20319e905e15c8 Mon Sep 17 00:00:00 2001 From: Jonas Rabenstein Date: Thu, 13 Aug 2026 02:43:17 +0200 Subject: [PATCH] sync --- Cargo.toml | 2 +- src/db.rs | 181 ++++++- src/main.rs | 1456 ++++++++++++++++++++++++++++++++++++++++----------- src/web.rs | 2 +- 4 files changed, 1306 insertions(+), 335 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index d1a0c45..2bc0ec5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ edition = "2024" [dependencies] anyhow = "1" -axum = "0.8" +axum = { version = "0.8", features = [ "macros" ] } axum-extra = { version = "0.10", features = ["cookie"] } chrono = { version = "0.4", features = ["serde"] } dotenvy = "0.15" diff --git a/src/db.rs b/src/db.rs index ae18974..7214d0b 100644 --- a/src/db.rs +++ b/src/db.rs @@ -8,16 +8,25 @@ 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() } -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( r#" - INSERT INTO sessions (token_hash, athlete_id) + INSERT INTO sessions ( + token_hash, + athlete_id + ) VALUES ($1, $2) "#, ) @@ -29,14 +38,20 @@ pub async fn create_session(db: &PgPool, athlete_id: i64, token: &str) -> Result 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( r#" - SELECT a.id, a.display_name + SELECT + a.id, + a.display_name FROM sessions s - JOIN athletes a ON a.id = s.athlete_id + JOIN athletes a + ON a.id = s.athlete_id WHERE s.token_hash = $1 "#, ) @@ -47,8 +62,11 @@ pub async fn athlete_for_session(db: &PgPool, token: &str) -> Result Result Result<()> { +#[derive(Debug, Clone)] +pub struct SessionAthlete { + pub id: i64, + pub intervals_athlete_id: String, + pub display_name: String, + pub access_token: String, +} + +pub async fn session_athlete( + db: &PgPool, + token: &str, +) -> Result> { + let hash = hash_token(token); + + let row = sqlx::query( + r#" + SELECT + a.id, + a.intervals_athlete_id, + a.display_name, + a.access_token + FROM sessions s + JOIN athletes a + ON a.id = s.athlete_id + WHERE s.token_hash = $1 + "#, + ) + .bind(&hash) + .fetch_optional(db) + .await?; + + use sqlx::Row; + + let Some(row) = row else { + return Ok(None); + }; + + sqlx::query( + r#" + UPDATE sessions + SET last_seen_at = now() + WHERE token_hash = $1 + "#, + ) + .bind(&hash) + .execute(db) + .await?; + + Ok(Some(SessionAthlete { + id: row.try_get("id")?, + intervals_athlete_id: row.try_get("intervals_athlete_id")?, + display_name: row.try_get("display_name")?, + access_token: row.try_get("access_token")?, + })) +} + +/// Return the start time of the most recently imported activity +/// for an athlete. +/// +/// This is used as the default `oldest` value on the sync page. +pub async fn last_synced_activity_time( + db: &PgPool, + athlete_id: i64, +) -> Result>> { + let row = sqlx::query( + r#" + SELECT MAX(start_time) AS last_synced + FROM activities + WHERE athlete_id = $1 + AND processed_at IS NOT NULL + "#, + ) + .bind(athlete_id) + .fetch_one(db) + .await?; + + use sqlx::Row; + + Ok(row.try_get("last_synced")?) +} + +pub async fn delete_old_oauth_states( + db: &PgPool, +) -> Result<()> { sqlx::query( r#" DELETE FROM oauth_states @@ -80,6 +181,40 @@ pub async fn delete_old_oauth_states(db: &PgPool) -> Result<()> { Ok(()) } +pub async fn ensure_dev_athlete( + db: &PgPool, + intervals_athlete_id: &str, + display_name: &str, + api_key: &str, +) -> Result { + let row = sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + display_name, + access_token, + scopes + ) + VALUES ($1, $2, $3, 'dev') + ON CONFLICT (intervals_athlete_id) + DO UPDATE SET + display_name = EXCLUDED.display_name, + access_token = EXCLUDED.access_token, + updated_at = now() + RETURNING id + "#, + ) + .bind(intervals_athlete_id) + .bind(display_name) + .bind(api_key) + .fetch_one(db) + .await?; + + use sqlx::Row; + + Ok(row.try_get("id")?) +} + pub async fn ensure_activity( tx: &mut Transaction<'_, Postgres>, athlete_id: i64, @@ -96,8 +231,17 @@ pub async fn ensure_activity( activity_json, processed_at ) - VALUES ($1, $2, $3, $4, NULL) - ON CONFLICT (athlete_id, intervals_activity_id) + VALUES ( + $1, + $2, + $3, + $4, + NULL + ) + ON CONFLICT ( + athlete_id, + intervals_activity_id + ) DO UPDATE SET start_time = EXCLUDED.start_time, activity_json = EXCLUDED.activity_json, @@ -113,6 +257,7 @@ pub async fn ensure_activity( .await?; use sqlx::Row; + Ok(row.try_get("id")?) } @@ -154,7 +299,10 @@ pub async fn save_crossing( $1, $2, $3, - ST_SetSRID(ST_MakePoint($4, $5), 4326), + ST_SetSRID( + ST_MakePoint($4, $5), + 4326 + ), $6, $7, $8 @@ -192,14 +340,3 @@ pub async fn mark_activity_processed( Ok(()) } - -pub async fn delete_session(db: &PgPool, token: &str) -> Result<()> { - let hash = hash_token(token); - - sqlx::query("DELETE FROM sessions WHERE token_hash = $1") - .bind(&hash) - .execute(db) - .await?; - - Ok(()) -} diff --git a/src/main.rs b/src/main.rs index 9fefc82..5dc451d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,23 +5,32 @@ mod geo; mod intervals; mod leaderboard; mod model; -mod web; mod webhook; +mod web; -use anyhow::{Context, Result, anyhow}; +use anyhow::{anyhow, Context, Result}; use axum::{ - Router, - extract::{Path, State}, - response::Json, + extract::{Path, Query, State}, + response::Html, routing::{get, post}, + Router, }; use chrono::{DateTime, Duration, Utc}; -use serde_json::{Value, json}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; use sqlx::PgPool; 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, + db::SessionAthlete, + intervals::IntervalsClient, + model::TrackPoint, +}; #[derive(Clone)] pub struct AppState { @@ -29,28 +38,56 @@ pub struct AppState { pub config: Config, } +#[derive(Debug, Deserialize)] +pub struct SyncQuery { + pub oldest: Option, + pub newest: Option, +} + +#[derive(Debug, Serialize)] +pub struct SyncActivity { + pub id: String, + pub name: String, + pub activity_type: String, + pub start_time: Option, + pub distance: Option, + pub imported: bool, +} + #[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()?; tracing::info!( - database_url = %redact_database_url(&config.database_url), + database_url = %redact_database_url( + &config.database_url + ), "connecting to PostgreSQL" ); - let db = PgPool::connect(&config.database_url) + let db = + PgPool::connect( + &config.database_url + ) .await - .context("cannot connect to PostgreSQL")?; + .context( + "cannot connect to PostgreSQL" + )?; sqlx::migrate!() .run(&db) .await - .context("database migration failed")?; + .context( + "database migration failed" + )?; let state = AppState { db, @@ -58,29 +95,72 @@ async fn main() -> Result<()> { }; let app = Router::new() - .route("/", get(web::index)) - .route("/health", get(health)) - .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("/activity/{activity_id}", get(web::activity)) - .route("/groups", get(web::groups).post(web::save_group)) - .route("/groups/delete", post(web::delete_group)) - .route("/dev/sync/{api_key}", get(dev_sync_one)) - .route("/dev/sync/{api_key}/{athlete_id}", get(dev_sync)) - .layer(TraceLayer::new_for_http()) + .route( + "/", + get(web::index), + ) + .route( + "/health", + get(health), + ) + .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( + "/sync", + get(sync_index), + ) + .route( + "/sync/{athlete_id}", + get(sync_athlete), + ) + .route( + "/sync/{athlete_id}/activities/{activity_id}/import", + post(sync_import_activity), + ) + .route( + "/sync/{athlete_id}/import-visible", + post(sync_import_visible), + ) + .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?; tracing::info!( address = %config.bind_address, "server listening" ); - axum::serve(listener, app).await?; + axum::serve( + listener, + app, + ) + .await?; + Ok(()) } @@ -88,258 +168,930 @@ async fn health() -> &'static str { "ok" } -fn redact_database_url(url: &str) -> String { - if let Some((prefix, _)) = url.split_once('@') { +fn redact_database_url( + url: &str, +) -> String { + if let Some((prefix, _)) = + url.split_once('@') + { if prefix.contains("://") { - return format!("{prefix}@***"); + return format!( + "{prefix}@***" + ); } } + url.to_string() } -async fn dev_sync( - State(state): State, - 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) - } - } +async fn current_session_athlete( + state: &AppState, + jar: &axum_extra::extract::cookie::CookieJar, +) -> Result { + let cookie = + jar.get("county_session") + .context( + "not authenticated" + )?; + + db::session_athlete( + &state.db, + cookie.value(), + ) + .await? + .context( + "session is invalid or expired" + ) } -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) - } - } -} - -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 - }; - - let client = IntervalsClient::with_api_key(state.config.clone(), api_key); - - let resolved_athlete_id = if athlete_id == "0" { - let owner = client.owner().await?; - - 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 - chrono::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" - ); - - let activities = client - .activities(&athlete_id, &oldest_string, &newest_string) +async fn require_athlete_access( + state: &AppState, + jar: &axum_extra::extract::cookie::CookieJar, + athlete_id: &str, +) -> Result { + let session = + current_session_athlete( + state, + jar, + ) .await?; - let activities_array = activities - .as_array() - .context("Intervals.icu activities response is not an array")?; + if session.intervals_athlete_id + != athlete_id + { + return Err(anyhow!( + "current authentication token has no access to athlete {}", + athlete_id + )); + } - ensure_dev_athlete(&state, &resolved_athlete_id, &client).await?; + Ok(session) +} - let mut processed = 0usize; - let mut skipped = 0usize; - let mut failed = 0usize; - let mut results = Vec::new(); +async fn sync_index( + State(state): State, + jar: axum_extra::extract::cookie::CookieJar, +) -> Result, (axum::http::StatusCode, String)> { + let session = + current_session_athlete( + &state, + &jar, + ) + .await + .map_err(http_error)?; - 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"})); + let html = format!( + r#" + + + + +Sync · Landkreis-Sprints + + + +

Synchronisation

+ +
+

{}

+
+ Intervals.icu Athlete: {} +
+ +

+ + Aktivitäten anzeigen + +

+
+ +

+ ← Leaderboard +

+ +"#, + web_escape( + &session.display_name + ), + web_escape( + &session.intervals_athlete_id + ), + url_segment( + &session.intervals_athlete_id + ), + ); + + Ok(Html(html)) +} + +async fn sync_athlete( + State(state): State, + Path(athlete_id): Path, + Query(query): Query, + jar: axum_extra::extract::cookie::CookieJar, +) -> Result, (axum::http::StatusCode, String)> { + let session = + require_athlete_access( + &state, + &jar, + &athlete_id, + ) + .await + .map_err(http_error)?; + + /* + * Default: + * + * oldest = start_time of last successfully imported activity + * newest = now + * + * An explicit query value overrides the default. + */ + let default_oldest = + db::last_synced_activity_time( + &state.db, + session.id, + ) + .await + .map_err(http_error)? + .unwrap_or_else(|| { + Utc::now() + - Duration::days(30) + }); + + let oldest = + parse_datetime_local( + query.oldest.as_deref() + ) + .unwrap_or( + default_oldest + ); + + let newest = + parse_datetime_local( + query.newest.as_deref() + ) + .unwrap_or_else( + || Utc::now() + ); + + let client = + IntervalsClient::with_api_key( + state.config.clone(), + session.access_token.clone(), + ); + + let activities = + client + .activities( + &athlete_id, + &oldest + .format("%Y-%m-%d") + .to_string(), + &newest + .format("%Y-%m-%d") + .to_string(), + ) + .await + .map_err(http_error)?; + + let activities = + activities + .as_array() + .ok_or_else(|| { + http_error(anyhow!( + "activities response is not an array" + )) + })?; + + let mut visible = + Vec::::new(); + + for activity in activities { + let Some(activity_id) = + activity + .get("id") + .and_then(Value::as_str) + else { continue; }; - match process_activity_with_client( - state.clone(), - resolved_athlete_id.clone(), - activity_id.to_string(), - &client, - ) - .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() - })); - } + let imported = + activity_exists( + &state.db, + session.id, + activity_id, + ) + .await + .map_err(http_error)?; + + if imported { + continue; } + + let name = + activity + .get("name") + .and_then(Value::as_str) + .unwrap_or( + "Aktivität" + ); + + let activity_type = + activity + .get("type") + .and_then(Value::as_str) + .unwrap_or(""); + + let start_time = + activity + .get("start_date_local") + .or_else(|| { + activity.get( + "start_date" + ) + }) + .and_then(Value::as_str); + + let distance = + activity + .get("distance") + .and_then(Value::as_f64); + + visible.push( + SyncActivity { + id: activity_id.to_string(), + name: name.to_string(), + activity_type: + activity_type.to_string(), + start_time: + start_time + .map(str::to_string), + distance, + imported: false, + } + ); } - tracing::info!( - athlete_id = %resolved_athlete_id, - found = activities_array.len(), - processed, - skipped, - failed, - "development activity sync finished" + let mut html = + String::from( + r#" + + + + +Sync · Landkreis-Sprints + + + +"#, + ); + + html.push_str(&format!( + "

Sync: {}

", + web_escape( + &session.display_name + ) + )); + + html.push_str( + r#"

+← Nutzer +

"#, ); - 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 - })) -} + let oldest_input = + format_datetime_local( + oldest + ); -async fn ensure_dev_athlete( - state: &AppState, - intervals_athlete_id: &str, - client: &IntervalsClient, -) -> Result<()> { - let athlete = client.athlete(intervals_athlete_id).await?; + let newest_input = + format_datetime_local( + newest + ); - let display_name = athlete - .get("name") - .or_else(|| athlete.get("display_name")) - .and_then(Value::as_str) - .unwrap_or("Intervals.icu athlete"); + html.push_str(&format!( + r#"
+
+
+
+ + +
- let exists = sqlx::query( - r#" - SELECT id FROM athletes WHERE intervals_athlete_id = $1 - "#, - ) - .bind(intervals_athlete_id) - .fetch_optional(&state.db) - .await?; +
+ + +
+
- if exists.is_some() { - return Ok(()); +

+ +

+
+
"#, + url_segment(&athlete_id), + oldest_input, + newest_input, + )); + + html.push_str(&format!( + r#"
+{} Aktivitäten gefunden, +{} davon noch nicht importiert. +
"#, + activities.len(), + visible.len(), + )); + + if !visible.is_empty() { + html.push_str(&format!( + r#"
+
+ + + +
+
"#, + url_segment( + &athlete_id + ), + web_escape( + &oldest_input + ), + web_escape( + &newest_input + ), + )); } - sqlx::query( - r#" - INSERT INTO athletes ( - intervals_athlete_id, - access_token, - display_name, - scopes - ) - VALUES ($1, $2, $3, $4) - "#, - ) - .bind(intervals_athlete_id) - .bind("") - .bind(display_name) - .bind("") - .execute(&state.db) - .await - .context("cannot create local dev athlete")?; + if visible.is_empty() { + html.push_str( + r#"
+ Keine nicht importierten Aktivitäten + im gewählten Zeitraum. +
"#, + ); + } - Ok(()) + for activity in &visible { + let distance = + activity.distance + .map(|m| { + format!( + "{:.1} km", + m / 1000.0 + ) + }) + .unwrap_or_else( + || "—".into() + ); + + html.push_str( + r#"
"#, + ); + + html.push_str(&format!( + r#"
+{} +
+{} · {} · {} +
+
"#, + web_escape( + &activity.name + ), + web_escape( + &activity.activity_type + ), + web_escape( + activity + .start_time + .as_deref() + .unwrap_or("") + ), + distance, + )); + + html.push_str(&format!( + r#"
+ +
"#, + url_segment( + &athlete_id + ), + url_segment( + &activity.id + ), + )); + + html.push_str( + "
", + ); + } + + html.push_str( + r#"

+← Leaderboard +

+ +"#, + ); + + Ok(Html(html)) +} + +async fn sync_import_activity( + State(state): State, + Path((athlete_id, activity_id)): Path<( + String, + String, + )>, + jar: axum_extra::extract::cookie::CookieJar, +) -> Result, (axum::http::StatusCode, String)> { + require_athlete_access( + &state, + &jar, + &athlete_id, + ) + .await + .map_err(http_error)?; + + process_activity( + state.clone(), + athlete_id.clone(), + activity_id.clone(), + ) + .await + .map_err(http_error)?; + + Ok(Html(format!( + r#" + + + + +Aktivität importiert + + +

+Aktivität {} wurde importiert. +

+

+Weiter +

+ +"#, + url_segment( + &athlete_id + ), + web_escape( + &activity_id + ), + url_segment( + &athlete_id + ), + ))) +} + +#[derive(Debug, Deserialize)] +struct ImportVisibleForm { + oldest: String, + newest: String, +} + +#[axum::debug_handler] +async fn sync_import_visible( + State(state): State, + Path(athlete_id): Path, + jar: axum_extra::extract::cookie::CookieJar, + axum::Form(form): axum::Form, +) -> Result, (axum::http::StatusCode, String)> { + let session = + require_athlete_access( + &state, + &jar, + &athlete_id, + ) + .await + .map_err(http_error)?; + + let oldest = + parse_datetime_local( + Some(&form.oldest) + ) + .unwrap_or_else( + || { + Utc::now() + - Duration::days(30) + } + ); + + let newest = + parse_datetime_local( + Some(&form.newest) + ) + .unwrap_or_else( + || Utc::now() + ); + + let client = + IntervalsClient::with_api_key( + state.config.clone(), + session.access_token.clone(), + ); + + let activities = + client.activities( + &athlete_id, + &oldest + .format("%Y-%m-%d") + .to_string(), + &newest + .format("%Y-%m-%d") + .to_string(), + ) + .await + .map_err(http_error)?; + + let activities = + activities + .as_array() + .ok_or_else(|| { + http_error(anyhow!( + "activities response is not an array" + )) + })?; + + for activity in activities { + let Some(activity_id) = + activity + .get("id") + .and_then(Value::as_str) + else { + continue; + }; + + let imported = + activity_exists( + &state.db, + session.id, + activity_id, + ) + .await + .map_err(http_error)?; + + if imported { + continue; + } + + /* + * The API's date parameters operate at day precision, + * so the exact time range is checked again here before + * importing. + */ + let Some(activity_time) = + activity_start_time(activity) + else { + continue; + }; + + if activity_time < oldest + || activity_time > newest + { + continue; + } + + process_activity( + state.clone(), + athlete_id.clone(), + activity_id.to_string(), + ) + .await + .map_err(http_error)?; + } + + Ok(Html(format!( + r#" + + + + +Synchronisation + + +

+Aktivitäten werden importiert … +

+

+Weiter +

+ +"#, + url_segment( + &athlete_id + ), + url_query( + &form.oldest + ), + url_query( + &form.newest + ), + url_segment( + &athlete_id + ), + ))) +} + +async fn activity_exists( + db: &PgPool, + athlete_id: i64, + activity_id: &str, +) -> Result { + let row = + sqlx::query( + r#" + SELECT EXISTS ( + SELECT 1 + FROM activities + WHERE athlete_id = $1 + AND intervals_activity_id = $2 + ) AS exists + "#, + ) + .bind(athlete_id) + .bind(activity_id) + .fetch_one(db) + .await?; + + use sqlx::Row; + + Ok(row.try_get("exists")?) +} + +fn activity_start_time( + activity: &Value, +) -> Option> { + let value = + activity + .get("start_date") + .or_else(|| { + activity.get( + "start_date_local" + ) + })?; + + value + .as_str() + .and_then(|value| { + parse_datetime_local( + Some(value) + ) + }) +} + +fn parse_datetime_local( + value: Option<&str>, +) -> Option> { + let value = value?; + + if let Ok(parsed) = + DateTime::parse_from_rfc3339( + value + ) + { + return Some( + parsed.with_timezone(&Utc) + ); + } + + if let Ok(parsed) = + DateTime::parse_from_str( + value, + "%Y-%m-%dT%H:%M", + ) + { + return Some( + parsed.with_timezone(&Utc) + ); + } + + /* + * datetime-local has no timezone information. + * + * We interpret it in the server/browser-independent + * local convention only for range comparison. For the UI, + * the selected values are sent back unchanged. + * + * Since the application is intended to run in one local + * environment for now, this is sufficient. + */ + chrono::NaiveDateTime::parse_from_str( + value, + "%Y-%m-%dT%H:%M", + ) + .ok() + .map(|naive| { + DateTime::::from_naive_utc_and_offset( + naive, + Utc, + ) + }) +} + +fn format_datetime_local( + value: DateTime, +) -> String { + value + .format("%Y-%m-%dT%H:%M") + .to_string() +} + +fn http_error( + error: anyhow::Error, +) -> ( + axum::http::StatusCode, + String, +) { + tracing::error!( + error = ?error, + "request failed" + ); + + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + error.to_string(), + ) +} + +fn web_escape( + value: &str, +) -> String { + value + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) + .replace('\'', "'") +} + +fn url_segment( + value: &str, +) -> String { + urlencoding::encode( + value + ) + .into_owned() +} + +fn url_query( + value: &str, +) -> String { + urlencoding::encode( + value + ) + .into_owned() } -/// Normal processing entry point used by the webhook path. -/// The OAuth access token is loaded from the local athlete row. pub async fn process_activity( state: AppState, intervals_athlete_id: String, activity_id: String, ) -> Result<()> { - let access_token: String = sqlx::query_scalar( - r#" - SELECT access_token - FROM athletes - WHERE intervals_athlete_id = $1 - "#, - ) - .bind(&intervals_athlete_id) - .fetch_optional(&state.db) - .await? - .ok_or_else(|| { - anyhow!( - "no local athlete for Intervals.icu athlete {}", - intervals_athlete_id + let athlete = + sqlx::query( + r#" + SELECT + id, + access_token + FROM athletes + WHERE intervals_athlete_id = $1 + "#, ) - })?; - - if access_token.trim().is_empty() { - return Err(anyhow!( - "local athlete {} has no access token", - intervals_athlete_id - )); - } - - let client = IntervalsClient::with_api_key(state.config.clone(), access_token); - - process_activity_with_client(state, intervals_athlete_id, activity_id, &client).await -} - -/// Same processing pipeline, but with an explicitly supplied Intervals client. -/// The development endpoint uses this so its personal API key is never -/// written into the database. -pub async fn process_activity_with_client( - state: AppState, - intervals_athlete_id: String, - activity_id: String, - client: &IntervalsClient, -) -> Result<()> { - let athlete = sqlx::query( - r#" - SELECT id - FROM athletes - WHERE intervals_athlete_id = $1 - "#, - ) - .bind(&intervals_athlete_id) - .fetch_optional(&state.db) - .await?; + .bind( + &intervals_athlete_id + ) + .fetch_optional(&state.db) + .await?; use sqlx::Row; @@ -350,18 +1102,44 @@ pub async fn process_activity_with_client( )); }; - let athlete_db_id: i64 = athlete.try_get("id")?; + let athlete_db_id: i64 = + athlete.try_get("id")?; - let activity = client.activity(&activity_id).await?; - let start_time = IntervalsClient::activity_start(&activity)?; - let streams = client.streams(&activity_id).await?; - let points = parse_track(&streams, start_time)?; + let access_token: String = + athlete.try_get( + "access_token" + )?; - tracing::debug!( - activity_id = %activity_id, - points = points.len(), - "parsed activity GPS points" - ); + let client = + IntervalsClient::with_api_key( + state.config.clone(), + access_token, + ); + + let activity = + client + .activity( + &activity_id + ) + .await?; + + let start_time = + IntervalsClient::activity_start( + &activity + )?; + + let streams = + client + .streams( + &activity_id + ) + .await?; + + let points = + parse_track( + &streams, + start_time + )?; if points.len() < 2 { tracing::info!( @@ -369,25 +1147,54 @@ pub async fn process_activity_with_client( points = points.len(), "activity has insufficient GPS data" ); + return Ok(()); } - let mut tx = state.db.begin().await?; + 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; @@ -397,14 +1204,17 @@ pub async fn process_activity_with_client( from = crossing.from_county_id, to = crossing.to_county_id, track_index = crossing.track_index, - lat = crossing.lat, - lon = crossing.lon, "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?; tracing::info!( @@ -416,71 +1226,95 @@ pub async fn process_activity_with_client( 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 (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 mut result = Vec::with_capacity(length); + let length = + times + .len() + .min(latitudes.len()) + .min(longitudes.len()); + + 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(lat) = + latitudes[index].as_f64() + else { continue; }; - if coordinate.len() != 2 { + let Some(lon) = + longitudes[index].as_f64() + else { + continue; + }; + + if !seconds.is_finite() + || !lat.is_finite() + || !lon.is_finite() + { continue; } - let Some(lat) = coordinate[0].as_f64() else { - continue; - }; - let Some(lon) = coordinate[1].as_f64() else { - continue; - }; - - if !lat.is_finite() || !lon.is_finite() || !seconds.is_finite() { + if !(-90.0..=90.0) + .contains(&lat) + { continue; } - let millis = (seconds * 1000.0).round() as i64; + if !(-180.0..=180.0) + .contains(&lon) + { + continue; + } - result.push(TrackPoint { - index, - time: start_time + Duration::milliseconds(millis), - lat, - lon, - }); + let millis = + (seconds * 1000.0) + .round() as i64; + + result.push( + TrackPoint { + index, + time: + start_time + + Duration::milliseconds( + millis + ), + lat, + lon, + } + ); } Ok(result) } - -#[cfg(test)] -mod tests { - use crate::geo::interpolate_time; - use chrono::{TimeZone, Utc}; - - #[test] - fn interpolation_is_millisecond_precise() { - let a = Utc.timestamp_millis_opt(1000).unwrap(); - let b = Utc.timestamp_millis_opt(2000).unwrap(); - let result = interpolate_time(a, b, 0.376); - assert_eq!(result.timestamp_millis(), 1376); - } - - #[test] - fn interpolation_rounds_to_nearest_ms() { - let a = Utc.timestamp_millis_opt(1000).unwrap(); - let b = Utc.timestamp_millis_opt(2000).unwrap(); - let result = interpolate_time(a, b, 0.3764); - assert_eq!(result.timestamp_millis(), 1376); - } -} diff --git a/src/web.rs b/src/web.rs index dd327f7..c092a02 100644 --- a/src/web.rs +++ b/src/web.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::HashSet; use chrono::{DateTime, Utc};