commit a9d22e831233fd47a202a6476a3a7939925b5ecf Author: Jonas Rabenstein Date: Wed Aug 12 13:24:01 2026 +0200 chatgpt diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..23cc376 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +DATABASE_URL=postgresql://county_sprints@/county_sprints?host=.pgsocket&port=5433 + +BIND_ADDRESS=127.0.0.1:8080 + +INTERVALS_BASE_URL=https://intervals.icu +INTERVALS_CLIENT_ID= +INTERVALS_CLIENT_SECRET= +INTERVALS_REDIRECT_URI=http://localhost:8080/oauth/callback +INTERVALS_WEBHOOK_SECRET= + +# false for local HTTP development, true behind HTTPS in production. +COOKIE_SECURE=false + +RUST_LOG=info diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c539170 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +/target +/.pgdata +/.pgsocket +.env diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..86a5324 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,44 @@ +[package] +name = "county-sprints" +version = "0.1.0" +edition = "2024" + +[dependencies] +anyhow = "1" +axum = "0.8" +axum-extra = { version = "0.10", features = ["cookie"] } +chrono = { version = "0.4", features = ["serde"] } +dotenvy = "0.15" +rand = "0.9" +reqwest = { + version = "0.12", + default-features = false, + features = ["json", "form", "rustls-tls"] +} +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.10" +sqlx = { + version = "0.8", + features = [ + "runtime-tokio-rustls", + "postgres", + "chrono", + "json", + "migrate" + ] +} +tokio = { + version = "1", + features = ["full"] +} +tower-http = { + version = "0.6", + features = ["trace"] +} +tracing = "0.1" +tracing-subscriber = { + version = "0.3", + features = ["env-filter"] +} +urlencoding = "2" diff --git a/README.md b/README.md new file mode 100644 index 0000000..dd9f37b --- /dev/null +++ b/README.md @@ -0,0 +1,43 @@ +# Landkreis-Sprints + +Automatische Auswertung von Landkreis-Sprints bei gemeinsamen Radausfahrten. + +## Architektur + +```text +Garmin / Wahoo / ... + | + v + Intervals.icu + | + | ACTIVITY_ANALYZED + v + /webhooks/intervals + | + v + Rust + | + +---- GET activity?intervals=true + | + +---- GET streams.json + | time + latlng + | + v + PostgreSQL + + PostGIS + | + v + Landkreis-Grenze + | + v + crossing_time + millisecond + | + v + icu_interval + | + v + 10-Minuten-Bucket + | + v + Leaderboard diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..b1ce508 --- /dev/null +++ b/flake.nix @@ -0,0 +1,127 @@ +{ + description = "Landkreis-Sprints"; + + inputs = { + nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable"; + rust-overlay.url = "github:oxalica/rust-overlay"; + flake-utils.url = "github:numtide/flake-utils"; + }; + + outputs = { nixpkgs, rust-overlay, flake-utils, ... }: + flake-utils.lib.eachDefaultSystem (system: + let + pkgs = import nixpkgs { + inherit system; + overlays = [ (import rust-overlay) ]; + }; + + rust = pkgs.rust-bin.stable.latest.default.override { + extensions = [ + "rustfmt" + "clippy" + ]; + }; + in { + devShells.default = pkgs.mkShell { + packages = with pkgs; [ + rust + + postgresql + postgis + + sqlx-cli + + gdal + geos + proj + + pkg-config + openssl + ]; + + shellHook = '' + export PGDATA="$PWD/.pgdata" + export PGHOST="$PWD/.pgsocket" + export PGPORT="5433" + export PGDATABASE="county_sprints" + export PGUSER="county_sprints" + + mkdir -p "$PGHOST" + + if [ ! -f "$PGDATA/PG_VERSION" ]; then + echo "initialising PostgreSQL..." + + initdb \ + --auth=trust \ + --no-locale \ + "$PGDATA" >/dev/null + + cat >> "$PGDATA/postgresql.conf" </dev/null 2>&1; then + pg_ctl \ + -D "$PGDATA" \ + -o "-k $PGHOST -p $PGPORT" \ + -l "$PGDATA/postgresql.log" \ + start >/dev/null + fi + + if ! psql \ + -h "$PGHOST" \ + -p "$PGPORT" \ + -d postgres \ + -tAc "SELECT 1 FROM pg_roles WHERE rolname='$PGUSER'" | + grep -q 1 + then + createuser \ + -h "$PGHOST" \ + -p "$PGPORT" \ + "$PGUSER" + fi + + if ! psql \ + -h "$PGHOST" \ + -p "$PGPORT" \ + -d postgres \ + -tAc "SELECT 1 FROM pg_database WHERE datname='$PGDATABASE'" | + grep -q 1 + then + createdb \ + -h "$PGHOST" \ + -p "$PGPORT" \ + -O "$PGUSER" \ + "$PGDATABASE" + fi + + if ! psql \ + -h "$PGHOST" \ + -p "$PGPORT" \ + -d "$PGDATABASE" \ + -tAc "SELECT 1 FROM pg_extension WHERE extname='postgis'" | + grep -q 1 + then + psql \ + -h "$PGHOST" \ + -p "$PGPORT" \ + -d "$PGDATABASE" \ + -c "CREATE EXTENSION postgis;" >/dev/null + fi + + export DATABASE_URL="postgresql://$PGUSER@/$PGDATABASE?host=$PGHOST&port=$PGPORT" + + echo + echo "========================================" + echo " county-sprints development environment" + echo "========================================" + echo "DATABASE_URL=$DATABASE_URL" + echo + ''; + }; + } + ); +} diff --git a/migrations/0001_initial.sql b/migrations/0001_initial.sql new file mode 100644 index 0000000..33df15d --- /dev/null +++ b/migrations/0001_initial.sql @@ -0,0 +1,105 @@ +CREATE EXTENSION IF NOT EXISTS postgis; + +CREATE TABLE athletes ( + id BIGSERIAL PRIMARY KEY, + + intervals_athlete_id TEXT NOT NULL UNIQUE, + display_name TEXT NOT NULL, + + access_token TEXT NOT NULL, + scopes TEXT NOT NULL, + + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE oauth_states ( + state TEXT PRIMARY KEY, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE sessions ( + token_hash BYTEA PRIMARY KEY, + + athlete_id BIGINT NOT NULL + REFERENCES athletes(id) + ON DELETE CASCADE, + + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX sessions_athlete_idx + ON sessions (athlete_id); + +CREATE TABLE activities ( + id BIGSERIAL PRIMARY KEY, + + athlete_id BIGINT NOT NULL + REFERENCES athletes(id) + ON DELETE CASCADE, + + intervals_activity_id TEXT NOT NULL, + + start_time TIMESTAMPTZ, + processed_at TIMESTAMPTZ, + + activity_json JSONB, + + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + + UNIQUE (athlete_id, intervals_activity_id) +); + +CREATE INDEX activities_start_time_idx + ON activities(start_time); + +CREATE TABLE county_boundaries ( + id BIGSERIAL PRIMARY KEY, + + name TEXT NOT NULL, + + geometry geometry(MultiPolygon, 4326) NOT NULL +); + +CREATE INDEX county_boundaries_geometry_idx + ON county_boundaries + USING GIST (geometry); + +CREATE TABLE county_crossings ( + id BIGSERIAL PRIMARY KEY, + + activity_id BIGINT NOT NULL + REFERENCES activities(id) + ON DELETE CASCADE, + + track_index BIGINT NOT NULL, + + crossing_time TIMESTAMPTZ(3) NOT NULL, + + location geometry(Point, 4326) NOT NULL, + + from_county_id BIGINT NOT NULL + REFERENCES county_boundaries(id), + + to_county_id BIGINT NOT NULL + REFERENCES county_boundaries(id), + + intervals_interval JSONB, + + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX county_crossings_time_idx + ON county_crossings(crossing_time); + +CREATE INDEX county_crossings_location_idx + ON county_crossings + USING GIST (location); + +CREATE INDEX county_crossings_direction_time_idx + ON county_crossings( + from_county_id, + to_county_id, + crossing_time + ); diff --git a/rust-toolchain.toml b/rust-toolchain.toml new file mode 100644 index 0000000..8917f32 --- /dev/null +++ b/rust-toolchain.toml @@ -0,0 +1,6 @@ +[toolchain] +channel = "stable" +components = [ + "rustfmt", + "clippy" +] diff --git a/scripts/import-vg250.sh b/scripts/import-vg250.sh new file mode 100755 index 0000000..5ea96ca --- /dev/null +++ b/scripts/import-vg250.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +set -euo pipefail + +if [ "$#" -ne 1 ]; then + echo "usage: $0 DE_VG250.gpkg" + exit 1 +fi + +INPUT="$1" + +if [ ! -f "$INPUT" ]; then + echo "file not found: $INPUT" + exit 1 +fi + +: "${DATABASE_URL:?DATABASE_URL must be set}" + +echo "Available layers:" +ogrinfo "$INPUT" | sed -n '1,80p' + +echo +echo "Importing vg250_krs..." + +ogr2ogr \ + -f PostgreSQL \ + "$DATABASE_URL" \ + "$INPUT" \ + -nln vg250_krs_import \ + -nlt PROMOTE_TO_MULTI \ + -t_srs EPSG:4326 \ + -overwrite + +echo "Import completed." + +psql "$DATABASE_URL" <<'SQL' +TRUNCATE county_boundaries RESTART IDENTITY; + +INSERT INTO county_boundaries ( + name, + geometry +) +SELECT + COALESCE( + NULLIF("GEN", ''), + NULLIF("BEZ", ''), + NULLIF("ARS", ''), + 'unknown' + ) AS name, + + ST_Multi( + ST_CollectionExtract( + ST_MakeValid(wkb_geometry), + 3 + ) + )::geometry(MultiPolygon,4326) + +FROM vg250_krs_import + +WHERE wkb_geometry IS NOT NULL; + +DROP TABLE vg250_krs_import; + +CREATE INDEX county_boundaries_geometry_idx + ON county_boundaries + USING GIST (geometry); + +ANALYZE county_boundaries; +SQL + +echo +echo "Imported counties:" +psql "$DATABASE_URL" \ + -c "SELECT id, name FROM county_boundaries ORDER BY name;" diff --git a/src/auth.rs b/src/auth.rs new file mode 100644 index 0000000..3fad639 --- /dev/null +++ b/src/auth.rs @@ -0,0 +1,220 @@ +use axum::{ + extract::{Query, State}, + response::{IntoResponse, Redirect}, +}; +use axum_extra::extract::cookie::{Cookie, CookieJar, SameSite}; +use rand::{ + distr::{Alphanumeric, SampleString}, + rng, +}; +use serde::Deserialize; + +use crate::{ + AppState, + db, + intervals::IntervalsClient, +}; + +#[derive(Debug, Deserialize)] +pub struct OAuthCallback { + pub code: Option, + pub state: Option, + pub error: Option, +} + +pub async fn start( + State(state): State, +) -> impl IntoResponse { + let mut rng = rng(); + + 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"); + 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 + { + tracing::error!(%error, "cannot create OAuth state"); + return "internal error".into_response(); + } + + let client = + IntervalsClient::new(state.config.clone()); + + Redirect::to( + &client.oauth_authorize_url(&oauth_state) + ) + .into_response() +} + +pub async fn callback( + State(state): State, + Query(query): Query, +) -> impl IntoResponse { + if query.error.is_some() { + return "Intervals.icu authorization denied".into_response(); + } + + let Some(code) = query.code else { + return "missing OAuth code".into_response(); + }; + + let Some(oauth_state) = query.state else { + return "missing OAuth state".into_response(); + }; + + let state_row = sqlx::query( + r#" + DELETE FROM oauth_states + WHERE + state = $1 + AND created_at > now() - interval '10 minutes' + RETURNING state + "#, + ) + .bind(oauth_state) + .fetch_optional(&state.db) + .await; + + match state_row { + Ok(Some(_)) => {} + Ok(None) => { + return "invalid or expired OAuth state".into_response(); + } + Err(error) => { + tracing::error!(%error, "OAuth state validation failed"); + return "internal error".into_response(); + } + } + + 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 (access_token, scope, athlete_id, display_name) = + match IntervalsClient::token_data(&response) { + Ok(value) => value, + Err(error) => { + tracing::error!(%error, "invalid OAuth token response"); + return "invalid OAuth response".into_response(); + } + }; + + let row = sqlx::query( + r#" + INSERT INTO athletes ( + intervals_athlete_id, + display_name, + access_token, + 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 + "#, + ) + .bind(&athlete_id) + .bind(&display_name) + .bind(&access_token) + .bind(&scope) + .fetch_one(&state.db) + .await; + + use sqlx::Row; + + 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"); + return "internal error".into_response(); + } + }, + Err(error) => { + tracing::error!(%error, "cannot store athlete"); + return "cannot store athlete".into_response(); + } + }; + + let mut rng = rng(); + + 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"); + 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 jar = CookieJar::new().add(cookie); + + ( + jar, + Redirect::to("/"), + ).into_response() +} + +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 jar = jar.remove( + Cookie::build("county_session") + .path("/") + ); + + ( + jar, + Redirect::to("/"), + ) +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..9ffce2f --- /dev/null +++ b/src/config.rs @@ -0,0 +1,41 @@ +use anyhow::Result; + +#[derive(Clone)] +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 cookie_secure: bool, +} + +impl Config { + pub fn from_env() -> Result { + dotenvy::dotenv().ok(); + + Ok(Self { + database_url: std::env::var("DATABASE_URL")?, + + bind_address: std::env::var("BIND_ADDRESS") + .unwrap_or_else(|_| "127.0.0.1:8080".into()), + + intervals_base_url: std::env::var("INTERVALS_BASE_URL") + .unwrap_or_else(|_| "https://intervals.icu".into()), + + 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), + }) + } +} diff --git a/src/db.rs b/src/db.rs new file mode 100644 index 0000000..9748e5a --- /dev/null +++ b/src/db.rs @@ -0,0 +1,210 @@ +use anyhow::Result; +use chrono::{DateTime, Utc}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use sqlx::{PgPool, Postgres, Transaction}; + +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<()> { + let hash = hash_token(token); + + sqlx::query( + r#" + INSERT INTO sessions ( + token_hash, + athlete_id + ) + VALUES ($1, $2) + "#, + ) + .bind(hash) + .bind(athlete_id) + .execute(db) + .await?; + + Ok(()) +} + +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 + 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; + + if let Some(row) = row { + let id: i64 = row.try_get("id")?; + let name: String = row.try_get("display_name")?; + + sqlx::query( + "UPDATE sessions SET last_seen_at = now() WHERE token_hash = $1", + ) + .bind(hash) + .execute(db) + .await?; + + Ok(Some((id, name))) + } else { + Ok(None) + } +} + +pub async fn delete_old_oauth_states( + db: &PgPool, +) -> Result<()> { + sqlx::query( + "DELETE FROM oauth_states WHERE created_at < now() - interval '10 minutes'", + ) + .execute(db) + .await?; + + Ok(()) +} + +pub async fn ensure_activity( + tx: &mut Transaction<'_, Postgres>, + athlete_id: i64, + intervals_activity_id: &str, + start_time: DateTime, + activity_json: &Value, +) -> Result { + let row = sqlx::query( + r#" + INSERT INTO activities ( + athlete_id, + intervals_activity_id, + start_time, + activity_json, + processed_at + ) + 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, + processed_at = NULL + + RETURNING id + "#, + ) + .bind(athlete_id) + .bind(intervals_activity_id) + .bind(start_time) + .bind(activity_json) + .fetch_one(&mut **tx) + .await?; + + use sqlx::Row; + + Ok(row.try_get("id")?) +} + +pub async fn delete_activity_crossings( + tx: &mut Transaction<'_, Postgres>, + activity_id: i64, +) -> Result<()> { + sqlx::query( + "DELETE FROM county_crossings WHERE activity_id = $1", + ) + .bind(activity_id) + .execute(&mut **tx) + .await?; + + Ok(()) +} + +pub async fn save_crossing( + tx: &mut Transaction<'_, Postgres>, + activity_id: i64, + crossing: &DetectedCrossing, + interval: Option<&Value>, +) -> Result<()> { + sqlx::query( + r#" + INSERT INTO county_crossings ( + activity_id, + track_index, + crossing_time, + location, + from_county_id, + to_county_id, + intervals_interval + ) + VALUES ( + $1, + $2, + $3, + + ST_SetSRID( + ST_MakePoint($4, $5), + 4326 + ), + + $6, + $7, + $8 + ) + "#, + ) + .bind(activity_id) + .bind(crossing.track_index as i64) + .bind(crossing.crossing_time) + .bind(crossing.lon) + .bind(crossing.lat) + .bind(crossing.from_county_id) + .bind(crossing.to_county_id) + .bind(interval) + .execute(&mut **tx) + .await?; + + Ok(()) +} + +pub async fn mark_activity_processed( + tx: &mut Transaction<'_, Postgres>, + activity_id: i64, +) -> Result<()> { + sqlx::query( + r#" + UPDATE activities + SET processed_at = now() + WHERE id = $1 + "#, + ) + .bind(activity_id) + .execute(&mut **tx) + .await?; + + Ok(()) +} diff --git a/src/geo.rs b/src/geo.rs new file mode 100644 index 0000000..3cb51ea --- /dev/null +++ b/src/geo.rs @@ -0,0 +1,167 @@ +use anyhow::Result; +use chrono::{DateTime, Duration, Utc}; +use sqlx::PgPool; + +use crate::model::{DetectedCrossing, TrackPoint}; + +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(); + + a + Duration::milliseconds( + (duration_ms as f64 * fraction).round() as i64 + ) +} + +/// Detects all county crossings between two consecutive GPS samples. +/// +/// The geometry itself is handled by PostGIS. Because the endpoint counties +/// are used to constrain the query, the GiST geometry index remains useful. +/// ST_LineLocatePoint gives the position of the intersection on the original +/// track segment; that fraction is then used for millisecond timestamp +/// interpolation. +pub async fn crossings_between( + db: &PgPool, + a: &TrackPoint, + b: &TrackPoint, +) -> Result> { + let rows = sqlx::query( + r#" + WITH points AS ( + SELECT + ST_SetSRID( + ST_MakePoint($1, $2), + 4326 + ) AS p1, + + ST_SetSRID( + ST_MakePoint($3, $4), + 4326 + ) AS p2 + ), + + track AS ( + SELECT + p1, + p2, + ST_MakeLine(p1, p2) AS line + FROM points + ), + + endpoint_counties AS ( + SELECT + a.id AS from_id, + b.id AS to_id, + t.line + FROM county_boundaries a + CROSS JOIN county_boundaries b + CROSS JOIN track t + WHERE + a.id <> b.id + AND ST_Covers(a.geometry, t.p1) + AND ST_Covers(b.geometry, t.p2) + AND ST_Intersects(a.geometry, t.line) + AND ST_Intersects(b.geometry, t.line) + ), + + intersections AS ( + SELECT + e.from_id, + e.to_id, + e.line, + dumped.geom AS intersection + FROM endpoint_counties e + CROSS JOIN LATERAL + ST_Dump( + ST_Intersection( + ST_Boundary( + ( + SELECT geometry + FROM county_boundaries + WHERE id = e.from_id + ) + ), + e.line + ) + ) AS dumped + ) + + SELECT + from_id, + to_id, + + ST_X(intersection) AS lon, + ST_Y(intersection) AS lat, + + ST_LineLocatePoint( + line, + intersection + ) AS fraction + + FROM intersections + + WHERE + ST_Dimension(intersection) = 0 + + ORDER BY fraction + "#, + ) + .bind(a.lon) + .bind(a.lat) + .bind(b.lon) + .bind(b.lat) + .fetch_all(db) + .await?; + + use sqlx::Row; + + let mut result = Vec::new(); + + for row in rows { + let fraction: f64 = row.try_get("fraction")?; + + // A sample exactly on a boundary is handled by the following + // segment. We only create crossings strictly inside a segment. + if !(fraction > 0.0 && fraction < 1.0) { + continue; + } + + let from_county_id: i64 = + row.try_get("from_id")?; + + let to_county_id: i64 = + row.try_get("to_id")?; + + let lon: f64 = + row.try_get("lon")?; + + let lat: f64 = + row.try_get("lat")?; + + result.push(DetectedCrossing { + track_index: a.index, + + crossing_time: interpolate_time( + a.time, + b.time, + fraction, + ), + + lat, + lon, + + from_county_id, + to_county_id, + + fraction, + }); + } + + Ok(result) +} diff --git a/src/intervals.rs b/src/intervals.rs new file mode 100644 index 0000000..141c5f5 --- /dev/null +++ b/src/intervals.rs @@ -0,0 +1,292 @@ +use anyhow::{Context, Result, anyhow}; +use chrono::{DateTime, Utc}; +use reqwest::Client; +use serde_json::{Value, json}; + +use crate::config::Config; + +#[derive(Clone)] +pub struct IntervalsClient { + client: Client, + config: Config, +} + +impl IntervalsClient { + pub fn new(config: Config) -> Self { + Self { + client: Client::new(), + config, + } + } + + pub fn oauth_authorize_url(&self, state: &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), + ) + } + + pub async fn exchange_code(&self, code: &str) -> Result { + let url = format!( + "{}/api/oauth/token", + self.config.intervals_base_url + ); + + 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?) + } + + 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> { + 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(); + } + } + } + + 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()); + } + + if let Some(data) = value.as_array() { + return Some(data.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()?); + } + + 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")) + } + + 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 + .get("access_token") + .and_then(Value::as_str) + .ok_or_else(|| anyhow!("OAuth response has no access_token"))?; + + let scope = response + .get("scope") + .and_then(Value::as_str) + .unwrap_or("") + .to_owned(); + + let athlete = response + .get("athlete") + .ok_or_else(|| anyhow!("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"))?; + + let athlete_name = athlete + .get("name") + .and_then(Value::as_str) + .unwrap_or(athlete_id); + + Ok(( + token.to_owned(), + scope, + athlete_id.to_owned(), + athlete_name.to_owned(), + )) + } + + pub fn _json_example() -> Value { + json!({ + "scope": "ACTIVITY:READ" + }) + } +} diff --git a/src/leaderboard.rs b/src/leaderboard.rs new file mode 100644 index 0000000..9183579 --- /dev/null +++ b/src/leaderboard.rs @@ -0,0 +1,206 @@ +use anyhow::Result; +use axum::{ + extract::State, + Json, +}; +use chrono::{DateTime, Utc}; +use serde_json::Value; +use sqlx::PgPool; + +use crate::{ + AppState, + model::{LeaderboardGroup, LeaderboardRow}, +}; + +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"); + + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + error.to_string(), + ) + }) +} + +pub async fn build( + db: &PgPool, +) -> Result> { + let rows = sqlx::query( + r#" + WITH bucketed AS ( + SELECT + cc.id, + cc.crossing_time, + cc.track_index, + cc.intervals_interval, + + a.intervals_activity_id, + ath.display_name, + + f.name AS from_county, + t.name AS to_county, + + to_timestamp( + floor( + extract(epoch FROM cc.crossing_time) + / 600.0 + ) * 600 + ) AS bucket_start + + FROM county_crossings cc + + JOIN activities a + ON a.id = cc.activity_id + + JOIN athletes ath + ON ath.id = a.athlete_id + + JOIN county_boundaries f + ON f.id = cc.from_county_id + + JOIN county_boundaries t + ON t.id = cc.to_county_id + ), + + ranked AS ( + SELECT + *, + row_number() OVER ( + PARTITION BY + bucket_start, + from_county, + to_county + ORDER BY crossing_time, id + ) AS rank + FROM bucketed + ) + + SELECT + rank, + display_name, + from_county, + to_county, + bucket_start, + crossing_time, + intervals_activity_id, + intervals_interval + + FROM ranked + + WHERE rank <= 20 + + ORDER BY + bucket_start DESC, + from_county, + to_county, + rank + "#, + ) + .fetch_all(db) + .await?; + + use sqlx::Row; + + let mut groups: Vec = Vec::new(); + + for row in rows { + let bucket_start: DateTime = + row.try_get("bucket_start")?; + + let from_county: String = + row.try_get("from_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 index = match group_index { + Some(index) => index, + + None => { + groups.push(LeaderboardGroup { + bucket_start, + from_county: from_county.clone(), + to_county: to_county.clone(), + rows: Vec::new(), + }); + + groups.len() - 1 + } + }; + + let rank: i64 = + row.try_get("rank")?; + + let athlete: String = + row.try_get("display_name")?; + + let crossing_time: DateTime = + row.try_get("crossing_time")?; + + let activity_id: String = + row.try_get("intervals_activity_id")?; + + 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 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 row_data = LeaderboardRow { + rank, + athlete, + + from_county, + to_county, + + crossing_time, + + activity_id: activity_id.clone(), + + activity_url: format!( + "https://intervals.icu/activities/{}", + activity_id + ), + + interval_url: interval_id.map(|id| { + format!( + "https://intervals.icu/activities/{}?interval={}", + activity_id, + id + ) + }), + + average_watts, + average_heartrate, + }; + + groups[index].rows.push(row_data); + } + + Ok(groups) +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..cf605c9 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,381 @@ +mod auth; +mod config; +mod db; +mod geo; +mod intervals; +mod leaderboard; +mod model; +mod webhook; +mod web; + +use std::sync::Arc; + +use anyhow::{Context, Result, anyhow}; +use axum::{ + Router, + routing::{get, post}, +}; +use chrono::{DateTime, Duration, Utc}; +use serde_json::Value; +use sqlx::PgPool; +use tower_http::trace::TraceLayer; +use tracing_subscriber::{ + layer::SubscriberExt, + util::SubscriberInitExt, +}; + +use crate::{ + config::Config, + intervals::IntervalsClient, + model::TrackPoint, +}; + +#[derive(Clone)] +pub struct AppState { + pub db: PgPool, + pub config: Config, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::registry() + .with( + tracing_subscriber::EnvFilter::from_default_env() + ) + .with(tracing_subscriber::fmt::layer()) + .init(); + + let config = + Config::from_env()?; + + let db = + PgPool::connect(&config.database_url) + .await + .context("cannot connect to PostgreSQL")?; + + sqlx::migrate!() + .run(&db) + .await + .context("database migration failed")?; + + let state = AppState { + db, + config: config.clone(), + }; + + 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), + ) + .layer(TraceLayer::new_for_http()) + .with_state(state); + + let listener = + tokio::net::TcpListener::bind( + &config.bind_address + ) + .await?; + + tracing::info!( + address = %config.bind_address, + "server listening" + ); + + axum::serve(listener, app).await?; + + Ok(()) +} + +async fn health() -> &'static str { + "ok" +} + +pub async fn process_activity( + state: AppState, + intervals_athlete_id: String, + activity_id: String, +) -> Result<()> { + let athlete = sqlx::query( + r#" + SELECT + id, + access_token + FROM athletes + WHERE intervals_athlete_id = $1 + "#, + ) + .bind(&intervals_athlete_id) + .fetch_optional(&state.db) + .await?; + + use sqlx::Row; + + let Some(athlete) = athlete else { + return Err(anyhow!( + "no OAuth connection for athlete {}", + intervals_athlete_id + )); + }; + + let athlete_db_id: i64 = + athlete.try_get("id")?; + + let access_token: String = + athlete.try_get("access_token")?; + + let client = + IntervalsClient::new(state.config.clone()); + + // + // First fetch the activity metadata including icu_intervals. + // + let activity = + client.activity( + &access_token, + &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, + )?; + + if points.len() < 2 { + tracing::info!( + activity_id = %activity_id, + "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?; + + let activity_db_id = + db::ensure_activity( + &mut tx, + athlete_db_id, + &activity_id, + start_time, + &activity, + ) + .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?; + + for crossing in crossings { + let interval = + IntervalsClient::matching_interval( + &activity, + crossing.track_index, + ); + + db::save_crossing( + &mut tx, + activity_db_id, + &crossing, + interval.as_ref(), + ) + .await?; + + crossing_count += 1; + + tracing::info!( + activity_id = %activity_id, + time = %crossing.crossing_time, + from = crossing.from_county_id, + to = crossing.to_county_id, + track_index = crossing.track_index, + "county crossing detected" + ); + } + } + + db::mark_activity_processed( + &mut tx, + activity_db_id, + ) + .await?; + + tx.commit().await?; + + tracing::info!( + activity_id = %activity_id, + crossings = crossing_count, + "activity processed" + ); + + 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" + ))?; + + 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 mut result = + Vec::with_capacity(length); + + for index in 0..length { + let Some(seconds) = + times[index].as_f64() + else { + continue; + }; + + let Some(coordinate) = + latlng[index].as_array() + else { + continue; + }; + + if coordinate.len() != 2 { + 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() + { + continue; + } + + 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 chrono::{TimeZone, Utc}; + + use crate::geo::interpolate_time; + + #[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/model.rs b/src/model.rs new file mode 100644 index 0000000..946bf68 --- /dev/null +++ b/src/model.rs @@ -0,0 +1,71 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +#[derive(Debug, Clone)] +pub struct TrackPoint { + pub index: usize, + pub time: DateTime, + pub lat: f64, + pub lon: f64, +} + +#[derive(Debug, Clone)] +pub struct DetectedCrossing { + pub track_index: usize, + pub crossing_time: DateTime, + pub lat: f64, + pub lon: f64, + pub from_county_id: i64, + pub to_county_id: i64, + pub fraction: f64, +} + +#[derive(Debug, Deserialize)] +pub struct WebhookEnvelope { + pub secret: String, + pub events: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct WebhookEvent { + pub athlete_id: String, + + #[serde(rename = "type")] + pub event_type: String, + + pub timestamp: DateTime, + + #[serde(default)] + pub activity: Option, +} + +#[derive(Debug, Serialize)] +pub struct LeaderboardRow { + pub rank: i64, + + pub athlete: String, + + pub from_county: String, + pub to_county: String, + + pub crossing_time: DateTime, + + pub activity_id: String, + pub activity_url: String, + + pub interval_url: Option, + + pub average_watts: Option, + pub average_heartrate: Option, +} + +#[derive(Debug, Serialize)] +pub struct LeaderboardGroup { + pub bucket_start: DateTime, + + pub from_county: String, + pub to_county: String, + + pub rows: Vec, +} diff --git a/src/web.rs b/src/web.rs new file mode 100644 index 0000000..b7566d8 --- /dev/null +++ b/src/web.rs @@ -0,0 +1,223 @@ +use axum::{ + extract::State, + response::Html, +}; +use axum_extra::extract::cookie::CookieJar; + +use crate::{ + AppState, + db, + leaderboard, +}; + +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 + .ok() + .flatten() + .map(|(_, name)| name), + + None => None, + }; + + let mut html = String::from( + r#" + + + + +Landkreis-Sprints + + + +
+
+

