county-sprints/src/db.rs

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
          AND s.created_at > now() - interval '1 day'
        "#,
    )
    .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(
            r#"
            UPDATE sessions
            SET last_seen_at = now()
            WHERE token_hash = $1
            "#,
        )
        .bind(&hash)
        .execute(db)
        .await?;

        Ok(Some((id, name)))
    } else {
        Ok(None)
    }
}

#[derive(Debug, Clone)]
pub struct SessionAthlete {
    pub id: i64,
    pub intervals_athlete_id: String,
    pub display_name: String,
    pub access_token: String,
}

pub async fn session_athlete(db: &PgPool, token: &str) -> Result<Option<SessionAthlete>> {
    let hash = hash_token(token);

    let row = sqlx::query(
        r#"
        SELECT
            a.id,
            a.intervals_athlete_id,
            a.display_name,
            a.access_token
        FROM sessions s
        JOIN athletes a
          ON a.id = s.athlete_id
        WHERE s.token_hash = $1
          AND s.created_at > now() - interval '1 day'
        "#,
    )
    .bind(&hash)
    .fetch_optional(db)
    .await?;

    use sqlx::Row;

    let Some(row) = row else {
        return Ok(None);
    };

    sqlx::query(
        r#"
        UPDATE sessions
        SET last_seen_at = now()
        WHERE token_hash = $1
        "#,
    )
    .bind(&hash)
    .execute(db)
    .await?;

    Ok(Some(SessionAthlete {
        id: row.try_get("id")?,
        intervals_athlete_id: row.try_get("intervals_athlete_id")?,
        display_name: row.try_get("display_name")?,
        access_token: row.try_get("access_token")?,
    }))
}

/// Return the start time of the most recently imported activity
/// for an athlete.
///
/// This is used as the default `oldest` value on the sync page.
pub async fn last_synced_activity_time(
    db: &PgPool,
    athlete_id: i64,
) -> Result<Option<DateTime<Utc>>> {
    let row = sqlx::query(
        r#"
        SELECT MAX(start_time) AS last_synced
        FROM activities
        WHERE athlete_id = $1
          AND processed_at IS NOT NULL
        "#,
    )
    .bind(athlete_id)
    .fetch_one(db)
    .await?;

    use sqlx::Row;

    Ok(row.try_get("last_synced")?)
}

pub async fn delete_old_oauth_states(db: &PgPool) -> Result<()> {
    sqlx::query(
        r#"
        DELETE FROM oauth_states
        WHERE created_at < now() - interval '10 minutes'
        "#,
    )
    .execute(db)
    .await?;

    Ok(())
}

/// Create or update an athlete authenticated with a personal API key.
pub async fn upsert_api_key_athlete(
    db: &PgPool,
    intervals_athlete_id: &str,
    display_name: &str,
    api_key: &str,
) -> Result<i64> {
    let row = sqlx::query(
        r#"
        INSERT INTO athletes (
            intervals_athlete_id,
            display_name,
            access_token,
            scopes
        )
        VALUES ($1, $2, $3, 'API_KEY')
        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(intervals_athlete_id)
    .bind(display_name)
    .bind(api_key)
    .fetch_one(db)
    .await?;

    use sqlx::Row;
    Ok(row.try_get("id")?)
}

/// Create or update an athlete authenticated through OAuth.
pub async fn upsert_oauth_athlete(
    db: &PgPool,
    intervals_athlete_id: &str,
    display_name: &str,
    access_token: &str,
    scopes: &str,
) -> Result<i64> {
    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(intervals_athlete_id)
    .bind(display_name)
    .bind(access_token)
    .bind(scopes)
    .fetch_one(db)
    .await?;

    use sqlx::Row;
    Ok(row.try_get("id")?)
}


/// Ensure that an athlete accessible through the current session exists
/// locally so activities can reference it. The session credential itself is
/// intentionally not stored here; callers continue to use their authenticated
/// IntervalsClient for API requests.
pub async fn ensure_local_athlete(
    db: &PgPool,
    intervals_athlete_id: &str,
    display_name: &str,
) -> Result<i64> {
    let row = sqlx::query(
        r#"
        INSERT INTO athletes (
            intervals_athlete_id,
            display_name,
            access_token,
            scopes
        )
        VALUES ($1, $2, '', 'SESSION_ACCESS')
        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(db)
    .await?;

    use sqlx::Row;
    Ok(row.try_get("id")?)
}

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(
        r#"
        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(())
}

/// Remove sessions past their 1-day validity window.
///
/// Expired sessions are already rejected by `session_athlete` /
/// `athlete_for_session`; this just keeps the table from growing forever.
pub async fn delete_expired_sessions(db: &PgPool) -> Result<()> {
    sqlx::query(
        r#"
        DELETE FROM sessions
        WHERE created_at <= now() - interval '1 day'
        "#,
    )
    .execute(db)
    .await?;

    Ok(())
}

pub async fn delete_session(db: &PgPool, token: &str) -> Result<()> {
    let hash = hash_token(token);

    sqlx::query("DELETE FROM sessions WHERE token_hash = $1")
        .bind(&hash)
        .execute(db)
        .await?;

    Ok(())
}