caveman/hunter/src/worker.rs

use std::collections::HashSet;
use std::time::Duration;
use crate::feed::Feed;

const DEFAULT_INTERVAL: Duration = Duration::from_secs(3600);
const MIN_INTERVAL: Duration = Duration::from_secs(10);

#[derive(Debug)]
pub enum Message {
    Fetched {
        host: String,
        new_posts: usize,
        next_interval: Duration,
    },
    Error { host: String },
    IntroduceHosts { hosts: Vec<String> },
}

pub fn fetch_and_process(
    message_tx: tokio::sync::mpsc::UnboundedSender<Message>,
    mut redis_man: redis::aio::ConnectionManager,
    client: reqwest::Client,
    host: String,
) {
    let url = format!("https://{}/api/v1/timelines/public", host);
    tokio::spawn(async move {
        match Feed::fetch(&client, &url).await {
            Ok(feed) => {
                // Analyze time intervals between posts to estimate when to fetch next
                let mut timestamps = feed.posts.iter()
                    .filter_map(|post| post.timestamp())
                    .collect::<Vec<_>>();
                timestamps.sort();
                let next_interval = if timestamps.len() > 1 {
                    ((*timestamps.last().unwrap() - timestamps[0]) / (timestamps.len() as i32)
                    ).to_std().unwrap_or(DEFAULT_INTERVAL)
                     .min(DEFAULT_INTERVAL)
                     .max(MIN_INTERVAL)
                } else {
                    DEFAULT_INTERVAL
                };

                // introduce new hosts, validate posts
                let mut hosts = HashSet::new();
                let mut new_posts = 0;
                for post in feed.posts.into_iter() {
                    // introduce instances from mentions
                    for mention in &post.mentions {
                        if let Some(user_host) = mention.user_host() {
                            hosts.insert(user_host);
                        }
                    }

                    if let Some(author_host) = post.account.host() {
                        if author_host == host && post.url_host().map(|s| s == host).unwrap_or(false) {
                            // send away to redis
                            if crate::redis_store::save_post(&mut redis_man, &host, post).await {
                                new_posts += 1;
                            }
                        }

                        // introduce instances from authors
                        hosts.insert(author_host);
                    }
                }
                let hosts = hosts.into_iter()
                    .map(|host| host.to_owned())
                    .collect();
                let _ = message_tx.send(Message::IntroduceHosts { hosts });

                // successfully fetched, save for future run
                crate::redis_store::save_host(&mut redis_man, &host).await;

                message_tx.send(Message::Fetched {
                    host: host.clone(),
                    new_posts,
                    next_interval,
                }).unwrap();
            }
            Err(e) => {
                log::error!("Failed fetching {}: {}", host, e);
                message_tx.send(Message::Error { host }).unwrap();
            }
        }
    });
}