ipx/crates/directory/src/directory.rs

//! The running directory: its store, the tokens it accepts, the refresher that fetches its feeds, and
//! the reloader that picks up a changed config.

use crate::config::Token;
use crate::wire::{Listing, Me};
use crate::{config, store};
use reqwest::redirect::Policy;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
use store::{Refreshed, Store};
use tokio::sync::Notify;
use tracing::Instrument;

/// How often each feed is fetched again. Metadata changes slowly, and every server subscribed
/// to a feed fetches it itself anyway: every six hours was 3.9 GB of feeds a day (#243).
pub const REFRESH_EVERY: i64 = 12 * 3600;

/// The running directory: its database, the tokens it accepts and the client its refresher
/// fetches feeds with.
pub struct Directory {
    pub store: Store,
    pub(crate) tokens: RwLock<Vec<Token>>,
    pub(crate) client: reqwest::Client,
    /// Wakes the refresher, so a feed a snapshot brings in is fetched now, not in a minute.
    pub(crate) wake: Notify,
    /// The public pages' copy of the list, and when it was read.
    pub(crate) pages_list: tokio::sync::Mutex<Option<(Instant, Arc<Vec<Listing>>)>>,
}

/// How long the public pages show the same copy of the list.
pub(crate) const PAGES_LIST_FOR: Duration = Duration::from_secs(60);

/// The directory as every handler and task holds it.
pub type Shared = Arc<Directory>;

impl Directory {
    /// A directory on `store`, accepting `tokens`, whose refresher may reach the internal
    /// addresses `fetch_internal` lists and no others.
    pub fn new(
        store: Store,
        tokens: Vec<Token>,
        fetch_internal: ipx_feed::guard::Allowed,
    ) -> Result<Shared, reqwest::Error> {
        // No redirects: ipx_feed::fetch follows them itself, so it sees each hop.
        let client = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(10))
            .read_timeout(ipx_feed::FEED_SILENCE)
            .redirect(Policy::none())
            .user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION")))
            // The feeds are named by whoever holds a write token, and redirect where their
            // publishers like: nothing on this network is fetched unless listed (#329).
            .connector_layer(ipx_feed::guard::Guard::new(fetch_internal))
            .build()?;
        Ok(Arc::new(Self {
            store,
            tokens: RwLock::new(tokens),
            client,
            wake: Notify::new(),
            pages_list: Default::default(),
        }))
    }

    /// The whole list for the public pages, read from the database at most once a minute. Anyone
    /// may ask for a page, and each asked the database for the whole catalogue on the pool the
    /// refresher and every server's /v1 calls share; a query string of its own got past any
    /// cache in front (#341). The lock is held through the read, so a burst waits on one query.
    pub async fn pages_list(&self) -> anyhow::Result<Arc<Vec<Listing>>> {
        let mut held = self.pages_list.lock().await;
        if let Some((at, list)) = held.as_ref()
            && at.elapsed() < PAGES_LIST_FOR
        {
            return Ok(list.clone());
        }
        let list = Arc::new(self.store.list().await?);
        *held = Some((Instant::now(), list.clone()));
        Ok(list)
    }

    /// Takes the tokens of a config file read again, and drops the snapshots of any that went.
    pub async fn set_tokens(&self, tokens: Vec<Token>) -> anyhow::Result<u64> {
        let names: Vec<String> = tokens.iter().map(|t| t.name.clone()).collect();
        *self.tokens.write().expect("a panic elsewhere poisoned this lock") = tokens;
        self.store.keep_instances(&names).await
    }

    pub(crate) fn caller(&self, presented: &str) -> Option<Me> {
        let h = config::hash(presented);
        // Every token compared, whether or not one matched, so timing says nothing either way.
        self.tokens.read().expect("a panic elsewhere poisoned this lock").iter().fold(None, |found, t| {
            if config::same_secret(&t.sha256, &h) { Some(Me { name: t.name.clone(), access: t.access }) } else { found }
        })
    }

    /// Fetches one feed and stores what it found.
    pub async fn refresh(&self, due: &store::Due) -> anyhow::Result<()> {
        let got = match ipx_feed::fetch(&self.client, &due.url, None, due.etag.as_deref(), due.last_modified.as_deref())
            .await
        {
            Ok((ipx_feed::Fetched::NotModified, _)) => Refreshed::NotModified,
            Ok((ipx_feed::Fetched::Body { bytes, etag, last_modified }, _)) => {
                if ipx_feed::is_opml(&bytes) {
                    // ipx sends the feeds inside an OPML, never the OPML.
                    Refreshed::Failed("an OPML list, not a feed".into())
                } else {
                    match ipx_feed::parse(&bytes) {
                        Ok(parsed) => Refreshed::Feed { parsed, etag, last_modified },
                        Err(e) => Refreshed::Failed(format!("{e:#}")),
                    }
                }
            }
            Err(e) => Refreshed::Failed(format!("{e:#}")),
        };
        if let Refreshed::Failed(e) = &got {
            crate::metrics::CHECKS_FAILED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            tracing::info!(feed = %due.id, error = %ipx_feed::redact_text(e), "feed check failed");
        }
        // ponytail: a feed that moved for good is still asked for at its old address, which
        // redirects; follow the move when a feed's id can change without losing its counts.
        self.store.save(&due.id, got).await
    }
}

