group-actor/src/group_handle.rs

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::store::{ConfigStore, GroupError};
use crate::store::data::GroupConfig;
use crate::utils::LogError;

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

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) {
        debug!("Handling notif #{}", n.id);
        let ts = n.timestamp_millis();
        self.config.set_last_notif(ts);

        match n.notification_type {
            NotificationType::Mention => {
                if let Some(status) = n.status {
                    if status.content.contains("/gi") || status.content.contains("\\gi") {
                        info!("Mention ignored by gi");
                    } else if status.content.contains("/gb") || status.content.contains("\\gb") {
                        if let Some(id) = status.in_reply_to_id {
                            info!("Boosting prev post by GB");
                            tokio::time::sleep(Duration::from_millis(500)).await;
//                            self.client.reblog(&id).await.log_error("Failed to boost");
                        }
                    } else {
                        info!("Boosting mention");
                        tokio::time::sleep(Duration::from_millis(500)).await;
//                        self.client.reblog(&status.id).await.log_error("Failed to boost");
                    }
                }
            }
            NotificationType::Follow => {
                info!("New follower!");
                tokio::time::sleep(Duration::from_millis(500)).await;

                /*
                let post = StatusBuilder::new()
                    .status(format!("@{} welcome to the group!", &n.account.acct))
                    .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<bool, GroupError> {
        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);
    }
}