ipodderx-rs/src/web.rs

//! Web front end. Runs inside the daemon so it reads SQLite and the event bus directly.

use anyhow::{Context, Result};
use axum::{
    Json, Router,
    extract::{Path, Query, Request, State},
    http::{StatusCode, header},
    middleware::{self, Next},
    response::{
        Html, IntoResponse, Response,
        sse::{Event as SseEvent, Sse},
    },
    routing::{delete, get, patch, post},
};
use futures_util::StreamExt;
use serde::Deserialize;
use tokio_stream::wrappers::BroadcastStream;
use tower::ServiceExt;
use tower_http::services::ServeFile;
use serde::Serialize;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::{broadcast, mpsc};

use crate::Ctx;
use crate::ipc::{Command, Event};

const COOKIE: &str = "ipx_token";

#[derive(Clone)]
pub struct WebState {
    pub ctx: Arc<Ctx>,
    pub config_path: PathBuf,
    pub cmds: mpsc::Sender<Command>,
    pub events: broadcast::Sender<Event>,
}

/// A 32-hex-character shared secret, generated when config.toml has none.
///
/// ponytail: /dev/urandom rather than a CSPRNG crate -- 16 bytes, once, on a Unix-only
/// binary. Falls back to the clock only if urandom is somehow unreadable, which would be a
/// weak token, so that case is logged loudly.
pub fn generate_token() -> String {
    use std::io::Read;
    let mut bytes = [0u8; 16];
    match std::fs::File::open("/dev/urandom").and_then(|mut f| f.read_exact(&mut bytes)) {
        Ok(()) => {}
        Err(e) => {
            tracing::error!(error = %e, "could not read /dev/urandom; token is NOT secure");
            let n = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos() as u64)
                .unwrap_or(0);
            bytes[..8].copy_from_slice(&n.to_le_bytes());
        }
    }
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

pub fn router(state: WebState) -> Router {
    Router::new()
        .route("/", get(index))
        .route("/api/feeds", get(feeds).post(add_feed))
        .route("/api/feeds/{id}", patch(patch_feed).delete(remove_feed))
        .route("/api/feeds/{id}/entries", get(entries))
        .route("/api/entries/{feed_id}/{guid}/flags", post(set_flags))
        .route("/api/entries/{feed_id}/{guid}/position", post(set_position))
        .route("/api/feeds/{id}/read-all", post(read_all))
        .route("/api/feeds/{id}/download-latest", post(download_latest))
        .route("/api/enclosures/{id}/download", post(download_now))
        .route("/api/enclosures/{id}", delete(delete_file))
        .route("/api/fetch", post(fetch_now))
        .route("/api/opml", get(export_opml).post(import_opml))
        .route("/api/events", get(events))
        .route("/media/{id}", get(media))
        .layer(middleware::from_fn_with_state(state.clone(), auth))
        .with_state(state)
}

pub async fn serve(state: WebState, bind: &str) -> Result<()> {
    let listener = tokio::net::TcpListener::bind(bind)
        .await
        .with_context(|| format!("binding {bind}"))?;
    tracing::info!(bind, "web ui listening");
    axum::serve(listener, router(state))
        .await
        .context("serving the web ui")
}

/// Token in `?token=` (which then sets a cookie) or in the cookie itself.
///
/// It has to be a cookie rather than a header: an `<audio src>` request is issued by the
/// browser, and there is no way to attach a header to it.
async fn auth(State(state): State<WebState>, req: Request, next: Next) -> Response {
    let expected = state.ctx.cfg().web.token.clone();
    if expected.is_empty() {
        // Refuse to serve rather than serve unauthenticated.
        return (StatusCode::INTERNAL_SERVER_ERROR, "no web token configured").into_response();
    }

    let from_query = req.uri().query().and_then(|q| {
        q.split('&')
            .find_map(|kv| kv.strip_prefix("token=").map(str::to_owned))
    });
    let from_cookie = req
        .headers()
        .get(header::COOKIE)
        .and_then(|v| v.to_str().ok())
        .and_then(|c| {
            c.split(';')
                .find_map(|kv| kv.trim().strip_prefix(&format!("{COOKIE}=")).map(str::to_owned))
        });

    let supplied = from_query.clone().or(from_cookie);
    if !supplied.is_some_and(|t| constant_time_eq(&t, &expected)) {
        return (StatusCode::UNAUTHORIZED, "bad or missing token").into_response();
    }

    let mut resp = next.run(req).await;
    if from_query.is_some() {
        // Remember it so the rest of the page (and the audio element) authenticates.
        if let Ok(v) = header::HeaderValue::from_str(&format!(
            "{COOKIE}={expected}; Path=/; SameSite=Lax; Max-Age=31536000"
        )) {
            resp.headers_mut().insert(header::SET_COOKIE, v);
        }
    }
    resp
}

