forked from MightyPork/group-actor
v0.2, refactor, improve some messages, fix lints
This commit is contained in:
@@ -0,0 +1,454 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use elefren::{FediClient, StatusBuilder};
|
||||
use elefren::debug::EventDisplay;
|
||||
use elefren::debug::NotificationDisplay;
|
||||
use elefren::debug::StatusDisplay;
|
||||
use elefren::entities::event::Event;
|
||||
use elefren::entities::notification::{Notification, NotificationType};
|
||||
use elefren::entities::status::Status;
|
||||
use elefren::status_builder::Visibility;
|
||||
use futures::StreamExt;
|
||||
|
||||
use handle_mention::ProcessMention;
|
||||
|
||||
use crate::error::GroupError;
|
||||
use crate::store::ConfigStore;
|
||||
use crate::store::data::GroupConfig;
|
||||
use crate::utils::{LogError, normalize_acct, VisExt};
|
||||
|
||||
mod handle_mention;
|
||||
|
||||
/// This is one group's config store capable of persistence
|
||||
#[derive(Debug)]
|
||||
pub struct GroupHandle {
|
||||
pub(crate) client: FediClient,
|
||||
pub(crate) config: GroupConfig,
|
||||
pub(crate) store: Arc<ConfigStore>,
|
||||
}
|
||||
|
||||
// const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250);
|
||||
const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(500);
|
||||
const MAX_CATCHUP_NOTIFS: usize = 25;
|
||||
// also statuses
|
||||
const MAX_CATCHUP_STATUSES: usize = 50;
|
||||
// higher because we can expect a lot of non-hashtag statuses here
|
||||
const PERIODIC_SAVE: Duration = Duration::from_secs(60);
|
||||
const PING_INTERVAL: Duration = Duration::from_secs(15); // must be < periodic save!
|
||||
|
||||
impl GroupHandle {
|
||||
pub async fn save(&mut self) -> Result<(), GroupError> {
|
||||
debug!("Saving group config & status");
|
||||
self.store.set_group_config(self.config.clone()).await?;
|
||||
trace!("Saved");
|
||||
self.config.clear_dirty_status();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn save_if_needed(&mut self) -> Result<(), GroupError> {
|
||||
if self.config.is_dirty() {
|
||||
self.save().await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/*
|
||||
pub async fn reload(&mut self) -> Result<(), GroupError> {
|
||||
if let Some(g) = self.store.get_group_config(self.config.get_acct()).await {
|
||||
self.config = g;
|
||||
Ok(())
|
||||
} else {
|
||||
Err(GroupError::GroupNotExist)
|
||||
}
|
||||
}
|
||||
*/
|
||||
}
|
||||
|
||||
trait NotifTimestamp {
|
||||
fn timestamp_millis(&self) -> u64;
|
||||
}
|
||||
|
||||
impl NotifTimestamp for Notification {
|
||||
fn timestamp_millis(&self) -> u64 {
|
||||
self.created_at.timestamp_millis().max(0) as u64
|
||||
}
|
||||
}
|
||||
|
||||
impl NotifTimestamp for Status {
|
||||
fn timestamp_millis(&self) -> u64 {
|
||||
// this may not work well for unseen status tracking,
|
||||
// if ancient statuses were to appear in the timeline :(
|
||||
self.created_at.timestamp_millis().max(0) as u64
|
||||
}
|
||||
}
|
||||
|
||||
impl GroupHandle {
|
||||
pub async fn run(&mut self) -> Result<(), GroupError> {
|
||||
assert!(PERIODIC_SAVE >= PING_INTERVAL);
|
||||
|
||||
loop {
|
||||
debug!("Opening streaming API socket");
|
||||
let mut next_save = Instant::now() + PERIODIC_SAVE; // so we save at start
|
||||
let mut events = self.client.streaming_user().await?;
|
||||
let socket_open_time = Instant::now();
|
||||
let mut last_rx = Instant::now();
|
||||
let mut last_ping = Instant::now();
|
||||
|
||||
match self.catch_up_with_missed_notifications().await {
|
||||
Ok(true) => {
|
||||
debug!("Some missed notifs handled");
|
||||
}
|
||||
Ok(false) => {
|
||||
debug!("No notifs missed");
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to handle missed notifs: {}", e);
|
||||
}
|
||||
}
|
||||
|
||||
match self.catch_up_with_missed_statuses().await {
|
||||
Ok(true) => {
|
||||
debug!("Some missed statuses handled");
|
||||
}
|
||||
Ok(false) => {
|
||||
debug!("No statuses missed");
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to handle missed statuses: {}", e);
|
||||
}
|
||||
}
|
||||
|
||||
if self.config.is_dirty() {
|
||||
// save asap
|
||||
next_save = Instant::now() - PERIODIC_SAVE
|
||||
}
|
||||
|
||||
'rx: loop {
|
||||
if next_save < Instant::now() {
|
||||
trace!("Save time elapsed, saving if needed");
|
||||
self.save_if_needed().await.log_error("Failed to save group");
|
||||
next_save = Instant::now() + PERIODIC_SAVE;
|
||||
}
|
||||
|
||||
if last_rx.elapsed() > PING_INTERVAL * 2 {
|
||||
warn!("Socket idle too long, close");
|
||||
break 'rx;
|
||||
}
|
||||
|
||||
if socket_open_time.elapsed() > Duration::from_secs(120) {
|
||||
debug!("Socket open too long, closing");
|
||||
break 'rx;
|
||||
}
|
||||
|
||||
trace!("Waiting for message");
|
||||
let timeout = next_save
|
||||
.saturating_duration_since(Instant::now())
|
||||
.min(PING_INTERVAL)
|
||||
.max(Duration::from_secs(1));
|
||||
|
||||
match tokio::time::timeout(timeout, events.next()).await {
|
||||
Ok(Some(event)) => {
|
||||
last_rx = Instant::now();
|
||||
debug!("(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event));
|
||||
match event {
|
||||
Event::Update(status) => {
|
||||
self.handle_status(status).await.log_error("Error handling a status");
|
||||
}
|
||||
Event::Notification(n) => {
|
||||
self.handle_notification(n).await.log_error("Error handling a notification");
|
||||
}
|
||||
Event::Delete(_id) => {}
|
||||
Event::FiltersChanged => {}
|
||||
Event::Heartbeat => {}
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
warn!("Group @{} socket closed, restarting...", self.config.get_acct());
|
||||
break 'rx;
|
||||
}
|
||||
Err(_) => {
|
||||
// Timeout so we can save if needed
|
||||
}
|
||||
}
|
||||
|
||||
if last_ping.elapsed() > PING_INTERVAL {
|
||||
last_ping = Instant::now();
|
||||
trace!("Pinging");
|
||||
if events.send_ping()
|
||||
.await.is_err() {
|
||||
break 'rx;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
warn!("Notif stream closed, will reopen");
|
||||
tokio::time::sleep(DELAY_REOPEN_STREAM).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_notification(&mut self, n: Notification) -> Result<(), GroupError> {
|
||||
debug!("Handling notif #{}", n.id);
|
||||
let ts = n.timestamp_millis();
|
||||
self.config.set_last_notif(ts);
|
||||
|
||||
let group_acct = self.config.get_acct().to_string();
|
||||
let notif_user_id = &n.account.id;
|
||||
let notif_acct = normalize_acct(&n.account.acct, &group_acct)?;
|
||||
|
||||
if notif_acct == group_acct {
|
||||
debug!("This is our post, ignore that");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if self.config.is_banned(¬if_acct) {
|
||||
warn!("Notification actor {} is banned!", notif_acct);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
match n.notification_type {
|
||||
NotificationType::Mention => {
|
||||
if let Some(status) = n.status {
|
||||
self.handle_mention_status(status).await?;
|
||||
}
|
||||
}
|
||||
NotificationType::Follow => {
|
||||
info!("New follower!");
|
||||
|
||||
// Just greet the user always
|
||||
self.handle_new_follow(¬if_acct, notif_user_id).await;
|
||||
|
||||
// if self.config.is_member_or_admin(¬if_acct) {
|
||||
// // Already joined, just doing something silly, ignore this
|
||||
// debug!("User already a member, ignoring");
|
||||
// } else {
|
||||
//
|
||||
// }
|
||||
}
|
||||
NotificationType::Favourite => {}
|
||||
NotificationType::Reblog => {}
|
||||
NotificationType::Other(_) => {}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle a non-mention status
|
||||
async fn handle_status(&mut self, s: Status) -> Result<(), GroupError> {
|
||||
debug!("Handling status #{}", s.id);
|
||||
let ts = s.timestamp_millis();
|
||||
self.config.set_last_status(ts);
|
||||
|
||||
if s.visibility.is_private() {
|
||||
debug!("Status is direct/private, discard");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if !s.content.contains('#') {
|
||||
debug!("No tags in status, discard");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let group_user = self.config.get_acct();
|
||||
let status_user = normalize_acct(&s.account.acct, group_user)?;
|
||||
|
||||
if status_user == group_user {
|
||||
debug!("This is our post, discard");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if s.content.contains("/add ")
|
||||
|| s.content.contains("/remove ")
|
||||
|| s.content.contains("\\add ")
|
||||
|| s.content.contains("\\remove ")
|
||||
{
|
||||
debug!("Looks like a hashtag manipulation command, discard");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if self.config.is_banned(&status_user) {
|
||||
debug!("Status author @{} is banned.", status_user);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if !self.config.is_member_or_admin(&status_user) {
|
||||
debug!("Status author @{} is not a member.", status_user);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let tags = crate::command::parse_status_tags(&s.content);
|
||||
debug!("Tags in status: {:?}", tags);
|
||||
|
||||
'tags: for t in tags {
|
||||
if self.config.is_tag_followed(&t) {
|
||||
info!("REBLOG #{} STATUS", &t);
|
||||
self.client.reblog(&s.id).await
|
||||
.log_error("Failed to reblog");
|
||||
break 'tags; // do not reblog multiple times!
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn follow_user(&mut self, id: &str) -> Result<(), GroupError> {
|
||||
self.client.follow(id).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Catch up with missed notifications, returns true if any were handled
|
||||
async fn catch_up_with_missed_notifications(&mut self) -> Result<bool, GroupError> {
|
||||
let last_notif = self.config.get_last_notif();
|
||||
|
||||
let notifications = self.client.notifications().await?;
|
||||
let mut iter = notifications.items_iter();
|
||||
|
||||
let mut notifs_to_handle = vec![];
|
||||
|
||||
// They are retrieved newest first, but we want oldest first for chronological handling
|
||||
|
||||
let mut num = 0;
|
||||
while let Some(n) = iter.next_item().await {
|
||||
let ts = n.timestamp_millis();
|
||||
if ts <= last_notif {
|
||||
break; // reached our last seen notif
|
||||
}
|
||||
|
||||
debug!("Inspecting notif {}", NotificationDisplay(&n));
|
||||
notifs_to_handle.push(n);
|
||||
num += 1;
|
||||
if num > MAX_CATCHUP_NOTIFS {
|
||||
warn!("Too many notifs missed to catch up!");
|
||||
break;
|
||||
}
|
||||
|
||||
// sleep so we dont make the api angry
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
|
||||
if notifs_to_handle.is_empty() {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
notifs_to_handle.reverse();
|
||||
|
||||
debug!("{} notifications to catch up!", notifs_to_handle.len());
|
||||
|
||||
for n in notifs_to_handle {
|
||||
debug!("Handling missed notification: {}", NotificationDisplay(&n));
|
||||
self.handle_notification(n).await.log_error("Error handling a notification");
|
||||
}
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Catch up with missed statuses, returns true if any were handled
|
||||
async fn catch_up_with_missed_statuses(&mut self) -> Result<bool, GroupError> {
|
||||
let last_status = self.config.get_last_status();
|
||||
|
||||
let statuses = self.client.get_home_timeline().await?;
|
||||
let mut iter = statuses.items_iter();
|
||||
|
||||
let mut statuses_to_handle = vec![];
|
||||
|
||||
// They are retrieved newest first, but we want oldest first for chronological handling
|
||||
|
||||
let mut newest_status = None;
|
||||
|
||||
let mut num = 0;
|
||||
while let Some(s) = iter.next_item().await {
|
||||
let ts = s.timestamp_millis();
|
||||
if ts <= last_status {
|
||||
break; // reached our last seen status (hopefully there arent any retro-bumped)
|
||||
}
|
||||
|
||||
debug!("Inspecting status {}", StatusDisplay(&s));
|
||||
|
||||
if newest_status.is_none() {
|
||||
newest_status = Some(ts);
|
||||
}
|
||||
|
||||
if s.content.contains('#') && !s.visibility.is_private() {
|
||||
statuses_to_handle.push(s);
|
||||
}
|
||||
num += 1;
|
||||
if num > MAX_CATCHUP_STATUSES {
|
||||
warn!("Too many statuses missed to catch up!");
|
||||
break;
|
||||
}
|
||||
|
||||
// sleep so we dont make the api angry
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
|
||||
if let Some(ts) = newest_status {
|
||||
self.config.set_last_status(ts);
|
||||
}
|
||||
|
||||
if statuses_to_handle.is_empty() {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
statuses_to_handle.reverse();
|
||||
|
||||
debug!("{} statuses to catch up!", statuses_to_handle.len());
|
||||
|
||||
for s in statuses_to_handle {
|
||||
debug!("Handling missed status: {}", StatusDisplay(&s));
|
||||
self.handle_status(s).await
|
||||
.log_error("Error handling a status");
|
||||
}
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
async fn handle_mention_status(&mut self, status: Status) -> Result<(), GroupError> {
|
||||
let res = ProcessMention::run(self, status).await;
|
||||
|
||||
self.save_if_needed().await
|
||||
.log_error("Failed to save");
|
||||
|
||||
res
|
||||
}
|
||||
|
||||
async fn handle_new_follow(&mut self, notif_acct: &str, notif_user_id: &str) {
|
||||
let mut follow_back = false;
|
||||
let text = if self.config.is_member_only() {
|
||||
// Admins are listed without @, so they won't become handles here.
|
||||
// Tagging all admins would be annoying.
|
||||
let mut admins = self.config.get_admins().cloned().collect::<Vec<_>>();
|
||||
admins.sort();
|
||||
|
||||
format!("\
|
||||
@{user} Welcome! This group has posting restricted to members. \
|
||||
If you'd like to join, please ask one of the group admins:\n\
|
||||
{admins}",
|
||||
user = notif_acct,
|
||||
admins = admins.join(", ")
|
||||
)
|
||||
} else {
|
||||
follow_back = true;
|
||||
format!("\
|
||||
@{user} Welcome to the group! The group user will now follow you back to complete the sign-up. \
|
||||
To share a post, tag the group user or use one of the group hashtags.\n\n\
|
||||
Use /help for more info.",
|
||||
user = notif_acct
|
||||
)
|
||||
};
|
||||
|
||||
let post = StatusBuilder::new()
|
||||
.status(text)
|
||||
.content_type("text/markdown")
|
||||
.visibility(Visibility::Direct)
|
||||
.build()
|
||||
.expect("error build status");
|
||||
|
||||
self.client.new_status(post).await
|
||||
.log_error("Failed to post");
|
||||
|
||||
if follow_back {
|
||||
self.follow_user(notif_user_id).await
|
||||
.log_error("Failed to follow back");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user