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::store::{ConfigStore, GroupError}; use crate::store::data::GroupConfig; use crate::utils::LogError; use std::collections::HashSet; /// 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, } 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> { const PERIODIC_SAVE: Duration = Duration::from_secs(60); const PING_INTERVAL: Duration = Duration::from_secs(15); 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; } 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(Duration::from_millis(1000)).await; } } async fn handle_notification(&mut self, n: Notification) { const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(500); debug!("Handling notif #{}", n.id); let ts = n.timestamp_millis(); self.config.set_last_notif(ts); let can_write = self.config.can_write(&n.account.acct); let is_admin = self.config.is_admin(&n.account.acct); if self.config.is_banned(&n.account.acct) { warn!("Notification actor {} is banned!", n.account.acct); return; } match n.notification_type { NotificationType::Mention => { if let Some(status) = n.status { if self.config.is_banned(&status.account.acct) { warn!("Status author {} is banned!", status.account.acct); return; } let commands = crate::command::parse_status(&status.content); if commands.is_empty() { debug!("No commands in post"); if !can_write { warn!("User {} not allowed to post in group", n.account.acct); return; } if status.in_reply_to_id.is_none() { // 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 { debug!("Not OP, ignore mention") } } else { let mut reply = vec![]; let mut boost_prev = 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![]; // TODO normalize local user handles let mut any_admin_cmd = false; for cmd in commands { match cmd { StatusCommand::Ignore => { debug!("Notif ignored because of ignore command"); return; } StatusCommand::Boost => { if !can_write { warn!("User {} not allowed to boost to group", n.account.acct); } else { boost_prev = status.in_reply_to_id.is_some(); } } StatusCommand::BanUser(u) => { if is_admin { if !self.config.is_banned(&u) { match self.config.ban_user(&u, true) { Ok(_) => { any_admin_cmd = true; reply.push(format!("User {} banned from group!", u)); } Err(e) => { reply.push(format!("Failed to ban user {}: {}", u, e)); } } } } else { warn!("Not admin, can't manage bans"); } } StatusCommand::UnbanUser(u) => { if is_admin { if self.config.is_banned(&u) { match self.config.ban_user(&u, false) { Ok(_) => { any_admin_cmd = true; reply.push(format!("User {} un-banned!", u)); } Err(e) => { unreachable!() } } } } else { warn!("Not admin, can't manage 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; instance_ban_announcements.push(s.clone()); reply.push(format!("Instance {} banned from group!", s)); } Err(e) => { reply.push(format!("Failed to ban instance {}: {}", s, e)); } } } } else { warn!("Not admin, can't manage 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; instance_unban_announcements.push(s.clone()); reply.push(format!("Instance {} un-banned!", s)); } Err(e) => { unreachable!() } } } } else { warn!("Not admin, can't manage bans"); } } StatusCommand::AddMember(u) => { if is_admin { if !self.config.is_member(&u) { match self.config.set_member(&u, true) { Ok(_) => { any_admin_cmd = true; reply.push(format!("User {} added to group!", u)); new_members.push(u); } Err(e) => { reply.push(format!("Failed to add user {} to group: {}", u, e)); } } } } else { warn!("Not admin, can't manage members"); } } StatusCommand::RemoveMember(u) => { if is_admin { if self.config.is_member(&u) { match self.config.set_member(&u, false) { Ok(_) => { any_admin_cmd = true; reply.push(format!("User {} removed from group.", u)); } Err(e) => { unreachable!() } } } } else { warn!("Not admin, can't manage members"); } } StatusCommand::GrantAdmin(u) => { if is_admin { if !self.config.is_admin(&u) { match self.config.set_admin(&u, true) { Ok(_) => { any_admin_cmd = true; reply.push(format!("User {} is now a group admin!", u)); new_admins.push(u); } Err(e) => { reply.push(format!("Failed to make user {} a group admin: {}", u, e)); } } } } else { warn!("Not admin, can't manage admin rights"); } } StatusCommand::RemoveAdmin(u) => { if is_admin { if self.config.is_admin(&u) { match self.config.set_admin(&u, false) { Ok(_) => { any_admin_cmd = true; reply.push(format!("User {} is no longer a group admin!", u)); removed_admins.push(u) } Err(e) => { reply.push(format!("Failed to revoke {}'s group admin: {}", u, e)); } } } } else { warn!("Not admin, can't manage admin rights"); } } StatusCommand::OpenGroup => { if is_admin { if self.config.is_member_only() { any_admin_cmd = true; self.config.set_member_only(false); reply.push(format!("Group changed to open-access")); } } else { warn!("Not admin, can't manage group mode"); } } StatusCommand::CloseGroup => { if is_admin { if !self.config.is_member_only() { any_admin_cmd = true; self.config.set_member_only(true); reply.push(format!("Group changed to member-only")); } } else { warn!("Not admin, can't manage group mode"); } } StatusCommand::Help => { reply.push("Mention the group user in a top-level post to share it with the group's members.".to_string()); reply.push("Posts with commands won't be shared. Supported commands:".to_string()); reply.push("/ignore, /ign, /i - don't run any commands in the post".to_string()); reply.push("/boost, /b - boost the replied-to post into the group".to_string()); reply.push("/leave - leave the group as a member".to_string()); if is_admin { reply.push("/members".to_string()); reply.push("/kick, /remove user - kick a member".to_string()); reply.push("/add user - add a member".to_string()); reply.push("/ban x - ban a user or a server".to_string()); reply.push("/unban x - lift a ban".to_string()); reply.push("/op, /admin user - give admin rights".to_string()); reply.push("/unop, /unadmin user - remove admin rights".to_string()); reply.push("/opengroup, /closegroup - control posting access".to_string()); } } StatusCommand::ListMembers => { if is_admin { reply.push("Member list:".to_string()); let admins = self.config.get_admins().collect::>(); let members = self.config.get_members().collect::>(); for m in members { if admins.contains(&m) { reply.push(format!("{} [admin]", m)); } else { reply.push(format!("{}", m)); } } } } StatusCommand::Leave => { if self.config.is_member(&n.account.acct) { any_admin_cmd = true; let _ = self.config.set_member(&n.account.acct, false); reply.push("You left the group.".to_string()); } } } } tokio::time::sleep(DELAY_BEFORE_ACTION).await; if boost_prev { self.client.reblog(&status.in_reply_to_id.as_ref().unwrap()).await .log_error("Failed to boost"); } if !reply.is_empty() { let r = reply.join("\n"); let post = StatusBuilder::new() .status(format!("@{user}\n{msg}", user=n.account.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 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 = &n.account.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 = &n.account.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"); } _ => {} } } /// Catch up with missed notifications, returns true if any were handled async fn catch_up_with_missed_notifications(&mut self) -> Result { const MAX_CATCHUP_NOTIFS: usize = 25; 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; } return Ok(true); } }