nonapode/county-sprints
Watch
1
0
Fork
You've already forked county-sprints
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

dev api key

Browse source
This commit is contained in:
Jonas Rabenstein 2026-08-12 14:02:12 +02:00
commit 89091a03d3
11 changed files with 791 additions and 762 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.toml
View file

@ -13,7 +13,7 @@ rand = "0.9"
reqwest = {
version = "0.12",
default-features = false,
features = ["json", "form", "rustls-tls"]
features = ["json", "rustls-tls"]
}
serde = { version = "1", features = ["derive"] }
serde_json = "1"

27
flake.nix
View file

@ -80,6 +80,9 @@
mkdir -p "$PGHOST"
# Use an absolute path for the PostgreSQL Unix socket.
PGHOST_ABS="$(cd "$PGHOST" && pwd)"
# ============================================================
# Initialise PostgreSQL cluster
# ============================================================
@ -101,8 +104,6 @@
# Start PostgreSQL
#
# PostgreSQL is deliberately restricted to the Unix socket.
# No TCP/IP listener is opened, so there are no port conflicts
# with another local PostgreSQL instance.
# ============================================================
if ! pg_ctl \
@ -116,7 +117,7 @@
pg_ctl \
-D "$PGDATA" \
-l "$PGDATA/postgresql.log" \
-o "-c listen_addresses= -k $PGHOST" \
-o "-c listen_addresses= -k $PGHOST_ABS" \
start \
>/dev/null
fi
@ -129,7 +130,7 @@
for _ in $(seq 1 50); do
if pg_isready \
-h "$PGHOST" \
-h "$PGHOST_ABS" \
-U "$PGUSER" \
>/dev/null 2>&1
then
@ -164,7 +165,7 @@
# ============================================================
if ! psql \
-h "$PGHOST" \
-h "$PGHOST_ABS" \
-U "$PGUSER" \
-d postgres \
-tAc \
@ -177,7 +178,7 @@
echo "==> Creating PostgreSQL database '$PGDATABASE'"
createdb \
-h "$PGHOST" \
-h "$PGHOST_ABS" \
-U "$PGUSER" \
"$PGDATABASE"
fi
@ -190,7 +191,7 @@
echo "==> Checking PostGIS"
psql \
-h "$PGHOST" \
-h "$PGHOST_ABS" \
-U "$PGUSER" \
-d "$PGDATABASE" \
-v ON_ERROR_STOP=1 \
@ -198,10 +199,13 @@
>/dev/null
# ============================================================
# Application connection string
# SQLx / application connection string
#
# localhost is only the syntactic host. The host query
# parameter tells libpq/SQLx to use our Unix socket.
# ============================================================
export DATABASE_URL="postgresql://$PGUSER@/$PGDATABASE?host=$PGHOST"
export DATABASE_URL="postgresql://$PGUSER@localhost/$PGDATABASE?host=$PGHOST_ABS"
# ============================================================
# Development information
@ -217,14 +221,14 @@
echo
echo "PostgreSQL:"
echo " $(psql --version)"
echo " socket: $PGHOST"
echo " socket: $PGHOST_ABS"
echo " database: $PGDATABASE"
echo " user: $PGUSER"
echo
echo "PostGIS:"
psql \
-h "$PGHOST" \
-h "$PGHOST_ABS" \
-U "$PGUSER" \
-d "$PGDATABASE" \
-tAc "SELECT PostGIS_Full_Version();" |
@ -242,6 +246,7 @@
echo
echo "============================================================"
echo
set +e
'';
};
}

179
src/auth.rs
View file

