group-actor/src/group_handler/mod.rs

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 = 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 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!

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;
                }

                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() {
                    warn!("Socket idle too long, close");
                    break 'rx;
                }
                if remains_to_retire.is_zero() {
                    debug!("Socket open too long, closing");
                    break 'rx;
                }

                trace!("Waiting for message");
                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

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

                /* ping is nice, but pleroma still sometimes doesnt send
                notifs after a while, just let it expire */

                // 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(&notif_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(&notif_acct, notif_user_id).await;

                // if self.config.is_member_or_admin(&notif_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 for tags
    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);

        // Short circuit checks
        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 self.config.is_banned(&status_user) {
            debug!("Status author @{} is banned, discard", status_user);
            return Ok(());
        }

        // optout does not work for members and admins, so don't check it

        if !self.config.is_member_or_admin(&status_user) {
            debug!("Status author @{} is not a member, discard", status_user);
            return Ok(());
        }

        let commands = crate::command::parse_slash_commands(&s.content);
        if commands.contains(&StatusCommand::Ignore) {
            debug!("Post has IGNORE command, discard");
            return Ok(());
        }

        for m in s.mentions {
            let mentioned_user = normalize_acct(&m.acct, group_user)?;
            if mentioned_user == group_user {
                if !commands.is_empty() {
                    debug!("Detected commands for this group, tags dont apply; discard");
                    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!
            } else {
                debug!("#{} 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
            }

            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 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");
        }
    }
}