irssc/
rss.rs

1use std::collections::HashSet;
2use std::sync::LazyLock;
3use std::{sync::Arc, time::Duration};
4
5use chrono::Utc;
6use feed_rs::model::Feed;
7use irc::client::Client;
8use log::{debug, error, info, trace};
9use tokio::sync::RwLock;
10
11use miette::{IntoDiagnostic, WrapErr, miette};
12
13use crate::bot::Subscriptions;
14use crate::feed_cache::CachedFeed;
15use crate::{APP_USER_AGENT, FEED_REFRESH_INTERVAL_SECONDS};
16use rand::random_range;
17
18async fn random_delay() {
19    let delay_seconds = random_range(1..=5);
20    tokio::time::sleep(Duration::from_secs(delay_seconds)).await;
21}
22
23/// Infinite loop tempered by an interval. runs [`update_feeds`] periodically
24pub(crate) async fn feed_watcher(subs: Arc<RwLock<Subscriptions>>, client: Arc<Client>) {
25    let mut feed_interval = tokio::time::interval(tokio::time::Duration::from_secs(
26        FEED_REFRESH_INTERVAL_SECONDS as u64,
27    ));
28
29    loop {
30        feed_interval.tick().await;
31        // Check if a reasonable lapse of time has occured since last check. Won't do anygood
32        if Utc::now()
33            < subs.read().await.last_check
34                + Duration::from_secs(FEED_REFRESH_INTERVAL_SECONDS as u64 / 2)
35        {
36            continue;
37        }
38        if let Err(e) = update_feeds(subs.clone(), &client).await {
39            error!("{:?}", e);
40        }
41    }
42}
43
44/// updates all feeds on a regular interval, until death.
45pub(crate) async fn update_feeds(
46    subscriptions: Arc<RwLock<Subscriptions>>,
47    client: &Client,
48) -> miette::Result<()> {
49    info!("Checking feeds updates…");
50    let subs = subscriptions.read().await;
51    let subs_clone = subs.subs.clone();
52    drop(subs);
53    for (sub_url, sub_users) in &subs_clone {
54        match update_feed(client, subscriptions.clone(), sub_url, sub_users).await {
55            Ok(_) => {
56                // great!
57            }
58            Err(e) => {
59                error!("Error while updating feed {sub_url}: {e}");
60            }
61        }
62    }
63    debug!("Feed checking finished, writing last check");
64    let mut subs = subscriptions.write().await;
65    subs.last_check = Utc::now();
66    subs.save().await?;
67    info!("Feed updates finished");
68    Ok(())
69}
70
71async fn update_feed(
72    client: &Client,
73    subs: Arc<RwLock<Subscriptions>>,
74    sub_url: &String,
75    sub_users: &HashSet<String>,
76) -> Result<(), miette::Error> {
77    let server_name = subs.read().await.server_name.clone();
78    info!("Feed update start: {sub_url}");
79    let feed = get_feed(sub_url).await?;
80    trace!("Feed parsed successfully");
81
82    let old_cached = CachedFeed::load(sub_url, &server_name)
83        .await
84        .unwrap_or(None);
85    let new_cached = CachedFeed::from_feed(sub_url, &feed, &server_name);
86    let new_articles = if let Some(old) = &old_cached {
87        trace!("Found cached feed for {}, comparing entries…", sub_url);
88        old.get_new_entries_from_feed(&feed)
89    } else {
90        trace!(
91            "No cached feed found for {}, treating all entries as read",
92            sub_url
93        );
94        vec![]
95    };
96
97    trace!("Found {} new articles for {}", new_articles.len(), sub_url);
98
99    for new in new_articles {
100        let s = format!(
101            "{}: {} {}",
102            feed.title
103                .as_ref()
104                .map(|t| t.content.as_str())
105                .unwrap_or("No title"),
106            new.title
107                .as_ref()
108                .map(|t| t.content.as_str())
109                .unwrap_or("New entry"),
110            new.links
111                .first()
112                .ok_or(miette!(
113                    "Could not find a link in entry {new:?} for feed {sub_url}"
114                ))?
115                .href
116        );
117        info!("Sending \"{s}\" to clients {sub_users:?}");
118        for u in sub_users {
119            client.send_privmsg(u, &s).into_diagnostic()?;
120        }
121        random_delay().await;
122    }
123    new_cached.save().await?;
124    info!("Feed update end: {sub_url}");
125    Ok(())
126}
127
128pub(crate) async fn get_feed(feed_url: &str) -> miette::Result<Feed> {
129    static HTTP_CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
130        reqwest::Client::builder()
131            .user_agent(APP_USER_AGENT)
132            .build()
133            .into_diagnostic()
134            .expect("Could not build http client")
135    });
136    trace!("update_feed {feed_url} making request…");
137    let response = HTTP_CLIENT
138        .get(feed_url)
139        .send()
140        .await
141        .into_diagnostic()
142        .wrap_err(format!("unable to fetch {feed_url}"))?
143        .bytes()
144        .await
145        .into_diagnostic()?;
146    trace!("update_feed: got response; parsing…");
147    let feed = feed_rs::parser::parse(response.as_ref())
148        .into_diagnostic()
149        .wrap_err("Could not parse feed")?;
150    trace!("Feed parsing OK");
151    Ok(feed)
152}
irssc/rss.rs

irssc/
rss.rs
