This commit is contained in:
Jonas Rabenstein 2026-08-12 13:24:01 +02:00
commit a9d22e8312
18 changed files with 2308 additions and 0 deletions

14
.env.example Normal file
View file

@ -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

4
.gitignore vendored Normal file
View file

@ -0,0 +1,4 @@
/target
/.pgdata
/.pgsocket
.env

44
Cargo.toml Normal file
View file

@ -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"

43
README.md Normal file
View file

@ -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

127
flake.nix Normal file
View file

@ -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" <<PGEOF
listen_addresses = ''
port = 5433
unix_socket_directories = '$PGHOST'
PGEOF
fi
if ! pg_ctl -D "$PGDATA" status >/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
'';
};
}
);
}

105
migrations/0001_initial.sql Normal file
View file

@ -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
);

6
rust-toolchain.toml Normal file
View file

@ -0,0 +1,6 @@
[toolchain]
channel = "stable"
components = [
"rustfmt",
"clippy"
]

73
scripts/import-vg250.sh Executable file
View file

@ -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;"

220
src/auth.rs Normal file
View file

@ -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<String>,
pub state: Option<String>,
pub error: Option<String>,
}
pub async fn start(
State(state): State<AppState>,
) -> 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<AppState>,
Query(query): Query<OAuthCallback>,
) -> 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<AppState>,
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("/"),
)
}

41
src/config.rs Normal file
View file

@ -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<Self> {
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),
})
}
}

210
src/db.rs Normal file
View file

@ -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<u8> {
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<Option<(i64, String)>> {
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<Utc>,
activity_json: &Value,
) -> Result<i64> {
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(())
}

167
src/geo.rs Normal file
View file

@ -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<Utc>,
b: DateTime<Utc>,
fraction: f64,
) -> DateTime<Utc> {
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<Vec<DetectedCrossing>> {
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)
}

292
src/intervals.rs Normal file
View file

@ -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<Value> {
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<Value> {
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<Value> {
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<Vec<Value>> {
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<DateTime<Utc>> {
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<Value> {
activity
.get("icu_intervals")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default()
}
pub fn matching_interval(
activity: &Value,
track_index: usize,
) -> Option<Value> {
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<f64>, Option<f64>, Option<i64>) {
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"
})
}
}

206
src/leaderboard.rs Normal file
View file

@ -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<AppState>,
) -> Result<Json<Vec<LeaderboardGroup>>, (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<Vec<LeaderboardGroup>> {
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<LeaderboardGroup> = Vec::new();
for row in rows {
let bucket_start: DateTime<Utc> =
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<Utc> =
row.try_get("crossing_time")?;
let activity_id: String =
row.try_get("intervals_activity_id")?;
let interval: Option<Value> =
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)
}

381
src/main.rs Normal file
View file

@ -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<Utc>,
) -> Result<Vec<TrackPoint>> {
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
);
}
}

71
src/model.rs Normal file
View file

@ -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<Utc>,
pub lat: f64,
pub lon: f64,
}
#[derive(Debug, Clone)]
pub struct DetectedCrossing {
pub track_index: usize,
pub crossing_time: DateTime<Utc>,
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<WebhookEvent>,
}
#[derive(Debug, Deserialize)]
pub struct WebhookEvent {
pub athlete_id: String,
#[serde(rename = "type")]
pub event_type: String,
pub timestamp: DateTime<Utc>,
#[serde(default)]
pub activity: Option<Value>,
}
#[derive(Debug, Serialize)]
pub struct LeaderboardRow {
pub rank: i64,
pub athlete: String,
pub from_county: String,
pub to_county: String,
pub crossing_time: DateTime<Utc>,
pub activity_id: String,
pub activity_url: String,
pub interval_url: Option<String>,
pub average_watts: Option<f64>,
pub average_heartrate: Option<f64>,
}
#[derive(Debug, Serialize)]
pub struct LeaderboardGroup {
pub bucket_start: DateTime<Utc>,
pub from_county: String,
pub to_county: String,
pub rows: Vec<LeaderboardRow>,
}

