forked from MightyPork/group-actor
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
573 lines
20 KiB
573 lines
20 KiB
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};
|
|
use crate::command::StatusCommand;
|
|
use elefren::entities::account::Account;
|
|
|
|
mod handle_mention;
|
|
|
|
/// This is one group's config store capable of persistence
|
|
#[derive(Debug)]
|
|
pub struct GroupHandle {
|
|
pub(crate) group_account: Account,
|
|
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 = 30;
|
|
// 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 SOCKET_ALIVE_TIMEOUT: Duration = Duration::from_secs(30);
|
|
const SOCKET_RETIRE_TIME: Duration = Duration::from_secs(120);
|
|
const PING_INTERVAL: Duration = Duration::from_secs(15); // must be < periodic save!
|
|
|
|
|
|
|
|
macro_rules! grp_debug {
|
|
($self:ident, $f:expr) => {
|
|
::log::debug!(concat!("(@{}) ", $f), $self.config.get_acct());
|
|
};
|
|
($self:ident, $f:expr, $($arg:tt)+) => {
|
|
::log::debug!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
|
|
};
|
|
}
|
|
|
|
macro_rules! grp_info {
|
|
($self:ident, $f:expr) => {
|
|
::log::info!(concat!("(@{}) ", $f), $self.config.get_acct());
|
|
};
|
|
($self:ident, $f:expr, $($arg:tt)+) => {
|
|
::log::info!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
|
|
};
|
|
}
|
|
|
|
macro_rules! grp_trace {
|
|
($self:ident, $f:expr) => {
|
|
::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct());
|
|
};
|
|
($self:ident, $f:expr, $($arg:tt)+) => {
|
|
::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
|
|
};
|
|
}
|
|
|
|
macro_rules! grp_warn {
|
|
($self:ident, $f:expr) => {
|
|
::log::warn!(concat!("(@{}) ", $f), $self.config.get_acct());
|
|
};
|
|
($self:ident, $f:expr, $($arg:tt)+) => {
|
|
::log::warn!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
|
|
};
|
|
}
|
|
|
|
macro_rules! grp_error {
|
|
($self:ident, $f:expr) => {
|
|
::log::error!(concat!("(@{}) ", $f), $self.config.get_acct());
|
|
};
|
|
($self:ident, $f:expr, $($arg:tt)+) => {
|
|
::log::error!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
|
|
};
|
|
}
|
|
|
|
|
|
|
|
impl GroupHandle {
|
|
pub async fn save(&mut self) -> Result<(), GroupError> {
|
|
grp_debug!(self, "Saving group config & status");
|
|
self.store.set_group_config(self.config.clone()).await?;
|
|
grp_trace!(self, "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
|
|
}
|
|
}
|
|
|
|
macro_rules! grp_trace {
|
|
($self:ident, $f:expr) => {
|
|
::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct());
|
|
};
|
|
($self:ident, $f:expr, $($arg:tt)+) => {
|
|
::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
|
|
};
|
|
}
|
|
|
|
impl GroupHandle {
|
|
pub async fn run(&mut self) -> Result<(), GroupError> {
|
|
loop {
|
|
match self.run_internal().await {
|
|
Ok(()) => unreachable!(),
|
|
Err(e @ GroupError::BadConfig(_)) => {
|
|
grp_error!(self, "ERROR in group handler, aborting! {}", e);
|
|
return Err(e);
|
|
}
|
|
Err(other) => {
|
|
grp_error!(self, "ERROR in group handler, will restart! {}", other);
|
|
tokio::time::sleep(Duration::from_secs(5)).await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn run_internal(&mut self) -> Result<(), GroupError> {
|
|
assert!(PERIODIC_SAVE >= PING_INTERVAL);
|
|
|
|
loop {
|
|
grp_debug!(self, "Opening streaming API socket");
|
|
let mut next_save = Instant::now() + PERIODIC_SAVE; // so we save at start
|
|
|
|
// wrapped in a timeout, this seems like the only place the group could hang
|
|
// (https://git.ondrovo.com/MightyPork/group-actor/issues/8)
|
|
let mut events = match tokio::time::timeout(Duration::from_secs(3), self.client.streaming_user()).await {
|
|
Ok(Ok(events)) => events,
|
|
Ok(Err(e)) => return Err(e.into()),
|
|
Err(_) => {
|
|
return Err(GroupError::ApiTimeout);
|
|
}
|
|
};
|
|
|
|
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) => {
|
|
grp_debug!(self, "Some missed notifs handled");
|
|
}
|
|
Ok(false) => {
|
|
grp_debug!(self, "No notifs missed");
|
|
}
|
|
Err(e) => {
|
|
grp_error!(self, "Failed to handle missed notifs: {}", e);
|
|
}
|
|
}
|
|
|
|
match self.catch_up_with_missed_statuses().await {
|
|
Ok(true) => {
|
|
grp_debug!(self, "Some missed statuses handled");
|
|
}
|
|
Ok(false) => {
|
|
grp_debug!(self, "No statuses missed");
|
|
}
|
|
Err(e) => {
|
|
grp_error!(self, "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() {
|
|
grp_trace!(self, "Save time elapsed, saving if needed");
|
|
self.save_if_needed().await.log_error("Failed to save group");
|
|
next_save = Instant::now() + PERIODIC_SAVE;
|
|
}
|
|
|
|
let remains_to_idle_close = SOCKET_ALIVE_TIMEOUT.saturating_sub(last_rx.elapsed());
|
|
let remains_to_retire = SOCKET_RETIRE_TIME.saturating_sub(socket_open_time.elapsed());
|
|
|
|
if remains_to_idle_close.is_zero() {
|
|
grp_warn!(self, "Socket idle too long, close");
|
|
break 'rx;
|
|
}
|
|
if remains_to_retire.is_zero() {
|
|
grp_debug!(self, "Socket open too long, closing");
|
|
break 'rx;
|
|
}
|
|
|
|
let timeout = next_save
|
|
.saturating_duration_since(Instant::now())
|
|
.min(remains_to_idle_close)
|
|
.min(remains_to_retire)
|
|
.max(Duration::from_secs(1)); // at least 1s
|
|
|
|
grp_debug!(self, "Wait for message {:?}", timeout);
|
|
match tokio::time::timeout(timeout, events.next()).await {
|
|
Ok(Some(event)) => {
|
|
last_rx = Instant::now();
|
|
grp_debug!(self, "(@{}) 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) => {
|
|
grp_warn!(self, "Group @{} socket closed, restarting...", self.config.get_acct());
|
|
break 'rx;
|
|
}
|
|
Err(_) => {
|
|
// Timeout so we can save if needed
|
|
}
|
|
}
|
|
}
|
|
|
|
grp_warn!(self, "Notif stream closed, will reopen");
|
|
tokio::time::sleep(DELAY_REOPEN_STREAM).await;
|
|
}
|
|
}
|
|
|
|
async fn handle_notification(&mut self, n: Notification) -> Result<(), GroupError> {
|
|
grp_debug!(self, "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 {
|
|
grp_debug!(self, "This is our post, ignore that");
|
|
return Ok(());
|
|
}
|
|
|
|
if self.config.is_banned(¬if_acct) {
|
|
grp_warn!(self, "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 => {
|
|
grp_info!(self, "New follower!");
|
|
// Just greet the user always
|
|
self.handle_new_follow(¬if_acct, notif_user_id).await;
|
|
}
|
|
NotificationType::Favourite => {}
|
|
NotificationType::Reblog => {}
|
|
NotificationType::Other(_) => {}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Handle a non-mention status for tags
|
|
async fn handle_status(&mut self, s: Status) -> Result<(), GroupError> {
|
|
grp_debug!(self, "Handling status #{}", s.id);
|
|
let ts = s.timestamp_millis();
|
|
self.config.set_last_status(ts);
|
|
|
|
let private = s.visibility.is_private();
|
|
let has_hashtags = s.content.contains('#');
|
|
let group_user = self.config.get_acct();
|
|
let status_user = normalize_acct(&s.account.acct, group_user)?;
|
|
let member_or_admin = self.config.is_member_or_admin(&status_user);
|
|
let commands = crate::command::parse_slash_commands(&s.content);
|
|
|
|
if status_user == group_user {
|
|
grp_debug!(self, "This is our post, discard");
|
|
return Ok(());
|
|
}
|
|
|
|
if self.config.is_banned(&status_user) {
|
|
grp_debug!(self, "Status author @{} is banned, discard", status_user);
|
|
return Ok(());
|
|
}
|
|
|
|
if self.config.is_optout(&status_user) && !member_or_admin
|
|
{
|
|
grp_debug!(self, "Status author @{} opted out, discard", status_user);
|
|
return Ok(());
|
|
}
|
|
|
|
if commands.contains(&StatusCommand::Ignore) {
|
|
grp_debug!(self, "Post has IGNORE command, discard");
|
|
return Ok(());
|
|
}
|
|
|
|
// Sometimes notifications don't work, but we see the mentions as statuses
|
|
for m in &s.mentions {
|
|
let mentioned_user = normalize_acct(&m.acct, group_user)?;
|
|
if mentioned_user == group_user {
|
|
let notif_time = self.config.get_last_notif();
|
|
if notif_time <= ts {
|
|
grp_debug!(self, "mentioned but status is older than last notif, can't be a valid notif, discard");
|
|
return Ok(());
|
|
} else {
|
|
if !commands.is_empty() {
|
|
grp_debug!(self, "Detected commands for this group, handle as notif");
|
|
|
|
return self.handle_notification(Notification {
|
|
id: s.id.clone(), // ???
|
|
notification_type: NotificationType::Mention,
|
|
created_at: s.created_at.clone(),
|
|
account: s.account.clone(),
|
|
status: Some(s)
|
|
}).await;
|
|
} else {
|
|
if !private {
|
|
grp_debug!(self, "Detected mention status, handle as notif");
|
|
} else {
|
|
grp_debug!(self, "mention in private without commands, discard, this is nothing");
|
|
return Ok(());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// optout does not work for members and admins, so don't check it
|
|
|
|
if !member_or_admin {
|
|
grp_debug!(self, "Status author @{} is not a member, discard", status_user);
|
|
return Ok(());
|
|
}
|
|
|
|
if private {
|
|
grp_debug!(self, "Status is private, discard");
|
|
return Ok(());
|
|
}
|
|
|
|
if !has_hashtags {
|
|
grp_debug!(self, "No hashtags, discard");
|
|
return Ok(());
|
|
}
|
|
|
|
let tags = crate::command::parse_status_tags(&s.content);
|
|
grp_debug!(self, "Tags in status: {:?}", tags);
|
|
|
|
'tags: for t in tags {
|
|
if self.config.is_tag_followed(&t) {
|
|
grp_info!(self, "REBLOG #{} STATUS", t);
|
|
self.client.reblog(&s.id).await
|
|
.log_error("Failed to reblog");
|
|
break 'tags; // do not reblog multiple times!
|
|
} else {
|
|
grp_debug!(self, "#{} is not a group tag", t);
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
grp_debug!(self, "Inspecting notif {}", NotificationDisplay(&n));
|
|
notifs_to_handle.push(n);
|
|
num += 1;
|
|
if num > MAX_CATCHUP_NOTIFS {
|
|
grp_warn!(self, "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();
|
|
|
|
grp_debug!(self, "{} notifications to catch up!", notifs_to_handle.len());
|
|
|
|
for n in notifs_to_handle {
|
|
grp_debug!(self, "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)
|
|
}
|
|
|
|
grp_debug!(self, "Inspecting status {}", StatusDisplay(&s));
|
|
|
|
if newest_status.is_none() {
|
|
newest_status = Some(ts);
|
|
}
|
|
|
|
statuses_to_handle.push(s);
|
|
num += 1;
|
|
if num > MAX_CATCHUP_STATUSES {
|
|
grp_warn!(self, "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() {
|
|
grp_debug!(self, "No statuses to handle");
|
|
return Ok(false);
|
|
}
|
|
|
|
statuses_to_handle.reverse();
|
|
|
|
grp_debug!(self, "{} statuses to catch up!", statuses_to_handle.len());
|
|
|
|
for s in statuses_to_handle {
|
|
grp_debug!(self, "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 to the group! 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;
|
|
|
|
self.config.set_member(notif_acct, true)
|
|
.log_error("Fail add a member");
|
|
|
|
format!("\
|
|
@{user} Welcome to the group! The group user will now follow you back to complete the sign-up. \
|
|
To share a post, @ the group user or use a group hashtag.\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");
|
|
}
|
|
}
|
|
}
|
|
|