@ -9,11 +9,7 @@ use rand::{
};
use serde::Deserialize;
use crate::{
AppState,
db,
intervals::IntervalsClient,
};
use crate::{AppState, db, intervals::IntervalsClient};
#[derive(Debug, Deserialize)]
pub struct OAuthCallback {
@ -22,39 +18,50 @@ pub struct OAuthCallback {
pub error: Option<String>,
}
pub async fn start(
State(state): State<AppState>,
) -> impl IntoResponse {
let mut rng = rng();
pub async fn start(State(state): State<AppState>) -> impl IntoResponse {
// Important: ThreadRng must not live across an await.
let oauth_state = {
let mut rng = rng();
Alphanumeric.sample_string(&mut rng, 48)
};
let oauth_state =
Alphanumeric.sample_string(&mut rng, 48);
if let Err(error) = db::delete_old_oauth_states(&state.db).await {
tracing::error!(
%error,
"cannot clean OAuth states"
);
if let Err(error) =
db::delete_old_oauth_states(&state.db).await
{
tracing::error!(%error, "cannot clean OAuth states");
return "internal error".into_response();
}
if let Err(error) = sqlx::query(
"INSERT INTO oauth_states(state) VALUES ($1)",
)
.bind(&oauth_state)
.execute(&state.db)
.await
if let Err(error) = sqlx::query("INSERT INTO oauth_states(state) VALUES ($1)")
.bind(&oauth_state)
.execute(&state.db)
.await
{
tracing::error!(%error, "cannot create OAuth state");
tracing::error!(
%error,
"cannot create OAuth state"
);
return "internal error".into_response();
}
let client =
IntervalsClient::new(state.config.clone());
let client = IntervalsClient::new(state.config.clone());
Redirect::to(
&client.oauth_authorize_url(&oauth_state)
)
.into_response()
let authorize_url = match client.oauth_authorize_url(&oauth_state) {
Ok(url) => url,
Err(error) => {
tracing::error!(
%error,
"cannot construct OAuth authorize URL"
);
return "cannot construct OAuth URL".into_response();
}
};
Redirect::to(&authorize_url).into_response()
}
pub async fn callback(
@ -88,32 +95,46 @@ pub async fn callback(
match state_row {
Ok(Some(_)) => {}
Ok(None) => {
return "invalid or expired OAuth state".into_response();
}
Err(error) => {
tracing::error!(%error, "OAuth state validation failed");
tracing::error!(
%error,
"OAuth state validation failed"
);
return "internal error".into_response();
}
}
let client =
IntervalsClient::new(state.config.clone());
let client = IntervalsClient::new(state.config.clone());
let response =
match client.exchange_code(&code).await {
Ok(response) => response,
Err(error) => {
tracing::error!(%error, "OAuth token exchange failed");
return "OAuth token exchange failed".into_response();
}
};
let response = match client.exchange_code(&code).await {
Ok(response) => response,
Err(error) => {
tracing::error!(
%error,
"OAuth token exchange failed"
);
return "OAuth token exchange failed".into_response();
}
};
let (access_token, scope, athlete_id, display_name) =
match IntervalsClient::token_data(&response) {
Ok(value) => value,
Err(error) => {
tracing::error!(%error, "invalid OAuth token response");
tracing::error!(
%error,
"invalid OAuth token response"
);
return "invalid OAuth response".into_response();
}
};
@ -127,14 +148,12 @@ pub async fn callback(
scopes
)
VALUES ($1, $2, $3, $4)
ON CONFLICT (intervals_athlete_id)
DO UPDATE SET
display_name = EXCLUDED.display_name,
access_token = EXCLUDED.access_token,
scopes = EXCLUDED.scopes,
updated_at = now()
RETURNING id
"#,
)
@ -150,71 +169,63 @@ pub async fn callback(
let athlete_db_id: i64 = match row {
Ok(row) => match row.try_get("id") {
Ok(id) => id,
Err(error) => {
tracing::error!(%error, "invalid athlete row");
tracing::error!(
%error,
"invalid athlete row"
);
return "internal error".into_response();
}
},
Err(error) => {
tracing::error!(%error, "cannot store athlete");
tracing::error!(
%error,
"cannot store athlete"
);
return "cannot store athlete".into_response();
}
};
let mut rng = rng();
let session_token = {
let mut rng = rng();
Alphanumeric.sample_string(&mut rng, 64)
};
let session_token =
Alphanumeric.sample_string(&mut rng, 64);
if let Err(error) = db::create_session(&state.db, athlete_db_id, &session_token).await {
tracing::error!(
%error,
"cannot create session"
);
if let Err(error) =
db::create_session(
&state.db,
athlete_db_id,
&session_token,
).await
{
tracing::error!(%error, "cannot create session");
return "cannot create session".into_response();
}
let cookie = Cookie::build(
("county_session", session_token)
)
.path("/")
.http_only(true)
.same_site(SameSite::Lax)
.secure(state.config.cookie_secure);
let cookie = Cookie::build(("county_session", session_token))
.path("/")
.http_only(true)
.same_site(SameSite::Lax)
.secure(state.config.cookie_secure);
let jar = CookieJar::new().add(cookie);
(
jar,
Redirect::to("/"),
).into_response()
(jar, Redirect::to("/")).into_response()
}
pub async fn logout(
State(state): State<AppState>,
jar: CookieJar,
) -> impl IntoResponse {
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 _ = sqlx::query("DELETE FROM sessions WHERE token_hash = $1")
.bind(&hash)
.execute(&state.db)
.await;
}
let jar = jar.remove(
Cookie::build("county_session")
.path("/")
);
let jar = jar.remove(Cookie::build("county_session").path("/"));
(
jar,
Redirect::to("/"),
)
(jar, Redirect::to("/")).into_response()
}

85
src/config.rs
View file

@ -1,15 +1,19 @@
use anyhow::Result;
use anyhow::{Context, Result};
use std::env;
#[derive(Clone)]
#[derive(Clone, Debug)]
pub struct Config {
pub database_url: String,
pub bind_address: String,
pub intervals_base_url: String,
pub intervals_client_id: String,
pub intervals_client_secret: String,
pub intervals_redirect_uri: String,
pub intervals_webhook_secret: String,
pub intervals_api_key: Option<String>,
pub intervals_athlete_id: String,
pub intervals_client_id: Option<String>,
pub intervals_client_secret: Option<String>,
pub intervals_redirect_uri: Option<String>,
pub intervals_webhook_secret: Option<String>,
pub cookie_secure: bool,
}
@ -18,24 +22,65 @@ impl Config {
pub fn from_env() -> Result<Self> {
dotenvy::dotenv().ok();
let database_url = required("DATABASE_URL")?;
let bind_address =
env::var("BIND_ADDRESS").unwrap_or_else(|_| "127.0.0.1:8080".to_string());
let intervals_base_url =
env::var("INTERVALS_BASE_URL").unwrap_or_else(|_| "https://intervals.icu".to_string());
let intervals_api_key = optional("INTERVALS_API_KEY");
let intervals_athlete_id =
env::var("INTERVALS_ATHLETE_ID").unwrap_or_else(|_| "0".to_string());
let intervals_client_id = optional("INTERVALS_CLIENT_ID");
let intervals_client_secret = optional("INTERVALS_CLIENT_SECRET");
let intervals_redirect_uri = optional("INTERVALS_REDIRECT_URI");
let intervals_webhook_secret = optional("INTERVALS_WEBHOOK_SECRET");
let cookie_secure = env::var("COOKIE_SECURE")
.unwrap_or_else(|_| "false".to_string())
.parse::<bool>()
.context("COOKIE_SECURE must be true or false")?;
Ok(Self {
database_url: std::env::var("DATABASE_URL")?,
database_url,
bind_address,
bind_address: std::env::var("BIND_ADDRESS")
.unwrap_or_else(|_| "127.0.0.1:8080".into()),
intervals_base_url,
intervals_api_key,
intervals_athlete_id,
intervals_base_url: std::env::var("INTERVALS_BASE_URL")
.unwrap_or_else(|_| "https://intervals.icu".into()),
intervals_client_id,
intervals_client_secret,
intervals_redirect_uri,
intervals_webhook_secret,
intervals_client_id: std::env::var("INTERVALS_CLIENT_ID")?,
intervals_client_secret: std::env::var("INTERVALS_CLIENT_SECRET")?,
intervals_redirect_uri: std::env::var("INTERVALS_REDIRECT_URI")?,
intervals_webhook_secret: std::env::var("INTERVALS_WEBHOOK_SECRET")?,
cookie_secure: std::env::var("COOKIE_SECURE")
.unwrap_or_else(|_| "true".into())
.parse()
.unwrap_or(true),
cookie_secure,
})
}
pub fn require_api_key(&self) -> Result<&str> {
self.intervals_api_key
.as_deref()
.context("INTERVALS_API_KEY is not configured")
}
}
fn required(name: &str) -> Result<String> {
let value = env::var(name).with_context(|| format!("{name} is not configured"))?;
if value.trim().is_empty() {
anyhow::bail!("{name} is empty");
}
Ok(value)
}
fn optional(name: &str) -> Option<String> {
env::var(name).ok().filter(|value| !value.trim().is_empty())
}

41
src/db.rs
View file

@ -12,11 +12,7 @@ pub fn hash_token(token: &str) -> Vec<u8> {
hasher.finalize().to_vec()
}
pub async fn create_session(
db: &PgPool,
athlete_id: i64,
token: &str,
) -> Result<()> {
pub async fn create_session(db: &PgPool, athlete_id: i64, token: &str) -> Result<()> {
let hash = hash_token(token);
sqlx::query(
@ -28,7 +24,7 @@ pub async fn create_session(
VALUES ($1, $2)
"#,
)
.bind(hash)
.bind(&hash)
.bind(athlete_id)
.execute(db)
.await?;
@ -36,10 +32,7 @@ pub async fn create_session(
Ok(())
}
pub async fn athlete_for_session(
db: &PgPool,
token: &str,
) -> Result<Option<(i64, String)>> {
pub async fn athlete_for_session(db: &PgPool, token: &str) -> Result<Option<(i64, String)>> {
let hash = hash_token(token);
let row = sqlx::query(
@ -53,7 +46,7 @@ pub async fn athlete_for_session(
WHERE s.token_hash = $1
"#,
)
.bind(hash)
.bind(&hash)
.fetch_optional(db)
.await?;
@ -64,9 +57,13 @@ pub async fn athlete_for_session(
let name: String = row.try_get("display_name")?;
sqlx::query(
"UPDATE sessions SET last_seen_at = now() WHERE token_hash = $1",
r#"
UPDATE sessions
SET last_seen_at = now()
WHERE token_hash = $1
"#,
)
.bind(hash)
.bind(&hash)
.execute(db)
.await?;
@ -76,11 +73,12 @@ pub async fn athlete_for_session(
}
}
pub async fn delete_old_oauth_states(
db: &PgPool,
) -> Result<()> {
pub async fn delete_old_oauth_states(db: &PgPool) -> Result<()> {
sqlx::query(
"DELETE FROM oauth_states WHERE created_at < now() - interval '10 minutes'",
r#"
DELETE FROM oauth_states
WHERE created_at < now() - interval '10 minutes'
"#,
)
.execute(db)
.await?;
@ -105,7 +103,6 @@ pub async fn ensure_activity(
processed_at
)
VALUES ($1, $2, $3, $4, NULL)
ON CONFLICT (
athlete_id,
intervals_activity_id
@ -114,7 +111,6 @@ pub async fn ensure_activity(
start_time = EXCLUDED.start_time,
activity_json = EXCLUDED.activity_json,
processed_at = NULL
RETURNING id
"#,
)
@ -135,7 +131,10 @@ pub async fn delete_activity_crossings(
activity_id: i64,
) -> Result<()> {
sqlx::query(
"DELETE FROM county_crossings WHERE activity_id = $1",
r#"
DELETE FROM county_crossings
WHERE activity_id = $1
"#,
)
.bind(activity_id)
.execute(&mut **tx)
@ -165,12 +164,10 @@ pub async fn save_crossing(
$1,
$2,
$3,
ST_SetSRID(
ST_MakePoint($4, $5),
4326
),
$6,
$7,
$8

31
src/geo.rs
View file

@ -4,19 +4,12 @@ use sqlx::PgPool;
use crate::model::{DetectedCrossing, TrackPoint};
pub fn interpolate_time(
a: DateTime<Utc>,
b: DateTime<Utc>,
fraction: f64,
) -> DateTime<Utc> {
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();
let duration_ms = (b - a).num_milliseconds();
a + Duration::milliseconds(
(duration_ms as f64 * fraction).round() as i64
)
a + Duration::milliseconds((duration_ms as f64 * fraction).round() as i64)
}
/// Detects all county crossings between two consecutive GPS samples.
@ -132,26 +125,18 @@ pub async fn crossings_between(
continue;
}
let from_county_id: i64 =
row.try_get("from_id")?;
let from_county_id: i64 = row.try_get("from_id")?;
let to_county_id: i64 =
row.try_get("to_id")?;
let to_county_id: i64 = row.try_get("to_id")?;
let lon: f64 =
row.try_get("lon")?;
let lon: f64 = row.try_get("lon")?;
let lat: f64 =
row.try_get("lat")?;
let lat: f64 = row.try_get("lat")?;
result.push(DetectedCrossing {
track_index: a.index,
crossing_time: interpolate_time(
a.time,
b.time,
fraction,
),
crossing_time: interpolate_time(a.time, b.time, fraction),
lat,
lon,

448
src/intervals.rs
View file

@ -1,6 +1,6 @@
use anyhow::{Context, Result, anyhow};
use chrono::{DateTime, Utc};
use reqwest::Client;
use chrono::{DateTime, NaiveDateTime, Utc};
use reqwest::{Client, StatusCode};
use serde_json::{Value, json};
use crate::config::Config;
@ -19,274 +19,266 @@ impl IntervalsClient {
}
}
pub fn oauth_authorize_url(&self, state: &str) -> String {
fn base_url(&self) -> &str {
self.config.intervals_base_url.trim_end_matches('/')
}
fn api_key(&self) -> Result<&str> {
self.config.require_api_key()
}
fn api_url(&self, path: &str) -> String {
format!(
"{}/oauth/authorize?client_id={}&redirect_uri={}&scope={}&state={}",
self.config.intervals_base_url,
urlencoding::encode(&self.config.intervals_client_id),
urlencoding::encode(&self.config.intervals_redirect_uri),
urlencoding::encode("ACTIVITY:READ"),
urlencoding::encode(state),
"{}/api/v1/{}",
self.base_url(),
path.trim_start_matches('/')
)
}
pub async fn exchange_code(&self, code: &str) -> Result<Value> {
fn authenticated(&self, request: reqwest::RequestBuilder) -> Result<reqwest::RequestBuilder> {
Ok(request.basic_auth("API_KEY", Some(self.api_key()?)))
}
async fn get_json(&self, url: String) -> Result<Value> {
let response = self
.authenticated(self.client.get(url))?
.send()
.await
.context("Intervals.icu request failed")?;
let status = response.status();
let body = response
.text()
.await
.context("cannot read Intervals.icu response")?;
if !status.is_success() {
return Err(anyhow!("Intervals.icu returned HTTP {}: {}", status, body));
}
serde_json::from_str(&body)
.with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body))
}
/// Get a single activity.
///
/// Uses:
/// GET /api/v1/activity/{id}
pub async fn activity(&self, activity_id: &str) -> Result<Value> {
let url = self.api_url(&format!("activity/{activity_id}"));
self.get_json(url).await
}
/// Get activity streams.
///
/// Uses:
/// GET /api/v1/activity/{id}/streams.json
///
/// The response is normally an array of stream objects:
///
/// [
/// {"type":"time","data":[...]},
/// {"type":"latlng","data":[...]},
/// ...
/// ]
pub async fn streams(&self, activity_id: &str) -> Result<Value> {
let url = self.api_url(&format!("activity/{activity_id}/streams.json"));
self.get_json(url).await
}
/// Fetch the athlete's activities.
///
/// `0` means the athlete belonging to the API key.
pub async fn activities(&self, oldest: &str, newest: &str) -> Result<Value> {
let athlete_id = &self.config.intervals_athlete_id;
let url = format!(
"{}/api/oauth/token",
self.config.intervals_base_url
"{}?oldest={}&newest={}",
self.api_url(&format!("athlete/{athlete_id}/activities")),
urlencoding::encode(oldest),
urlencoding::encode(newest),
);
Ok(self
.client
.post(url)
.form(&[
("client_id", self.config.intervals_client_id.as_str()),
(
"client_secret",
self.config.intervals_client_secret.as_str(),
),
("code", code),
])
.send()
.await?
.error_for_status()?
.json()
.await?)
self.get_json(url).await
}
pub async fn activity(
&self,
access_token: &str,
activity_id: &str,
) -> Result<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>> {
/// Return a stream's values independent of whether the API
/// returned an object or an array of stream objects.
pub fn stream_values<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Vec<Value>> {
// Current API format: array of streams.
if let Some(array) = streams.as_array() {
for entry in array {
if entry.get("type")
.and_then(Value::as_str)
== Some(wanted)
{
return entry
.get("data")
.and_then(Value::as_array)
.cloned();
for stream in array {
if stream.get("type").and_then(Value::as_str) == Some(stream_type) {
return stream.get("data").and_then(Value::as_array);
}
}
}
if let Some(object) = streams.as_object() {
if let Some(value) = object.get(wanted) {
if let Some(data) = value.get("data").and_then(Value::as_array) {
return Some(data.clone());
}
// Also support a dictionary-like response:
//
// {
// "time": [...],
// "latlng": [...]
// }
streams.get(stream_type).and_then(Value::as_array)
}
if let Some(data) = value.as_array() {
return Some(data.clone());
}
/// Extract the activity start timestamp.
pub fn activity_start(activity: &Value) -> Result<DateTime<Utc>> {
let value = activity
.get("start_date")
.or_else(|| activity.get("start_date_local"))
.or_else(|| activity.get("start_time"))
.ok_or_else(|| anyhow!("activity has no start timestamp"))?;
let string = value
.as_str()
.ok_or_else(|| anyhow!("activity start timestamp is not a string"))?;
parse_datetime(string)
}
/// Try to find an Intervals.icu interval corresponding
/// to a GPS stream index.
///
/// The exact interval representation has changed over
/// time, so this deliberately supports the common forms.
pub fn matching_interval(activity: &Value, track_index: usize) -> Option<Value> {
let intervals = activity.get("icu_intervals")?;
let array = intervals.as_array()?;
for interval in array {
let start = interval
.get("start_index")
.or_else(|| interval.get("start"));
let end = interval.get("end_index").or_else(|| interval.get("end"));
let start = start.and_then(Value::as_u64)?;
let end = end.and_then(Value::as_u64)?;
if (start as usize) <= track_index && track_index <= end as usize {
return Some(interval.clone());
}
}
None
}
pub fn activity_start(activity: &Value) -> Result<DateTime<Utc>> {
if let Some(value) = activity
.get("start_date")
.and_then(Value::as_str)
{
return Ok(value.parse()?);
/// OAuth methods are intentionally retained so we can
/// re-enable multi-user OAuth later.
pub fn oauth_authorize_url(&self, state: &str) -> Result<String> {
let client_id = self
.config
.intervals_client_id
.as_deref()
.context("INTERVALS_CLIENT_ID is not configured")?;
let redirect_uri = self
.config
.intervals_redirect_uri
.as_deref()
.context("INTERVALS_REDIRECT_URI is not configured")?;
Ok(format!(
"{}/oauth/authorize?client_id={}&redirect_uri={}&scope={}&state={}",
self.base_url(),
urlencoding::encode(client_id),
urlencoding::encode(redirect_uri),
urlencoding::encode("ACTIVITY:READ"),
urlencoding::encode(state),
))
}
pub async fn exchange_code(&self, code: &str) -> Result<Value> {
let client_id = self
.config
.intervals_client_id
.as_deref()
.context("INTERVALS_CLIENT_ID is not configured")?;
let client_secret = self
.config
.intervals_client_secret
.as_deref()
.context("INTERVALS_CLIENT_SECRET is not configured")?;
let response = self
.client
.post(format!("{}/api/oauth/token", self.base_url()))
.form(&[
("client_id", client_id),
("client_secret", client_secret),
("code", code),
])
.send()
.await
.context("OAuth token request failed")?;
let status = response.status();
let body = response.text().await?;
if !status.is_success() {
return Err(anyhow!(
"OAuth token exchange returned HTTP {}: {}",
status,
body
));
}
if let Some(value) = activity
.get("start_date_local")
.and_then(Value::as_str)
{
let naive = chrono::NaiveDateTime::parse_from_str(
value,
"%Y-%m-%dT%H:%M:%S",
)?;
return Ok(
naive
.and_utc()
);
}
Err(anyhow!("activity contains no usable start_date"))
Ok(serde_json::from_str(&body)?)
}
pub fn icu_intervals(activity: &Value) -> Vec<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
pub fn token_data(response: &Value) -> Result<(String, String, String, String)> {
let access_token = response
.get("access_token")
.and_then(Value::as_str)
.ok_or_else(|| anyhow!("OAuth response has no access_token"))?;
.context("OAuth response has no access_token")?
.to_string();
let scope = response
.get("scope")
.and_then(Value::as_str)
.unwrap_or("")
.to_owned();
.to_string();
let athlete = response
.get("athlete")
.ok_or_else(|| anyhow!("OAuth response has no athlete"))?;
.context("OAuth response has no athlete")?;
let athlete_id = athlete
.get("id")
.and_then(Value::as_str)
.ok_or_else(|| anyhow!("OAuth response has no athlete.id"))?;
.or_else(|| athlete.get("id").and_then(Value::as_i64).map(|_| ""))
.unwrap_or("")
.to_string();
let athlete_name = athlete
let display_name = athlete
.get("name")
.or_else(|| athlete.get("display_name"))
.and_then(Value::as_str)
.unwrap_or(athlete_id);
.unwrap_or("Intervals.icu athlete")
.to_string();
Ok((
token.to_owned(),
scope,
athlete_id.to_owned(),
athlete_name.to_owned(),
))
}
pub fn _json_example() -> Value {
json!({
"scope": "ACTIVITY:READ"
})
Ok((access_token, scope, athlete_id, display_name))
}
}
fn parse_datetime(value: &str) -> Result<DateTime<Utc>> {
if let Ok(value) = DateTime::parse_from_rfc3339(value) {
return Ok(value.with_timezone(&Utc));
}
if let Ok(value) = DateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S%z") {
return Ok(value.with_timezone(&Utc));
}
if let Ok(value) = NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S") {
return Ok(DateTime::<Utc>::from_naive_utc_and_offset(value, Utc));
}
Err(anyhow!("cannot parse activity timestamp: {}", value))
}

93
src/leaderboard.rs
View file

@ -1,8 +1,5 @@
use anyhow::Result;
use axum::{
extract::State,
Json,
};
use axum::{Json, extract::State};
use chrono::{DateTime, Utc};
use serde_json::Value;
use sqlx::PgPool;
@ -15,22 +12,17 @@ use crate::{
pub async fn api(
State(state): State<AppState>,
) -> Result<Json<Vec<LeaderboardGroup>>, (axum::http::StatusCode, String)> {
build(&state.db)
.await
.map(Json)
.map_err(|error| {
tracing::error!(%error, "leaderboard query failed");
build(&state.db).await.map(Json).map_err(|error| {
tracing::error!(%error, "leaderboard query failed");
(
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
error.to_string(),
)
})
(
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
error.to_string(),
)
})
}
pub async fn build(
db: &PgPool,
) -> Result<Vec<LeaderboardGroup>> {
pub async fn build(db: &PgPool) -> Result<Vec<LeaderboardGroup>> {
let rows = sqlx::query(
r#"
WITH bucketed AS (
@ -110,21 +102,17 @@ pub async fn build(
let mut groups: Vec<LeaderboardGroup> = Vec::new();
for row in rows {
let bucket_start: DateTime<Utc> =
row.try_get("bucket_start")?;
let bucket_start: DateTime<Utc> = row.try_get("bucket_start")?;
let from_county: String =
row.try_get("from_county")?;
let from_county: String = row.try_get("from_county")?;
let to_county: String =
row.try_get("to_county")?;
let to_county: String = row.try_get("to_county")?;
let group_index =
groups.iter().position(|group| {
group.bucket_start == bucket_start
&& group.from_county == from_county
&& group.to_county == to_county
});
let group_index = groups.iter().position(|group| {
group.bucket_start == bucket_start
&& group.from_county == from_county
&& group.to_county == to_county
});
let index = match group_index {
Some(index) => index,
@ -141,35 +129,30 @@ pub async fn build(
}
};
let rank: i64 =
row.try_get("rank")?;
let rank: i64 = row.try_get("rank")?;
let athlete: String =
row.try_get("display_name")?;
let athlete: String = row.try_get("display_name")?;
let crossing_time: DateTime<Utc> =
row.try_get("crossing_time")?;
let crossing_time: DateTime<Utc> = row.try_get("crossing_time")?;
let activity_id: String =
row.try_get("intervals_activity_id")?;
let activity_id: String = row.try_get("intervals_activity_id")?;
let interval: Option<Value> =
row.try_get("intervals_interval")?;
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 interval_id = interval
.as_ref()
.and_then(|v| v.get("id"))
.and_then(Value::as_i64);
let average_watts =
interval.as_ref()
.and_then(|v| v.get("average_watts"))
.and_then(Value::as_f64);
let average_watts = interval
.as_ref()
.and_then(|v| v.get("average_watts"))
.and_then(Value::as_f64);
let average_heartrate =
interval.as_ref()
.and_then(|v| v.get("average_heartrate"))
.and_then(Value::as_f64);
let average_heartrate = interval
.as_ref()
.and_then(|v| v.get("average_heartrate"))
.and_then(Value::as_f64);
let row_data = LeaderboardRow {
rank,
@ -182,16 +165,12 @@ pub async fn build(
activity_id: activity_id.clone(),
activity_url: format!(
"https://intervals.icu/activities/{}",
activity_id
),
activity_url: format!("https://intervals.icu/activities/{}", activity_id),
interval_url: interval_id.map(|id| {
format!(
"https://intervals.icu/activities/{}?interval={}",
activity_id,
id
activity_id, id
)
}),

428
src/main.rs
View file

@ -5,30 +5,22 @@ mod geo;
mod intervals;
mod leaderboard;
mod model;
mod webhook;
mod web;
use std::sync::Arc;
mod webhook;
use anyhow::{Context, Result, anyhow};
use axum::{
Router,
extract::{Path, State},
routing::{get, post},
};
use chrono::{DateTime, Duration, Utc};
use serde_json::Value;
use sqlx::PgPool;
use sqlx::{PgPool, Row};
use tower_http::trace::TraceLayer;
use tracing_subscriber::{
layer::SubscriberExt,
util::SubscriberInitExt,
};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use crate::{
config::Config,
intervals::IntervalsClient,
model::TrackPoint,
};
use crate::{config::Config, intervals::IntervalsClient, model::TrackPoint};
#[derive(Clone)]
pub struct AppState {
@ -39,19 +31,20 @@ pub struct AppState {
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::from_default_env()
)
.with(tracing_subscriber::EnvFilter::from_default_env())
.with(tracing_subscriber::fmt::layer())
.init();
let config =
Config::from_env()?;
let config = Config::from_env()?;
let db =
PgPool::connect(&config.database_url)
.await
.context("cannot connect to PostgreSQL")?;
tracing::info!(
database_url = %redact_database_url(&config.database_url),
"connecting to PostgreSQL"
);
let db = PgPool::connect(&config.database_url)
.await
.context("cannot connect to PostgreSQL")?;
sqlx::migrate!()
.run(&db)
@ -69,22 +62,29 @@ async fn main() -> Result<()> {
.route("/oauth/start", get(auth::start))
.route("/oauth/callback", get(auth::callback))
.route("/logout", get(auth::logout))
.route(
"/webhooks/intervals",
post(webhook::receive),
)
.route(
"/api/leaderboard",
get(leaderboard::api),
)
.route("/webhooks/intervals", post(webhook::receive))
.route("/api/leaderboard", get(leaderboard::api))
/*
* Development-only endpoint.
*
* This lets us test the complete activity-processing
* pipeline without having to trigger an Intervals.icu
* webhook.
*
* Example:
*
* curl -X POST \
* http://127.0.0.1:8080/dev/process/12345678
*
* Do not expose this endpoint publicly in production.
*/
.route("/dev/process/:activity_id", post(dev_process_activity))
.layer(TraceLayer::new_for_http())
.with_state(state);
let listener =
tokio::net::TcpListener::bind(
&config.bind_address
)
.await?;
let listener = tokio::net::TcpListener::bind(&config.bind_address)
.await
.with_context(|| format!("cannot bind to {}", config.bind_address))?;
tracing::info!(
address = %config.bind_address,
@ -100,16 +100,64 @@ async fn health() -> &'static str {
"ok"
}
/// Development endpoint for processing one real Intervals.icu
/// activity using INTERVALS_API_KEY.
///
/// This intentionally bypasses OAuth and webhooks.
async fn dev_process_activity(
State(state): State<AppState>,
Path(activity_id): Path<String>,
) -> Result<String, (axum::http::StatusCode, String)> {
process_activity(state, configured_athlete_id(), activity_id)
.await
.map(|_| "activity processed\n".to_string())
.map_err(internal_error)
}
fn configured_athlete_id() -> String {
std::env::var("INTERVALS_ATHLETE_ID").unwrap_or_else(|_| "0".to_string())
}
fn internal_error(error: anyhow::Error) -> (axum::http::StatusCode, String) {
tracing::error!(
%error,
"request failed"
);
(
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
error.to_string(),
)
}
/// Process one Intervals.icu activity.
///
/// In development mode the Intervals.icu personal API key from
/// INTERVALS_API_KEY is used by IntervalsClient.
///
/// The activity is:
///
/// 1. fetched from Intervals.icu
/// 2. its GPS/time streams are fetched
/// 3. GPS points are converted to TrackPoint values
/// 4. county-border crossings are detected
/// 5. crossing times are interpolated to millisecond precision
/// 6. matching Intervals.icu intervals are attached where possible
/// 7. everything is stored transactionally
pub async fn process_activity(
state: AppState,
intervals_athlete_id: String,
activity_id: String,
) -> Result<()> {
/*
* The development API key represents one athlete.
*
* We still keep the athlete in our database because activities
* and crossings belong to an athlete.
*/
let athlete = sqlx::query(
r#"
SELECT
id,
access_token
SELECT id
FROM athletes
WHERE intervals_athlete_id = $1
"#,
@ -118,113 +166,127 @@ pub async fn process_activity(
.fetch_optional(&state.db)
.await?;
use sqlx::Row;
let athlete_db_id: i64 = match athlete {
Some(row) => row.try_get("id")?,
let Some(athlete) = athlete else {
return Err(anyhow!(
"no OAuth connection for athlete {}",
intervals_athlete_id
));
None => {
let client = IntervalsClient::new(state.config.clone());
/*
* Fetch the activity first so we can get a useful
* athlete/display name if available.
*/
let activity = client.activity(&activity_id).await?;
let display_name = activity
.get("athlete")
.and_then(|athlete| athlete.get("name").or_else(|| athlete.get("display_name")))
.and_then(Value::as_str)
.unwrap_or("Intervals.icu athlete");
let row = sqlx::query(
r#"
INSERT INTO athletes (
intervals_athlete_id,
display_name,
access_token,
scopes
)
VALUES ($1, $2, '', 'DEV_API_KEY')
ON CONFLICT (intervals_athlete_id)
DO UPDATE SET
display_name = EXCLUDED.display_name,
updated_at = now()
RETURNING id
"#,
)
.bind(&intervals_athlete_id)
.bind(display_name)
.fetch_one(&state.db)
.await?;
row.try_get("id")?
}
};
let athlete_db_id: i64 =
athlete.try_get("id")?;
let client = IntervalsClient::new(state.config.clone());
let access_token: String =
athlete.try_get("access_token")?;
/*
* First fetch the activity metadata.
*
* Among other things this may contain Intervals.icu's
* automatically detected intervals.
*/
let activity = client.activity(&activity_id).await?;
let client =
IntervalsClient::new(state.config.clone());
let start_time = IntervalsClient::activity_start(&activity)?;
//
// First fetch the activity metadata including icu_intervals.
//
let activity =
client.activity(
&access_token,
&activity_id,
)
.await?;
/*
* Then fetch only the activity streams we need.
*
* For county crossings:
*
* time
* latlng
*
* Optional streams such as watts/hr can be added later
* without changing the crossing algorithm.
*/
let streams = client.streams(&activity_id).await?;
let start_time =
IntervalsClient::activity_start(
&activity
)?;
//
// Then fetch only what we actually need from the
// activity stream: time + GPS.
//
let streams =
client.streams(
&access_token,
&activity_id,
)
.await?;
let points =
parse_track(
&streams,
start_time,
)?;
let points = parse_track(&streams, start_time)?;
if points.len() < 2 {
tracing::info!(
activity_id = %activity_id,
points = points.len(),
"activity has insufficient GPS data"
);
return Ok(());
}
//
// Use a transaction so re-processing an activity is idempotent:
// old crossings disappear and are replaced atomically.
//
let mut tx =
state.db.begin().await?;
tracing::info!(
activity_id = %activity_id,
points = points.len(),
start_time = %start_time,
"processing activity"
);
/*
* Use a transaction so re-processing an activity is
* idempotent:
*
* old crossings are deleted
* new crossings are inserted
* activity is marked processed
*
* all atomically.
*/
let mut tx = state.db.begin().await?;
let activity_db_id =
db::ensure_activity(
&mut tx,
athlete_db_id,
&activity_id,
start_time,
&activity,
)
.await?;
db::ensure_activity(&mut tx, athlete_db_id, &activity_id, start_time, &activity).await?;
db::delete_activity_crossings(
&mut tx,
activity_db_id,
)
.await?;
db::delete_activity_crossings(&mut tx, activity_db_id).await?;
let mut crossing_count = 0usize;
for pair in points.windows(2) {
let crossings =
geo::crossings_between(
&state.db,
&pair[0],
&pair[1],
)
.await?;
let crossings = geo::crossings_between(&state.db, &pair[0], &pair[1]).await?;
for crossing in crossings {
let interval =
IntervalsClient::matching_interval(
&activity,
crossing.track_index,
);
/*
* Try to associate this county crossing with an
* automatically detected Intervals.icu interval.
*
* This is deliberately optional: a crossing does
* not become invalid merely because no interval
* was detected.
*/
let interval = IntervalsClient::matching_interval(&activity, crossing.track_index);
db::save_crossing(
&mut tx,
activity_db_id,
&crossing,
interval.as_ref(),
)
.await?;
db::save_crossing(&mut tx, activity_db_id, &crossing, interval.as_ref()).await?;
crossing_count += 1;
@ -234,16 +296,13 @@ pub async fn process_activity(
from = crossing.from_county_id,
to = crossing.to_county_id,
track_index = crossing.track_index,
interval_found = interval.is_some(),
"county crossing detected"
);
}
}
db::mark_activity_processed(
&mut tx,
activity_db_id,
)
.await?;
db::mark_activity_processed(&mut tx, activity_db_id).await?;
tx.commit().await?;
@ -256,44 +315,23 @@ pub async fn process_activity(
Ok(())
}
fn parse_track(
streams: &Value,
start_time: DateTime<Utc>,
) -> Result<Vec<TrackPoint>> {
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<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 latlng = IntervalsClient::stream_values(streams, "latlng")
.ok_or_else(|| anyhow!("Intervals.icu response has no latlng stream"))?;
let length =
times.len().min(latlng.len());
let length = times.len().min(latlng.len());
let mut result =
Vec::with_capacity(length);
let mut result = Vec::with_capacity(length);
for index in 0..length {
let Some(seconds) =
times[index].as_f64()
else {
let Some(seconds) = times[index].as_f64() else {
continue;
};
let Some(coordinate) =
latlng[index].as_array()
else {
let Some(coordinate) = latlng[index].as_array() else {
continue;
};
@ -301,32 +339,32 @@ fn parse_track(
continue;
}
let Some(lat) =
coordinate[0].as_f64()
else {
let Some(lat) = coordinate[0].as_f64() else {
continue;
};
let Some(lon) =
coordinate[1].as_f64()
else {
let Some(lon) = coordinate[1].as_f64() else {
continue;
};
if !lat.is_finite()
|| !lon.is_finite()
|| !seconds.is_finite()
{
if !lat.is_finite() || !lon.is_finite() || !seconds.is_finite() {
continue;
}
let millis =
(seconds * 1000.0).round() as i64;
/*
* Intervals.icu's time stream is relative to the
* activity start.
*
* We retain millisecond precision. Nanosecond
* precision would not provide meaningful additional
* accuracy because GPS samples themselves are much
* less precise.
*/
let millis = (seconds * 1000.0).round() as i64;
result.push(TrackPoint {
index,
time: start_time
+ Duration::milliseconds(millis),
time: start_time + Duration::milliseconds(millis),
lat,
lon,
});
@ -335,6 +373,30 @@ fn parse_track(
Ok(result)
}
/// Avoid accidentally logging database passwords.
fn redact_database_url(database_url: &str) -> String {
/*
* For our current local Unix-socket URL there normally is
* no password. For safety, don't print anything after the
* first '@' if a password is present.
*/
if let Some(at) = database_url.rfind('@') {
if let Some(scheme) = database_url.find("://") {
let prefix_end = scheme + 3;
if at > prefix_end {
return format!(
"{}***@{}",
&database_url[..prefix_end],
&database_url[at + 1..]
);
}
}
}
database_url.to_string()
}
#[cfg(test)]
mod tests {
use chrono::{TimeZone, Utc};
@ -343,39 +405,23 @@ mod tests {
#[test]
fn interpolation_is_millisecond_precise() {
let a =
Utc.timestamp_millis_opt(1000)
.unwrap();
let a = Utc.timestamp_millis_opt(1000).unwrap();
let b =
Utc.timestamp_millis_opt(2000)
.unwrap();
let b = Utc.timestamp_millis_opt(2000).unwrap();
let result =
interpolate_time(a, b, 0.376);
let result = interpolate_time(a, b, 0.376);
assert_eq!(
result.timestamp_millis(),
1376
);
assert_eq!(result.timestamp_millis(), 1376);
}
#[test]
fn interpolation_rounds_to_nearest_ms() {
let a =
Utc.timestamp_millis_opt(1000)
.unwrap();
let a = Utc.timestamp_millis_opt(1000).unwrap();
let b =
Utc.timestamp_millis_opt(2000)
.unwrap();
let b = Utc.timestamp_millis_opt(2000).unwrap();
let result =
interpolate_time(a, b, 0.3764);
let result = interpolate_time(a, b, 0.3764);
assert_eq!(
result.timestamp_millis(),
1376
);
assert_eq!(result.timestamp_millis(), 1376);
}
}

83
src/web.rs
View file

@ -1,38 +1,20 @@
use axum::{
extract::State,
response::Html,
};
use axum::{extract::State, response::Html};
use axum_extra::extract::cookie::CookieJar;
use crate::{
AppState,
db,
leaderboard,
};
use crate::{AppState, db, leaderboard};
pub async fn index(
State(state): State<AppState>,
jar: CookieJar,
) -> Html<String> {
let groups =
leaderboard::build(&state.db)
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
.unwrap_or_default();
.ok()
.flatten()
.map(|(_, name)| name),
let current_user =
match jar.get("county_session") {
Some(cookie) =>
db::athlete_for_session(
&state.db,
cookie.value(),
)
.await
.ok()
.flatten()
.map(|(_, name)| name),
None => None,
};
None => None,
};
let mut html = String::from(
r#"<!doctype html>
@ -113,7 +95,7 @@ button, .button {
html.push_str(
r#"<a class="button" href="/oauth/start">
Mit Intervals.icu verbinden
</a>"#
</a>"#,
);
}
@ -123,7 +105,7 @@ button, .button {
html.push_str(
r#"<div class="card">
Noch keine Landkreisüberquerungen.
</div>"#
</div>"#,
);
}
@ -153,31 +135,28 @@ button, .button {
<th>HR</th>
</tr>
</thead>
<tbody>"#
<tbody>"#,
);
for row in group.rows {
let interval =
match row.interval_url {
Some(url) =>
format!(
"<a href=\"{}\" target=\"_blank\">Intervall</a>",
escape_html(&url)
),
let interval = match row.interval_url {
Some(url) => format!(
"<a href=\"{}\" target=\"_blank\">Intervall</a>",
escape_html(&url)
),
None =>
"<span class=\"small\">—</span>".into(),
};
None => "<span class=\"small\">—</span>".into(),
};
let power =
row.average_watts
.map(|v| format!("{:.0} W", v))
.unwrap_or_else(|| "—".into());
let power = row
.average_watts
.map(|v| format!("{:.0} W", v))
.unwrap_or_else(|| "—".into());
let hr =
row.average_heartrate
.map(|v| format!("{:.0} bpm", v))
.unwrap_or_else(|| "—".into());
let hr = row
.average_heartrate
.map(|v| format!("{:.0} bpm", v))
.unwrap_or_else(|| "—".into());
html.push_str(&format!(
r#"<tr>
@ -207,7 +186,7 @@ button, .button {
Datenquelle: Intervals.icu · Landkreisgeometrien: VG250
</footer>
</body>
</html>"#
</html>"#,
);
Html(html)

122
src/webhook.rs
View file

@ -1,81 +1,71 @@
use axum::{
extract::State,
http::StatusCode,
Json,
};
use axum::{extract::State, http::StatusCode, response::IntoResponse};
use serde::Deserialize;
use crate::{
AppState,
model::WebhookEnvelope,
};
use crate::AppState;
#[derive(Debug, Deserialize)]
pub struct WebhookPayload {
pub secret: String,
#[serde(default)]
pub athlete_id: Option<String>,
#[serde(default)]
pub activity_id: Option<String>,
}
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);
axum::Json(payload): axum::Json<WebhookPayload>,
) -> impl IntoResponse {
let Some(expected_secret) = state.config.intervals_webhook_secret.as_deref() else {
tracing::error!("Intervals.icu webhook secret is not configured");
return (
StatusCode::INTERNAL_SERVER_ERROR,
"webhook secret not configured",
)
.into_response();
};
if payload.secret != expected_secret {
tracing::warn!("received webhook with invalid secret");
return (StatusCode::UNAUTHORIZED, "invalid webhook secret").into_response();
}
for event in payload.events {
match event.event_type.as_str() {
"ACTIVITY_UPLOADED" |
"ACTIVITY_ANALYZED" => {
let activity_id = event
.activity
.as_ref()
.and_then(|activity| {
activity.get("id")
})
.and_then(|id| {
id.as_str()
});
let Some(activity_id) = payload.activity_id else {
tracing::warn!("received webhook without activity_id");
let Some(activity_id) = activity_id else {
tracing::warn!(
athlete_id = %event.athlete_id,
"activity webhook without activity id"
);
return (StatusCode::BAD_REQUEST, "missing activity_id").into_response();
};
continue;
};
let athlete_id = payload
.athlete_id
.or_else(|| std::env::var("INTERVALS_ATHLETE_ID").ok());
if let Err(error) =
crate::process_activity(
state.clone(),
event.athlete_id.clone(),
activity_id.to_owned(),
).await
{
tracing::error!(
%error,
athlete_id = %event.athlete_id,
activity_id = %activity_id,
"activity processing failed"
);
let Some(athlete_id) = athlete_id else {
tracing::error!(
"webhook has no athlete_id and \
INTERVALS_ATHLETE_ID is not configured"
);
// Return 500 so Intervals.icu can retry the webhook.
return Err(StatusCode::INTERNAL_SERVER_ERROR);
}
}
return (StatusCode::BAD_REQUEST, "missing athlete_id").into_response();
};
"APP_SCOPE_CHANGED" => {
tracing::info!(
athlete_id = %event.athlete_id,
"OAuth scopes changed"
);
}
let state_clone = state.clone();
_ => {
tracing::debug!(
event_type = %event.event_type,
"ignoring Intervals.icu webhook"
);
}
tokio::spawn(async move {
if let Err(error) =
crate::process_activity(state_clone, athlete_id, activity_id.clone()).await
{
tracing::error!(
%error,
activity_id = %activity_id,
"failed to process webhook activity"
);
}
}
});
// Intervals.icu has historically retried on 204; use an ordinary 200.
Ok("ok")
(StatusCode::ACCEPTED, "activity queued").into_response()
}