add notif dedup to avoid duplicate announcement

.gitignore

@@ -5,3 +5,6 @@ group-actor-data.toml
 groups.json
 fedigroups
 *.bak
+*.old
+/groups
+/groups.d

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## v0.3.0
+- Changed config/storage format to directory-based, removed shared config mutex
+- Made more options configurable (timeouts, catch-up limits, etc)
+- Changed default log level to Debug, added `-q` to reduce it (opposite of `-v`)
+- Code cleaning
+
+## v0.2.8
+- fix error processing statuses when a misskey poll has infinite run time
+
 ## v0.2.7
 - Fix some wrong responses to admin commands
 - Remove automatic announcements from some admin commands

Cargo.lock

@@ -276,7 +276,7 @@ checksum = "ea57b42383d091c85abcc2706240b94ab2a8fa1fc81c10ff23c4de06e2a90b5e"
 [[package]]
 name = "elefren"
 version = "0.22.0"
-source = "git+https://git.ondrovo.com/MightyPork/elefren-fork.git?rev=7847df0#7847df0e2b2c0f6b9a089e2db2f9b10dfe070a45"
+source = "git+https://git.ondrovo.com/MightyPork/elefren-fork.git?rev=b10e5935ae32f4756b19e9ca58b78a5382f865d1#b10e5935ae32f4756b19e9ca58b78a5382f865d1"
 dependencies = [
  "chrono",
  "doc-comment",
@@ -328,7 +328,7 @@ checksum = "e88a8acf291dafb59c2d96e8f59828f3838bb1a70398823ade51a84de6a6deed"
 
 [[package]]
 name = "fedigroups"
-version = "0.2.7"
+version = "0.3.0"
 dependencies = [
  "anyhow",
  "clap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fedigroups"
-version = "0.2.7"
+version = "0.3.0"
 authors = ["Ondřej Hruška <ondra@ondrovo.com>"]
 edition = "2018"
 publish = false
@@ -10,7 +10,7 @@ build = "build.rs"
 
 [dependencies]
 #elefren = { path = "../elefren22-fork" }
-elefren = { git = "https://git.ondrovo.com/MightyPork/elefren-fork.git", rev = "7847df0" }
+elefren = { git = "https://git.ondrovo.com/MightyPork/elefren-fork.git", rev = "b10e5935ae32f4756b19e9ca58b78a5382f865d1" }
 
 env_logger = "0.9.0"
 

README.md

@@ -36,7 +36,7 @@ You can also run the program using Cargo, that is handy for development: `cargo
 3. **Make sure you auth as the correct user!** 
 4. Paste the Oauth2 token you got into the terminal, hit enter.
 
-The program now ends. The credentials are saved in a file `groups.json`.
+The program now ends. The credentials are saved in the directory `groups.d/account@server/`, which is created if missing.
 
 You can repeat this for any number of groups.
 
@@ -44,55 +44,150 @@ In case you need to re-authenticate an existing group, do the same but use `-A`
 
 ### Editing config
 
-**Do not edit the config while the group service is running, it will overwrite your changes!**
+**JSON does not support comments! Remove comments before using examples copied from this guide!**
 
-The JSON file is easily editable, you can e.g. add yourself as an admin (use the e-mail format, e.g. `piggo@piggo.space`). 
+A typical setup could look like this:
-The file format is quite self-explanatory.
+
+```
+├── groups.d
+│   ├── betty@piggo.space
+│   │   ├── config.json
+│   │   ├── control.json
+│   │   └── state.json
+│   └── chatterbox@botsin.space
+│       ├── config.json
+│       ├── control.json
+│       └── state.json
+└── groups.json
+```
+
+#### Common config
+
+There is one shared config file: `groups.json`
+
+- If the file does not exist, default settings are used. This is usually good enough.
+- This file applies to all groups
+- Prior to 0.3, the groups were also configured here. 
+- Running 0.3+ with the old file will produce an error, you need to update the config before continuing - move groups to subfolders
+
+```
+{
+    // Max number of missed notifs to process after connect
+    max_catchup_notifs: 30,
+    // Max number of missed statuses to process after connect
+    max_catchup_statuses: 50,
+    // Delay between fetched pages when catching up
+    delay_fetch_page_s: 0.25,
+    // Delay after sending a status, making a follow or some other action.
+    // Set if there are Throttled errors and you need to slow the service down.
+    delay_after_post_s: 0.0,
+    // Delay before trying to re-connect after the server closed the socket
+    delay_reopen_closed_s: 0.5,
+    // Delay before trying to re-connect after an error
+    delay_reopen_error_s: 5.0,
+    // Timeout for a notification/timeline socket to be considered alive.
+    // If nothing arrives in this interval, reopen it. Some servers have a buggy socket
+    // implementation where it stays open but no longer works.
+    socket_alive_timeout_s: 30.0,
+    // Time after which a socket is always closed, even if seemingly alive.
+    // This is a work-around for servers that stop sending notifs after a while.
+    socket_retire_time_s: 120.0,
+}
+```
+
+#### Per-group config
+
+Each group is stored as a sub-directory of `groups.d/`. The sub-directories are normally named after their accounts, 
+but this is not required. For example, `groups.d/betty@piggo.space/`.
+
+The group's config and state is split into three files in a way that minimizes the risk of data loss.
+
+Only the `config.json` file with credentials is required; the others will be created as needed by the group daemon.
+
+- `config.json` - immutable config, never changed beside when you run the `-A` command to reauth a group.
+  This is where the account name, the auth token and the `enabled` flag are stored.
+- `control.json` - settings and state that change at runtime or can be set using slash commands.
+  This file is overwritten by the group service when needed.
+- `state.json` - frequently changing state data. The last-seen status/notification timestamps are kept here.
+  State is split from Control to limit the write frequency of the control file. Timestamps can be updated multiple times
+  per minute.
+
+**Do not edit the control and state files while the group service is running, it may overwrite your changes!**
+
+The JSON files are easily editable, you can e.g. add yourself as an admin (use the e-mail format, e.g. `piggo@piggo.space`).
+Note that changing config externally requires a restart. It's better to use slash commands and update the config at run-time.
+
+When adding hashtags, *they must be entered as lowercase* and without the `#` symbol!
+
+The file formats are quite self-explanatory (again, remove comments before copying, JSON does not support comments!)
+
+**config.json**
 
 ```json
 {
-  "groups": {
+  // Enable or disable the group service
-    "group@myserver.xyz": {
+  "enabled": true,
-      "enabled": true,
+  // Group account name
-      "acct": "group@myserver.xyz",
+  "acct": "group@myserver.xyz",
-      "appdata": {
+  // Saved mastodon API credentials
-        "base": "https://myserver.xyz",
+  "appdata": {
-        "client_id": "...",
+    "base": "https://myserver.xyz",
-        "client_secret": "...",
+    "client_id": "...",
-        "redirect": "urn:ietf:wg:oauth:2.0:oob",
+    "client_secret": "...",
-        "token": "..."
+    "redirect": "urn:ietf:wg:oauth:2.0:oob",
-      },
+    "token": "..."
-      "group_tags": [
-        "grouptest"
-      ],
-      "admin_users": [
-        "admin@myserver.xyz"
-      ],
-      "member_only": false,
-      "member_users": [],
-      "banned_users": [],
-      "banned_servers": [
-        "bad-stuff-here.cc"
-      ],
-      "last_notif_ts": 1630011219000,
-      "last_status_ts": 1630011362000
-    }
   }
 }
 ```
 
-- `group_tags` - group hashtags (without the `#`). The group reblogs anything with these hashtags if the author is a member.
+**control.json**
-- `member_users` - group members, used to track whose hashtags should be reblogged; in member-only groups, this is also a user whitelist.
+
-- `banned_users` - can't post or interact with the group service
+```json
-- `banned_servers` - work like an instance block
+{
+  // List of group hashtags, lowercase.
+  // The group reblogs anything with these hashtags if the author is a member.
+  "group_tags": [
+    "grouptest"
+  ],
+  // List of admin users (e-mail format)
+  "admin_users": [
+    "admin@myserver.xyz"
+  ],
+  // Restrict write access to manually added members
+  "member_only": false,
+  // List of member users (e-mail format)
+  "member_users": [],
+  // List of banned users (e-mail format), their posts and actions are ignored by the group
+  "banned_users": [],
+  // List of banned servers, users from there can't interact with the group and their posts can't be shared.
+  "banned_servers": [
+    "bad-stuff-here.cc"
+  ]
+}
+```
+
+**state.json**
+
+Internal use, millisecond timestamps of the last-seen status and notification.
+
+```json
+{
+  "last_notif_ts": 1630011219000,
+  "last_status_ts": 1630011362000
+}
+```
 
 ### Running
 
-To run the group service, simply run it with no arguments. It will read what to do from `groups.json`.
+To run the group service, simply run it with no arguments. 
+It will read the `groups.json` file (if present), find groups in `groups.d/` and start the services for you.
 
-Note that the file must be writable, it is updated at run-time.
+Note that the control and status files must be writable, they are updated at run-time.
+Config files can have limited permissions to avoid accidental overwrite.
 
-An example systemd service file is included in the repository as well. Make sure to set up the system user/group and file permissions according to your needs. You can use targets in the included `Makefile` to manage the systemd service and look at logs.
+An example systemd service file is included in the repository as well.
+Make sure to set up the system user/group and file permissions according to your needs. 
+You can use targets in the included `Makefile` to manage the systemd service and look at logs.
 
 ## Group usage
 
@@ -116,7 +211,6 @@ These won't be shared:
 - `ducks suck`
 - `@group #ducks /i` (anything with the "ignore" command is ignored)
 - `@group /remove #ducks` (admin command, even if it includes a group hashtag)
-- `@otheruser tell me about ducks` (in a thread)
 - `@otheruser @group tell me about ducks` (in a thread)
 
 ### Commands

src/command.rs

@@ -1,6 +1,6 @@
+use crate::utils;
 use once_cell::sync::Lazy;
 use regex::Regex;
-use crate::utils;
 
 #[derive(Debug, Clone, PartialEq)]
 pub enum StatusCommand {
@@ -101,7 +101,8 @@ static RE_UNBAN_SERVER: once_cell::sync::Lazy<Regex> = Lazy::new(|| command!(r"u
 
 static RE_ADD_MEMBER: once_cell::sync::Lazy<Regex> = Lazy::new(|| command!(r"(?:add|follow)\s+", p_user!()));
 
-static RE_REMOVE_MEMBER: once_cell::sync::Lazy<Regex> = Lazy::new(|| command!(r"(?:kick|unfollow|remove)\s+", p_user!()));
+static RE_REMOVE_MEMBER: once_cell::sync::Lazy<Regex> =
+    Lazy::new(|| command!(r"(?:kick|unfollow|remove)\s+", p_user!()));
 
 static RE_ADD_TAG: once_cell::sync::Lazy<Regex> = Lazy::new(|| command!(r"(?:add|follow)\s+", p_hashtag!()));
 
@@ -134,8 +135,7 @@ static RE_PING: once_cell::sync::Lazy<Regex> = Lazy::new(|| command!(r"ping"));
 static RE_ANNOUNCE: once_cell::sync::Lazy<Regex> =
     Lazy::new(|| Regex::new(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(r"(?:^|\b|\s|>|\n)#(\w+)").unwrap());
-    Lazy::new(|| Regex::new(r"(?:^|\b|\s|>|\n)#(\w+)").unwrap());
 
 pub static RE_NOBOT_TAG: once_cell::sync::Lazy<Regex> =
     Lazy::new(|| Regex::new(r"(?:^|\b|\s|>|\n)#nobot(?:\b|$)").unwrap());
@@ -336,11 +336,15 @@ pub fn parse_slash_commands(content: &str) -> Vec<StatusCommand> {
 
 #[cfg(test)]
 mod test {
-    use crate::command::{parse_slash_commands, RE_A_HASHTAG, RE_HASHTAG_TRIGGERING_PLEROMA_BUG, RE_NOBOT_TAG, RE_ADD_TAG, RE_JOIN, StatusCommand};
+    use crate::command::{
+        parse_slash_commands, StatusCommand, RE_ADD_TAG, RE_A_HASHTAG, RE_HASHTAG_TRIGGERING_PLEROMA_BUG, RE_JOIN,
+        RE_NOBOT_TAG,
+    };
 
     use super::{
         RE_ADD_MEMBER, RE_ANNOUNCE, RE_BAN_SERVER, RE_BAN_USER, RE_BOOST, RE_CLOSE_GROUP, RE_GRANT_ADMIN, RE_HELP,
-        RE_IGNORE, RE_LEAVE, RE_OPTOUT, RE_OPTIN, RE_MEMBERS, RE_OPEN_GROUP, RE_REMOVE_MEMBER, RE_REVOKE_ADMIN, RE_TAGS, RE_UNDO,
+        RE_IGNORE, RE_LEAVE, RE_MEMBERS, RE_OPEN_GROUP, RE_OPTIN, RE_OPTOUT, RE_REMOVE_MEMBER, RE_REVOKE_ADMIN,
+        RE_TAGS, RE_UNDO,
     };
 
     #[test]
@@ -556,7 +560,9 @@ mod test {
         assert!(RE_A_HASHTAG.is_match("#banana"));
         assert!(RE_A_HASHTAG.is_match("#ласточка"));
         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 #городДляЛюдей aaa"));
 
         for (i, c) in RE_A_HASHTAG.captures_iter("foo #banana #χαλβάς #ласточка").enumerate() {
             if i == 0 {

src/error.rs

@@ -10,8 +10,6 @@ pub enum GroupError {
     UserOptedOut,
     #[error("Server could not be banned because there are admin users on it")]
     AdminsOnServer,
-    #[error("Group config is missing in the config store")]
-    GroupNotExist,
     #[error("Config error: {0}")]
     BadConfig(Cow<'static, str>),
     #[error("API request timed out")]
@@ -32,7 +30,6 @@ impl PartialEq for GroupError {
             (Self::UserIsAdmin, Self::UserIsAdmin)
                 | (Self::UserIsBanned, Self::UserIsBanned)
                 | (Self::AdminsOnServer, Self::AdminsOnServer)
-                | (Self::GroupNotExist, Self::GroupNotExist)
                 | (Self::BadConfig(_), Self::BadConfig(_))
         )
     }

src/group_handler/handle_mention.rs

@@ -1,22 +1,25 @@
+use std::cmp::Ordering;
 use std::collections::HashSet;
 use std::time::Duration;
 
-use elefren::{FediClient, SearchType, StatusBuilder};
 use elefren::entities::account::Account;
 use elefren::entities::prelude::Status;
 use elefren::status_builder::Visibility;
+use elefren::{FediClient, SearchType, StatusBuilder};
 
-use crate::command::{RE_NOBOT_TAG, StatusCommand};
+use crate::command::{StatusCommand, RE_NOBOT_TAG};
 use crate::error::GroupError;
 use crate::group_handler::GroupHandle;
-use crate::store::data::GroupConfig;
+use crate::store::group_config::GroupConfig;
-use crate::utils::{LogError, normalize_acct};
+use crate::store::CommonConfig;
 use crate::utils;
+use crate::utils::{normalize_acct, LogError};
 
 pub struct ProcessMention<'a> {
     status: Status,
     group_account: &'a Account,
     config: &'a mut GroupConfig,
+    cc: &'a CommonConfig,
     client: &'a mut FediClient,
     group_acct: String,
     status_acct: String,
@@ -33,13 +36,13 @@ impl<'a> ProcessMention<'a> {
     async fn lookup_acct_id(&self, acct: &str, followed: bool) -> Result<Option<String>, GroupError> {
         debug!("Looking up user ID by acct: {}", acct);
 
-        match tokio::time::timeout(Duration::from_secs(5), self.client.search_v2(
+        match tokio::time::timeout(
-            acct,
+            Duration::from_secs(5),
-            !followed,
+            self.client
-            Some(SearchType::Accounts),
+                .search_v2(acct, !followed, Some(SearchType::Accounts), Some(1), followed),
-            Some(1),
+        )
-            followed,
+        .await
-        )).await {
+        {
             Err(_) => {
                 warn!("Account lookup timeout!");
                 Err(GroupError::ApiTimeout)
@@ -50,6 +53,7 @@ impl<'a> ProcessMention<'a> {
             }
             Ok(Ok(res)) => {
                 for item in res.accounts {
+                    // XXX limit is 1!
                     let acct_normalized = normalize_acct(&item.acct, &self.group_acct)?;
                     if acct_normalized == acct {
                         debug!("Search done, account found: {}", item.acct);
@@ -91,12 +95,14 @@ impl<'a> ProcessMention<'a> {
     async fn follow_user_by_id(&self, id: &str) -> Result<(), GroupError> {
         debug!("Trying to follow user #{}", id);
         self.client.follow(id).await?;
+        self.delay_after_post().await;
         Ok(())
     }
 
     async fn unfollow_user_by_id(&self, id: &str) -> Result<(), GroupError> {
         debug!("Trying to unfollow user #{}", id);
         self.client.unfollow(id).await?;
+        self.delay_after_post().await;
         Ok(())
     }
 
@@ -113,6 +119,7 @@ impl<'a> ProcessMention<'a> {
             group_account: &gh.group_account,
             status_user_id: status.account.id.to_string(),
             client: &mut gh.client,
+            cc: &gh.cc,
             can_write: gh.config.can_write(&status_acct),
             is_admin: gh.config.is_admin(&status_acct),
             replies: vec![],
@@ -129,16 +136,15 @@ impl<'a> ProcessMention<'a> {
     }
 
     async fn reblog_status(&self) {
-        self.client.reblog(&self.status.id)
+        self.client.reblog(&self.status.id).await.log_error("Failed to reblog status");
-            .await
+        self.delay_after_post().await;
-            .log_error("Failed to reblog status")
     }
 
     fn add_reply(&mut self, line: impl AsRef<str>) {
         self.replies.push(line.as_ref().trim_matches(' ').to_string())
     }
 
-    fn add_announcement<'t>(&mut self, line: impl AsRef<str>) {
+    fn add_announcement(&mut self, line: impl AsRef<str>) {
         self.announcements.push(line.as_ref().trim_matches(' ').to_string())
     }
 
@@ -156,8 +162,7 @@ impl<'a> ProcessMention<'a> {
             for cmd in commands {
                 match cmd {
                     StatusCommand::Undo => {
-                        self.cmd_undo().await
+                        self.cmd_undo().await.log_error("Error handling undo cmd");
-                            .log_error("Error handling undo cmd");
                     }
                     StatusCommand::Ignore => {
                         unreachable!(); // Handled above
@@ -169,12 +174,10 @@ impl<'a> ProcessMention<'a> {
                         self.cmd_boost().await;
                     }
                     StatusCommand::BanUser(u) => {
-                        self.cmd_ban_user(&u).await
+                        self.cmd_ban_user(&u).await.log_error("Error handling ban-user cmd");
-                            .log_error("Error handling ban-user cmd");
                     }
                     StatusCommand::UnbanUser(u) => {
-                        self.cmd_unban_user(&u).await
+                        self.cmd_unban_user(&u).await.log_error("Error handling unban-user cmd");
-                            .log_error("Error handling unban-user cmd");
                     }
                     StatusCommand::OptOut => {
                         self.cmd_optout().await;
@@ -189,12 +192,10 @@ impl<'a> ProcessMention<'a> {
                         self.cmd_unban_server(&s).await;
                     }
                     StatusCommand::AddMember(u) => {
-                        self.cmd_add_member(&u).await
+                        self.cmd_add_member(&u).await.log_error("Error handling add-member cmd");
-                            .log_error("Error handling add-member cmd");
                     }
                     StatusCommand::RemoveMember(u) => {
-                        self.cmd_remove_member(&u).await
+                        self.cmd_remove_member(&u).await.log_error("Error handling remove-member cmd");
-                            .log_error("Error handling remove-member cmd");
                     }
                     StatusCommand::AddTag(tag) => {
                         self.cmd_add_tag(tag).await;
@@ -203,12 +204,10 @@ impl<'a> ProcessMention<'a> {
                         self.cmd_remove_tag(tag).await;
                     }
                     StatusCommand::GrantAdmin(u) => {
-                        self.cmd_grant_admin(&u).await
+                        self.cmd_grant_admin(&u).await.log_error("Error handling grant-admin cmd");
-                            .log_error("Error handling grant-admin cmd");
                     }
                     StatusCommand::RemoveAdmin(u) => {
-                        self.cmd_revoke_member(&u).await
+                        self.cmd_revoke_member(&u).await.log_error("Error handling grant-admin cmd");
-                            .log_error("Error handling grant-admin cmd");
                     }
                     StatusCommand::OpenGroup => {
                         self.cmd_open_group().await;
@@ -239,13 +238,14 @@ impl<'a> ProcessMention<'a> {
         }
 
         if self.do_boost_prev_post {
-            if let (Some(prev_acct_id), Some(prev_status_id)) = (self.status.in_reply_to_account_id.as_ref(), self.status.in_reply_to_id.as_ref()) {
+            if let (Some(prev_acct_id), Some(prev_status_id)) = (
+                self.status.in_reply_to_account_id.as_ref(),
+                self.status.in_reply_to_id.as_ref(),
+            ) {
                 match self.id_to_acct_check_boostable(prev_acct_id).await {
                     Ok(_acct) => {
-                        self.client
+                        self.client.reblog(prev_status_id).await.log_error("Failed to boost");
-                            .reblog(prev_status_id)
+                        self.delay_after_post().await;
-                            .await
-                            .log_error("Failed to boost");
                     }
                     Err(e) => {
                         warn!("Can't reblog: {}", e);
@@ -262,19 +262,8 @@ impl<'a> ProcessMention<'a> {
                 apply_trailing_hashtag_pleroma_bug_workaround(&mut msg);
             }
 
-            if let Ok(post) = StatusBuilder::new()
+            let mention = format!("@{user} ", user = self.status_acct);
-                .status(format!("@{user} {msg}", user = self.status_acct, msg = msg))
+            self.send_reply_multipart(mention, msg).await?;
-                .content_type(if self.want_markdown {
-                    "text/markdown"
-                } else {
-                    "text/plain"
-                })
-                .visibility(Visibility::Direct)
-                .build()
-            {
-                let _ = self.client.new_status(post)
-                    .await.log_error("Failed to post");
-            }
         }
 
         if !self.announcements.is_empty() {
@@ -285,15 +274,62 @@ impl<'a> ProcessMention<'a> {
                 apply_trailing_hashtag_pleroma_bug_workaround(&mut msg);
             }
 
-            let post = StatusBuilder::new()
+            let msg = format!("**📢 Group announcement**\n{msg}", msg = msg);
-                .status(format!("**📢 Group announcement**\n{msg}", msg = msg))
+            self.send_announcement_multipart(&msg).await?;
-                .content_type("text/markdown")
+        }
-                .visibility(Visibility::Public)
-                .build()
-                .expect("error build status");
 
-            let _ = self.client.new_status(post)
+        Ok(())
-                .await.log_error("Failed to post");
+    }
+
+    async fn send_reply_multipart(&self, mention: String, msg: String) -> Result<(), GroupError> {
+        let parts = smart_split(&msg, Some(mention), self.config.get_character_limit());
+
+        let mut parent = self.status.id.clone();
+        for p in parts {
+            if let Ok(post) = StatusBuilder::new()
+                .status(p)
+                .content_type(if self.want_markdown {
+                    "text/markdown"
+                } else {
+                    "text/plain"
+                })
+                .in_reply_to(&parent)
+                .visibility(Visibility::Direct)
+                .build()
+            {
+                let status = self.client.new_status(post).await?;
+                self.delay_after_post().await;
+                parent = status.id;
+            }
+
+            // Sleep a bit to avoid throttling
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+
+        Ok(())
+    }
+
+    async fn send_announcement_multipart(&self, msg: &str) -> Result<(), GroupError> {
+        let parts = smart_split(msg, None, self.config.get_character_limit());
+
+        let mut parent = None;
+        for p in parts {
+            let mut builder = StatusBuilder::new();
+
+            builder.status(p).content_type("text/markdown").visibility(Visibility::Public);
+
+            if let Some(p) = parent.as_ref() {
+                builder.in_reply_to(p);
+            }
+
+            let post = builder.build().expect("error build status");
+
+            let status = self.client.new_status(post).await?;
+            self.delay_after_post().await;
+            parent = Some(status.id);
+
+            // Sleep a bit to avoid throttling
+            tokio::time::sleep(Duration::from_secs(1)).await;
         }
 
         Ok(())
@@ -352,26 +388,28 @@ impl<'a> ProcessMention<'a> {
     }
 
     async fn cmd_undo(&mut self) -> Result<(), GroupError> {
-        if let (Some(ref parent_account_id), Some(ref parent_status_id)) = (&self.status.in_reply_to_account_id, &self.status.in_reply_to_id) {
+        if let (Some(ref parent_account_id), Some(ref parent_status_id)) =
+            (&self.status.in_reply_to_account_id, &self.status.in_reply_to_id)
+        {
             if parent_account_id == &self.group_account.id {
                 // This is a post sent by the group user, likely an announcement.
                 // Undo here means delete it.
                 if self.is_admin {
                     info!("Deleting group post #{}", parent_status_id);
                     self.client.delete_status(parent_status_id).await?;
+                    self.delay_after_post().await;
                 } else {
                     warn!("Only admin can delete posts made by the group user");
                 }
+            } else if self.is_admin || parent_account_id == &self.status_user_id {
+                info!("Un-reblogging post #{}", parent_status_id);
+                // User unboosting own post boosted by accident, or admin doing it
+                self.client.unreblog(parent_status_id).await?;
+                self.delay_after_post().await;
             } else {
-                if self.is_admin || parent_account_id == &self.status_user_id {
+                warn!("Only the author and admins can undo reblogs");
-                    info!("Un-reblogging post #{}", parent_status_id);
+                // XXX this means when someone /b's someone else's post to a group,
-                    // User unboosting own post boosted by accident, or admin doing it
+                //  they then can't reverse that (only admin or the post's author can)
-                    self.client.unreblog(parent_status_id).await?;
-                } else {
-                    warn!("Only the author and admins can undo reblogs");
-                    // XXX this means when someone /b's someone else's post to a group,
-                    //  they then can't reverse that (only admin or the post's author can)
-                }
             }
         }
 
@@ -385,8 +423,7 @@ impl<'a> ProcessMention<'a> {
                 match self.config.ban_user(&u, true) {
                     Ok(_) => {
                         self.add_reply(format!("User {} banned from group!", u));
-                        self.unfollow_by_acct(&u).await
+                        self.unfollow_by_acct(&u).await.log_error("Failed to unfollow banned user");
-                            .log_error("Failed to unfollow banned user");
                     }
                     Err(e) => {
                         self.add_reply(format!("Failed to ban user {}: {}", u, e));
@@ -468,8 +505,7 @@ impl<'a> ProcessMention<'a> {
             match self.config.set_member(&u, true) {
                 Ok(_) => {
                     self.add_reply(format!("User {} added to the group!", u));
-                    self.follow_by_acct(&u)
+                    self.follow_by_acct(&u).await.log_error("Failed to follow");
-                        .await.log_error("Failed to follow");
                 }
                 Err(e) => {
                     self.add_reply(format!("Failed to add user {} to group: {}", u, e));
@@ -487,8 +523,7 @@ impl<'a> ProcessMention<'a> {
             match self.config.set_member(&u, false) {
                 Ok(_) => {
                     self.add_reply(format!("User {} removed from the group.", u));
-                    self.unfollow_by_acct(&u).await
+                    self.unfollow_by_acct(&u).await.log_error("Failed to unfollow removed user");
-                        .log_error("Failed to unfollow removed user");
                 }
                 Err(e) => {
                     self.add_reply(format!("Unexpected error occured: {}", e));
@@ -538,10 +573,7 @@ impl<'a> ProcessMention<'a> {
                         self.add_reply(format!("User {} is now a group admin!", u));
                     }
                     Err(e) => {
-                        self.add_reply(format!(
+                        self.add_reply(format!("Failed to make user {} a group admin: {}", u, e));
-                            "Failed to make user {} a group admin: {}",
-                            u, e
-                        ));
                     }
                 }
             } else {
@@ -619,7 +651,8 @@ impl<'a> ProcessMention<'a> {
             self.add_reply(format!("This is a public-access group. {}", membership_line));
         }
 
-        self.add_reply("\
+        self.add_reply(
+            "\
             To share a post, @ the group user or use a group hashtag.\n\
             \n\
             **Supported commands:**\n\
@@ -628,7 +661,8 @@ impl<'a> ProcessMention<'a> {
             `/tags` - show group hashtags\n\
             `/join` - (re-)join the group\n\
             `/leave` - leave the group\n\
-            `/optout` - forbid sharing of your posts");
+            `/optout` - forbid sharing of your posts",
+        );
 
         if self.is_admin {
             self.add_reply("`/members`, `/who` - show group members / admins");
@@ -641,7 +675,8 @@ impl<'a> ProcessMention<'a> {
         // XXX when used on instance with small character limit, this won't fit!
 
         if self.is_admin {
-            self.add_reply("\n\
+            self.add_reply(
+                "\n\
                 **Admin commands:**\n\
                 `/ping` - check the group works\n\
                 `/add user` - add a member (user@domain)\n\
@@ -655,7 +690,8 @@ impl<'a> ProcessMention<'a> {
                 `/deadmin user` - revoke admin rights\n\
                 `/closegroup` - make member-only\n\
                 `/opengroup` - make public-access\n\
-                `/announce x` - make a public announcement");
+                `/announce x` - make a public announcement",
+            );
         }
     }
 
@@ -684,10 +720,13 @@ impl<'a> ProcessMention<'a> {
         if self.config.is_member_or_admin(&self.status_acct) {
             // admin can leave but that's a bad idea
             let _ = self.config.set_member(&self.status_acct, false);
-            self.add_reply("You're no longer a group member. Unfollow the group user to stop receiving group messages.");
+            self.add_reply(
+                "You're no longer a group member. Unfollow the group user to stop receiving group messages.",
+            );
         }
 
-        self.unfollow_user_by_id(&self.status_user_id).await
+        self.unfollow_user_by_id(&self.status_user_id)
+            .await
             .log_error("Failed to unfollow");
     }
 
@@ -696,34 +735,39 @@ impl<'a> ProcessMention<'a> {
             debug!("Already member or admin, try to follow-back again");
             // Already a member, so let's try to follow the user
             // again, maybe first time it failed
-            self.follow_user_by_id(&self.status_user_id).await
+            self.follow_user_by_id(&self.status_user_id).await.log_error("Failed to follow");
-                .log_error("Failed to follow");
         } else {
             // Not a member yet
             if self.config.is_member_only() {
                 // No you can't
-                self.add_reply("\
+                self.add_reply(
+                    "\
                     Sorry, this group is closed to new sign-ups.\n\
-                    Please ask one of the group admins to add you:");
+                    Please ask one of the group admins to add you:",
+                );
 
                 self.append_admin_list_to_reply();
             } else {
                 // Open access, try to follow back
-                self.follow_user_by_id(&self.status_user_id).await
+                self.follow_user_by_id(&self.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
                 let _ = self.config.set_member(&self.status_acct, true);
-                self.add_reply("\
+                self.add_reply(
+                    "\
                     Welcome to the group! The group user will now follow you to complete the sign-up. \
                     Make sure you follow back to receive shared posts!\n\n\
-                    Use /help for more info.");
+                    Use /help for more info.",
+                );
             }
         }
     }
 
     async fn cmd_ping(&mut self) {
-        self.add_reply(format!("pong, this is fedigroups service v{}", env!("CARGO_PKG_VERSION")));
+        self.add_reply(format!(
+            "pong, this is fedigroups service v{}",
+            env!("CARGO_PKG_VERSION")
+        ));
     }
 
     async fn unfollow_by_acct(&self, acct: &str) -> Result<(), GroupError> {
@@ -753,20 +797,223 @@ impl<'a> ProcessMention<'a> {
         } else {
             let normalized = normalize_acct(&account.acct, &self.group_acct)?;
             if self.config.is_banned(&normalized) {
-                return Err(GroupError::UserIsBanned);
+                Err(GroupError::UserIsBanned)
             } else if self.config.is_optout(&normalized) {
-                return Err(GroupError::UserOptedOut);
+                Err(GroupError::UserOptedOut)
             } else {
                 Ok(normalized)
             }
         }
     }
+
+    async fn delay_after_post(&self) {
+        tokio::time::sleep(Duration::from_secs_f64(self.cc.delay_after_post_s)).await;
+    }
 }
 
 fn apply_trailing_hashtag_pleroma_bug_workaround(msg: &mut String) {
-    if crate::command::RE_HASHTAG_TRIGGERING_PLEROMA_BUG.is_match(&msg) {
+    if crate::command::RE_HASHTAG_TRIGGERING_PLEROMA_BUG.is_match(msg) {
         // if a status ends with a hashtag, pleroma will fuck it up
         debug!("Adding \" .\" to fix pleroma hashtag eating bug!");
         msg.push_str(" .");
     }
 }
+
+fn smart_split(msg: &str, prefix: Option<String>, limit: usize) -> Vec<String> {
+    let prefix = prefix.unwrap_or_default();
+
+    if msg.len() + prefix.len() < limit {
+        return vec![format!("{}{}", prefix, msg)];
+    }
+
+    let mut parts_to_send = vec![];
+    let mut this_piece = prefix.clone();
+    for l in msg.split('\n') {
+        println!("* Line: {:?}", l);
+
+        match (this_piece.len() + l.len()).cmp(&limit) {
+            Ordering::Less => {
+                println!("append line");
+                // this line still fits comfortably
+                this_piece.push_str(l);
+                this_piece.push('\n');
+            }
+            Ordering::Equal => {
+                println!("exactly fits within limit");
+                // this line exactly reaches the limit
+                this_piece.push_str(l);
+                parts_to_send.push(std::mem::take(&mut this_piece).trim().to_owned());
+                this_piece.push_str(&prefix);
+            }
+            Ordering::Greater => {
+                println!("too long to append (already {} + new {})", this_piece.len(), l.len());
+                // line too long to append
+                if this_piece != prefix {
+                    let trimmed = this_piece.trim();
+                    if !trimmed.is_empty() {
+                        println!("flush buffer: {:?}", trimmed);
+                        parts_to_send.push(trimmed.to_owned());
+                    }
+                }
+
+                // start new piece with the line. If the line is too long, break it up.
+                this_piece = format!("{}{}", prefix, l);
+
+                while this_piece.len() > limit {
+                    // line too long, try splitting at the last space, if any
+                    let to_send = if let Some(last_space) = (&this_piece[..=limit]).rfind(' ') {
+                        println!("line split at word boundary");
+                        let mut p = this_piece.split_off(last_space + 1);
+                        std::mem::swap(&mut p, &mut this_piece);
+                        p
+                    } else {
+                        println!("line split at exact len (no word boundary found)");
+                        let mut p = this_piece.split_off(limit);
+                        std::mem::swap(&mut p, &mut this_piece);
+                        p
+                    };
+                    let part_trimmed = to_send.trim();
+                    println!("flush buffer: {:?}", part_trimmed);
+                    parts_to_send.push(part_trimmed.to_owned());
+                    this_piece = format!("{}{}", prefix, this_piece.trim());
+                }
+                this_piece.push('\n');
+            }
+        }
+    }
+
+    if this_piece != prefix {
+        let leftover_trimmed = this_piece.trim();
+        if !leftover_trimmed.is_empty() {
+            println!("flush buffer: {:?}", leftover_trimmed);
+            parts_to_send.push(leftover_trimmed.to_owned());
+        }
+    }
+
+    parts_to_send
+}
+
+#[cfg(test)]
+mod test {
+    #[test]
+    fn test_smart_split_lines() {
+        let to_split = "a234567890\nb234567890\nc234567890\nd234\n67890\ne234567890\n";
+
+        let parts = super::smart_split(to_split, None, 10);
+        assert_eq!(
+            vec![
+                "a234567890".to_string(),
+                "b234567890".to_string(),
+                "c234567890".to_string(),
+                "d234\n67890".to_string(),
+                "e234567890".to_string(),
+            ],
+            parts
+        );
+    }
+
+    #[test]
+    fn test_smart_split_nosplit() {
+        let to_split = "foo\nbar\nbaz";
+
+        let parts = super::smart_split(to_split, None, 1000);
+        assert_eq!(vec!["foo\nbar\nbaz".to_string(),], parts);
+    }
+
+    #[test]
+    fn test_smart_split_nosplit_prefix() {
+        let to_split = "foo\nbar\nbaz";
+        let parts = super::smart_split(to_split, Some("PREFIX".to_string()), 1000);
+        assert_eq!(vec!["PREFIXfoo\nbar\nbaz".to_string(),], parts);
+    }
+
+    #[test]
+    fn test_smart_split_prefix_each() {
+        let to_split = "1234\n56\n7";
+        let parts = super::smart_split(to_split, Some("PREFIX".to_string()), 10);
+        assert_eq!(vec!["PREFIX1234".to_string(), "PREFIX56\n7".to_string(),], parts);
+    }
+
+    #[test]
+    fn test_smart_split_words() {
+        let to_split = "one two three four five six seven eight nine ten";
+        let parts = super::smart_split(to_split, None, 10);
+        assert_eq!(
+            vec![
+                "one two".to_string(),
+                "three four".to_string(),
+                "five six".to_string(),
+                "seven".to_string(),
+                "eight nine".to_string(),
+                "ten".to_string(),
+            ],
+            parts
+        );
+    }
+
+    #[test]
+    fn test_smart_split_words_multispace() {
+        let to_split = "one two      three four    five six     seven    eight nine     ten   ";
+        let parts = super::smart_split(to_split, None, 10);
+        assert_eq!(
+            vec![
+                "one two".to_string(),
+                "three four".to_string(),
+                "five six".to_string(),
+                "seven".to_string(),
+                "eight nine".to_string(),
+                "ten".to_string(),
+            ],
+            parts
+        );
+    }
+
+    #[test]
+    fn test_smart_split_words_longword() {
+        let to_split = "one two threefourfive six";
+        let parts = super::smart_split(to_split, None, 10);
+        assert_eq!(
+            vec!["one two".to_string(), "threefourf".to_string(), "ive six".to_string(),],
+            parts
+        );
+    }
+
+    #[test]
+    fn test_smart_split_words_prefix() {
+        let to_split = "one two three four five six seven eight nine ten";
+        let parts = super::smart_split(to_split, Some("PREFIX".to_string()), 15);
+        assert_eq!(
+            vec![
+                "PREFIXone two".to_string(),
+                "PREFIXthree".to_string(),
+                "PREFIXfour five".to_string(),
+                "PREFIXsix seven".to_string(),
+                "PREFIXeight".to_string(),
+                "PREFIXnine ten".to_string(),
+            ],
+            parts
+        );
+    }
+
+    #[test]
+    fn test_smart_split_realistic() {
+        let to_split = "\
+            Lorem ipsum dolor sit amet, consectetur adipiscing elit.\n\
+            Aenean venenatis libero ac ex suscipit, nec efficitur arcu convallis.\n\
+            Nulla ante neque, efficitur nec fermentum a, fermentum nec nisl.\n\
+            Sed dolor ex, vestibulum at malesuada ut, faucibus ac ante.\n\
+            Nullam scelerisque magna dui, id tempor purus faucibus sit amet.\n\
+            Curabitur pretium condimentum pharetra.\n\
+            Aenean dictum, tortor et ultrices fermentum, mauris erat vehicula lectus.\n\
+            Nec varius mauris sem sollicitudin dolor. Nunc porta in urna nec vulputate.";
+        let parts = super::smart_split(to_split, Some("@pepa@pig.club ".to_string()), 140);
+        assert_eq!(vec![
+            "@pepa@pig.club Lorem ipsum dolor sit amet, consectetur adipiscing elit.".to_string(),
+            "@pepa@pig.club Aenean venenatis libero ac ex suscipit, nec efficitur arcu convallis.".to_string(),
+            "@pepa@pig.club Nulla ante neque, efficitur nec fermentum a, fermentum nec nisl.\nSed dolor ex, vestibulum at malesuada ut, faucibus ac ante.".to_string(),
+            "@pepa@pig.club Nullam scelerisque magna dui, id tempor purus faucibus sit amet.\nCurabitur pretium condimentum pharetra.".to_string(),
+            "@pepa@pig.club Aenean dictum, tortor et ultrices fermentum, mauris erat vehicula lectus.".to_string(),
+            "@pepa@pig.club Nec varius mauris sem sollicitudin dolor. Nunc porta in urna nec vulputate.".to_string(),
+        ], parts);
+    }
+}

src/group_handler/mod.rs

@@ -1,73 +1,119 @@
+use std::collections::VecDeque;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use elefren::{FediClient, StatusBuilder};
 use elefren::debug::EventDisplay;
 use elefren::debug::NotificationDisplay;
 use elefren::debug::StatusDisplay;
+use elefren::entities::account::Account;
 use elefren::entities::event::Event;
 use elefren::entities::notification::{Notification, NotificationType};
 use elefren::entities::status::Status;
 use elefren::status_builder::Visibility;
+use elefren::{FediClient, StatusBuilder};
 use futures::StreamExt;
 
 use handle_mention::ProcessMention;
 
-use crate::error::GroupError;
-use crate::store::ConfigStore;
-use crate::store::data::GroupConfig;
-use crate::utils::{LogError, normalize_acct, VisExt};
 use crate::command::StatusCommand;
-use elefren::entities::account::Account;
+use crate::error::GroupError;
+use crate::store::CommonConfig;
+use crate::store::GroupConfig;
+use crate::utils::{normalize_acct, LogError, VisExt};
 
 mod handle_mention;
 
 /// This is one group's config store capable of persistence
 #[derive(Debug)]
 pub struct GroupHandle {
-    pub(crate) group_account: Account,
+    pub group_account: Account,
-    pub(crate) client: FediClient,
+    pub client: FediClient,
-    pub(crate) config: GroupConfig,
+    pub config: GroupConfig,
-    pub(crate) store: Arc<ConfigStore>,
+    pub cc: Arc<CommonConfig>,
+    pub internal: GroupInternal,
 }
 
-// const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250);
+#[derive(Debug)]
-const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(500);
+pub struct GroupInternal {
-const MAX_CATCHUP_NOTIFS: usize = 25;
+    recently_seen_notif_statuses: VecDeque<String>,
-// also statuses
+}
-const MAX_CATCHUP_STATUSES: usize = 50;
+
-// higher because we can expect a lot of non-hashtag statuses here
+impl Default for GroupInternal {
-const PERIODIC_SAVE: Duration = Duration::from_secs(60);
+    fn default() -> Self {
-const SOCKET_ALIVE_TIMEOUT: Duration = Duration::from_secs(30);
+        Self {
-const SOCKET_RETIRE_TIME: Duration = Duration::from_secs(120);
+            recently_seen_notif_statuses: VecDeque::new()
-const PING_INTERVAL: Duration = Duration::from_secs(15); // must be < periodic save!
+        }
+    }
+}
+
+// TODO move other options to common_config!
+
+// // const DELAY_BEFORE_ACTION: Duration = Duration::from_millis(250);
+// const DELAY_REOPEN_STREAM: Duration = Duration::from_millis(500);
+// // higher because we can expect a lot of non-hashtag statuses here
+// const SOCKET_ALIVE_TIMEOUT: Duration = Duration::from_secs(30);
+// const SOCKET_RETIRE_TIME: Duration = Duration::from_secs(120);
+
+macro_rules! grp_debug {
+    ($self:ident, $f:expr) => {
+        ::log::debug!(concat!("(@{}) ", $f), $self.config.get_acct());
+    };
+    ($self:ident, $f:expr, $($arg:tt)+) => {
+        ::log::debug!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
+    };
+}
+
+macro_rules! grp_info {
+    ($self:ident, $f:expr) => {
+        ::log::info!(concat!("(@{}) ", $f), $self.config.get_acct());
+    };
+    ($self:ident, $f:expr, $($arg:tt)+) => {
+        ::log::info!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
+    };
+}
+
+macro_rules! grp_trace {
+    ($self:ident, $f:expr) => {
+        ::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct());
+    };
+    ($self:ident, $f:expr, $($arg:tt)+) => {
+        ::log::trace!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
+    };
+}
+
+macro_rules! grp_warn {
+    ($self:ident, $f:expr) => {
+        ::log::warn!(concat!("(@{}) ", $f), $self.config.get_acct());
+    };
+    ($self:ident, $f:expr, $($arg:tt)+) => {
+        ::log::warn!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
+    };
+}
+
+macro_rules! grp_error {
+    ($self:ident, $f:expr) => {
+        ::log::error!(concat!("(@{}) ", $f), $self.config.get_acct());
+    };
+    ($self:ident, $f:expr, $($arg:tt)+) => {
+        ::log::error!(concat!("(@{}) ", $f), $self.config.get_acct(), $($arg)+);
+    };
+}
 
 impl GroupHandle {
+    #[allow(unused)]
     pub async fn save(&mut self) -> Result<(), GroupError> {
-        debug!("Saving group config & status");
+        grp_debug!(self, "Saving group state unconditionally");
-        self.store.set_group_config(self.config.clone()).await?;
+        self.config.save(false).await?;
-        trace!("Saved");
-        self.config.clear_dirty_status();
         Ok(())
     }
 
     pub async fn save_if_needed(&mut self) -> Result<(), GroupError> {
         if self.config.is_dirty() {
-            self.save().await?;
+            grp_debug!(self, "Saving group state due to changes");
+            self.config.save_if_needed(false).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 {
@@ -90,75 +136,87 @@ impl NotifTimestamp for Status {
 
 impl GroupHandle {
     pub async fn run(&mut self) -> Result<(), GroupError> {
-        assert!(PERIODIC_SAVE >= PING_INTERVAL);
-
         loop {
-            debug!("Opening streaming API socket");
+            match self.run_internal().await {
-            let mut next_save = Instant::now() + PERIODIC_SAVE; // so we save at start
+                Ok(()) => unreachable!(),
-            let mut events = self.client.streaming_user().await?;
+                Err(e @ GroupError::BadConfig(_)) => {
+                    grp_error!(self, "ERROR in group handler, aborting! {}", e);
+                    return Err(e);
+                }
+                Err(other) => {
+                    grp_error!(self, "ERROR in group handler, will restart! {}", other);
+                    tokio::time::sleep(Duration::from_secs_f64(self.cc.delay_reopen_error_s)).await;
+                }
+            }
+        }
+    }
+
+    pub async fn run_internal(&mut self) -> Result<(), GroupError> {
+        loop {
+            grp_debug!(self, "Opening streaming API socket");
+
+            // wrapped in a timeout, this seems like the only place the group could hang
+            // (https://git.ondrovo.com/MightyPork/group-actor/issues/8)
+            let mut events = match tokio::time::timeout(Duration::from_secs(3), self.client.streaming_user()).await {
+                Ok(Ok(events)) => events,
+                Ok(Err(e)) => return Err(e.into()),
+                Err(_) => {
+                    return Err(GroupError::ApiTimeout);
+                }
+            };
+
             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 {
                 Ok(true) => {
-                    debug!("Some missed notifs handled");
+                    grp_debug!(self, "Some missed notifs handled");
                 }
                 Ok(false) => {
-                    debug!("No notifs missed");
+                    grp_debug!(self, "No notifs missed");
                 }
                 Err(e) => {
-                    error!("Failed to handle missed notifs: {}", e);
+                    grp_error!(self, "Failed to handle missed notifs: {}", e);
                 }
             }
 
             match self.catch_up_with_missed_statuses().await {
                 Ok(true) => {
-                    debug!("Some missed statuses handled");
+                    grp_debug!(self, "Some missed statuses handled");
                 }
                 Ok(false) => {
-                    debug!("No statuses missed");
+                    grp_debug!(self, "No statuses missed");
                 }
                 Err(e) => {
-                    error!("Failed to handle missed statuses: {}", e);
+                    grp_error!(self, "Failed to handle missed statuses: {}", e);
                 }
             }
 
-            if self.config.is_dirty() {
+            self.save_if_needed().await.log_error("Failed to save");
-                // save asap
-                next_save = Instant::now() - PERIODIC_SAVE
-            }
 
             'rx: loop {
-                if next_save < Instant::now() {
+                let remains_to_idle_close =
-                    trace!("Save time elapsed, saving if needed");
+                    Duration::from_secs_f64(self.cc.socket_alive_timeout_s).saturating_sub(last_rx.elapsed());
-                    self.save_if_needed().await.log_error("Failed to save group");
-                    next_save = Instant::now() + PERIODIC_SAVE;
-                }
 
-                let remains_to_idle_close = SOCKET_ALIVE_TIMEOUT.saturating_sub(last_rx.elapsed());
+                let remains_to_retire =
-                let remains_to_retire = SOCKET_RETIRE_TIME.saturating_sub(socket_open_time.elapsed());
+                    Duration::from_secs_f64(self.cc.socket_retire_time_s).saturating_sub(socket_open_time.elapsed());
 
                 if remains_to_idle_close.is_zero() {
-                    warn!("Socket idle too long, close");
+                    grp_warn!(self, "Socket idle too long, close");
                     break 'rx;
                 }
                 if remains_to_retire.is_zero() {
-                    debug!("Socket open too long, closing");
+                    grp_debug!(self, "Socket open too long, closing");
                     break 'rx;
                 }
 
-                trace!("Waiting for message");
+                let timeout = remains_to_idle_close.min(remains_to_retire).max(Duration::from_secs(1)); // at least 1s
-                let timeout = next_save
-                    .saturating_duration_since(Instant::now())
-                    .min(remains_to_idle_close)
-                    .min(remains_to_retire)
-                    .max(Duration::from_secs(1)); // at least 1s
 
+                grp_debug!(self, "Wait for message {:?}", timeout);
                 match tokio::time::timeout(timeout, events.next()).await {
                     Ok(Some(event)) => {
                         last_rx = Instant::now();
-                        debug!("(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event));
+                        grp_debug!(self, "(@{}) Event: {}", self.config.get_acct(), EventDisplay(&event));
                         match event {
                             Event::Update(status) => {
                                 self.handle_status(status).await.log_error("Error handling a status");
@@ -170,37 +228,33 @@ impl GroupHandle {
                             Event::FiltersChanged => {}
                             Event::Heartbeat => {}
                         }
+
+                        self.save_if_needed().await.log_error("Failed to save");
                     }
                     Ok(None) => {
-                        warn!("Group @{} socket closed, restarting...", self.config.get_acct());
+                        grp_warn!(self, "Group @{} socket closed, restarting...", self.config.get_acct());
                         break 'rx;
                     }
                     Err(_) => {
                         // Timeout so we can save if needed
                     }
                 }
-
-                /* ping is nice, but pleroma still sometimes doesnt send
-                notifs after a while, just let it expire */
-
-                // if last_ping.elapsed() > PING_INTERVAL {
-                //     last_ping = Instant::now();
-                //     trace!("Pinging");
-                //     if events.send_ping()
-                //         .await.is_err() {
-                //         break 'rx;
-                //     }
-                // }
             }
 
-            warn!("Notif stream closed, will reopen");
+            grp_warn!(self, "Notif stream closed, will reopen");
-            tokio::time::sleep(DELAY_REOPEN_STREAM).await;
+            tokio::time::sleep(Duration::from_secs_f64(self.cc.delay_reopen_closed_s)).await;
         }
     }
 
     async fn handle_notification(&mut self, n: Notification) -> Result<(), GroupError> {
-        debug!("Handling notif #{}", n.id);
+        grp_debug!(self, "Handling notif #{}", n.id);
+        grp_trace!(self, "{:?}", n);
+
         let ts = n.timestamp_millis();
+        if ts < self.config.get_last_notif() {
+            grp_debug!(self, "Notif is too old, discard");
+            return Ok(());
+        }
         self.config.set_last_notif(ts);
 
         let group_acct = self.config.get_acct().to_string();
@@ -208,33 +262,34 @@ impl GroupHandle {
         let notif_acct = normalize_acct(&n.account.acct, &group_acct)?;
 
         if notif_acct == group_acct {
-            debug!("This is our post, ignore that");
+            grp_debug!(self, "This is our post, ignore that");
             return Ok(());
         }
 
         if self.config.is_banned(&notif_acct) {
-            warn!("Notification actor {} is banned!", notif_acct);
+            grp_warn!(self, "Notification actor {} is banned!", notif_acct);
             return Ok(());
         }
 
         match n.notification_type {
             NotificationType::Mention => {
                 if let Some(status) = n.status {
-                    self.handle_mention_status(status).await?;
+                    if self.internal.recently_seen_notif_statuses.contains(&status.id) {
+                        grp_debug!(self, "Already saw this notif, discard");
+                    } else {
+                        self.internal.recently_seen_notif_statuses.push_front(status.id.clone());
+                        while self.internal.recently_seen_notif_statuses.len() > 64 {
+                            let _ = self.internal.recently_seen_notif_statuses.pop_back();
+                        }
+
+                        self.handle_mention_status(status).await?;
+                    }
                 }
             }
             NotificationType::Follow => {
-                info!("New follower!");
+                grp_info!(self, "New follower!");
-
                 // Just greet the user always
                 self.handle_new_follow(&notif_acct, notif_user_id).await;
-
-                // if self.config.is_member_or_admin(&notif_acct) {
-                //     // Already joined, just doing something silly, ignore this
-                //     debug!("User already a member, ignoring");
-                // } else {
-                //
-                // }
             }
             NotificationType::Favourite => {}
             NotificationType::Reblog => {}
@@ -246,68 +301,100 @@ impl GroupHandle {
 
     /// Handle a non-mention status for tags
     async fn handle_status(&mut self, s: Status) -> Result<(), GroupError> {
-        debug!("Handling status #{}", s.id);
+        grp_debug!(self, "Handling status #{}", s.id);
+        grp_trace!(self, "{:?}", s);
+
         let ts = s.timestamp_millis();
         self.config.set_last_status(ts);
 
-        // Short circuit checks
+        let private = s.visibility.is_private();
-        if s.visibility.is_private() {
+        let has_hashtags = s.content.contains('#');
-            debug!("Status is direct/private, discard");
-            return Ok(());
-        }
-
-        if !s.content.contains('#') {
-            debug!("No tags in status, discard");
-            return Ok(());
-        }
-
         let group_user = self.config.get_acct();
         let status_user = normalize_acct(&s.account.acct, group_user)?;
+        let member_or_admin = self.config.is_member_or_admin(&status_user);
+        let commands = crate::command::parse_slash_commands(&s.content);
 
         if status_user == group_user {
-            debug!("This is our post, discard");
+            grp_debug!(self, "This is our post, discard");
             return Ok(());
         }
 
         if self.config.is_banned(&status_user) {
-            debug!("Status author @{} is banned, discard", status_user);
+            grp_debug!(self, "Status author @{} is banned, discard", status_user);
             return Ok(());
         }
 
-        // optout does not work for members and admins, so don't check it
+        if self.config.is_optout(&status_user) && !member_or_admin {
-
+            grp_debug!(self, "Status author @{} opted out, discard", status_user);
-        if !self.config.is_member_or_admin(&status_user) {
-            debug!("Status author @{} is not a member, discard", status_user);
             return Ok(());
         }
 
-        let commands = crate::command::parse_slash_commands(&s.content);
         if commands.contains(&StatusCommand::Ignore) {
-            debug!("Post has IGNORE command, discard");
+            grp_debug!(self, "Post has IGNORE command, discard");
             return Ok(());
         }
 
-        for m in s.mentions {
+        // Sometimes notifications don't work, but we see the mentions as statuses
+        for m in &s.mentions {
             let mentioned_user = normalize_acct(&m.acct, group_user)?;
             if mentioned_user == group_user {
+                let notif_time = self.config.get_last_notif();
+
+                if notif_time <= ts {
+                    grp_debug!(
+                        self,
+                        "mentioned but status is older than last notif, can't be a valid notif, discard"
+                    );
+                    return Ok(());
+                }
+
                 if !commands.is_empty() {
-                    debug!("Detected commands for this group, tags dont apply; discard");
+                    grp_debug!(self, "Detected commands for this group, handle as notif");
+
+                    return self
+                        .handle_notification(Notification {
+                            id: s.id.clone(), // ???
+                            notification_type: NotificationType::Mention,
+                            created_at: s.created_at,
+                            account: s.account.clone(),
+                            status: Some(s),
+                        })
+                        .await;
+                } else if private {
+                    grp_debug!(self, "mention in private without commands, discard, this is nothing");
                     return Ok(());
                 }
             }
         }
 
+        // optout does not work for members and admins, so don't check it
+
+        if !member_or_admin {
+            grp_debug!(self, "Status author @{} is not a member, discard", status_user);
+            return Ok(());
+        }
+
+        if private {
+            grp_debug!(self, "Status is private, discard");
+            return Ok(());
+        }
+
+        if !has_hashtags {
+            grp_debug!(self, "No hashtags, discard");
+            return Ok(());
+        }
+
         let tags = crate::command::parse_status_tags(&s.content);
-        debug!("Tags in status: {:?}", tags);
+        grp_debug!(self, "Tags in status: {:?}", tags);
 
         'tags: for t in tags {
             if self.config.is_tag_followed(&t) {
-                info!("REBLOG #{} STATUS", t);
+                grp_info!(self, "REBLOG #{} STATUS", t);
-                self.client.reblog(&s.id).await
+                self.client.reblog(&s.id).await.log_error("Failed to reblog");
-                    .log_error("Failed to reblog");
+                self.delay_after_post().await;
                 break 'tags; // do not reblog multiple times!
             } else {
-                debug!("#{} is not a group tag", t);
+                grp_debug!(self, "#{} is not a group tag", t);
             }
         }
 
@@ -316,6 +403,7 @@ impl GroupHandle {
 
     async fn follow_user(&mut self, id: &str) -> Result<(), GroupError> {
         self.client.follow(id).await?;
+        self.delay_after_post().await;
         Ok(())
     }
 
@@ -331,22 +419,28 @@ impl GroupHandle {
         // They are retrieved newest first, but we want oldest first for chronological handling
 
         let mut num = 0;
+        let mut old_pn = 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));
+            grp_debug!(self, "Inspecting notif {}", NotificationDisplay(&n));
+            grp_trace!(self, "{:?}", n);
             notifs_to_handle.push(n);
             num += 1;
-            if num > MAX_CATCHUP_NOTIFS {
+            if num > self.cc.max_catchup_notifs {
-                warn!("Too many notifs missed to catch up!");
+                grp_warn!(self, "Too many notifs missed to catch up!");
                 break;
             }
 
-            // sleep so we dont make the api angry
+            let pn = iter.page_num();
-            tokio::time::sleep(Duration::from_millis(250)).await;
+            if pn != old_pn {
+                old_pn = pn;
+                // sleep so we dont make the api angry
+                tokio::time::sleep(Duration::from_secs_f64(self.cc.delay_fetch_page_s)).await;
+            }
         }
 
         if notifs_to_handle.is_empty() {
@@ -355,10 +449,10 @@ impl GroupHandle {
 
         notifs_to_handle.reverse();
 
-        debug!("{} notifications to catch up!", notifs_to_handle.len());
+        grp_debug!(self, "{} notifications to catch up!", notifs_to_handle.len());
 
         for n in notifs_to_handle {
-            debug!("Handling missed notification: {}", NotificationDisplay(&n));
+            grp_debug!(self, "Handling missed notification: {}", NotificationDisplay(&n));
             self.handle_notification(n).await.log_error("Error handling a notification");
         }
 
@@ -379,29 +473,33 @@ impl GroupHandle {
         let mut newest_status = None;
 
         let mut num = 0;
+        let mut old_pn = 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));
+            grp_debug!(self, "Inspecting status {}", StatusDisplay(&s));
+            grp_trace!(self, "{:?}", s);
 
             if newest_status.is_none() {
                 newest_status = Some(ts);
             }
 
-            if s.content.contains('#') && !s.visibility.is_private() {
+            statuses_to_handle.push(s);
-                statuses_to_handle.push(s);
-            }
             num += 1;
-            if num > MAX_CATCHUP_STATUSES {
+            if num > self.cc.max_catchup_statuses {
-                warn!("Too many statuses missed to catch up!");
+                grp_warn!(self, "Too many statuses missed to catch up!");
                 break;
             }
 
-            // sleep so we dont make the api angry
+            let pn = iter.page_num();
-            tokio::time::sleep(Duration::from_millis(250)).await;
+            if pn != old_pn {
+                old_pn = pn;
+                // sleep so we dont make the api angry
+                tokio::time::sleep(Duration::from_secs_f64(self.cc.delay_fetch_page_s)).await;
+            }
         }
 
         if let Some(ts) = newest_status {
@@ -409,17 +507,17 @@ impl GroupHandle {
         }
 
         if statuses_to_handle.is_empty() {
+            grp_debug!(self, "No statuses to handle");
             return Ok(false);
         }
 
         statuses_to_handle.reverse();
 
-        debug!("{} statuses to catch up!", statuses_to_handle.len());
+        grp_debug!(self, "{} statuses to catch up!", statuses_to_handle.len());
 
         for s in statuses_to_handle {
-            debug!("Handling missed status: {}", StatusDisplay(&s));
+            grp_debug!(self, "Handling missed status: {}", StatusDisplay(&s));
-            self.handle_status(s).await
+            self.handle_status(s).await.log_error("Error handling a status");
-                .log_error("Error handling a status");
         }
 
         Ok(true)
@@ -428,8 +526,7 @@ impl GroupHandle {
     async fn handle_mention_status(&mut self, status: Status) -> Result<(), GroupError> {
         let res = ProcessMention::run(self, status).await;
 
-        self.save_if_needed().await
+        self.save_if_needed().await.log_error("Failed to save");
-            .log_error("Failed to save");
 
         res
     }
@@ -442,24 +539,25 @@ impl GroupHandle {
             let mut admins = self.config.get_admins().cloned().collect::<Vec<_>>();
             admins.sort();
 
-            format!("\
+            format!(
+                "\
                 @{user} Welcome to the group! This group has posting restricted to members. \
                 If you'd like to join, please ask one of the group admins:\n\
                 {admins}",
-                    user = notif_acct,
+                user = notif_acct,
-                    admins = admins.join(", ")
+                admins = admins.join(", ")
             )
         } else {
             follow_back = true;
 
-            self.config.set_member(notif_acct, true)
+            self.config.set_member(notif_acct, true).log_error("Fail add a member");
-                .log_error("Fail add a member");
 
-            format!("\
+            format!(
+                "\
                 @{user} Welcome to the group! The group user will now follow you back to complete the sign-up. \
                 To share a post, @ the group user or use a group hashtag.\n\n\
                 Use /help for more info.",
-                    user = notif_acct
+                user = notif_acct
             )
         };
 
@@ -470,12 +568,15 @@ impl GroupHandle {
             .build()
             .expect("error build status");
 
-        self.client.new_status(post).await
+        self.client.new_status(post).await.log_error("Failed to post");
-            .log_error("Failed to post");
+        self.delay_after_post().await;
 
         if follow_back {
-            self.follow_user(notif_user_id).await
+            self.follow_user(notif_user_id).await.log_error("Failed to follow back");
-                .log_error("Failed to follow back");
         }
     }
+
+    async fn delay_after_post(&self) {
+        tokio::time::sleep(Duration::from_secs_f64(self.cc.delay_after_post_s)).await;
+    }
 }

src/main.rs

@@ -31,12 +31,18 @@ async fn main() -> anyhow::Result<()> {
                 .multiple(true)
                 .help("increase logging, can be repeated"),
         )
+        .arg(
+            Arg::with_name("quiet")
+                .short("q")
+                .multiple(true)
+                .help("decrease logging, can be repeated"),
+        )
         .arg(
             Arg::with_name("config")
                 .short("c")
                 .long("config")
                 .takes_value(true)
-                .help("set custom storage file, defaults to groups.json"),
+                .help("set custom config directory, defaults to the current folder"),
         )
         .arg(
             Arg::with_name("auth")
@@ -64,20 +70,28 @@ async fn main() -> anyhow::Result<()> {
         LevelFilter::Trace,
     ];
 
-    let default_level = 2;
+    let default_level = 3;
 
-    let level = (default_level + args.occurrences_of("verbose") as usize).min(LEVELS.len());
+    let level = (
+        default_level as isize
+            + args.occurrences_of("verbose") as isize
+            - args.occurrences_of("quiet") as isize)
+        .clamp(0, LEVELS.len() as isize) as usize;
 
     env_logger::Builder::new()
         .filter_level(LEVELS[level])
         .write_style(env_logger::WriteStyle::Always)
         .filter_module("rustls", LevelFilter::Warn)
         .filter_module("reqwest", LevelFilter::Warn)
+        .filter_module("tungstenite", LevelFilter::Warn)
+        .filter_module("tokio_tungstenite", LevelFilter::Warn)
+        .filter_module("tokio_util", LevelFilter::Warn)
+        .filter_module("want", LevelFilter::Warn)
+        .filter_module("mio", LevelFilter::Warn)
         .init();
 
-    let store = store::ConfigStore::new(StoreOptions {
+    let store = store::ConfigStore::load_from_fs(StoreOptions {
-        store_path: args.value_of("config").unwrap_or("groups.json").to_string(),
+        store_dir: args.value_of("config").unwrap_or(".").to_string(),
-        save_pretty: true,
     })
     .await?;
 
@@ -112,11 +126,16 @@ async fn main() -> anyhow::Result<()> {
     }
 
     // Start
-    let groups = store.spawn_groups().await;
+    let groups = store.spawn_groups().await?;
 
     let mut handles = vec![];
     for mut g in groups {
-        handles.push(tokio::spawn(async move { g.run().await }));
+        handles.push(tokio::spawn(async move {
+            match g.run().await {
+                Ok(()) => unreachable!(),
+                Err(e) => error!("GROUP FAILED! {}", e),
+            }
+        }));
     }
 
     futures::future::join_all(handles).await;

src/store/common_config.rs

@@ -0,0 +1,39 @@
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct CommonConfig {
+    /// Max number of missed notifs to process after connect
+    pub max_catchup_notifs: usize,
+    /// Max number of missed statuses to process after connect
+    pub max_catchup_statuses: usize,
+    /// Delay between fetched pages when catching up
+    pub delay_fetch_page_s: f64,
+    /// Delay after sending a status, making a follow or some other action.
+    /// Set if there are Throttled errors and you need to slow the service down.
+    pub delay_after_post_s: f64,
+    /// Delay before trying to re-connect after the server closed the socket
+    pub delay_reopen_closed_s: f64,
+    /// Delay before trying to re-connect after an error
+    pub delay_reopen_error_s: f64,
+    /// Timeout for a notification/timeline socket to be considered alive.
+    /// If nothing arrives in this interval, reopen it. Some servers have a buggy socket
+    /// implementation where it stays open but no longer works.
+    pub socket_alive_timeout_s: f64,
+    /// Time after which a socket is always closed, even if seemingly alive.
+    /// This is a work-around for servers that stop sending notifs after a while.
+    pub socket_retire_time_s: f64,
+}
+
+impl Default for CommonConfig {
+    fn default() -> Self {
+        Self {
+            max_catchup_notifs: 30,
+            max_catchup_statuses: 50,
+            delay_fetch_page_s: 0.25,
+            delay_after_post_s: 0.0,
+            delay_reopen_closed_s: 0.5,
+            delay_reopen_error_s: 5.0,
+            socket_alive_timeout_s: 30.0,
+            socket_retire_time_s: 120.0,
+        }
+    }
+}

src/store/data.rs

@@ -1,455 +0,0 @@
-use std::collections::{HashMap, HashSet};
-
-use elefren::AppData;
-
-use crate::error::GroupError;
-
-/// This is the inner data struct holding the config
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
-pub(crate) struct Config {
-    pub(crate) groups: HashMap<String, GroupConfig>,
-}
-
-impl Config {
-    // pub(crate) fn iter_groups(&self) -> impl Iterator<Item = &GroupConfig> {
-    //     self.groups.values()
-    // }
-
-    pub(crate) fn get_group_config(&self, acct: &str) -> Option<&GroupConfig> {
-        self.groups.get(acct)
-    }
-
-    pub(crate) fn set_group_config(&mut self, grp: GroupConfig) {
-        self.groups.insert(grp.acct.clone(), grp);
-    }
-}
-
-/// This is the inner data struct holding a group's config
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(default)]
-pub(crate) struct GroupConfig {
-    enabled: bool,
-    /// Group actor's acct
-    acct: String,
-    /// elefren data
-    appdata: AppData,
-    /// Hashtags the group will auto-boost from it's members
-    group_tags: HashSet<String>,
-    /// List of admin account "acct" names, e.g. piggo@piggo.space
-    admin_users: HashSet<String>,
-    /// List of users allowed to post to the group, if it is member-only
-    member_users: HashSet<String>,
-    /// List of users banned from posting to the group
-    banned_users: HashSet<String>,
-    /// Users who decided they don't want to be shared to the group (does not apply to members)
-    optout_users: HashSet<String>,
-    /// True if only members should be allowed to write
-    member_only: bool,
-    /// Banned domain names, e.g. kiwifarms.cc
-    banned_servers: HashSet<String>,
-    /// Last seen notification timestamp (millis)
-    last_notif_ts: u64,
-    /// Last seen status timestamp (millis)
-    last_status_ts: u64,
-    #[serde(skip)]
-    dirty: bool,
-}
-
-impl Default for GroupConfig {
-    fn default() -> Self {
-        Self {
-            enabled: true,
-            acct: "".to_string(),
-            appdata: AppData {
-                base: Default::default(),
-                client_id: Default::default(),
-                client_secret: Default::default(),
-                redirect: Default::default(),
-                token: Default::default(),
-            },
-            group_tags: Default::default(),
-            admin_users: Default::default(),
-            member_users: Default::default(),
-            banned_users: Default::default(),
-            optout_users: Default::default(),
-            member_only: false,
-            banned_servers: Default::default(),
-            last_notif_ts: 0,
-            last_status_ts: 0,
-            dirty: false,
-        }
-    }
-}
-
-impl GroupConfig {
-    pub(crate) fn new(acct: String, appdata: AppData) -> Self {
-        Self {
-            acct,
-            appdata,
-            ..Default::default()
-        }
-    }
-
-    pub(crate) fn is_enabled(&self) -> bool {
-        self.enabled
-    }
-
-    /*
-    pub(crate) fn set_enabled(&mut self, ena: bool) {
-        self.enabled = ena;
-        self.mark_dirty();
-    }
-    */
-
-    pub(crate) fn get_appdata(&self) -> &AppData {
-        &self.appdata
-    }
-
-    pub(crate) fn set_appdata(&mut self, appdata: AppData) {
-        if self.appdata != appdata {
-            self.mark_dirty();
-        }
-        self.appdata = appdata;
-    }
-
-    pub(crate) fn get_admins(&self) -> impl Iterator<Item = &String> {
-        self.admin_users.iter()
-    }
-
-    pub(crate) fn get_members(&self) -> impl Iterator<Item = &String> {
-        self.member_users.iter()
-    }
-
-    pub(crate) fn get_tags(&self) -> impl Iterator<Item = &String> {
-        self.group_tags.iter()
-    }
-
-    pub(crate) fn set_last_notif(&mut self, ts: u64) {
-        if self.last_notif_ts != ts {
-            self.mark_dirty();
-        }
-        self.last_notif_ts = self.last_notif_ts.max(ts);
-    }
-
-    pub(crate) fn get_last_notif(&self) -> u64 {
-        self.last_notif_ts
-    }
-
-    pub(crate) fn set_last_status(&mut self, ts: u64) {
-        if self.last_status_ts != ts {
-            self.mark_dirty();
-        }
-        self.last_status_ts = self.last_status_ts.max(ts);
-    }
-
-    pub(crate) fn get_last_status(&self) -> u64 {
-        self.last_status_ts
-    }
-
-    pub(crate) fn get_acct(&self) -> &str {
-        &self.acct
-    }
-
-    pub(crate) fn is_optout(&self, acct: &str) -> bool {
-        self.optout_users.contains(acct)
-    }
-
-    pub(crate) fn is_admin(&self, acct: &str) -> bool {
-        self.admin_users.contains(acct)
-    }
-
-    pub(crate) fn is_member(&self, acct: &str) -> bool {
-        self.member_users.contains(acct)
-    }
-
-    pub(crate) fn is_member_or_admin(&self, acct: &str) -> bool {
-        self.is_member(acct)
-            || self.is_admin(acct)
-    }
-
-    pub(crate) fn is_banned(&self, acct: &str) -> bool {
-        self.banned_users.contains(acct) || self.is_users_server_banned(acct)
-    }
-
-    pub(crate) fn is_server_banned(&self, server: &str) -> bool {
-        self.banned_servers.contains(server)
-    }
-
-    /// Check if the user's server is banned
-    fn is_users_server_banned(&self, acct: &str) -> bool {
-        let server = acct_to_server(acct);
-        self.is_server_banned(server)
-    }
-
-    pub(crate) fn can_write(&self, acct: &str) -> bool {
-        if self.is_admin(acct) {
-            true
-        } else {
-            !self.is_banned(acct) && (!self.is_member_only() || self.is_member(acct))
-        }
-    }
-
-    pub(crate) fn set_admin(&mut self, acct: &str, admin: bool) -> Result<(), GroupError> {
-        let change = if admin {
-            if self.is_banned(acct) {
-                return Err(GroupError::UserIsBanned);
-            }
-            self.admin_users.insert(acct.to_owned())
-        } else {
-            self.admin_users.remove(acct)
-        };
-        if change {
-            self.mark_dirty();
-        }
-        Ok(())
-    }
-
-    pub(crate) fn set_member(&mut self, acct: &str, member: bool) -> Result<(), GroupError> {
-        let change = if member {
-            if self.is_banned(acct) {
-                return Err(GroupError::UserIsBanned);
-            }
-            self.member_users.insert(acct.to_owned())
-        } else {
-            self.member_users.remove(acct)
-        };
-        if change {
-            self.mark_dirty();
-        }
-        Ok(())
-    }
-
-    pub(crate) fn set_optout(&mut self, acct: &str, optout: bool) {
-        let change = if optout {
-            self.optout_users.insert(acct.to_owned())
-        } else {
-            self.optout_users.remove(acct)
-        };
-        if change {
-            self.mark_dirty();
-        }
-    }
-
-    pub(crate) fn ban_user(&mut self, acct: &str, ban: bool) -> Result<(), GroupError> {
-        let mut change = false;
-        if ban {
-            if self.is_admin(acct) {
-                return Err(GroupError::UserIsAdmin);
-            }
-            // Banned user is also kicked
-            change |= self.member_users.remove(acct);
-            change |= self.banned_users.insert(acct.to_owned());
-        } else {
-            change |= self.banned_users.remove(acct);
-        }
-        if change {
-            self.mark_dirty();
-        }
-        Ok(())
-    }
-
-    pub(crate) fn ban_server(&mut self, server: &str, ban: bool) -> Result<(), GroupError> {
-        let changed = if ban {
-            for acct in &self.admin_users {
-                let acct_server = acct_to_server(acct);
-                if acct_server == server {
-                    return Err(GroupError::AdminsOnServer);
-                }
-            }
-            self.banned_servers.insert(server.to_owned())
-        } else {
-            self.banned_servers.remove(server)
-        };
-        if changed {
-            self.mark_dirty();
-        }
-        Ok(())
-    }
-
-    pub(crate) fn add_tag(&mut self, tag: &str) {
-        if self.group_tags.insert(tag.to_string()) {
-            self.mark_dirty();
-        }
-    }
-
-    pub(crate) fn remove_tag(&mut self, tag: &str) {
-        if self.group_tags.remove(tag) {
-            self.mark_dirty();
-        }
-    }
-
-    pub(crate) fn is_tag_followed(&self, tag: &str) -> bool {
-        self.group_tags.contains(tag)
-    }
-
-    pub(crate) fn set_member_only(&mut self, member_only: bool) {
-        if self.member_only != member_only {
-            self.mark_dirty();
-        }
-        self.member_only = member_only;
-    }
-
-    pub(crate) fn is_member_only(&self) -> bool {
-        self.member_only
-    }
-
-    pub(crate) fn mark_dirty(&mut self) {
-        self.dirty = true;
-    }
-
-    pub(crate) fn is_dirty(&self) -> bool {
-        self.dirty
-    }
-
-    pub(crate) fn clear_dirty_status(&mut self) {
-        self.dirty = false;
-    }
-}
-
-fn acct_to_server(acct: &str) -> &str {
-    acct.split('@').nth(1).unwrap_or_default()
-}
-
-#[cfg(test)]
-mod tests {
-    use crate::error::GroupError;
-    use crate::store::data::{acct_to_server, GroupConfig};
-
-    #[test]
-    fn test_acct_to_server() {
-        assert_eq!("pikachu.rocks", acct_to_server("raichu@pikachu.rocks"));
-        assert_eq!("pikachu.rocks", acct_to_server("m@pikachu.rocks"));
-        assert_eq!("", acct_to_server("what"));
-    }
-
-    #[test]
-    fn test_default_rules() {
-        let group = GroupConfig::default();
-        assert!(!group.is_member_only());
-        assert!(!group.is_member("piggo@piggo.space"));
-        assert!(!group.is_admin("piggo@piggo.space"));
-        assert!(group.can_write("piggo@piggo.space"), "anyone can post by default");
-    }
-
-    #[test]
-    fn test_member_only() {
-        let mut group = GroupConfig::default();
-        assert!(group.can_write("piggo@piggo.space"), "rando can write in public group");
-
-        group.set_member_only(true);
-        assert!(
-            !group.can_write("piggo@piggo.space"),
-            "rando can't write in member-only group"
-        );
-
-        // Admin in member only
-        group.set_admin("piggo@piggo.space", true).unwrap();
-        assert!(
-            group.can_write("piggo@piggo.space"),
-            "admin non-member can write in member-only group"
-        );
-        group.set_admin("piggo@piggo.space", false).unwrap();
-        assert!(
-            !group.can_write("piggo@piggo.space"),
-            "removed admin removes privileged write access"
-        );
-
-        // Member in member only
-        group.set_member("piggo@piggo.space", true).unwrap();
-        assert!(
-            group.can_write("piggo@piggo.space"),
-            "member can post in member-only group"
-        );
-        group.set_admin("piggo@piggo.space", true).unwrap();
-        assert!(
-            group.can_write("piggo@piggo.space"),
-            "member+admin can post in member-only group"
-        );
-    }
-
-    #[test]
-    fn test_banned_users() {
-        // Banning single user
-        let mut group = GroupConfig::default();
-        group.ban_user("piggo@piggo.space", true).unwrap();
-        assert!(!group.can_write("piggo@piggo.space"), "banned user can't post");
-        group.ban_user("piggo@piggo.space", false).unwrap();
-        assert!(group.can_write("piggo@piggo.space"), "un-ban works");
-    }
-
-    #[test]
-    fn test_banned_members() {
-        // Banning single user
-        let mut group = GroupConfig::default();
-        group.set_member_only(true);
-
-        group.set_member("piggo@piggo.space", true).unwrap();
-        assert!(group.can_write("piggo@piggo.space"), "member can write");
-        assert!(group.is_member("piggo@piggo.space"), "member is member");
-        assert!(!group.is_banned("piggo@piggo.space"), "user not banned by default");
-
-        group.ban_user("piggo@piggo.space", true).unwrap();
-        assert!(group.is_member("piggo@piggo.space"), "still member even if banned");
-        assert!(group.is_banned("piggo@piggo.space"), "banned user is banned");
-
-        assert!(!group.can_write("piggo@piggo.space"), "banned member can't post");
-
-        // unban
-        group.ban_user("piggo@piggo.space", false).unwrap();
-        assert!(group.can_write("piggo@piggo.space"), "un-ban works");
-    }
-
-    #[test]
-    fn test_server_ban() {
-        let mut group = GroupConfig::default();
-        assert!(group.can_write("hitler@nazi.camp"), "randos can write");
-
-        group.ban_server("nazi.camp", true).unwrap();
-        assert!(
-            !group.can_write("hitler@nazi.camp"),
-            "users from banned server can't write"
-        );
-        assert!(
-            !group.can_write("1488@nazi.camp"),
-            "users from banned server can't write"
-        );
-        assert!(group.can_write("troll@freezepeach.xyz"), "other users can still write");
-
-        group.ban_server("nazi.camp", false).unwrap();
-        assert!(group.can_write("hitler@nazi.camp"), "server unban works");
-    }
-
-    #[test]
-    fn test_sanity() {
-        let mut group = GroupConfig::default();
-
-        group.set_admin("piggo@piggo.space", true).unwrap();
-        assert_eq!(
-            Err(GroupError::UserIsAdmin),
-            group.ban_user("piggo@piggo.space", true),
-            "can't bad admin users"
-        );
-        group.ban_user("piggo@piggo.space", false).expect("can unbad admin");
-
-        group.ban_user("hitler@nazi.camp", true).unwrap();
-        assert_eq!(
-            Err(GroupError::UserIsBanned),
-            group.set_admin("hitler@nazi.camp", true),
-            "can't make banned users admins"
-        );
-
-        group.ban_server("freespeechextremist.com", true).unwrap();
-        assert_eq!(
-            Err(GroupError::UserIsBanned),
-            group.set_admin("nibber@freespeechextremist.com", true),
-            "can't make server-banned users admins"
-        );
-
-        assert!(group.is_admin("piggo@piggo.space"));
-        assert_eq!(
-            Err(GroupError::AdminsOnServer),
-            group.ban_server("piggo.space", true),
-            "can't bad server with admins"
-        );
-    }
-}

src/store/group_config.rs

@@ -0,0 +1,702 @@
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+
+use elefren::AppData;
+
+use crate::error::GroupError;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+struct FixedConfig {
+    enabled: bool,
+    /// Group actor's acct
+    acct: String,
+    /// elefren data
+    appdata: AppData,
+    /// Server's character limit
+    character_limit: usize,
+    #[serde(skip)]
+    _dirty: bool,
+    #[serde(skip)]
+    _path: PathBuf,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+struct MutableConfig {
+    /// Hashtags the group will auto-boost from it's members
+    group_tags: HashSet<String>,
+    /// List of admin account "acct" names, e.g. piggo@piggo.space
+    admin_users: HashSet<String>,
+    /// List of users allowed to post to the group, if it is member-only
+    member_users: HashSet<String>,
+    /// List of users banned from posting to the group
+    banned_users: HashSet<String>,
+    /// Users who decided they don't want to be shared to the group (does not apply to members)
+    optout_users: HashSet<String>,
+    /// True if only members should be allowed to write
+    member_only: bool,
+    /// Banned domain names, e.g. kiwifarms.cc
+    banned_servers: HashSet<String>,
+    #[serde(skip)]
+    _dirty: bool,
+    #[serde(skip)]
+    _path: PathBuf,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+struct StateConfig {
+    /// Last seen notification timestamp (millis)
+    last_notif_ts: u64,
+    /// Last seen status timestamp (millis)
+    last_status_ts: u64,
+    #[serde(skip)]
+    _dirty: bool,
+    #[serde(skip)]
+    _path: PathBuf,
+}
+
+/// This is the inner data struct holding a group's config
+#[derive(Debug, Clone)]
+pub struct GroupConfig {
+    /// Fixed config that we only read
+    config: FixedConfig,
+    /// Mutable config we can write
+    control: MutableConfig,
+    /// State config with timestamps and transient data that is changed frequently
+    state: StateConfig,
+}
+
+impl Default for FixedConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            acct: "".to_string(),
+            appdata: AppData {
+                base: Default::default(),
+                client_id: Default::default(),
+                client_secret: Default::default(),
+                redirect: Default::default(),
+                token: Default::default(),
+            },
+            character_limit: 5000,
+            _dirty: false,
+            _path: PathBuf::default(),
+        }
+    }
+}
+
+impl Default for MutableConfig {
+    fn default() -> Self {
+        Self {
+            group_tags: Default::default(),
+            admin_users: Default::default(),
+            member_users: Default::default(),
+            banned_users: Default::default(),
+            optout_users: Default::default(),
+            member_only: false,
+            banned_servers: Default::default(),
+            _dirty: false,
+            _path: PathBuf::default(),
+        }
+    }
+}
+
+impl Default for StateConfig {
+    fn default() -> Self {
+        Self {
+            last_notif_ts: 0,
+            last_status_ts: 0,
+            _dirty: false,
+            _path: PathBuf::default(),
+        }
+    }
+}
+
+macro_rules! impl_change_tracking {
+    ($struc:ident) => {
+        impl $struc {
+            pub(crate) fn mark_dirty(&mut self) {
+                self._dirty = true;
+            }
+
+            pub(crate) fn is_dirty(&self) -> bool {
+                self._dirty
+            }
+
+            pub(crate) fn clear_dirty_status(&mut self) {
+                self._dirty = false;
+            }
+
+            pub(crate) async fn save_if_needed(&mut self) -> Result<bool, GroupError> {
+                if self.is_dirty() {
+                    self.save().await?;
+                    Ok(true)
+                } else {
+                    Ok(false)
+                }
+            }
+
+            pub(crate) async fn save(&mut self) -> Result<(), GroupError> {
+                tokio::fs::write(&self._path, serde_json::to_string_pretty(&self)?.as_bytes()).await?;
+                self.clear_dirty_status();
+                Ok(())
+            }
+        }
+    };
+}
+
+impl_change_tracking!(FixedConfig);
+impl_change_tracking!(MutableConfig);
+impl_change_tracking!(StateConfig);
+
+impl Default for GroupConfig {
+    fn default() -> Self {
+        Self {
+            config: Default::default(),
+            control: Default::default(),
+            state: Default::default(),
+        }
+    }
+}
+
+async fn load_or_create_control_file(control_path: impl AsRef<Path>) -> Result<MutableConfig, GroupError> {
+    let control_path = control_path.as_ref();
+    let mut dirty = false;
+    let mut control: MutableConfig = if control_path.is_file() {
+        let f = tokio::fs::read(&control_path).await?;
+        let mut control: MutableConfig = serde_json::from_slice(&f)?;
+        control._path = control_path.to_owned();
+        control
+    } else {
+        debug!("control file missing, creating empty");
+        dirty = true;
+        MutableConfig {
+            _path: control_path.to_owned(),
+            ..Default::default()
+        }
+    };
+    if dirty {
+        control.save().await?;
+    }
+    Ok(control)
+}
+
+async fn load_or_create_state_file(state_path: impl AsRef<Path>) -> Result<StateConfig, GroupError> {
+    let state_path = state_path.as_ref();
+    let mut dirty = false;
+    let mut state: StateConfig = if state_path.is_file() {
+        let f = tokio::fs::read(&state_path).await?;
+        let mut control: StateConfig = serde_json::from_slice(&f)?;
+        control._path = state_path.to_owned();
+        control
+    } else {
+        debug!("state file missing, creating empty");
+        dirty = true;
+        StateConfig {
+            _path: state_path.to_owned(),
+            ..Default::default()
+        }
+    };
+    if dirty {
+        state.save().await?;
+    }
+    Ok(state)
+}
+
+impl GroupConfig {
+    pub(crate) fn is_dirty(&self) -> bool {
+        self.config.is_dirty() || self.control.is_dirty() || self.state.is_dirty()
+    }
+
+    /// Save only what changed
+    pub(crate) async fn save_if_needed(&mut self, danger_allow_overwriting_config: bool) -> Result<(), GroupError> {
+        #[allow(clippy::collapsible_if)]
+        if danger_allow_overwriting_config {
+            if self.config.save_if_needed().await? {
+                debug!(
+                    "Written {} config file {}",
+                    self.config.acct,
+                    self.config._path.display()
+                );
+            }
+        }
+        if self.control.save_if_needed().await? {
+            debug!(
+                "Written {} control file {}",
+                self.config.acct,
+                self.control._path.display()
+            );
+        }
+        if self.state.save_if_needed().await? {
+            debug!("Written {} state file {}", self.config.acct, self.state._path.display());
+        }
+        Ok(())
+    }
+
+    /// Save all unconditionally
+    #[allow(unused)]
+    pub(crate) async fn save(&mut self, danger_allow_overwriting_config: bool) -> Result<(), GroupError> {
+        if danger_allow_overwriting_config {
+            self.config.save().await?;
+        }
+        self.control.save().await?;
+        self.state.save().await?;
+        Ok(())
+    }
+
+    /// (re)init using new authorization
+    pub(crate) async fn from_appdata(acct: String, appdata: AppData, group_dir: PathBuf) -> Result<Self, GroupError> {
+        if !group_dir.is_dir() {
+            debug!("Creating group directory");
+            tokio::fs::create_dir_all(&group_dir).await?;
+        }
+
+        let config_path = group_dir.join("config.json");
+        let control_path = group_dir.join("control.json");
+        let state_path = group_dir.join("state.json");
+
+        // try to reuse content of the files, if present
+
+        /* config */
+        let mut dirty = false;
+        let mut config: FixedConfig = if config_path.is_file() {
+            let f = tokio::fs::read(&config_path).await?;
+            let mut config: FixedConfig = serde_json::from_slice(&f)?;
+            config._path = config_path;
+            if config.appdata != appdata {
+                config.appdata = appdata;
+                dirty = true;
+            }
+            if config.acct != acct {
+                config.acct = acct.clone();
+                dirty = true;
+            }
+            config
+        } else {
+            dirty = true;
+            FixedConfig {
+                acct: acct.clone(),
+                appdata,
+                _path: config_path,
+                ..Default::default()
+            }
+        };
+        if dirty {
+            debug!("config file for {} changed, creating/updating", acct);
+            config.save().await?;
+        }
+
+        /* control */
+        let control = load_or_create_control_file(control_path).await?;
+
+        /* state */
+        let state = load_or_create_state_file(state_path).await?;
+
+        let g = GroupConfig { config, control, state };
+        g.warn_of_bad_config();
+        Ok(g)
+    }
+
+    pub(crate) async fn from_dir(group_dir: PathBuf) -> Result<Self, GroupError> {
+        let config_path = group_dir.join("config.json");
+        let control_path = group_dir.join("control.json");
+        let state_path = group_dir.join("state.json");
+
+        // try to reuse content of the files, if present
+
+        /* config */
+        let f = tokio::fs::read(&config_path).await?;
+        let mut config: FixedConfig = serde_json::from_slice(&f)?;
+        config._path = config_path;
+
+        /* control */
+        let control = load_or_create_control_file(control_path).await?;
+
+        /* state */
+        let state = load_or_create_state_file(state_path).await?;
+
+        let g = GroupConfig { config, control, state };
+        g.warn_of_bad_config();
+        Ok(g)
+    }
+
+    fn warn_of_bad_config(&self) {
+        for t in &self.control.group_tags {
+            if &t.to_lowercase() != t {
+                warn!(
+                    "Group {} hashtag \"{}\" is not lowercase, it won't work!",
+                    self.config.acct, t
+                );
+            }
+        }
+
+        for u in self
+            .control
+            .admin_users
+            .iter()
+            .chain(self.control.member_users.iter())
+            .chain(self.control.banned_users.iter())
+            .chain(self.control.optout_users.iter())
+        {
+            if &u.to_lowercase() != u {
+                warn!(
+                    "Group {} config contains a user with non-lowercase name \"{}\", it won't work!",
+                    self.config.acct, u
+                );
+            }
+
+            if u.starts_with('@') || u.chars().filter(|c| *c == '@').count() != 1 {
+                warn!("Group {} config contains an invalid user name: {}", self.config.acct, u);
+            }
+        }
+    }
+
+    pub(crate) fn get_character_limit(&self) -> usize {
+        self.config.character_limit
+    }
+
+    pub(crate) fn is_enabled(&self) -> bool {
+        self.config.enabled
+    }
+
+    pub(crate) fn get_appdata(&self) -> &AppData {
+        &self.config.appdata
+    }
+
+    pub(crate) fn set_appdata(&mut self, appdata: AppData) {
+        if self.config.appdata != appdata {
+            self.config.mark_dirty();
+        }
+        self.config.appdata = appdata;
+    }
+
+    pub(crate) fn get_admins(&self) -> impl Iterator<Item = &String> {
+        self.control.admin_users.iter()
+    }
+
+    pub(crate) fn get_members(&self) -> impl Iterator<Item = &String> {
+        self.control.member_users.iter()
+    }
+
+    pub(crate) fn get_tags(&self) -> impl Iterator<Item = &String> {
+        self.control.group_tags.iter()
+    }
+
+    pub(crate) fn set_last_notif(&mut self, ts: u64) {
+        if self.state.last_notif_ts != ts {
+            self.state.mark_dirty();
+        }
+        self.state.last_notif_ts = self.state.last_notif_ts.max(ts);
+    }
+
+    pub(crate) fn get_last_notif(&self) -> u64 {
+        self.state.last_notif_ts
+    }
+
+    pub(crate) fn set_last_status(&mut self, ts: u64) {
+        if self.state.last_status_ts != ts {
+            self.state.mark_dirty();
+        }
+        self.state.last_status_ts = self.state.last_status_ts.max(ts);
+    }
+
+    pub(crate) fn get_last_status(&self) -> u64 {
+        self.state.last_status_ts
+    }
+
+    pub(crate) fn get_acct(&self) -> &str {
+        &self.config.acct
+    }
+
+    pub(crate) fn is_optout(&self, acct: &str) -> bool {
+        self.control.optout_users.contains(acct)
+    }
+
+    pub(crate) fn is_admin(&self, acct: &str) -> bool {
+        self.control.admin_users.contains(acct)
+    }
+
+    pub(crate) fn is_member(&self, acct: &str) -> bool {
+        self.control.member_users.contains(acct)
+    }
+
+    pub(crate) fn is_member_or_admin(&self, acct: &str) -> bool {
+        self.is_member(acct) || self.is_admin(acct)
+    }
+
+    pub(crate) fn is_banned(&self, acct: &str) -> bool {
+        self.control.banned_users.contains(acct) || self.is_users_server_banned(acct)
+    }
+
+    pub(crate) fn is_server_banned(&self, server: &str) -> bool {
+        self.control.banned_servers.contains(server)
+    }
+
+    /// Check if the user's server is banned
+    fn is_users_server_banned(&self, acct: &str) -> bool {
+        let server = acct_to_server(acct);
+        self.is_server_banned(server)
+    }
+
+    pub(crate) fn can_write(&self, acct: &str) -> bool {
+        if self.is_admin(acct) {
+            true
+        } else {
+            !self.is_banned(acct) && (!self.is_member_only() || self.is_member(acct))
+        }
+    }
+
+    pub(crate) fn set_admin(&mut self, acct: &str, admin: bool) -> Result<(), GroupError> {
+        let change = if admin {
+            if self.is_banned(acct) {
+                return Err(GroupError::UserIsBanned);
+            }
+            self.control.admin_users.insert(acct.to_owned())
+        } else {
+            self.control.admin_users.remove(acct)
+        };
+        if change {
+            self.control.mark_dirty();
+        }
+        Ok(())
+    }
+
+    pub(crate) fn set_member(&mut self, acct: &str, member: bool) -> Result<(), GroupError> {
+        let change = if member {
+            if self.is_banned(acct) {
+                return Err(GroupError::UserIsBanned);
+            }
+            self.control.member_users.insert(acct.to_owned())
+        } else {
+            self.control.member_users.remove(acct)
+        };
+        if change {
+            self.control.mark_dirty();
+        }
+        Ok(())
+    }
+
+    pub(crate) fn set_optout(&mut self, acct: &str, optout: bool) {
+        let change = if optout {
+            self.control.optout_users.insert(acct.to_owned())
+        } else {
+            self.control.optout_users.remove(acct)
+        };
+        if change {
+            self.control.mark_dirty();
+        }
+    }
+
+    pub(crate) fn ban_user(&mut self, acct: &str, ban: bool) -> Result<(), GroupError> {
+        let mut change = false;
+        if ban {
+            if self.is_admin(acct) {
+                return Err(GroupError::UserIsAdmin);
+            }
+            // Banned user is also kicked
+            change |= self.control.member_users.remove(acct);
+            change |= self.control.banned_users.insert(acct.to_owned());
+        } else {
+            change |= self.control.banned_users.remove(acct);
+        }
+        if change {
+            self.control.mark_dirty();
+        }
+        Ok(())
+    }
+
+    pub(crate) fn ban_server(&mut self, server: &str, ban: bool) -> Result<(), GroupError> {
+        let changed = if ban {
+            for acct in &self.control.admin_users {
+                let acct_server = acct_to_server(acct);
+                if acct_server == server {
+                    return Err(GroupError::AdminsOnServer);
+                }
+            }
+            self.control.banned_servers.insert(server.to_owned())
+        } else {
+            self.control.banned_servers.remove(server)
+        };
+        if changed {
+            self.control.mark_dirty();
+        }
+        Ok(())
+    }
+
+    pub(crate) fn add_tag(&mut self, tag: &str) {
+        if self.control.group_tags.insert(tag.to_string()) {
+            self.control.mark_dirty();
+        }
+    }
+
+    pub(crate) fn remove_tag(&mut self, tag: &str) {
+        if self.control.group_tags.remove(tag) {
+            self.control.mark_dirty();
+        }
+    }
+
+    pub(crate) fn is_tag_followed(&self, tag: &str) -> bool {
+        self.control.group_tags.contains(tag)
+    }
+
+    pub(crate) fn set_member_only(&mut self, member_only: bool) {
+        if self.control.member_only != member_only {
+            self.control.mark_dirty();
+        }
+        self.control.member_only = member_only;
+    }
+
+    pub(crate) fn is_member_only(&self) -> bool {
+        self.control.member_only
+    }
+}
+
+fn acct_to_server(acct: &str) -> &str {
+    acct.split('@').nth(1).unwrap_or_default()
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::error::GroupError;
+    use crate::store::group_config::{acct_to_server, GroupConfig};
+
+    #[test]
+    fn test_acct_to_server() {
+        assert_eq!("pikachu.rocks", acct_to_server("raichu@pikachu.rocks"));
+        assert_eq!("pikachu.rocks", acct_to_server("m@pikachu.rocks"));
+        assert_eq!("", acct_to_server("what"));
+    }
+
+    #[test]
+    fn test_default_rules() {
+        let group = GroupConfig::default();
+        assert!(!group.is_member_only());
+        assert!(!group.is_member("piggo@piggo.space"));
+        assert!(!group.is_admin("piggo@piggo.space"));
+        assert!(group.can_write("piggo@piggo.space"), "anyone can post by default");
+    }
+
+    #[test]
+    fn test_member_only() {
+        let mut group = GroupConfig::default();
+        assert!(group.can_write("piggo@piggo.space"), "rando can write in public group");
+
+        group.set_member_only(true);
+        assert!(
+            !group.can_write("piggo@piggo.space"),
+            "rando can't write in member-only group"
+        );
+
+        // Admin in member only
+        group.set_admin("piggo@piggo.space", true).unwrap();
+        assert!(
+            group.can_write("piggo@piggo.space"),
+            "admin non-member can write in member-only group"
+        );
+        group.set_admin("piggo@piggo.space", false).unwrap();
+        assert!(
+            !group.can_write("piggo@piggo.space"),
+            "removed admin removes privileged write access"
+        );
+
+        // Member in member only
+        group.set_member("piggo@piggo.space", true).unwrap();
+        assert!(
+            group.can_write("piggo@piggo.space"),
+            "member can post in member-only group"
+        );
+        group.set_admin("piggo@piggo.space", true).unwrap();
+        assert!(
+            group.can_write("piggo@piggo.space"),
+            "member+admin can post in member-only group"
+        );
+    }
+
+    #[test]
+    fn test_banned_users() {
+        // Banning single user
+        let mut group = GroupConfig::default();
+        group.ban_user("piggo@piggo.space", true).unwrap();
+        assert!(!group.can_write("piggo@piggo.space"), "banned user can't post");
+        group.ban_user("piggo@piggo.space", false).unwrap();
+        assert!(group.can_write("piggo@piggo.space"), "un-ban works");
+    }
+
+    #[test]
+    fn test_banned_members() {
+        // Banning single user
+        let mut group = GroupConfig::default();
+        group.set_member_only(true);
+
+        group.set_member("piggo@piggo.space", true).unwrap();
+        assert!(group.can_write("piggo@piggo.space"), "member can write");
+        assert!(group.is_member("piggo@piggo.space"), "member is member");
+        assert!(!group.is_banned("piggo@piggo.space"), "user not banned by default");
+
+        group.ban_user("piggo@piggo.space", true).unwrap();
+        assert!(group.is_member("piggo@piggo.space"), "still member even if banned");
+        assert!(group.is_banned("piggo@piggo.space"), "banned user is banned");
+
+        assert!(!group.can_write("piggo@piggo.space"), "banned member can't post");
+
+        // unban
+        group.ban_user("piggo@piggo.space", false).unwrap();
+        assert!(group.can_write("piggo@piggo.space"), "un-ban works");
+    }
+
+    #[test]
+    fn test_server_ban() {
+        let mut group = GroupConfig::default();
+        assert!(group.can_write("hitler@nazi.camp"), "randos can write");
+
+        group.ban_server("nazi.camp", true).unwrap();
+        assert!(
+            !group.can_write("hitler@nazi.camp"),
+            "users from banned server can't write"
+        );
+        assert!(
+            !group.can_write("1488@nazi.camp"),
+            "users from banned server can't write"
+        );
+        assert!(group.can_write("troll@freezepeach.xyz"), "other users can still write");
+
+        group.ban_server("nazi.camp", false).unwrap();
+        assert!(group.can_write("hitler@nazi.camp"), "server unban works");
+    }
+
+    #[test]
+    fn test_sanity() {
+        let mut group = GroupConfig::default();
+
+        group.set_admin("piggo@piggo.space", true).unwrap();
+        assert_eq!(
+            Err(GroupError::UserIsAdmin),
+            group.ban_user("piggo@piggo.space", true),
+            "can't bad admin users"
+        );
+        group.ban_user("piggo@piggo.space", false).expect("can unbad admin");
+
+        group.ban_user("hitler@nazi.camp", true).unwrap();
+        assert_eq!(
+            Err(GroupError::UserIsBanned),
+            group.set_admin("hitler@nazi.camp", true),
+            "can't make banned users admins"
+        );
+
+        group.ban_server("freespeechextremist.com", true).unwrap();
+        assert_eq!(
+            Err(GroupError::UserIsBanned),
+            group.set_admin("nibber@freespeechextremist.com", true),
+            "can't make server-banned users admins"
+        );
+
+        assert!(group.is_admin("piggo@piggo.space"));
+        assert_eq!(
+            Err(GroupError::AdminsOnServer),
+            group.ban_server("piggo.space", true),
+            "can't bad server with admins"
+        );
+    }
+}

src/store/mod.rs

@@ -3,21 +3,20 @@ 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 crate::group_handler::{GroupHandle, GroupInternal};
-use std::time::Duration;
 
-pub(crate) mod data;
+pub mod common_config;
+pub mod group_config;
+pub use common_config::CommonConfig;
+pub use group_config::GroupConfig;
 
 #[derive(Debug, Default)]
 pub struct ConfigStore {
     store_path: PathBuf,
-    save_pretty: bool,
+    groups_path: PathBuf,
-    data: tokio::sync::RwLock<Config>,
+    config: Arc<CommonConfig>,
 }
 
 #[derive(Debug)]
@@ -28,29 +27,63 @@ pub struct NewGroupOptions {
 
 #[derive(Debug)]
 pub struct StoreOptions {
-    pub store_path: String,
+    pub store_dir: 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> {
+    pub async fn load_from_fs(options: StoreOptions) -> Result<Arc<Self>, GroupError> {
-        let path: &Path = options.store_path.as_ref();
+        let given_path: &Path = options.store_dir.as_ref();
 
-        let config = if path.is_file() {
+        let mut common_file: Option<PathBuf> = None;
-            let f = tokio::fs::read(path).await?;
+        let base_dir: PathBuf;
+        if given_path.is_file() {
+            if given_path.extension().unwrap_or_default().to_string_lossy() == "json" {
+                // this is a groups.json file
+                common_file = Some(given_path.to_owned());
+                base_dir = given_path
+                    .parent()
+                    .ok_or_else(|| GroupError::BadConfig("no parent dir".into()))?
+                    .to_owned();
+            } else {
+                return Err(GroupError::BadConfig("bad config file, should be JSON".into()));
+            }
+        } else if given_path.is_dir() {
+            let cf = given_path.join("groups.json");
+            if cf.is_file() {
+                common_file = Some(cf);
+            }
+            base_dir = given_path.to_owned();
+        } else {
+            return Err(GroupError::BadConfig("bad config file/dir".into()));
+        }
+
+        if !base_dir.is_dir() {
+            return Err(GroupError::BadConfig("base dir does not exist".into()));
+        }
+
+        let config: CommonConfig = if let Some(cf) = &common_file {
+            debug!("Loading common config from {}", cf.display());
+            let f = tokio::fs::read(&cf).await?;
             serde_json::from_slice(&f)?
         } else {
-            let empty = Config::default();
+            debug!("No common config file, using defaults");
-            tokio::fs::write(path, serde_json::to_string(&empty)?.as_bytes()).await?;
+            CommonConfig::default()
-            empty
         };
 
+        debug!("Using common config:\n{:#?}", config);
+
+        let groups_path = base_dir.join("groups.d");
+        if !groups_path.exists() {
+            debug!("Creating groups directory");
+            tokio::fs::create_dir_all(&groups_path).await?;
+        }
+
         Ok(Arc::new(Self {
-            store_path: path.to_owned(),
+            store_path: base_dir.to_owned(),
-            save_pretty: options.save_pretty,
+            groups_path,
-            data: RwLock::new(config),
+            config: Arc::new(config),
         }))
     }
 
@@ -67,10 +100,11 @@ impl ConfigStore {
         let client = elefren::helpers::cli::authenticate(registration).await?;
         let appdata = client.data.clone();
 
-        let data = GroupConfig::new(opts.acct.clone(), appdata);
+        let group_dir = self.groups_path.join(&opts.acct);
+
+        let data = GroupConfig::from_appdata(opts.acct.clone(), appdata, group_dir).await?;
 
         // save & persist
-        self.set_group_config(data.clone()).await?;
 
         let group_account = match client.verify_credentials().await {
             Ok(account) => {
@@ -90,15 +124,16 @@ impl ConfigStore {
             group_account,
             client,
             config: data,
-            store: self.clone(),
+            cc: self.config.clone(),
+            internal: GroupInternal::default(),
         })
     }
 
     /// 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 group_dir = self.groups_path.join(&acct);
-        let mut config = groups.get_group_config(acct).ok_or(GroupError::GroupNotExist)?.clone();
+
-        drop(groups);
+        let mut config = GroupConfig::from_dir(group_dir).await?;
 
         println!("--- Re-authenticating bot user @{} ---", acct);
         let registration = Registration::new(config.get_appdata().base.to_string())
@@ -114,7 +149,7 @@ impl ConfigStore {
         let appdata = client.data.clone();
 
         config.set_appdata(appdata);
-        self.set_group_config(config.clone()).await?;
+        config.save_if_needed(true).await?;
 
         let group_account = match client.verify_credentials().await {
             Ok(account) => {
@@ -134,95 +169,70 @@ impl ConfigStore {
             group_account,
             client,
             config,
-            store: self.clone(),
+            cc: self.config.clone(),
+            internal: GroupInternal::default(),
         })
     }
 
     /// Spawn existing group using saved creds
-    pub async fn spawn_groups(self: Arc<Self>) -> Vec<GroupHandle> {
+    pub async fn spawn_groups(self: Arc<Self>) -> Result<Vec<GroupHandle>, GroupError> {
-        let groups = self.data.read().await.clone();
+        info!("Starting group services for groups in {}", self.groups_path.display());
-        let groups_iter = groups.groups.into_values();
+        let dirs = std::fs::read_dir(&self.groups_path)?;
 
         // Connect in parallel
-        futures::stream::iter(groups_iter)
+        Ok(futures::stream::iter(dirs)
-            .map(|gc| async {
+            .map(|entry_maybe: Result<std::fs::DirEntry, std::io::Error>| async {
-                if !gc.is_enabled() {
+                match entry_maybe {
-                    debug!("Group @{} is DISABLED", gc.get_acct());
+                    Ok(entry) => {
-                    return None;
+                        let gc = GroupConfig::from_dir(entry.path()).await.ok()?;
-                }
 
-                debug!("Connecting to @{}", gc.get_acct());
+                        if !gc.is_enabled() {
+                            debug!("Group @{} is DISABLED", gc.get_acct());
+                            return None;
+                        }
 
-                let client = FediClient::from(gc.get_appdata().clone());
+                        debug!("Connecting to @{}", gc.get_acct());
 
-                let my_account = match client.verify_credentials().await {
+                        let client = FediClient::from(gc.get_appdata().clone());
-                    Ok(account) => {
+
-                        info!(
+                        let my_account = match client.verify_credentials().await {
-                            "Group account verified: @{}, \"{}\"",
+                            Ok(account) => {
-                            account.acct, account.display_name
+                                info!(
-                        );
+                                    "Group account verified: @{}, \"{}\"",
-                        account
+                                    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,
+                            cc: self.config.clone(),
+                            internal: GroupInternal::default(),
+                        })
                     }
                     Err(e) => {
-                        error!("Group @{} auth error: {}", gc.get_acct(), e);
+                        error!("{}", e);
-                        return None;
+                        None
                     }
-                };
+                }
-
-                Some(GroupHandle {
-                    group_account: my_account,
-                    client,
-                    config: gc,
-                    store: self.clone(),
-                })
             })
             .buffer_unordered(8)
             .collect::<Vec<_>>()
             .await
             .into_iter()
             .flatten()
-            .collect()
+            .collect())
     }
 
-    pub async fn group_exists(&self, acct : &str) -> bool {
+    pub async fn group_exists(&self, acct: &str) -> bool {
-        self.data.read().await.groups.contains_key(acct)
+        self.store_path.join(acct).join("config.json").is_file()
-    }
-
-    /*
-    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(())
     }
 }
 
@@ -232,19 +242,3 @@ fn make_scopes() -> Scopes {
         | Scopes::write(scopes::Write::Media)
         | Scopes::write(scopes::Write::Follows)
 }
-
-// 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)
-//         }
-//     }
-// }

src/utils.rs

@@ -105,9 +105,7 @@ impl VisExt for Visibility {
 
     fn make_unlisted(self) -> Self {
         match self {
-            Visibility::Public => {
+            Visibility::Public => Visibility::Unlisted,
-                Visibility::Unlisted
-            }
             other => other,
         }
     }