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.
680 lines
33 KiB
680 lines
33 KiB
use std::collections::HashSet;
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, Instant};
|
|
|
|
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 elefren::{FediClient, StatusBuilder};
|
|
use futures::StreamExt;
|
|
|
|
use crate::command::StatusCommand;
|
|
use crate::error::GroupError;
|
|
use crate::store::data::GroupConfig;
|
|
use crate::store::ConfigStore;
|
|
use crate::utils::{normalize_acct, 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>,
|
|
}
|
|
|
|
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(¬if_acct);
|
|
let is_admin = self.config.is_admin(¬if_acct);
|
|
|
|
if self.config.is_banned(¬if_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 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("You are not allowed to post to this group".to_string());
|
|
}
|
|
} 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) => {
|
|
info!("Sending PSA");
|
|
announcements.push(a);
|
|
}
|
|
StatusCommand::Boost => {
|
|
if can_write {
|
|
do_boost_prev_post = status.in_reply_to_id.is_some();
|
|
} else {
|
|
replies.push(
|
|
"You are not allowed to share to this group"
|
|
.to_string(),
|
|
);
|
|
}
|
|
}
|
|
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("Only admins can manage user bans".to_string());
|
|
}
|
|
}
|
|
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(_) => {
|
|
unreachable!()
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
replies
|
|
.push("Only admins can manage user bans".to_string());
|
|
}
|
|
}
|
|
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("Only admins can manage server bans".to_string());
|
|
}
|
|
}
|
|
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(_) => {
|
|
unreachable!()
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
replies
|
|
.push("Only admins can manage server bans".to_string());
|
|
}
|
|
}
|
|
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("Only admins can manage members".to_string());
|
|
}
|
|
}
|
|
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(_) => {
|
|
unreachable!()
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
replies.push("Only admins can manage members".to_string());
|
|
}
|
|
}
|
|
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("Only admins can manage admins".to_string());
|
|
}
|
|
}
|
|
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("Only admins can manage admins".to_string());
|
|
}
|
|
}
|
|
StatusCommand::OpenGroup => {
|
|
if is_admin {
|
|
if self.config.is_member_only() {
|
|
any_admin_cmd = true;
|
|
self.config.set_member_only(false);
|
|
replies
|
|
.push("Group changed to open-access".to_string());
|
|
announcements
|
|
.push("This group is now open-access!".to_string());
|
|
}
|
|
} else {
|
|
replies
|
|
.push("Only admins can set group options".to_string());
|
|
}
|
|
}
|
|
StatusCommand::CloseGroup => {
|
|
if is_admin {
|
|
if !self.config.is_member_only() {
|
|
any_admin_cmd = true;
|
|
self.config.set_member_only(true);
|
|
replies
|
|
.push("Group changed to member-only".to_string());
|
|
announcements
|
|
.push("This group is now member-only!".to_string());
|
|
}
|
|
} else {
|
|
replies
|
|
.push("Only admins can set group options".to_string());
|
|
}
|
|
}
|
|
StatusCommand::Help => {
|
|
if self.config.is_member_only() {
|
|
let mut s = "This is a member-only group. ".to_string();
|
|
if self.config.can_write(¬if_acct) {
|
|
if is_admin {
|
|
s.push_str("*You are an admin.*");
|
|
} else {
|
|
s.push_str("*You are a member.*");
|
|
}
|
|
} else {
|
|
s.push_str("*You are not a member, ask one of the admins to add you.*");
|
|
}
|
|
replies.push(s);
|
|
} else {
|
|
let mut s = "This is a public-access group. ".to_string();
|
|
if is_admin {
|
|
s.push_str("*You are an admin.*");
|
|
}
|
|
replies.push(s);
|
|
}
|
|
|
|
replies.push(
|
|
"\nTo share an original post, mention the group user.\n\
|
|
Posts with commands, and replies, won't be shared.\n\
|
|
\n\
|
|
**Supported commands:**\n\
|
|
`/ignore, /i` - make the group completely ignore the post\n\
|
|
`/members, /who` - show group members / admins\n\
|
|
`/boost, /b` - boost the replied-to post into the group"
|
|
.to_string(),
|
|
);
|
|
|
|
if self.config.is_member_only() {
|
|
replies.push("`/leave` - leave the group".to_string());
|
|
}
|
|
|
|
if is_admin {
|
|
replies.push(
|
|
"`/add user` - add a member (use e-mail style address)\n\
|
|
`/kick, /remove user` - kick a member\n\
|
|
`/ban x` - ban a user or a server\n\
|
|
`/unban x` - lift a ban\n\
|
|
`/op, /admin user` - grant admin rights\n\
|
|
`/deop, /deadmin user` - revoke admin rights\n\
|
|
`/opengroup` - make member-only\n\
|
|
`/closegroup` - make public-access\n\
|
|
`/announce x` - make a public announcement from the rest of the status".to_string());
|
|
}
|
|
}
|
|
StatusCommand::ListMembers => {
|
|
let mut show_admins = false;
|
|
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(m.to_string());
|
|
}
|
|
}
|
|
} else {
|
|
show_admins = true;
|
|
}
|
|
} else {
|
|
show_admins = true;
|
|
}
|
|
|
|
if show_admins {
|
|
replies.push("Group admins:".to_string());
|
|
let mut admins =
|
|
self.config.get_admins().collect::<Vec<_>>();
|
|
admins.sort();
|
|
for a in admins {
|
|
replies.push(a.to_string());
|
|
}
|
|
}
|
|
}
|
|
StatusCommand::Leave => {
|
|
if self.config.is_member(¬if_acct) {
|
|
any_admin_cmd = true;
|
|
let _ = self.config.set_member(¬if_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::Direct)
|
|
.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");
|
|
}
|
|
|
|
Ok(true)
|
|
}
|
|
}
|
|
|