use std::collections::HashSet; use std::sync::Arc; use std::time::{Duration, Instant}; use elefren::{FediClient, StatusBuilder}; use elefren::debug::EventDisplay; use elefren::debug::NotificationDisplay; use elefren::entities::event::Event; use elefren::entities::notification::{Notification, NotificationType}; use elefren::status_builder::Visibility; use futures::StreamExt; use crate::command::StatusCommand; use crate::error::GroupError; use crate::store::ConfigStore; use crate::store::data::GroupConfig; use crate::utils::{LogError, normalize_acct}; /// 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, } const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250); const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(1000); const MAX_CATCHUP_NOTIFS: usize = 25; 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?; 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 GroupHandle { pub async fn run(&mut self) -> Result<(), GroupError> { assert!(PERIODIC_SAVE >= PING_INTERVAL); let mut next_save = Instant::now() + PERIODIC_SAVE; // so we save at start loop { debug!("Opening streaming API socket"); let mut events = self.client.streaming_user().await?; match self.catch_up_with_missed_notifications().await { Ok(true) => { debug!("Some missed notifs handled"); // Save asap! next_save = Instant::now() - PERIODIC_SAVE } Ok(false) => { debug!("No notifs missed"); } Err(e) => { error!("Failed to handle missed notifs: {}", e); } } loop { if next_save < Instant::now() { self.save_if_needed().await .log_error("Failed to save group"); next_save = Instant::now() + PERIODIC_SAVE; } 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)) => { debug!("(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event)); match event { Event::Update(_status) => {} Event::Notification(n) => { self.handle_notification(n).await .log_error("Error handling a notification"); } Event::Delete(_id) => {} Event::FiltersChanged => {} } } Ok(None) => { warn!("Group @{} socket closed, restarting...", self.config.get_acct()); break; } Err(_) => { // Timeout so we can save if needed } } trace!("Pinging"); events.send_ping().await.log_error("Fail to send ping"); } 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_acct = normalize_acct(&n.account.acct, &group_acct)?; let can_write = self.config.can_write(¬if_acct); let is_admin = self.config.is_admin(¬if_acct); 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 { let status_acct = normalize_acct(&status.account.acct, &group_acct)?; if self.config.is_banned(&status_acct) { warn!("Status author {} is banned!", status_acct); return Ok(()); } let commands = crate::command::parse_status(&status.content); let mut replies = vec![]; let mut announcements = vec![]; let mut do_boost_prev_post = false; // let mut new_members = vec![]; // let mut new_admins = vec![]; // let mut removed_admins = vec![]; // let mut instance_ban_announcements = vec![]; // let mut instance_unban_announcements = vec![]; let mut any_admin_cmd = false; if commands.is_empty() { debug!("No commands in post"); if status.in_reply_to_id.is_none() { if can_write { // Someone tagged the group in OP, boost it. info!("Boosting OP mention"); tokio::time::sleep(DELAY_BEFORE_ACTION).await; self.client.reblog(&status.id).await .log_error("Failed to boost"); } else { replies.push(format!("You are not allowed to post to this group")); } } else { debug!("Not OP, ignore mention"); } } else { for cmd in commands { match cmd { // ignore is first on purpose StatusCommand::Ignore => { debug!("Notif ignored because of ignore command"); return Ok(()); } StatusCommand::Announce(a) => { debug!("Sending PSA"); announcements.push(a); } StatusCommand::Boost => { if can_write { do_boost_prev_post = status.in_reply_to_id.is_some(); } else { replies.push(format!("You are not allowed to share to this group")); } } StatusCommand::BanUser(u) => { let u = normalize_acct(&u, &group_acct)?; if is_admin { if !self.config.is_banned(&u) { match self.config.ban_user(&u, true) { Ok(_) => { any_admin_cmd = true; replies.push(format!("User {} banned from group!", u)); // no announcement here } Err(e) => { replies.push(format!("Failed to ban user {}: {}", u, e)); } } } } else { replies.push(format!("Only admins can manage user bans")); } } StatusCommand::UnbanUser(u) => { let u = normalize_acct(&u, &group_acct)?; if is_admin { if self.config.is_banned(&u) { match self.config.ban_user(&u, false) { Ok(_) => { any_admin_cmd = true; replies.push(format!("User {} un-banned!", u)); // no announcement here } Err(e) => { unreachable!() } } } } else { replies.push(format!("Only admins can manage user bans")); } } StatusCommand::BanServer(s) => { if is_admin { if !self.config.is_server_banned(&s) { match self.config.ban_server(&s, true) { Ok(_) => { any_admin_cmd = true; announcements.push(format!("Server \"{}\" has been banned.", s)); replies.push(format!("Server {} banned from group!", s)); } Err(e) => { replies.push(format!("Failed to ban server {}: {}", s, e)); } } } } else { replies.push(format!("Only admins can manage server bans")); } } StatusCommand::UnbanServer(s) => { if is_admin { if self.config.is_server_banned(&s) { match self.config.ban_server(&s, false) { Ok(_) => { any_admin_cmd = true; announcements.push(format!("Server \"{}\" has been un-banned.", s)); replies.push(format!("Server {} un-banned!", s)); } Err(e) => { unreachable!() } } } } else { replies.push(format!("Only admins can manage server bans")); } } StatusCommand::AddMember(u) => { let u = normalize_acct(&u, &group_acct)?; if is_admin { if !self.config.is_member(&u) { match self.config.set_member(&u, true) { Ok(_) => { any_admin_cmd = true; replies.push(format!("User {} added to the group!", u)); if self.config.is_member_only() { announcements.push(format!("Welcome new member @{} to the group!", u)); } } Err(e) => { replies.push(format!("Failed to add user {} to group: {}", u, e)); } } } } else { replies.push(format!("Only admins can manage members")); } } StatusCommand::RemoveMember(u) => { let u = normalize_acct(&u, &group_acct)?; if is_admin { if self.config.is_member(&u) { match self.config.set_member(&u, false) { Ok(_) => { any_admin_cmd = true; replies.push(format!("User {} removed from the group.", u)); } Err(e) => { unreachable!() } } } } else { replies.push(format!("Only admins can manage members")); } } StatusCommand::GrantAdmin(u) => { let u = normalize_acct(&u, &group_acct)?; if is_admin { if !self.config.is_admin(&u) { match self.config.set_admin(&u, true) { Ok(_) => { any_admin_cmd = true; replies.push(format!("User {} is now a group admin!", u)); announcements.push(format!("User @{} can now manage this group!", u)); } Err(e) => { replies.push(format!("Failed to make user {} a group admin: {}", u, e)); } } } } else { replies.push(format!("Only admins can manage admins")); } } StatusCommand::RemoveAdmin(u) => { let u = normalize_acct(&u, &group_acct)?; if is_admin { if self.config.is_admin(&u) { match self.config.set_admin(&u, false) { Ok(_) => { any_admin_cmd = true; replies.push(format!("User {} is no longer a group admin!", u)); announcements.push(format!("User @{} no longer manages this group.", u)); } Err(e) => { replies.push(format!("Failed to revoke {}'s group admin: {}", u, e)); } } } } else { replies.push(format!("Only admins can manage admins")); } } StatusCommand::OpenGroup => { if is_admin { if self.config.is_member_only() { any_admin_cmd = true; self.config.set_member_only(false); replies.push(format!("Group changed to open-access")); announcements.push(format!("This group is now open-access!")); } } else { replies.push(format!("Only admins can set group options")); } } StatusCommand::CloseGroup => { if is_admin { if !self.config.is_member_only() { any_admin_cmd = true; self.config.set_member_only(true); replies.push(format!("Group changed to member-only")); announcements.push(format!("This group is now member-only!")); } } else { replies.push(format!("Only admins can set group options")); } } StatusCommand::Help => { replies.push("Mention the group user in a top-level post to share it with the group's members.".to_string()); replies.push("Posts with commands won't be shared. Supported commands:".to_string()); replies.push("/ignore, /ign, /i - don't run any commands in the post".to_string()); replies.push("/boost, /b - boost the replied-to post into the group".to_string()); replies.push("/leave - leave the group as a member".to_string()); if is_admin { replies.push("/members".to_string()); replies.push("/kick, /remove user - kick a member".to_string()); replies.push("/add user - add a member".to_string()); replies.push("/ban x - ban a user or a server".to_string()); replies.push("/unban x - lift a ban".to_string()); replies.push("/op, /admin user - give admin rights".to_string()); replies.push("/unop, /unadmin user - remove admin rights".to_string()); replies.push("/opengroup, /closegroup - control posting access".to_string()); } } StatusCommand::ListMembers => { if is_admin { if self.config.is_member_only() { replies.push("Group members:".to_string()); let admins = self.config.get_admins().collect::>(); let mut members = self.config.get_members().collect::>(); members.extend(admins.iter()); members.sort(); members.dedup(); for m in members { if admins.contains(&m) { replies.push(format!("{} [admin]", m)); } else { replies.push(format!("{}", m)); } } } else { replies.push("Group admins:".to_string()); let mut admins = self.config.get_admins().collect::>(); admins.sort(); for a in admins { replies.push(format!("{}", a)); } } } } StatusCommand::Leave => { if self.config.is_member(¬if_acct) { any_admin_cmd = true; let _ = self.config.set_member(¬if_acct, false); replies.push("You left the group.".to_string()); } } } } tokio::time::sleep(DELAY_BEFORE_ACTION).await; } if do_boost_prev_post { self.client.reblog(&status.in_reply_to_id.as_ref().unwrap()).await .log_error("Failed to boost"); } if !replies.is_empty() { let r = replies.join("\n"); let post = StatusBuilder::new() .status(format!("@{user}\n{msg}", user = notif_acct, msg = r)) .content_type("text/markdown") .visibility(Visibility::Direct) .build().expect("error build status"); let _ = self.client.new_status(post).await.log_error("Failed to post"); } if !announcements.is_empty() { let msg = announcements.join("\n"); let post = StatusBuilder::new() .status(format!("**📢 Group announcement**\n{msg}", msg = msg)) .content_type("text/markdown") .visibility(Visibility::Unlisted) .build().expect("error build status"); let _ = self.client.new_status(post).await.log_error("Failed to post"); } if any_admin_cmd { self.save_if_needed().await.log_error("Failed to save"); } } } NotificationType::Follow => { info!("New follower!"); tokio::time::sleep(Duration::from_millis(500)).await; 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::>(); admins.sort(); format!( "@{user} welcome to the group! This is a member-only group, you won't be \ able to post. Ask the group admins if you wish to join!\n\n\ Admins: {admins}", user = notif_acct, admins = admins.join(", ")) } else { format!( "@{user} welcome to the group! This is a public-access group. \ To share a post, tag the group user. Use /help for more info.", user = notif_acct) }; let post = StatusBuilder::new() .status(text) .content_type("text/markdown") .visibility(Visibility::Unlisted) .build().expect("error build status"); let _ = self.client.new_status(post).await.log_error("Failed to post"); } _ => {} } Ok(()) } /// Catch up with missed notifications, returns true if any were handled async fn catch_up_with_missed_notifications(&mut self) -> Result { 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 } notifs_to_handle.push(n); num += 1; if num > MAX_CATCHUP_NOTIFS { warn!("Too many notifs missed to catch up!"); break; } } 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"); } return Ok(true); } }