93 lines
3.3 KiB
Rust
93 lines
3.3 KiB
Rust
use std::collections::HashSet;
|
|
use std::time::Duration;
|
|
use crate::feed::Feed;
|
|
|
|
// timeouts are fairly low as they will be multiplied with the amount
|
|
// of sequential fetches without new posts by the scheduler.
|
|
|
|
const DEFAULT_INTERVAL: Duration = Duration::from_secs(30);
|
|
const MIN_INTERVAL: Duration = Duration::from_secs(10);
|
|
const ERROR_INTERVAL: Duration = Duration::from_secs(180);
|
|
|
|
#[derive(Debug)]
|
|
pub enum Message {
|
|
Fetched {
|
|
host: String,
|
|
new_posts: usize,
|
|
next_interval: Duration,
|
|
},
|
|
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);
|
|
}
|
|
}
|
|
|
|
// check if it's an actual post, not a repost
|
|
if let Some(author_host) = post.account.host() {
|
|
// 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::Fetched {
|
|
host,
|
|
new_posts: 0,
|
|
next_interval: ERROR_INTERVAL,
|
|
}).unwrap();
|
|
}
|
|
}
|
|
});
|
|
}
|