/// Compares without leaking length or position through timing.
fn constant_time_eq(a: &str, b: &str) -> bool {
    let (a, b) = (a.as_bytes(), b.as_bytes());
    if a.len() != b.len() {
        return false;
    }
    a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}

async fn index() -> Html<&'static str> {
    Html(include_str!("../web/index.html"))
}

#[derive(Serialize)]
struct FeedRow {
    id: String,
    url: String,
    title: Option<String>,
    image: Option<String>,
    folder: Option<String>,
    keywords: Vec<String>,
    allow_explicit: bool,
    auto_download: bool,
    max_new_per_check: Option<usize>,
    last_checked: Option<i64>,
    last_error: Option<String>,
    entries: i64,
    downloaded: i64,
    unread: i64,
}

async fn feeds(State(state): State<WebState>) -> Result<Json<Vec<FeedRow>>, ApiError> {
    let cfg = state.ctx.cfg();
    let mut out = Vec::with_capacity(cfg.feeds.len());
    for (id, feed) in &cfg.feeds {
        let s = state.ctx.db.feed_summary(id)?;
        out.push(FeedRow {
            id: id.clone(),
            url: feed.url.clone(),
            title: s.title,
            image: s.image,
            folder: feed.folder.clone(),
            keywords: feed.keywords.clone(),
            allow_explicit: feed.allow_explicit,
            auto_download: feed.auto_download,
            max_new_per_check: feed.max_new_per_check,
            last_checked: s.last_checked,
            last_error: s.last_error,
            entries: s.entries,
            downloaded: s.downloaded,
            unread: state.ctx.db.unread_count(id)?,
        });
    }
    Ok(Json(out))
}

/// Turns anyhow errors into a 500 with a readable body.
pub struct ApiError(anyhow::Error);