223
src/web.rs Normal file
View file

@ -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<AppState>,
jar: CookieJar,
) -> Html<String> {
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#"<!doctype html>
<html lang="de">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Landkreis-Sprints</title>
<style>
body {
font-family: system-ui, sans-serif;
max-width: 1100px;
margin: 0 auto;
padding: 1rem;
background: #111827;
color: #f9fafb;
}
a { color: #93c5fd; }
header {
display: flex;
justify-content: space-between;
align-items: center;
gap: 1rem;
}
.card {
background: #1f2937;
border-radius: 12px;
padding: 1rem;
margin: 1rem 0;
}
table {
width: 100%;
border-collapse: collapse;
}
th, td {
text-align: left;
padding: .55rem;
border-bottom: 1px solid #374151;
}
.rank {
font-weight: bold;
}
.time {
font-family: ui-monospace, monospace;
}
.small {
color: #9ca3af;
font-size: .9rem;
}
button, .button {
background: #2563eb;
color: white;
border: 0;
border-radius: 8px;
padding: .6rem .9rem;
text-decoration: none;
}
</style>
</head>
<body>
<header>
<div>
<h1>🚴 Landkreis-Sprints</h1>
<div class="small">
Grenzübertritte · Millisekunden · Intervals.icu
</div>
</div>
"#,
);
if let Some(name) = current_user {
html.push_str(&format!(
"<div>Angemeldet als <strong>{}</strong> · \
<a class=\"button\" href=\"/logout\">Logout</a></div>",
escape_html(&name)
));
} else {
html.push_str(
r#"<a class="button" href="/oauth/start">
Mit Intervals.icu verbinden
</a>"#
);
}
html.push_str("</header>");
if groups.is_empty() {
html.push_str(
r#"<div class="card">
Noch keine Landkreisüberquerungen.
</div>"#
);
}
for group in groups {
html.push_str("<div class=\"card\">");
html.push_str(&format!(
"<h2>{} → {}</h2>",
escape_html(&group.from_county),
escape_html(&group.to_county)
));
html.push_str(&format!(
"<div class=\"small\">10-Minuten-Fenster ab {}</div>",
group.bucket_start.format("%Y-%m-%d %H:%M:%S UTC")
));
html.push_str(
r#"<table>
<thead>
<tr>
<th>#</th>
<th>Fahrer</th>
<th>Übertritt</th>
<th>Intervall</th>
<th>Power</th>
<th>HR</th>
</tr>
</thead>
<tbody>"#
);
for row in group.rows {
let interval =
match row.interval_url {
Some(url) =>
format!(
"<a href=\"{}\" target=\"_blank\">Intervall</a>",
escape_html(&url)
),
None =>
"<span class=\"small\">—</span>".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#"<tr>
<td class="rank">{}</td>
<td>{}</td>
<td class="time">{}</td>
<td>{}</td>
<td>{}</td>
<td>{}</td>
</tr>"#,
row.rank,
escape_html(&row.athlete),
row.crossing_time.format("%H:%M:%S%.3f"),
interval,
power,
hr
));
}
html.push_str("</tbody></table>");
html.push_str("</div>");
}
html.push_str(
r#"<footer class="small">
Datenquelle: Intervals.icu · Landkreisgeometrien: VG250
</footer>
</body>
</html>"#
);
Html(html)
}
fn escape_html(value: &str) -> String {
value
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&#39;")
}

81
src/webhook.rs Normal file
View file

@ -0,0 +1,81 @@
use axum::{
extract::State,
http::StatusCode,
Json,
};
use crate::{
AppState,
model::WebhookEnvelope,
};
pub async fn receive(
State(state): State<AppState>,
Json(payload): Json<WebhookEnvelope>,
) -> 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")
}