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

fixups

Browse source
This commit is contained in:
Jonas Rabenstein 2026-08-12 14:24:50 +02:00
commit eda8e76a20
4 changed files with 481 additions and 218 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/config.rs
View file

@ -16,6 +16,8 @@ pub struct Config {
pub intervals_webhook_secret: Option<String>,
pub cookie_secure: bool,
pub dev_sync_days: i64,
}
impl Config {
@ -23,6 +25,7 @@ impl Config {
dotenvy::dotenv().ok();
let database_url = required("DATABASE_URL")?;
let bind_address =
env::var("BIND_ADDRESS").unwrap_or_else(|_| "127.0.0.1:8080".to_string());
@ -47,6 +50,15 @@ impl Config {
.parse::<bool>()
.context("COOKIE_SECURE must be true or false")?;
let dev_sync_days = env::var("DEV_SYNC_DAYS")
.unwrap_or_else(|_| "30".to_string())
.parse::<i64>()
.context("DEV_SYNC_DAYS must be an integer")?;
if dev_sync_days <= 0 {
anyhow::bail!("DEV_SYNC_DAYS must be greater than zero");
}
Ok(Self {
database_url,
bind_address,
@ -61,6 +73,8 @@ impl Config {
intervals_webhook_secret,
cookie_secure,
dev_sync_days,
})
}

11
src/db.rs
View file

@ -8,7 +8,9 @@ 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()
}
@ -54,6 +56,7 @@ pub async fn athlete_for_session(db: &PgPool, token: &str) -> Result<Option<(i64
if let Some(row) = row {
let id: i64 = row.try_get("id")?;
let name: String = row.try_get("display_name")?;
sqlx::query(
@ -102,7 +105,13 @@ pub async fn ensure_activity(
activity_json,
processed_at
)
VALUES ($1, $2, $3, $4, NULL)
VALUES (
$1,
$2,
$3,
$4,
NULL
)
ON CONFLICT (
athlete_id,
intervals_activity_id

161
src/intervals.rs
View file

@ -1,7 +1,7 @@
use anyhow::{Context, Result, anyhow};
use chrono::{DateTime, NaiveDateTime, Utc};
use reqwest::{Client, StatusCode};
use serde_json::{Value, json};
use reqwest::Client;
use serde_json::Value;
use crate::config::Config;
@ -9,13 +9,28 @@ use crate::config::Config;
pub struct IntervalsClient {
client: Client,
config: Config,
api_key: Option<String>,
}
impl IntervalsClient {
pub fn new(config: Config) -> Self {
let api_key = config.intervals_api_key.clone();
Self {
client: Client::new(),
config,
api_key,
}
}
/// Create a client using an explicitly supplied personal API key.
///
/// Primarily used by /dev/sync/{api_key}/{athlete_id}.
pub fn with_api_key(config: Config, api_key: impl Into<String>) -> Self {
Self {
client: Client::new(),
config,
api_key: Some(api_key.into()),
}
}
@ -24,7 +39,9 @@ impl IntervalsClient {
}
fn api_key(&self) -> Result<&str> {
self.config.require_api_key()
self.api_key
.as_deref()
.context("Intervals.icu API key is not configured")
}
fn api_url(&self, path: &str) -> String {
@ -41,12 +58,13 @@ impl IntervalsClient {
async fn get_json(&self, url: String) -> Result<Value> {
let response = self
.authenticated(self.client.get(url))?
.authenticated(self.client.get(&url))?
.send()
.await
.context("Intervals.icu request failed")?;
let status = response.status();
let body = response
.text()
.await
@ -60,40 +78,103 @@ impl IntervalsClient {
.with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body))
}
/// Get the athlete belonging to the configured API key.
///
/// GET /api/v1/athlete/0
pub async fn owner(&self) -> Result<Value> {
let url = self.api_url("athlete/0");
self.get_json(url).await
}
/// Get a specific athlete.
pub async fn athlete(&self, athlete_id: &str) -> Result<Value> {
let url = self.api_url(&format!("athlete/{athlete_id}"));
self.get_json(url).await
}
/// Get a single activity.
///
/// Uses:
/// GET /api/v1/activity/{id}
pub async fn activity(&self, activity_id: &str) -> Result<Value> {
let url = self.api_url(&format!("activity/{activity_id}"));
let url = self.api_url(&format!("activity/{activity_id}?intervals=true"));
self.get_json(url).await
}
/// Get activity streams.
///
/// Uses:
/// GET /api/v1/activity/{id}/streams.json
///
/// The response is normally an array of stream objects:
///
/// [
/// {"type":"time","data":[...]},
/// {"type":"latlng","data":[...]},
/// ...
/// ]
/// We explicitly request the streams needed for GPS processing.
pub async fn streams(&self, activity_id: &str) -> Result<Value> {
let url = self.api_url(&format!("activity/{activity_id}/streams.json"));
let url = self.api_url(&format!(
"activity/{activity_id}/streams.json?types=time,latlng"
));
self.get_json(url).await
}
/// Fetch the athlete's activities.
///
/// `0` means the athlete belonging to the API key.
pub async fn activities(&self, oldest: &str, newest: &str) -> Result<Value> {
let athlete_id = &self.config.intervals_athlete_id;
/// Find a stream object by type.
pub fn find_stream<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Value> {
// Normal API response:
//
// [
// {
// "type": "time",
// "data": [...]
// },
// {
// "type": "latlng",
// "data": [...],
// "data2": [...]
// }
// ]
if let Some(array) = streams.as_array() {
return array
.iter()
.find(|stream| stream.get("type").and_then(Value::as_str) == Some(stream_type));
}
// Also tolerate:
//
// {
// "time": {...},
// "latlng": {...}
// }
streams.get(stream_type)
}
/// Get the `data` array of a normal stream.
pub fn stream_values<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Vec<Value>> {
let stream = Self::find_stream(streams, stream_type)?;
stream.get("data").and_then(Value::as_array)
}
/// Get the latitude and longitude arrays from the
/// Intervals.icu `latlng` stream.
///
/// Intervals.icu represents latlng as:
///
/// data = latitude
/// data2 = longitude
///
pub fn latlng_values<'a>(streams: &'a Value) -> Option<(&'a Vec<Value>, &'a Vec<Value>)> {
let stream = Self::find_stream(streams, "latlng")?;
let lat = stream.get("data")?.as_array()?;
let lon = stream.get("data2")?.as_array()?;
Some((lat, lon))
}
/// Fetch an athlete's activities.
///
/// `athlete_id = "0"` means the athlete belonging to
/// the API key.
pub async fn activities(&self, athlete_id: &str, oldest: &str, newest: &str) -> Result<Value> {
let url = format!(
"{}?oldest={}&newest={}",
self.api_url(&format!("athlete/{athlete_id}/activities")),
@ -104,27 +185,6 @@ impl IntervalsClient {
self.get_json(url).await
}
/// Return a stream's values independent of whether the API
/// returned an object or an array of stream objects.
pub fn stream_values<'a>(streams: &'a Value, stream_type: &str) -> Option<&'a Vec<Value>> {
// Current API format: array of streams.
if let Some(array) = streams.as_array() {
for stream in array {
if stream.get("type").and_then(Value::as_str) == Some(stream_type) {
return stream.get("data").and_then(Value::as_array);
}
}
}
// Also support a dictionary-like response:
//
// {
// "time": [...],
// "latlng": [...]
// }
streams.get(stream_type).and_then(Value::as_array)
}
/// Extract the activity start timestamp.
pub fn activity_start(activity: &Value) -> Result<DateTime<Utc>> {
let value = activity
@ -142,9 +202,6 @@ impl IntervalsClient {
/// Try to find an Intervals.icu interval corresponding
/// to a GPS stream index.
///
/// The exact interval representation has changed over
/// time, so this deliberately supports the common forms.
pub fn matching_interval(activity: &Value, track_index: usize) -> Option<Value> {
let intervals = activity.get("icu_intervals")?;
@ -158,6 +215,7 @@ impl IntervalsClient {
let end = interval.get("end_index").or_else(|| interval.get("end"));
let start = start.and_then(Value::as_u64)?;
let end = end.and_then(Value::as_u64)?;
if (start as usize) <= track_index && track_index <= end as usize {
@ -168,8 +226,7 @@ impl IntervalsClient {
None
}
/// OAuth methods are intentionally retained so we can
/// re-enable multi-user OAuth later.
/// OAuth methods retained for the normal multi-user flow.
pub fn oauth_authorize_url(&self, state: &str) -> Result<String> {
let client_id = self
.config
@ -219,6 +276,7 @@ impl IntervalsClient {
.context("OAuth token request failed")?;
let status = response.status();
let body = response.text().await?;
if !status.is_success() {
@ -252,9 +310,14 @@ impl IntervalsClient {
let athlete_id = athlete
.get("id")
.and_then(Value::as_str)
.or_else(|| athlete.get("id").and_then(Value::as_i64).map(|_| ""))
.unwrap_or("")
.to_string();
.map(str::to_string)
.or_else(|| {
athlete
.get("id")
.and_then(Value::as_i64)
.map(|id| id.to_string())
})
.unwrap_or_default();
let display_name = athlete
.get("name")

515
src/main.rs
View file

@ -12,11 +12,12 @@ use anyhow::{Context, Result, anyhow};
use axum::{
Router,
extract::{Path, State},
response::Json,
routing::{get, post},
};
use chrono::{DateTime, Duration, Utc};
use serde_json::Value;
use sqlx::{PgPool, Row};
use serde_json::{Value, json};
use sqlx::PgPool;
use tower_http::trace::TraceLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@ -64,27 +65,16 @@ async fn main() -> Result<()> {
.route("/logout", get(auth::logout))
.route("/webhooks/intervals", post(webhook::receive))
.route("/api/leaderboard", get(leaderboard::api))
/*
* Development-only endpoint.
*
* This lets us test the complete activity-processing
* pipeline without having to trigger an Intervals.icu
* webhook.
*
* Example:
*
* curl -X POST \
* http://127.0.0.1:8080/dev/process/12345678
*
* Do not expose this endpoint publicly in production.
*/
.route("/dev/process/:activity_id", post(dev_process_activity))
// Development-only activity synchronisation.
//
// /dev/sync/{api_key}
// /dev/sync/{api_key}/{athlete_id}
.route("/dev/sync/{api_key}", get(dev_sync_one))
.route("/dev/sync/{api_key}/{athlete_id}", get(dev_sync))
.layer(TraceLayer::new_for_http())
.with_state(state);
let listener = tokio::net::TcpListener::bind(&config.bind_address)
.await
.with_context(|| format!("cannot bind to {}", config.bind_address))?;
let listener = tokio::net::TcpListener::bind(&config.bind_address).await?;
tracing::info!(
address = %config.bind_address,
@ -100,64 +90,327 @@ async fn health() -> &'static str {
"ok"
}
/// Development endpoint for processing one real Intervals.icu
/// activity using INTERVALS_API_KEY.
fn redact_database_url(url: &str) -> String {
if let Some((prefix, _)) = url.split_once('@') {
if prefix.contains("://") {
return format!("{prefix}@***");
}
}
url.to_string()
}
/// Development endpoint for importing/analyzing activities.
///
/// This intentionally bypasses OAuth and webhooks.
async fn dev_process_activity(
/// Examples:
///
/// GET /dev/sync/MY_API_KEY
/// GET /dev/sync/MY_API_KEY/0
/// GET /dev/sync/MY_API_KEY/i123456
///
/// `0` means the owner of the supplied API key.
async fn dev_sync(
State(state): State<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)
Path((api_key, athlete_id)): Path<(String, String)>,
) -> Result<Json<Value>, axum::http::StatusCode> {
match dev_sync_inner(state, api_key, athlete_id).await {
Ok(result) => Ok(Json(result)),
Err(error) => {
tracing::error!(
error = %error,
"development sync failed"
);
Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR)
}
}
}
fn configured_athlete_id() -> String {
std::env::var("INTERVALS_ATHLETE_ID").unwrap_or_else(|_| "0".to_string())
/// One-segment variant:
///
/// /dev/sync/{api_key}
///
/// Equivalent to:
///
/// /dev/sync/{api_key}/0
async fn dev_sync_one(
State(state): State<AppState>,
Path(api_key): Path<String>,
) -> Result<Json<Value>, axum::http::StatusCode> {
match dev_sync_inner(state, api_key, "0".to_string()).await {
Ok(result) => Ok(Json(result)),
Err(error) => {
tracing::error!(
error = %error,
"development sync failed"
);
Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR)
}
}
}
fn internal_error(error: anyhow::Error) -> (axum::http::StatusCode, String) {
tracing::error!(
%error,
"request failed"
async fn dev_sync_inner(state: AppState, api_key: String, athlete_id: String) -> Result<Value> {
if api_key.trim().is_empty() {
return Err(anyhow!("API key is empty"));
}
let athlete_id = if athlete_id.trim().is_empty() {
"0".to_string()
} else {
athlete_id
};
/*
* This client is explicitly authenticated using the API key
* supplied to the development endpoint.
*/
let client = IntervalsClient::with_api_key(state.config.clone(), api_key.clone());
/*
* Resolve athlete 0 through Intervals.icu.
*
* Intervals.icu supports athlete/0 for personal API keys;
* it refers to the athlete belonging to that key.
*/
let resolved_athlete_id = if athlete_id == "0" {
let owner = client
.owner()
.await
.context("cannot resolve API-key owner")?;
owner
.get("id")
.and_then(Value::as_str)
.map(str::to_string)
.or_else(|| {
owner
.get("id")
.and_then(Value::as_i64)
.map(|id| id.to_string())
})
.context("Intervals.icu owner response has no athlete id")?
} else {
athlete_id.clone()
};
let newest = Utc::now().date_naive();
let oldest = newest - Duration::days(state.config.dev_sync_days);
let oldest_string = oldest.format("%Y-%m-%d").to_string();
let newest_string = newest.format("%Y-%m-%d").to_string();
tracing::info!(
athlete_id = %resolved_athlete_id,
oldest = %oldest_string,
newest = %newest_string,
"starting development activity sync"
);
(
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
error.to_string(),
)
/*
* IMPORTANT:
*
* Use the resolved athlete ID here, not the original
* "0" value.
*/
let activities = client
.activities(&resolved_athlete_id, &oldest_string, &newest_string)
.await
.context("cannot fetch development activities")?;
let activities_array = activities
.as_array()
.context("Intervals.icu activities response is not an array")?;
let mut processed = 0usize;
let mut skipped = 0usize;
let mut failed = 0usize;
let mut results = Vec::new();
/*
* Make sure there is a corresponding local athlete row.
*
* We store the development API key as access_token because
* process_activity() uses the local athlete record and the
* existing activity-processing pipeline expects that field.
*/
ensure_dev_athlete(&state, &resolved_athlete_id, &client, &api_key).await?;
for activity in activities_array {
let Some(activity_id) = activity.get("id").and_then(Value::as_str) else {
skipped += 1;
results.push(json!({
"status": "skipped",
"reason": "activity has no id"
}));
continue;
};
match process_activity(
state.clone(),
resolved_athlete_id.clone(),
activity_id.to_string(),
)
.await
{
Ok(()) => {
processed += 1;
results.push(json!({
"id": activity_id,
"status": "processed"
}));
}
Err(error) => {
failed += 1;
tracing::error!(
activity_id = %activity_id,
error = %error,
"development activity processing failed"
);
results.push(json!({
"id": activity_id,
"status": "failed",
"error": error.to_string()
}));
}
}
}
tracing::info!(
athlete_id = %resolved_athlete_id,
found = activities_array.len(),
processed,
skipped,
failed,
"development activity sync finished"
);
Ok(json!({
"athlete_id": resolved_athlete_id,
"oldest": oldest_string,
"newest": newest_string,
"found": activities_array.len(),
"processed": processed,
"skipped": skipped,
"failed": failed,
"activities": results
}))
}
/// Process one Intervals.icu activity.
/// Make sure the athlete used by /dev/sync exists locally.
///
/// In development mode the Intervals.icu personal API key from
/// INTERVALS_API_KEY is used by IntervalsClient.
/// If the athlete already exists, update its development API
/// credential so process_activity() can use it.
///
/// The activity is:
///
/// 1. fetched from Intervals.icu
/// 2. its GPS/time streams are fetched
/// 3. GPS points are converted to TrackPoint values
/// 4. county-border crossings are detected
/// 5. crossing times are interpolated to millisecond precision
/// 6. matching Intervals.icu intervals are attached where possible
/// 7. everything is stored transactionally
/// We deliberately do not print the API key.
async fn ensure_dev_athlete(
state: &AppState,
intervals_athlete_id: &str,
client: &IntervalsClient,
api_key: &str,
) -> Result<()> {
let athlete = client
.athlete(intervals_athlete_id)
.await
.with_context(|| {
format!(
"cannot fetch Intervals.icu athlete {}",
intervals_athlete_id
)
})?;
let display_name = athlete
.get("name")
.or_else(|| athlete.get("display_name"))
.and_then(Value::as_str)
.unwrap_or("Intervals.icu athlete");
/*
* If this athlete already exists, update the development
* credential.
*
* This is particularly useful if the row was originally
* created by an earlier development sync with an empty token.
*/
let existing = sqlx::query(
r#"
SELECT id
FROM athletes
WHERE intervals_athlete_id = $1
"#,
)
.bind(intervals_athlete_id)
.fetch_optional(&state.db)
.await?;
if existing.is_some() {
sqlx::query(
r#"
UPDATE athletes
SET
access_token = $2,
display_name = $3,
updated_at = now()
WHERE intervals_athlete_id = $1
"#,
)
.bind(intervals_athlete_id)
.bind(api_key)
.bind(display_name)
.execute(&state.db)
.await
.context("cannot update local dev athlete")?;
return Ok(());
}
/*
* `scopes` is NOT NULL in the initial migration, so it must
* be supplied here.
*/
sqlx::query(
r#"
INSERT INTO athletes (
intervals_athlete_id,
display_name,
access_token,
scopes
)
VALUES ($1, $2, $3, $4)
"#,
)
.bind(intervals_athlete_id)
.bind(display_name)
.bind(api_key)
.bind("dev")
.execute(&state.db)
.await
.context("cannot create local dev athlete")?;
Ok(())
}
pub async fn process_activity(
state: AppState,
intervals_athlete_id: String,
activity_id: String,
) -> Result<()> {
/*
* The development API key represents one athlete.
*
* We still keep the athlete in our database because activities
* and crossings belong to an athlete.
*/
let athlete = sqlx::query(
r#"
SELECT id
SELECT
id,
access_token
FROM athletes
WHERE intervals_athlete_id = $1
"#,
@ -166,73 +419,45 @@ pub async fn process_activity(
.fetch_optional(&state.db)
.await?;
let athlete_db_id: i64 = match athlete {
Some(row) => row.try_get("id")?,
use sqlx::Row;
None => {
let client = IntervalsClient::new(state.config.clone());
/*
* Fetch the activity first so we can get a useful
* athlete/display name if available.
*/
let activity = client.activity(&activity_id).await?;
let display_name = activity
.get("athlete")
.and_then(|athlete| athlete.get("name").or_else(|| athlete.get("display_name")))
.and_then(Value::as_str)
.unwrap_or("Intervals.icu athlete");
let row = sqlx::query(
r#"
INSERT INTO athletes (
intervals_athlete_id,
display_name,
access_token,
scopes
)
VALUES ($1, $2, '', 'DEV_API_KEY')
ON CONFLICT (intervals_athlete_id)
DO UPDATE SET
display_name = EXCLUDED.display_name,
updated_at = now()
RETURNING id
"#,
)
.bind(&intervals_athlete_id)
.bind(display_name)
.fetch_one(&state.db)
.await?;
row.try_get("id")?
}
let Some(athlete) = athlete else {
return Err(anyhow!(
"no local athlete for Intervals.icu athlete {}",
intervals_athlete_id
));
};
let athlete_db_id: i64 = athlete.try_get("id")?;
/*
* The normal activity-processing path uses the configured
* Intervals client.
*
* The development endpoint has already ensured that the
* athlete exists locally and that its credential is stored.
*/
let client = IntervalsClient::new(state.config.clone());
/*
* First fetch the activity metadata.
*
* Among other things this may contain Intervals.icu's
* automatically detected intervals.
* Fetch activity metadata.
*/
let activity = client.activity(&activity_id).await?;
let activity = client
.activity(&activity_id)
.await
.with_context(|| format!("cannot fetch Intervals.icu activity {}", activity_id))?;
let start_time = IntervalsClient::activity_start(&activity)?;
/*
* Then fetch only the activity streams we need.
*
* For county crossings:
*
* time
* latlng
*
* Optional streams such as watts/hr can be added later
* without changing the crossing algorithm.
* Fetch GPS/time streams.
*/
let streams = client.streams(&activity_id).await?;
let streams = client.streams(&activity_id).await.with_context(|| {
format!(
"cannot fetch streams for Intervals.icu activity {}",
activity_id
)
})?;
let points = parse_track(&streams, start_time)?;
@ -246,22 +471,16 @@ pub async fn process_activity(
return Ok(());
}
tracing::info!(
activity_id = %activity_id,
points = points.len(),
start_time = %start_time,
"processing activity"
);
/*
* Use a transaction so re-processing an activity is
* idempotent:
* Re-processing is idempotent:
*
* old crossings are deleted
* new crossings are inserted
* activity is marked processed
* - update activity
* - remove existing crossings
* - calculate crossings
* - insert crossings
* - mark processed
*
* all atomically.
* All changes happen inside one transaction.
*/
let mut tx = state.db.begin().await?;
@ -276,14 +495,6 @@ pub async fn process_activity(
let crossings = geo::crossings_between(&state.db, &pair[0], &pair[1]).await?;
for crossing in crossings {
/*
* Try to associate this county crossing with an
* automatically detected Intervals.icu interval.
*
* This is deliberately optional: a crossing does
* not become invalid merely because no interval
* was detected.
*/
let interval = IntervalsClient::matching_interval(&activity, crossing.track_index);
db::save_crossing(&mut tx, activity_db_id, &crossing, interval.as_ref()).await?;
@ -296,7 +507,6 @@ pub async fn process_activity(
from = crossing.from_county_id,
to = crossing.to_county_id,
track_index = crossing.track_index,
interval_found = interval.is_some(),
"county crossing detected"
);
}
@ -351,15 +561,6 @@ fn parse_track(streams: &Value, start_time: DateTime<Utc>) -> Result<Vec<TrackPo
continue;
}
/*
* 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 {
@ -373,30 +574,6 @@ fn parse_track(streams: &Value, start_time: DateTime<Utc>) -> Result<Vec<TrackPo
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};