group-actor/src/store/mod.rs

use std::path::{Path, PathBuf};
use std::sync::Arc;

use elefren::{scopes, FediClient, Registration, Scopes};
use futures::StreamExt;
use tokio::sync::RwLock;

use data::{Config, GroupConfig};

use crate::error::GroupError;
use crate::group_handler::GroupHandle;
use std::time::Duration;

pub(crate) mod data;

#[derive(Debug, Default)]
pub struct ConfigStore {
    store_path: PathBuf,
    save_pretty: bool,
    data: tokio::sync::RwLock<Config>,
}

#[derive(Debug)]
pub struct NewGroupOptions {
    pub server: String,
    pub acct: String,
}

#[derive(Debug)]
pub struct StoreOptions {
    pub store_path: String,
    pub save_pretty: bool,
}

impl ConfigStore {
    /// Create a new instance of the store.
    /// If a path is given, it will try to load the content from a file.
    pub async fn new(options: StoreOptions) -> Result<Arc<Self>, GroupError> {
        let path: &Path = options.store_path.as_ref();

        let config = if path.is_file() {
            let f = tokio::fs::read(path).await?;
            serde_json::from_slice(&f)?
        } else {
            let empty = Config::default();
            tokio::fs::write(path, serde_json::to_string(&empty)?.as_bytes()).await?;
            empty
        };

        Ok(Arc::new(Self {
            store_path: path.to_owned(),
            save_pretty: options.save_pretty,
            data: RwLock::new(config),
        }))
    }

    /// Spawn a new group
    pub async fn auth_new_group(self: &Arc<Self>, opts: NewGroupOptions) -> Result<GroupHandle, GroupError> {
        let registration = Registration::new(&opts.server)
            .client_name("group-actor")
            .force_login(true)
            .scopes(make_scopes())
            .build()
            .await?;

        println!("--- Authenticating NEW bot user @{} ---", opts.acct);
        let client = elefren::helpers::cli::authenticate(registration).await?;
        let appdata = client.data.clone();

        let data = GroupConfig::new(opts.acct.clone(), appdata);

        // save & persist
        self.set_group_config(data.clone()).await?;

        let group_account = match client.verify_credentials().await {
            Ok(account) => {
                info!(
                    "Group account verified: @{}, \"{}\"",
                    account.acct, account.display_name
                );
                account
            }
            Err(e) => {
                error!("Group @{} auth error: {}", opts.acct, e);
                return Err(e.into());
            }
        };

        Ok(GroupHandle {
            group_account,
            client,
            config: data,
            store: self.clone(),
        })
    }

    /// Re-auth an existing group
    pub async fn reauth_group(self: &Arc<Self>, acct: &str) -> Result<GroupHandle, GroupError> {
        let groups = self.data.read().await;
        let mut config = groups.get_group_config(acct).ok_or(GroupError::GroupNotExist)?.clone();
        drop(groups);

        println!("--- Re-authenticating bot user @{} ---", acct);
        let registration = Registration::new(config.get_appdata().base.to_string())
            .client_name("group-actor")
            .force_login(true)
            .scopes(make_scopes())
            .build()
            .await?;

        let client = elefren::helpers::cli::authenticate(registration).await?;
        println!("Auth complete");

        let appdata = client.data.clone();

        config.set_appdata(appdata);
        self.set_group_config(config.clone()).await?;

        let group_account = match client.verify_credentials().await {
            Ok(account) => {
                info!(
                    "Group account verified: @{}, \"{}\"",
                    account.acct, account.display_name
                );
                account
            }
            Err(e) => {
                error!("Group @{} auth error: {}", acct, e);
                return Err(e.into());
            }
        };

        Ok(GroupHandle {
            group_account,
            client,
            config,
            store: self.clone(),
        })
    }

    /// Spawn existing group using saved creds
    pub async fn spawn_groups(self: Arc<Self>) -> Vec<GroupHandle> {
        let groups = self.data.read().await.clone();
        let groups_iter = groups.groups.into_values();

        // Connect in parallel
        futures::stream::iter(groups_iter)
            .map(|gc| async {
                if !gc.is_enabled() {
                    debug!("Group @{} is DISABLED", gc.get_acct());
                    return None;
                }

                debug!("Connecting to @{}", gc.get_acct());

                let client = FediClient::from(gc.get_appdata().clone());

                let my_account = match client.verify_credentials().await {
                    Ok(account) => {
                        info!(
                            "Group account verified: @{}, \"{}\"",
                            account.acct, account.display_name
                        );
                        account
                    }
                    Err(e) => {
                        error!("Group @{} auth error: {}", gc.get_acct(), e);
                        return None;
                    }
                };

                Some(GroupHandle {
                    group_account: my_account,
                    client,
                    config: gc,
                    store: self.clone(),
                })
            })
            .buffer_unordered(8)
            .collect::<Vec<_>>()
            .await
            .into_iter()
            .flatten()
            .collect()
    }

    pub async fn group_exists(&self, acct : &str) -> bool {
        self.data.read().await.groups.contains_key(acct)
    }

    /*
    pub(crate) async fn get_group_config(&self, group: &str) -> Option<GroupConfig> {
        let c = self.data.read().await;
        c.get_group_config(group).cloned()
    }
    */

    //noinspection RsSelfConvention
    /// Set group config to the store. The store then saved.
    pub(crate) async fn set_group_config(&self, config: GroupConfig) -> Result<(), GroupError> {
        trace!("Locking mutex");
        if let Ok(mut data) = tokio::time::timeout(Duration::from_secs(1), self.data.write()).await {
            trace!("Locked");
            data.set_group_config(config);
            trace!("Writing file");
            self.persist(&data).await?;
        } else {
            error!("DEADLOCK? Timeout waiting for data RW Lock in settings store");
        }
        Ok(())
    }

    /// Persist the store
    async fn persist(&self, data: &Config) -> Result<(), GroupError> {
        tokio::fs::write(
            &self.store_path,
            if self.save_pretty {
                serde_json::to_string_pretty(&data)
            } else {
                serde_json::to_string(&data)
            }?
            .as_bytes(),
        )
        .await?;
        Ok(())
    }
}

fn make_scopes() -> Scopes {
    Scopes::read_all()
        | Scopes::write(scopes::Write::Statuses)
        | Scopes::write(scopes::Write::Media)
        | Scopes::write(scopes::Write::Follows)
}