pleroma groups!!!!!! try it -> https://piggo.space/hob
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
group-actor/src/store/mod.rs

196 lines
5.9 KiB

use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use elefren::{FediClient, Registration, Scopes, scopes};
use elefren::entities::event::Event;
use futures::StreamExt;
use tokio::sync::RwLock;
use data::{Config, GroupConfig};
use crate::error::GroupError;
use crate::group_handle::GroupHandle;
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, appdata);
// save & persist
self.set_group_config(data.clone()).await?;
Ok(GroupHandle {
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();
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?;
let appdata = client.data.clone();
config.set_appdata(appdata);
self.set_group_config(config.clone()).await?;
Ok(GroupHandle {
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;
let groups_iter = groups.iter_groups().cloned();
// 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());
match client.verify_credentials().await {
Ok(account) => {
info!("Group account verified: @{}, {}", account.acct, account.display_name);
}
Err(e) => {
error!("Group @{} auth error: {}", gc.get_acct(), e);
return None;
}
};
Some(GroupHandle {
client,
config: gc,
store: self.clone(),
})
}).buffer_unordered(8).collect::<Vec<_>>().await
.into_iter().flatten().collect()
}
pub(crate) async fn get_group_config(&self, group: &str) -> Option<GroupConfig> {
let c = self.data.read().await;
c.get_group_config(group).map(|inner| {
inner.clone()
})
}
//noinspection RsSelfConvention
/// Set group config to the store. The store then saved.
pub(crate) async fn set_group_config<'a>(&'a self, config: GroupConfig) -> Result<(), GroupError> {
let mut data = self.data.write().await;
data.set_group_config(config);
self.persist(&data).await?;
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(scopes::Read::Accounts)
| Scopes::read(scopes::Read::Notifications)
| Scopes::read(scopes::Read::Statuses)
| Scopes::read(scopes::Read::Follows)
| Scopes::write(scopes::Write::Statuses)
| Scopes::write(scopes::Write::Media)
}
// trait TapOk<T> {
// fn tap_ok<F: FnOnce(&T)>(self, f: F) -> Self;
// }
//
// impl<T, E> TapOk<T> for Result<T, E> {
// fn tap_ok<F: FnOnce(&T)>(self, f: F) -> Self {
// match self {
// Ok(v) => {
// f(&v);
// Ok(v)
// }
// Err(e) => Err(e)
// }
// }
// }