🚴 Landkreis-Sprints

+
+ Grenzübertritte · Millisekunden · Intervals.icu +
+
+"#, + ); + + if let Some(name) = current_user { + html.push_str(&format!( + "
Angemeldet als {} · \ + Logout
", + escape_html(&name) + )); + } else { + html.push_str( + r#" + Mit Intervals.icu verbinden + "# + ); + } + + html.push_str("
"); + + if groups.is_empty() { + html.push_str( + r#"
+ Noch keine Landkreisüberquerungen. +
"# + ); + } + + for group in groups { + html.push_str("
"); + + html.push_str(&format!( + "

{} → {}

", + escape_html(&group.from_county), + escape_html(&group.to_county) + )); + + html.push_str(&format!( + "
10-Minuten-Fenster ab {}
", + group.bucket_start.format("%Y-%m-%d %H:%M:%S UTC") + )); + + html.push_str( + r#" + + + + + + + + + + +"# + ); + + for row in group.rows { + let interval = + match row.interval_url { + Some(url) => + format!( + "Intervall", + escape_html(&url) + ), + + None => + "—".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()); + + html.push_str(&format!( + r#" + + + + + + +"#, + row.rank, + escape_html(&row.athlete), + row.crossing_time.format("%H:%M:%S%.3f"), + interval, + power, + hr + )); + } + + html.push_str("
#FahrerÃœbertrittIntervallPowerHR
{}{}{}{}{}{}
"); + + html.push_str("
"); + } + + html.push_str( + r#"
+ Datenquelle: Intervals.icu · Landkreisgeometrien: VG250 +
+ +"# + ); + + Html(html) +} + +fn escape_html(value: &str) -> String { + value + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) + .replace('\'', "'") +} diff --git a/src/webhook.rs b/src/webhook.rs new file mode 100644 index 0000000..ec10b98 --- /dev/null +++ b/src/webhook.rs @@ -0,0 +1,81 @@ +use axum::{ + extract::State, + http::StatusCode, + Json, +}; + +use crate::{ + AppState, + model::WebhookEnvelope, +}; + +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); + } + + 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) = activity_id else { + tracing::warn!( + athlete_id = %event.athlete_id, + "activity webhook without activity id" + ); + + continue; + }; + + 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" + ); + + // Return 500 so Intervals.icu can retry the webhook. + return Err(StatusCode::INTERNAL_SERVER_ERROR); + } + } + + "APP_SCOPE_CHANGED" => { + tracing::info!( + athlete_id = %event.athlete_id, + "OAuth scopes changed" + ); + } + + _ => { + tracing::debug!( + event_type = %event.event_type, + "ignoring Intervals.icu webhook" + ); + } + } + } + + // Intervals.icu has historically retried on 204; use an ordinary 200. + Ok("ok") +}