master
Ondřej Hruška 3 years ago
parent 3b7700a4b1
commit 8afc77dd60
Signed by untrusted user: MightyPork
GPG Key ID: 2C5FD5035250423D
  1. 1
      .gitignore
  2. 2
      Cargo.lock
  3. 2
      Cargo.toml
  4. 37
      src/command.rs
  5. 591
      src/group_handle.rs
  6. 2
      src/store/data.rs
  7. 17
      src/store/mod.rs

1
.gitignore vendored

@ -4,3 +4,4 @@ group-actor-data.toml
.idea/ .idea/
groups.json groups.json
fedigroups fedigroups
*.bak

2
Cargo.lock generated

@ -327,7 +327,7 @@ checksum = "e88a8acf291dafb59c2d96e8f59828f3838bb1a70398823ade51a84de6a6deed"
[[package]] [[package]]
name = "fedigroups" name = "fedigroups"
version = "0.1.0" version = "0.2.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"clap", "clap",

@ -1,6 +1,6 @@
[package] [package]
name = "fedigroups" name = "fedigroups"
version = "0.1.0" version = "0.2.0"
authors = ["Ondřej Hruška <ondra@ondrovo.com>"] authors = ["Ondřej Hruška <ondra@ondrovo.com>"]
edition = "2018" edition = "2018"
publish = false publish = false

