group-actor/src/group_handle.rs

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<ConfigStore>,
}

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(&notif_acct);
        let is_admin = self.config.is_admin(&notif_acct);

        if self.config.is_banned(&notif_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(
                                        "To share an original post with the group, mention the group user.\n\
                                        Posts with commands won't be shared. Supported commands:\n\
                                        `/ignore, /i` - make the group completely ignore the post\n\
                                        `/boost, /b` - boost the replied-to post into the group\n\
                                        `/leave` - leave the group as a member".to_string());

                                    if is_admin {
                                        replies.push(
                                            "`/members` - list members and admins\n\
                                            `/kick, /remove user` - kick a member\n\
                                            `/add user` - add a member\n\
                                            `/ban x` - ban a user or a server\n\
                                            `/unban x` - lift a ban\n\
                                            `/op, /admin user` - give admin rights\n\
                                            `/deop, /deadmin user` - remove admin rights\n\
                                            `/opengroup` - make member-only\n\
                                            `/closegroup` - make public-access\n\
                                            `/announce x` - make a public announcement from all that follows".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::<HashSet<_>>();
                                            let mut members = self.config.get_members().collect::<Vec<_>>();
                                            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::<Vec<_>>();
                                            admins.sort();
                                            for a in admins {
                                                replies.push(format!("{}", a));
                                            }
                                        }
                                    }
                                }
                                StatusCommand::Leave => {
                                    if self.config.is_member(&notif_acct) {
                                        any_admin_cmd = true;
                                        let _ = self.config.set_member(&notif_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::<Vec<_>>();
                    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! \
                         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<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
            }
            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);
    }
}