impl<E: Into<anyhow::Error>> From<E> for ApiError {
    fn from(e: E) -> Self {
        Self(e.into())
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        tracing::warn!(error = ?self.0, "api error");
        (StatusCode::INTERNAL_SERVER_ERROR, format!("{:#}", self.0)).into_response()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn token_comparison_rejects_mismatches_and_length_differences() {
        assert!(constant_time_eq("abc123", "abc123"));
        assert!(!constant_time_eq("abc123", "abc124"));
        assert!(!constant_time_eq("abc", "abc123"));
        assert!(!constant_time_eq("", "abc"));
        assert!(constant_time_eq("", ""));
    }

    #[test]
    fn generated_tokens_are_32_hex_chars_and_not_repeated() {
        let a = generate_token();
        let b = generate_token();
        assert_eq!(a.len(), 32);
        assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
        assert_ne!(a, b);
    }
}

#[derive(Deserialize)]
struct Page {
    #[serde(default)]
    offset: i64,
    #[serde(default = "fifty")]
    limit: i64,
    #[serde(default)]
    filter: Option<String>,
    #[serde(default)]
    q: Option<String>,
}

fn fifty() -> i64 {
    50
}

#[derive(Serialize)]
struct EntryPage {
    total: i64,
    entries: Vec<crate::db::EntryRow>,
}

async fn entries(
    State(state): State<WebState>,
    Path(id): Path<String>,
    Query(page): Query<Page>,
) -> Result<Json<EntryPage>, ApiError> {
    let filter = crate::db::Filter::parse(page.filter.as_deref().unwrap_or("all"));
    let search = page.q.as_deref().map(str::trim).filter(|q| !q.is_empty());
    let mut rows = state
        .ctx
        .db
        .entries(&id, filter, search, page.offset, page.limit.clamp(1, 200))?;
    // Feed HTML is untrusted: it reaches the page only after ammonia has been through it.
    for row in &mut rows {
        if let Some(d) = &row.description {
            row.description = Some(ammonia::clean(d));
        }
    }
    let total = state.ctx.db.count_entries(&id, filter, search)?;
    Ok(Json(EntryPage { total, entries: rows }))
}

#[derive(Deserialize)]
struct NewFeed {
    url: String,
    #[serde(default)]
    folder: Option<String>,
    #[serde(default)]
    keywords: Vec<String>,
}

async fn add_feed(
    State(state): State<WebState>,
    Json(body): Json<NewFeed>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let mut cfg = (*state.ctx.cfg()).clone();
    if let Some((id, _)) = cfg.feeds.iter().find(|(_, f)| f.url == body.url) {
        return Ok(Json(serde_json::json!({ "id": id, "existing": true })));
    }
    let id = crate::add_one(&state.ctx, &mut cfg, &body.url, body.folder, body.keywords).await?;
    cfg.save(&state.config_path)?;
    state.ctx.reload_cfg(&state.config_path)?;
    Ok(Json(serde_json::json!({ "id": id, "existing": false })))
}

/// Only the fields that are present are changed.
#[derive(Deserialize)]
struct FeedPatch {
    folder: Option<Option<String>>,
    keywords: Option<Vec<String>>,
    allow_explicit: Option<bool>,
    auto_download: Option<bool>,
    max_new_per_check: Option<Option<usize>>,
}

async fn patch_feed(
    State(state): State<WebState>,
    Path(id): Path<String>,
    Json(body): Json<FeedPatch>,
) -> Result<StatusCode, ApiError> {
    let mut cfg = (*state.ctx.cfg()).clone();
    let feed = cfg
        .feeds
        .get_mut(&id)
        .ok_or_else(|| anyhow::anyhow!("no feed with id {id:?}"))?;

    if let Some(v) = body.folder {
        feed.folder = v.filter(|s| !s.trim().is_empty());
    }
    if let Some(v) = body.keywords {
        feed.keywords = v.into_iter().filter(|k| !k.trim().is_empty()).collect();
    }
    if let Some(v) = body.allow_explicit {
        feed.allow_explicit = v;
    }
    if let Some(v) = body.auto_download {
        feed.auto_download = v;
    }
    if let Some(v) = body.max_new_per_check {
        feed.max_new_per_check = v;
    }
    cfg.save(&state.config_path)?;
    state.ctx.reload_cfg(&state.config_path)?;
    Ok(StatusCode::NO_CONTENT)
}

async fn remove_feed(
    State(state): State<WebState>,
    Path(id): Path<String>,
) -> Result<StatusCode, ApiError> {
    let mut cfg = (*state.ctx.cfg()).clone();
    if cfg.feeds.remove(&id).is_none() {
        return Err(anyhow::anyhow!("no feed with id {id:?}").into());
    }
    // Downloads and history stay, so re-adding does not re-pull the back catalogue.
    cfg.save(&state.config_path)?;
    state.ctx.reload_cfg(&state.config_path)?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
struct Flags {
    read: Option<bool>,
    flagged: Option<bool>,
}

async fn set_flags(
    State(state): State<WebState>,
    Path((feed_id, guid)): Path<(String, String)>,
    Json(body): Json<Flags>,
) -> Result<StatusCode, ApiError> {
    use crate::db::EntryFlag;
    if let Some(v) = body.read {
        state.ctx.db.set_entry_flag(&feed_id, &guid, EntryFlag::Read, v)?;
    }
    if let Some(v) = body.flagged {
        state.ctx.db.set_entry_flag(&feed_id, &guid, EntryFlag::Flagged, v)?;
    }
    Ok(StatusCode::NO_CONTENT)
}

/// Downloads one enclosure now. This cannot be "requeue and scan": a scan takes the
/// lowest-id pending rows up to max_new_per_check, so with a big backlog it would download
/// other episodes and leave the requested one pending.
async fn download_now(
    State(state): State<WebState>,
    Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
    let enc = state
        .ctx
        .db
        .enclosure(id)?
        .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
    if enc.path.is_some() {
        return Ok(StatusCode::NO_CONTENT); // Already here.
    }
    state.ctx.db.requeue(id)?;
    state
        .cmds
        .send(Command::Download { enclosure: id })
        .await
        .map_err(|_| anyhow::anyhow!("the daemon is not accepting commands"))?;
    Ok(StatusCode::ACCEPTED)
}

async fn delete_file(
    State(state): State<WebState>,
    Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
    let enc = state
        .ctx
        .db
        .enclosure(id)?
        .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
    if let Some(path) = &enc.path
        && let Err(e) = std::fs::remove_file(path)
        && e.kind() != std::io::ErrorKind::NotFound
    {
        return Err(e.into());
    }
    // The row survives as 'reaped', which is what stops the next scan re-downloading it.
    state.ctx.db.mark_reaped(id)?;
    state.events.send(Event::Reaped {
        path: enc.path.unwrap_or_default(),
        bytes: enc.length.unwrap_or(0).max(0) as u64,
    }).ok();
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
struct FetchBody {
    #[serde(default)]
    feed: Option<String>,
    #[serde(default)]
    force: bool,
}

async fn fetch_now(
    State(state): State<WebState>,
    Json(body): Json<FetchBody>,
) -> Result<StatusCode, ApiError> {
    state
        .cmds
        .send(Command::Fetch { feed: body.feed, force: body.force })
        .await
        .map_err(|_| anyhow::anyhow!("the daemon is not accepting commands"))?;
    Ok(StatusCode::ACCEPTED)
}

/// The same broadcast the socket clients read, as server-sent events.
async fn events(State(state): State<WebState>) -> Sse<impl futures_util::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
    let stream = BroadcastStream::new(state.events.subscribe()).filter_map(|ev| async move {
        let ev = ev.ok()?;
        Some(Ok(SseEvent::default().data(serde_json::to_string(&ev).ok()?)))
    });
    Sse::new(stream).keep_alive(axum::response::sse::KeepAlive::default())
}

/// Audio, served by ServeFile so Range requests work and the player can seek.
async fn media(
    State(state): State<WebState>,
    Path(id): Path<i64>,
    req: Request,
) -> Response {
    let Ok(Some(enc)) = state.ctx.db.enclosure(id) else {
        return (StatusCode::NOT_FOUND, "no such enclosure").into_response();
    };
    let Some(path) = enc.path else {
        return (StatusCode::NOT_FOUND, "not downloaded").into_response();
    };
    match ServeFile::new(path).oneshot(req).await {
        Ok(r) => r.into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

#[derive(Deserialize)]
struct Position {
    secs: i64,
}

async fn set_position(
    State(state): State<WebState>,
    Path((feed_id, guid)): Path<(String, String)>,
    Json(body): Json<Position>,
) -> Result<StatusCode, ApiError> {
    state.ctx.db.set_position(&feed_id, &guid, body.secs)?;
    Ok(StatusCode::NO_CONTENT)
}

async fn read_all(
    State(state): State<WebState>,
    Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let n = state.ctx.db.mark_all_read(&id)?;
    Ok(Json(serde_json::json!({ "marked": n })))
}

#[derive(Deserialize)]
struct HowMany {
    #[serde(default = "five")]
    count: i64,
}

fn five() -> i64 {
    5
}

/// Queues the newest N undownloaded episodes, each as its own explicit Download command so
/// none of them is subject to the per-scan cap.
async fn download_latest(
    State(state): State<WebState>,
    Path(id): Path<String>,
    Json(body): Json<HowMany>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let ids = state.ctx.db.undownloaded(&id, body.count.clamp(1, 100))?;
    for enc in &ids {
        state.ctx.db.requeue(*enc)?;
        state
            .cmds
            .send(Command::Download { enclosure: *enc })
            .await
            .map_err(|_| anyhow::anyhow!("the daemon is not accepting commands"))?;
    }
    Ok(Json(serde_json::json!({ "queued": ids.len() })))
}

/// Subscriptions as OPML, so they can move to another podcast app.
async fn export_opml(State(state): State<WebState>) -> Result<Response, ApiError> {
    let cfg = state.ctx.cfg();
    let mut doc = opml::OPML {
        head: Some(opml::Head {
            title: Some("ipx subscriptions".into()),
            ..Default::default()
        }),
        ..Default::default()
    };
    for (id, feed) in &cfg.feeds {
        let title = state
            .ctx
            .db
            .feed_summary(id)
            .ok()
            .and_then(|s| s.title)
            .unwrap_or_else(|| id.clone());
        doc.add_feed(&title, &feed.url);
    }
    let xml = doc.to_string().map_err(|e| anyhow::anyhow!("writing OPML: {e}"))?;
    Ok((
        [
            (header::CONTENT_TYPE, "text/x-opml; charset=utf-8"),
            (
                header::CONTENT_DISPOSITION,
                "attachment; filename=\"ipx-subscriptions.opml\"",
            ),
        ],
        xml,
    )
        .into_response())
}

#[derive(Deserialize)]
struct OpmlBody {
    xml: String,
}

async fn import_opml(
    State(state): State<WebState>,
    Json(body): Json<OpmlBody>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let doc = opml::OPML::from_str(&body.xml)
        .map_err(|e| anyhow::anyhow!("that does not parse as OPML: {e}"))?;
    let mut found = vec![];
    crate::collect_outlines(&doc.body.outlines, &mut found);

    let mut cfg = (*state.ctx.cfg()).clone();
    let mut added = 0;
    for (title, url) in found {
        if cfg.feeds.values().any(|f| f.url == url) {
            continue;
        }
        let id = crate::config::unique_slug(&title, &cfg.feeds);
        cfg.feeds.insert(
            id,
            crate::config::Feed {
                url,
                folder: None,
                keywords: vec![],
                allow_explicit: false,
                auto_download: true,
                max_new_per_check: None,
                username: None,
                password: None,
                password_env: None,
            },
        );
        added += 1;
    }
    cfg.save(&state.config_path)?;
    state.ctx.reload_cfg(&state.config_path)?;
    Ok(Json(serde_json::json!({ "added": added })))
}