@ -122,11 +122,12 @@ static RE_ANNOUNCE: once_cell::sync::Lazy<Regex> =
Lazy::new(|| Regex::new(concat!(r"(?:^|\s|>|\n)[\\/]announce\s+(.*)$")).unwrap()); Lazy::new(|| Regex::new(concat!(r"(?:^|\s|>|\n)[\\/]announce\s+(.*)$")).unwrap());
static RE_A_HASHTAG: once_cell::sync::Lazy<Regex> = static RE_A_HASHTAG: once_cell::sync::Lazy<Regex> =
Lazy::new(|| Regex::new(concat!(r"\b#(\w+)")).unwrap()); Lazy::new(|| Regex::new(concat!(r"(?:^|\b|\s|>|\n)#(\w+)")).unwrap());
pub fn parse_status_tags(content: &str) -> Vec<String> { pub fn parse_status_tags(content: &str) -> Vec<String> {
debug!("Raw content: {}", content); debug!("Raw content: {}", content);
let content = content.replace("<br/>", " "); let content = content.replace("<br/>", "<br/> ");
let content = content.replace("</p>", "</p> ");
let content = voca_rs::strip::strip_tags(&content); let content = voca_rs::strip::strip_tags(&content);
debug!("Stripped tags: {}", content); debug!("Stripped tags: {}", content);
@ -143,10 +144,8 @@ pub fn parse_status_tags(content: &str) -> Vec<String> {
pub fn parse_slash_commands(content: &str) -> Vec<StatusCommand> { pub fn parse_slash_commands(content: &str) -> Vec<StatusCommand> {
debug!("Raw content: {}", content); debug!("Raw content: {}", content);
let content = content.replace("<br/>", " "); let content = content.replace("<br/>", "<br/> ");
// let content = content.replace("<br />", " "); let content = content.replace("</p>", "</p> ");
// let content = content.replace("<BR/>", " ");
// let content = content.replace("<BR />", " ");
let content = voca_rs::strip::strip_tags(&content); let content = voca_rs::strip::strip_tags(&content);
debug!("Stripped tags: {}", content); debug!("Stripped tags: {}", content);
@ -423,13 +422,14 @@ mod test {
} }
#[test] #[test]
fn test_add_member() { fn test_add_tag() {
assert!(RE_ADD_TAG.is_match("/add #breadposting")); assert!(RE_ADD_TAG.is_match("/add #breadposting"));
assert!(RE_ADD_TAG.is_match("/add #čučkaři")); assert!(RE_ADD_TAG.is_match("/add #čučkaři"));
assert!(RE_ADD_TAG.is_match("/add #χαλβάς")); assert!(RE_ADD_TAG.is_match("/add #χαλβάς"));
assert!(RE_ADD_TAG.is_match("\\add #ласточка")); assert!(RE_ADD_TAG.is_match("\\add #ласточка"));
assert!(RE_ADD_TAG.is_match("/add #nya.")); assert!(RE_ADD_TAG.is_match("/add #nya."));
assert!(RE_ADD_TAG.is_match("/add #nya)")); assert!(RE_ADD_TAG.is_match("/add #nya)"));
assert!(RE_ADD_TAG.is_match("/add #nya and more)"));
let c = RE_ADD_TAG.captures("/add #breadposting"); let c = RE_ADD_TAG.captures("/add #breadposting");
assert_eq!(c.unwrap().get(1).unwrap().as_str(), "breadposting"); assert_eq!(c.unwrap().get(1).unwrap().as_str(), "breadposting");
@ -440,10 +440,10 @@ mod test {
let c = RE_ADD_TAG.captures("/add #ласточка"); let c = RE_ADD_TAG.captures("/add #ласточка");
assert_eq!(c.unwrap().get(1).unwrap().as_str(), "ласточка"); assert_eq!(c.unwrap().get(1).unwrap().as_str(), "ласточка");
let c = RE_ADD_TAG.captures("#nya."); let c = RE_ADD_TAG.captures("/add #nya.");
assert_eq!(c.unwrap().get(1).unwrap().as_str(), "nya"); assert_eq!(c.unwrap().get(1).unwrap().as_str(), "nya");
let c = RE_ADD_TAG.captures("#nya)"); let c = RE_ADD_TAG.captures("/add #nya)");
assert_eq!(c.unwrap().get(1).unwrap().as_str(), "nya"); assert_eq!(c.unwrap().get(1).unwrap().as_str(), "nya");
} }
@ -510,7 +510,7 @@ mod test {
} }
#[test] #[test]
fn test_members() { fn test_tags() {
assert!(!RE_TAGS.is_match("/members")); assert!(!RE_TAGS.is_match("/members"));
assert!(RE_TAGS.is_match("/hashtags")); assert!(RE_TAGS.is_match("/hashtags"));
assert!(RE_TAGS.is_match("dsfsd /tags dfgd d")); assert!(RE_TAGS.is_match("dsfsd /tags dfgd d"));
@ -524,10 +524,17 @@ mod test {
assert!(RE_A_HASHTAG.is_match("#χαλβάς")); assert!(RE_A_HASHTAG.is_match("#χαλβάς"));
assert!(RE_A_HASHTAG.is_match("foo #banana gfdfgd")); assert!(RE_A_HASHTAG.is_match("foo #banana gfdfgd"));
let c = RE_GRANT_ADMIN.captures("foo #banana #χαλβάς #ласточка."); for (i, c) in RE_A_HASHTAG.captures_iter("foo #banana #χαλβάς #ласточка").enumerate() {
assert_eq!(c.unwrap().get(1).unwrap().as_str(), "banana"); if i == 0 {
assert_eq!(c.unwrap().get(2).unwrap().as_str(), "χαλβάς"); assert_eq!(c.get(1).unwrap().as_str(), "banana");
assert_eq!(c.unwrap().get(3).unwrap().as_str(), "ласточка"); }
else if i == 1 {
assert_eq!(c.get(1).unwrap().as_str(), "χαλβάς");
}
else if i == 2 {
assert_eq!(c.get(1).unwrap().as_str(), "ласточка");
}
}
} }
#[test] #[test]
@ -540,7 +547,7 @@ mod test {
} }
#[test] #[test]
fn test_leave() { fn test_join() {
assert!(!RE_JOIN.is_match("/list")); assert!(!RE_JOIN.is_match("/list"));
assert!(RE_JOIN.is_match("/join")); assert!(RE_JOIN.is_match("/join"));
assert!(RE_JOIN.is_match("/join")); assert!(RE_JOIN.is_match("/join"));

@ -27,10 +27,10 @@ pub struct GroupHandle {
} }
const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250); const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250);
const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(1000); const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(500);
const MAX_CATCHUP_NOTIFS: usize = 25; const MAX_CATCHUP_NOTIFS: usize = 25;
// also statuses // also statuses
const MAX_CATCHUP_STATUSES: usize = 100; const MAX_CATCHUP_STATUSES: usize = 50;
// higher because we can expect a lot of non-hashtag statuses here // higher because we can expect a lot of non-hashtag statuses here
const PERIODIC_SAVE: Duration = Duration::from_secs(60); const PERIODIC_SAVE: Duration = Duration::from_secs(60);
const PING_INTERVAL: Duration = Duration::from_secs(15); // must be < periodic save! const PING_INTERVAL: Duration = Duration::from_secs(15); // must be < periodic save!
@ -39,6 +39,7 @@ impl GroupHandle {
pub async fn save(&mut self) -> Result<(), GroupError> { pub async fn save(&mut self) -> Result<(), GroupError> {
debug!("Saving group config & status"); debug!("Saving group config & status");
self.store.set_group_config(self.config.clone()).await?; self.store.set_group_config(self.config.clone()).await?;
debug!("Saved");
self.config.clear_dirty_status(); self.config.clear_dirty_status();
Ok(()) Ok(())
} }
@ -84,17 +85,17 @@ impl GroupHandle {
pub async fn run(&mut self) -> Result<(), GroupError> { pub async fn run(&mut self) -> Result<(), GroupError> {
assert!(PERIODIC_SAVE >= PING_INTERVAL); assert!(PERIODIC_SAVE >= PING_INTERVAL);
let mut next_save = Instant::now() + PERIODIC_SAVE; // so we save at start
loop { loop {
debug!("Opening streaming API socket"); 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 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 { match self.catch_up_with_missed_notifications().await {
Ok(true) => { Ok(true) => {
debug!("Some missed notifs handled"); debug!("Some missed notifs handled");
// Save asap!
next_save = Instant::now() - PERIODIC_SAVE
} }
Ok(false) => { Ok(false) => {
debug!("No notifs missed"); debug!("No notifs missed");
@ -107,8 +108,6 @@ impl GroupHandle {
match self.catch_up_with_missed_statuses().await { match self.catch_up_with_missed_statuses().await {
Ok(true) => { Ok(true) => {
debug!("Some missed statuses handled"); debug!("Some missed statuses handled");
// Save asap!
next_save = Instant::now() - PERIODIC_SAVE
} }
Ok(false) => { Ok(false) => {
debug!("No statuses missed"); debug!("No statuses missed");
@ -118,12 +117,29 @@ impl GroupHandle {
} }
} }
loop { if self.config.is_dirty() {
// save asap
next_save = Instant::now() - PERIODIC_SAVE
}
'rx: loop {
if next_save < Instant::now() { if next_save < Instant::now() {
debug!("Save time elapsed, saving if needed");
self.save_if_needed().await.log_error("Failed to save group"); self.save_if_needed().await.log_error("Failed to save group");
next_save = Instant::now() + PERIODIC_SAVE; next_save = Instant::now() + PERIODIC_SAVE;
} }
if last_rx.elapsed() > PING_INTERVAL * 2 {
warn!("Socket idle too long, close");
break 'rx;
}
if socket_open_time.elapsed() > Duration::from_secs(120) {
debug!("Socket open too long, closing");
break 'rx;
}
debug!("Await msg");
let timeout = next_save let timeout = next_save
.saturating_duration_since(Instant::now()) .saturating_duration_since(Instant::now())
.min(PING_INTERVAL) .min(PING_INTERVAL)
@ -131,6 +147,7 @@ impl GroupHandle {
match tokio::time::timeout(timeout, events.next()).await { match tokio::time::timeout(timeout, events.next()).await {
Ok(Some(event)) => { Ok(Some(event)) => {
last_rx = Instant::now();
debug!("(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event)); debug!("(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event));
match event { match event {
Event::Update(status) => { Event::Update(status) => {
@ -141,19 +158,26 @@ impl GroupHandle {
} }
Event::Delete(_id) => {} Event::Delete(_id) => {}
Event::FiltersChanged => {} Event::FiltersChanged => {}
Event::Heartbeat => {}
} }
} }
Ok(None) => { Ok(None) => {
warn!("Group @{} socket closed, restarting...", self.config.get_acct()); warn!("Group @{} socket closed, restarting...", self.config.get_acct());
break; break 'rx;
} }
Err(_) => { Err(_) => {
// Timeout so we can save if needed // Timeout so we can save if needed
} }
} }
trace!("Pinging"); if last_ping.elapsed() > PING_INTERVAL {
events.send_ping().await.log_error("Fail to send ping"); last_ping = Instant::now();
debug!("Pinging");
if events.send_ping()
.await.is_err() {
break 'rx;
}
}
} }
warn!("Notif stream closed, will reopen"); warn!("Notif stream closed, will reopen");
@ -167,10 +191,13 @@ impl GroupHandle {
self.config.set_last_notif(ts); self.config.set_last_notif(ts);
let group_acct = self.config.get_acct().to_string(); 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)?; let notif_acct = normalize_acct(&n.account.acct, &group_acct)?;
let can_write = self.config.can_write(&notif_acct); if notif_acct == group_acct {
let is_admin = self.config.is_admin(&notif_acct); debug!("This is our post, ignore that");
return Ok(());
}
if self.config.is_banned(&notif_acct) { if self.config.is_banned(&notif_acct) {
warn!("Notification actor {} is banned!", notif_acct); warn!("Notification actor {} is banned!", notif_acct);
@ -180,7 +207,260 @@ impl GroupHandle {
match n.notification_type { match n.notification_type {
NotificationType::Mention => { NotificationType::Mention => {
if let Some(status) = n.status { if let Some(status) = n.status {
self.handle_mention_status(status).await?;
}
}
NotificationType::Follow => {
info!("New follower!");
if self.config.is_member_or_admin(&notif_acct) {
// Already joined, just doing something silly, ignore this
debug!("User already a member, ignoring");
} else {
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} Hi, this is a member-only group, you won't be \
able to post. You can still receive group posts though. If you'd like to join, \
please ask one of the group admins to add you:\n\n\
{admins}",
user = notif_acct,
admins = admins.join(", ")
)
} else {
self.follow_user(notif_user_id).await
.log_error("Failed to follow");
make_welcome_text(&notif_acct)
};
let post = StatusBuilder::new()
.status(text)
.content_type("text/markdown")
.visibility(Visibility::Direct)
.build()
.expect("error build status");
// tokio::time::sleep(Duration::from_millis(500)).await;
let _ = self.client.new_status(post).await.log_error("Failed to post");
}
}
NotificationType::Favourite => {}
NotificationType::Reblog => {}
}
Ok(())
}
/// Handle a non-mention status
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);
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, ignore that");
return Ok(());
}
// for m in &s.mentions {
// let ma = normalize_acct(&m.acct, gu)?;
// if ma == gu {
// debug!("Mention detected, handle status as mention notification!");
// return self.handle_mention_status(s).await;
// }
// }
if !s.content.contains('#') {
debug!("No tags in status");
return Ok(());
}
if s.visibility.is_private() {
debug!("Status is direct/private, not boosting");
return Ok(());
}
if s.content.contains("/add ") || s.content.contains("/remove ") {
debug!("Discard, looks like a hashtag manipulation command");
return Ok(());
}
if self.config.is_banned(&status_user) {
debug!("Status author @{} is banned.", status_user);
return Ok(());
}
if !self.config.is_member_or_admin(&status_user) {
debug!("Status author @{} is not a member.", status_user);
return Ok(());
}
let tags = crate::command::parse_status_tags(&s.content);
debug!("Tags in status: {:?}", tags);
for t in tags {
if self.config.is_tag_followed(&t) {
self.client.reblog(&s.id).await
.log_error("Failed to reblog");
break;
}
}
Ok(())
}
async fn follow_user(&mut self, id: &str) -> Result<(), GroupError> {
self.client.follow(id).await?;
Ok(())
}
async fn unfollow_user(&mut self, id: &str) -> Result<(), GroupError> {
self.client.unfollow(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)
}
fn list_admins(&self, replies: &mut Vec<String>) {
let mut admins = self.config.get_admins().collect::<Vec<_>>();
admins.sort();
for a in admins {
replies.push(a.to_string());
}
}
fn list_members(&self, replies: &mut Vec<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());
}
}
}
async fn handle_mention_status(&mut self, status: Status) -> Result<(), GroupError> {
let group_acct = self.config.get_acct().to_string();
let status_acct = normalize_acct(&status.account.acct, &group_acct)?; let status_acct = normalize_acct(&status.account.acct, &group_acct)?;
let status_user_id = &status.account.id;
let can_write = self.config.can_write(&status_acct);
let is_admin = self.config.is_admin(&status_acct);
if self.config.is_banned(&status_acct) { if self.config.is_banned(&status_acct) {
warn!("Status author {} is banned!", status_acct); warn!("Status author {} is banned!", status_acct);
@ -193,6 +473,7 @@ impl GroupHandle {
let mut announcements = vec![]; let mut announcements = vec![];
let mut do_boost_prev_post = false; let mut do_boost_prev_post = false;
let mut any_admin_cmd = false; let mut any_admin_cmd = false;
let mut want_markdown = false;
if commands.is_empty() { if commands.is_empty() {
debug!("No commands in post"); debug!("No commands in post");
@ -200,7 +481,7 @@ impl GroupHandle {
if can_write { if can_write {
// Someone tagged the group in OP, boost it. // Someone tagged the group in OP, boost it.
info!("Boosting OP mention"); info!("Boosting OP mention");
tokio::time::sleep(DELAY_BEFORE_ACTION).await; // tokio::time::sleep(DELAY_BEFORE_ACTION).await;
self.client.reblog(&status.id).await.log_error("Failed to boost"); self.client.reblog(&status.id).await.log_error("Failed to boost");
} else { } else {
replies.push("You are not allowed to post to this group".to_string()); replies.push("You are not allowed to post to this group".to_string());
@ -236,8 +517,9 @@ impl GroupHandle {
any_admin_cmd = true; any_admin_cmd = true;
replies.push(format!("User {} banned from group!", u)); replies.push(format!("User {} banned from group!", u));
self.unfollow_user(&u).await // FIXME we need user ID, not handle - get it via API?
.log_error("Failed to unfollow"); // self.unfollow_user(&u).await
// .log_error("Failed to unfollow");
// no announcement here // no announcement here
} }
@ -314,7 +596,7 @@ impl GroupHandle {
Ok(_) => { Ok(_) => {
any_admin_cmd = true; any_admin_cmd = true;
replies.push(format!("User {} added to the group!", u)); replies.push(format!("User {} added to the group!", u));
self.follow_user(&u).await.log_error("Failed to follow"); self.follow_user(status_user_id).await.log_error("Failed to follow");
} }
Err(e) => { Err(e) => {
replies.push(format!("Failed to add user {} to group: {}", u, e)); replies.push(format!("Failed to add user {} to group: {}", u, e));
@ -334,8 +616,9 @@ impl GroupHandle {
any_admin_cmd = true; any_admin_cmd = true;
replies.push(format!("User {} removed from the group.", u)); replies.push(format!("User {} removed from the group.", u));
self.unfollow_user(&u).await // FIXME we need user ID, not handle - get it via API?
.log_error("Failed to unfollow"); // self.unfollow_user(&u).await
// .log_error("Failed to unfollow");
} }
Err(_) => { Err(_) => {
unreachable!() unreachable!()
@ -436,24 +719,26 @@ impl GroupHandle {
} }
} }
StatusCommand::Help => { StatusCommand::Help => {
want_markdown = true;
if self.config.is_member_only() { if self.config.is_member_only() {
let mut s = "This is a member-only group. ".to_string(); replies.push("This is a member-only group. ".to_string());
if self.config.can_write(&notif_acct) {
if is_admin {
s.push_str("*You are an admin.*");
} else { } else {
s.push_str("*You are a member.*"); replies.push("This is a public-access group. ".to_string());
} }
if self.config.can_write(&status_acct) {
if is_admin {
replies.push("*You are an admin.*".to_string());
} else { } else {
s.push_str("*You are not a member, ask one of the admins to add you.*"); replies.push("*You are a member.*".to_string());
} }
replies.push(s);
} else { } else {
let mut s = "This is a public-access group. ".to_string(); if self.config.is_member_only() {
if is_admin { replies.push("*You are not a member, ask one of the admins to add you.*".to_string());
s.push_str("*You are an admin.*"); } else {
replies.push("*You are not a member, follow or use /join to join the group.*".to_string());
} }
replies.push(s);
} }
replies.push( replies.push(
@ -494,17 +779,9 @@ impl GroupHandle {
StatusCommand::ListMembers => { StatusCommand::ListMembers => {
let mut show_admins = false; let mut show_admins = false;
if is_admin { if is_admin {
if self.config.is_member_only() {
replies.push("Group members:".to_string()); replies.push("Group members:".to_string());
self.list_members(&mut replies); self.list_members(&mut replies);
} else { } else {
show_admins = true;
}
} else {
show_admins = true;
}
if show_admins {
replies.push("Group admins:".to_string()); replies.push("Group admins:".to_string());
self.list_admins(&mut replies); self.list_admins(&mut replies);
} }
@ -518,22 +795,22 @@ impl GroupHandle {
} }
} }
StatusCommand::Leave => { StatusCommand::Leave => {
if self.config.is_member_or_admin(&notif_acct) { if self.config.is_member_or_admin(&status_acct) {
// admin can leave but that's a bad idea // admin can leave but that's a bad idea
any_admin_cmd = true; any_admin_cmd = true;
let _ = self.config.set_member(&notif_acct, false); let _ = self.config.set_member(&status_acct, false);
replies.push("You're no longer a group member. Unfollow the group user to stop receiving group messages.".to_string()); replies.push("You're no longer a group member. Unfollow the group user to stop receiving group messages.".to_string());
self.unfollow_user(&notif_acct).await self.unfollow_user(&status_user_id).await
.log_error("Failed to unfollow"); .log_error("Failed to unfollow");
} }
} }
StatusCommand::Join => { StatusCommand::Join => {
if self.config.is_member_or_admin(&notif_acct) { if self.config.is_member_or_admin(&status_acct) {
// Already a member, so let's try to follow the user // Already a member, so let's try to follow the user
// again, maybe first time it failed // again, maybe first time it failed
self.follow_user(&notif_acct).await self.follow_user(status_user_id).await
.log_error("Failed to follow"); .log_error("Failed to follow");
} else { } else {
// Not a member yet // Not a member yet
@ -545,11 +822,11 @@ impl GroupHandle {
self.list_admins(&mut replies); self.list_admins(&mut replies);
} else { } else {
// Open access // Open access
self.follow_user(&notif_acct).await self.follow_user(status_user_id).await
.log_error("Failed to follow"); .log_error("Failed to follow");
// This only fails if the user is banned, but that is filtered above // This only fails if the user is banned, but that is filtered above
let _ = self.config.set_member(&notif_acct, true); let _ = self.config.set_member(&status_acct, true);
replies.push(format!("\ replies.push(format!("\
Thanks for joining, you are now a member and the group user will \ Thanks for joining, you are now a member and the group user will \
follow you so you can use group hashtags. Make sure you follow the \ follow you so you can use group hashtags. Make sure you follow the \
@ -563,7 +840,7 @@ impl GroupHandle {
} }
} }
tokio::time::sleep(DELAY_BEFORE_ACTION).await; // tokio::time::sleep(DELAY_BEFORE_ACTION).await;
} }
if do_boost_prev_post { if do_boost_prev_post {
@ -574,11 +851,17 @@ impl GroupHandle {
} }
if !replies.is_empty() { if !replies.is_empty() {
debug!("replies={:?}", replies);
let r = replies.join("\n"); let r = replies.join("\n");
debug!("r={}", r);
let post = StatusBuilder::new() let post = StatusBuilder::new()
.status(format!("@{user}\n{msg}", user = notif_acct, msg = r)) .status(format!("@{user}\n{msg}", user = status_acct, msg = r))
.content_type("text/markdown") .content_type(if want_markdown {
"text/markdown"
} else {
"text/plain"
})
.visibility(Visibility::Direct) .visibility(Visibility::Direct)
.build() .build()
.expect("error build status"); .expect("error build status");
@ -599,220 +882,12 @@ impl GroupHandle {
} }
if any_admin_cmd { if any_admin_cmd {
debug!("Saving after admin cmd");
self.save_if_needed().await.log_error("Failed to save"); self.save_if_needed().await.log_error("Failed to save");
} }
}
}
NotificationType::Follow => {
info!("New follower!");
if self.config.is_member_or_admin(&notif_acct) {
// Already joined, just doing something silly, ignore this
debug!("User already a member, ignoring");
} else {
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} Hi, this is a member-only group, you won't be \
able to post. You can still receive group posts though. If you'd like to join, \
please ask one of the group admins to add you:\n\n\
{admins}",
user = notif_acct,
admins = admins.join(", ")
)
} else {
self.follow_user(&notif_acct).await
.log_error("Failed to follow");
make_welcome_text(&notif_acct)
};
let post = StatusBuilder::new()
.status(text)
.content_type("text/markdown")
.visibility(Visibility::Direct)
.build()
.expect("error build status");
tokio::time::sleep(Duration::from_millis(500)).await;
let _ = self.client.new_status(post).await.log_error("Failed to post");
}
}
NotificationType::Favourite => {}
NotificationType::Reblog => {}
}
Ok(())
}
/// Handle a non-mention status
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);
if !s.content.contains('#') {
debug!("No tags in status");
return Ok(());
}
if s.visibility.is_private() {
debug!("Status is direct/private, not boosting");
return Ok(());
}
if s.content.contains("/add ") || s.content.contains("/remove ") {
debug!("Discard, looks like a hashtag manipulation command");
return Ok(());
}
let gu = self.config.get_acct();
let su = normalize_acct(&s.account.acct, gu)?;
if self.config.is_banned(&su) {
debug!("Status author @{} is banned.", su);
return Ok(());
}
if !self.config.is_member_or_admin(&su) {
debug!("Status author @{} is not a member.", su);
return Ok(());
}
let tags = crate::command::parse_status_tags(&s.content);
debug!("Tags in status: {:?}", tags);
for t in tags {
if self.config.is_tag_followed(&t) {
self.client.reblog(&s.id).await
.log_error("Failed to reblog");
break;
}
}
Ok(())
}
async fn follow_user(&mut self, acct: &str) -> Result<(), GroupError> {
self.client.follow(acct).await?;
Ok(())
}
async fn unfollow_user(&mut self, acct: &str) -> Result<(), GroupError> {
self.client.unfollow(acct).await?;
Ok(()) 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)
}
/// 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 notifications = self.client.get_home_timeline().await?;
let mut iter = notifications.items_iter();
let mut statuses_to_handle = vec![];
// They are retrieved newest first, but we want oldest first for chronological handling
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)
}
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;
}
}
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)
}
fn list_admins(&self, replies: &mut Vec<String>) {
let mut admins = self.config.get_admins().collect::<Vec<_>>();
admins.sort();
for a in admins {
replies.push(a.to_string());
}
}
fn list_members(&self, replies: &mut Vec<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());
}
}
}
} }
fn make_welcome_text(user: &str) -> String { fn make_welcome_text(user: &str) -> String {

@ -7,7 +7,7 @@ use crate::error::GroupError;
/// This is the inner data struct holding the config /// This is the inner data struct holding the config
#[derive(Debug, Clone, Serialize, Deserialize, Default)] #[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub(crate) struct Config { pub(crate) struct Config {
groups: HashMap<String, GroupConfig>, pub(crate) groups: HashMap<String, GroupConfig>,
} }
impl Config { impl Config {

@ -9,6 +9,7 @@ use data::{Config, GroupConfig};
use crate::error::GroupError; use crate::error::GroupError;
use crate::group_handle::GroupHandle; use crate::group_handle::GroupHandle;
use std::time::Duration;
pub(crate) mod data; pub(crate) mod data;
@ -82,6 +83,7 @@ impl ConfigStore {
pub async fn reauth_group(self: &Arc<Self>, acct: &str) -> Result<GroupHandle, GroupError> { pub async fn reauth_group(self: &Arc<Self>, acct: &str) -> Result<GroupHandle, GroupError> {
let groups = self.data.read().await; let groups = self.data.read().await;
let mut config = groups.get_group_config(acct).ok_or(GroupError::GroupNotExist)?.clone(); let mut config = groups.get_group_config(acct).ok_or(GroupError::GroupNotExist)?.clone();
drop(groups);
println!("--- Re-authenticating bot user @{} ---", acct); println!("--- Re-authenticating bot user @{} ---", acct);
let registration = Registration::new(config.get_appdata().base.to_string()) let registration = Registration::new(config.get_appdata().base.to_string())
@ -92,6 +94,8 @@ impl ConfigStore {
.await?; .await?;
let client = elefren::helpers::cli::authenticate(registration).await?; let client = elefren::helpers::cli::authenticate(registration).await?;
println!("Auth complete");
let appdata = client.data.clone(); let appdata = client.data.clone();
config.set_appdata(appdata); config.set_appdata(appdata);
@ -106,8 +110,8 @@ impl ConfigStore {
/// Spawn existing group using saved creds /// Spawn existing group using saved creds
pub async fn spawn_groups(self: Arc<Self>) -> Vec<GroupHandle> { pub async fn spawn_groups(self: Arc<Self>) -> Vec<GroupHandle> {
let groups = self.data.read().await; let groups = self.data.read().await.clone();
let groups_iter = groups.iter_groups().cloned(); let groups_iter = groups.groups.into_values();
// Connect in parallel // Connect in parallel
futures::stream::iter(groups_iter) futures::stream::iter(groups_iter)
@ -158,9 +162,15 @@ impl ConfigStore {
//noinspection RsSelfConvention //noinspection RsSelfConvention
/// Set group config to the store. The store then saved. /// Set group config to the store. The store then saved.
pub(crate) async fn set_group_config(&self, config: GroupConfig) -> Result<(), GroupError> { pub(crate) async fn set_group_config(&self, config: GroupConfig) -> Result<(), GroupError> {
let mut data = self.data.write().await; debug!("Locking mutex");
if let Ok(mut data) = tokio::time::timeout(Duration::from_secs(1), self.data.write()).await {
debug!("Locked");
data.set_group_config(config); data.set_group_config(config);
debug!("Writing file");
self.persist(&data).await?; self.persist(&data).await?;
} else {
error!("DEADLOCK? Timeout waiting for data RW Lock in settings store");
}
Ok(()) Ok(())
} }
@ -187,6 +197,7 @@ fn make_scopes() -> Scopes {
| Scopes::read(scopes::Read::Follows) | Scopes::read(scopes::Read::Follows)
| Scopes::write(scopes::Write::Statuses) | Scopes::write(scopes::Write::Statuses)
| Scopes::write(scopes::Write::Media) | Scopes::write(scopes::Write::Media)
| Scopes::write(scopes::Write::Follows)
} }
// trait TapOk<T> { // trait TapOk<T> {

Loading…
Cancel
Save