caveman/src/worker.rs

use std::collections::HashSet;
use std::time::Duration;
use chrono::{DateTime, FixedOffset};
use crate::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,
        next_interval: Duration,
        latest_timestamp: Option<DateTime<FixedOffset>>,
    },
    Error { host: String },
    IntroduceHosts { hosts: Vec<String> },
    Posts { posts: Vec<feed::Post> },
}

pub fn fetch_and_process(
    tx: tokio::sync::mpsc::UnboundedSender<Message>,
    client: reqwest::Client,
    host: String,
) {
    let url = format!("https://{}/api/v1/timelines/public", host);
    tokio::spawn(async move {
        match feed::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()
                     .min(DEFAULT_INTERVAL)
                     .max(MIN_INTERVAL)
                } else {
                    DEFAULT_INTERVAL
                };
                let latest_timestamp = timestamps.iter().last().cloned();

                // introduce new hosts, validate posts
                let mut posts = Vec::with_capacity(feed.posts.len());
                let mut hosts = HashSet::new();
                for post in feed.posts.into_iter() {
                    if let Some(author_host) = post.account.host() {
                        if author_host == host && post.url_host().as_ref() == Some(&host) {
                            posts.push(post);
                        }
                        hosts.insert(author_host);
                    }
                }
                let hosts = hosts.into_iter().collect();
                tx.send(Message::IntroduceHosts { hosts }).unwrap();
                tx.send(Message::Posts { posts }).unwrap();

                tx.send(Message::Fetched {
                    host: host.clone(),
                    next_interval,
                    latest_timestamp,
                }).unwrap();
            }
            Err(e) => {
                println!("Failed fetching {}: {}", host, e);
                tx.send(Message::Error { host }).unwrap();
            }
        }
    });
}