/// Fetches whatever is due, a few at a time, forever.
pub async fn refresher(dir: Shared) {
    loop {
        match dir.store.due(REFRESH_EVERY, 25).await {
            Ok(due) => {
                if !due.is_empty() {
                    // A round is a trace, each feed's check a span in it (#213).
                    let round = tracing::info_span!("refresh", feeds = due.len());
                    Instrument::instrument(
                        async {
                            // A few at a time: in turn, two publishers timing out held a round
                            // for a minute and more (#227).
                            futures_util::StreamExt::for_each_concurrent(futures_util::stream::iter(&due), 5, |d| {
                                let check = tracing::info_span!("feed", feed = %d.id);
                                async {
                                    if let Err(e) = Instrument::instrument(dir.refresh(d), check).await {
                                        tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check");
                                    }
                                }
                            })
                            .await;
                            // One line a round, for the dashboard's refresher panel (#210); failures have their own.
                            tracing::info!(checked = due.len(), "feeds checked");
                            crate::metrics::CHECKED.fetch_add(due.len() as u64, std::sync::atomic::Ordering::Relaxed);
                        },
                        round,
                    )
                    .await;
                    continue;
                }
            }
            Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not read which feeds are due"),
        }
        tokio::select! {
            _ = tokio::time::sleep(Duration::from_secs(60)) => {}
            _ = dir.wake.notified() => {}
        }
    }
}

/// Reads the config file again whenever it changes, checked each minute. A file that fails to
/// load leaves the tokens as they were, so a typo locks nobody out.
pub async fn reloader(dir: Shared, path: PathBuf) {
    let mtime = |p: &std::path::Path| std::fs::metadata(p).and_then(|m| m.modified()).ok();
    let mut last = mtime(&path);
    loop {
        tokio::time::sleep(Duration::from_secs(60)).await;
        let now = mtime(&path);
        if now == last {
            continue;
        }
        last = now;
        match config::Config::load(&path).and_then(|c| c.tokens()) {
            Ok(tokens) => {
                let n = tokens.len();
                match dir.set_tokens(tokens).await {
                    Ok(dropped) => tracing::info!(tokens = n, dropped, "config read again"),
                    Err(e) => {
                        tracing::warn!(error = %format!("{e:#}"), "config read again, but old snapshots could not be dropped")
                    }
                }
            }
            Err(e) => {
                tracing::warn!(error = %format!("{e:#}"), "config changed but does not load; keeping the tokens as they were")
            }
        }
    }
}

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

    async fn scratch() -> Shared {
        Directory::new(Store::scratch().await, vec![], Default::default()).unwrap()
    }

    #[tokio::test]
    async fn the_refresher_fetches_nothing_on_this_network() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { while listener.accept().await.is_ok() {} });
        let dir = scratch().await;
        let err = dir.client.get(format!("http://{addr}/feed.xml")).send().await.unwrap_err();
        assert!(format!("{err:?}").contains("internal address"), "{err:?}");
    }

    #[tokio::test]
    async fn the_public_pages_read_the_list_once_a_minute() {
        let dir = scratch().await;
        let first = dir.pages_list().await.unwrap();
        assert!(Arc::ptr_eq(&first, &dir.pages_list().await.unwrap()), "a second page asks the database nothing");
        // Once the copy is a minute old, the next page reads the list again.
        dir.pages_list.lock().await.as_mut().unwrap().0 -= PAGES_LIST_FOR;
        assert!(!Arc::ptr_eq(&first, &dir.pages_list().await.unwrap()));
    }

    #[tokio::test]
    async fn a_refresh_lists_a_feed_and_not_one_that_could_not_be_read() {
        use axum::routing::get;
        let rss = r#"<?xml version="1.0"?><rss version="2.0"><channel><title>Good Show</title>
            <item><title>One</title><guid>g1</guid><enclosure url="https://x/1.mp3" type="audio/mpeg" length="1"/></item>
            </channel></rss>"#;
        let opml =
            r#"<?xml version="1.0"?><opml version="2.0"><head/><body><outline xmlUrl="https://x/f"/></body></opml>"#;
        let app = axum::Router::new()
            .route("/good.xml", get(move || async move { rss }))
            .route("/list.opml", get(move || async move { opml }))
            .route("/garbled.xml", get(|| async { "<rss><channel" }))
            .route("/gone.xml", get(|| async { axum::http::StatusCode::NOT_FOUND }));
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let at = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await });

        let allowed = ipx_feed::guard::Allowed::parse(&["127.0.0.1".into()]).unwrap();
        let dir = Directory::new(Store::scratch().await, vec![], allowed).unwrap();
        let paths = ["/good.xml", "/list.opml", "/garbled.xml", "/gone.xml"];
        let entries: Vec<crate::SnapshotEntry> = paths
            .iter()
            .map(|p| crate::SnapshotEntry { url: format!("http://{at}{p}"), subscribers: 1, category: None })
            .collect();
        assert_eq!(dir.store.snapshot("tower", &entries).await.unwrap().accepted, 4);
        let due = dir.store.due(REFRESH_EVERY, 10).await.unwrap();
        assert_eq!(due.len(), 4);
        for d in &due {
            dir.refresh(d).await.unwrap();
        }
        assert!(dir.store.due(REFRESH_EVERY, 10).await.unwrap().is_empty(), "each was checked, failed or not");

        let page = |path: &str| {
            let d = due.iter().find(|d| d.url.ends_with(path)).unwrap();
            dir.store.feed(&d.id)
        };
        let good = page("/good.xml").await.unwrap().unwrap();
        assert_eq!(good.listing.title.as_deref(), Some("Good Show"));
        assert!(good.listing.podcast && good.failing.is_none() && good.items.len() == 1);
        for bad in ["/list.opml", "/garbled.xml", "/gone.xml"] {
            // Never read, so never listed: a page of nothing is no use to anyone browsing.
            assert!(page(bad).await.unwrap().is_none(), "{bad} is not listed");
        }
    }
}