Block statuses from blocking users

This commit fixes an issue where a status from A would be displayed on
B's public timelines even when A had B blocked (i.e., it would treat B
as though they were muted rather than blocked for the purpose of
public timelines).

src/parse_client_request/user/mod.rs

@@ -39,8 +39,9 @@ impl Subscription {
             timeline: Timeline::from_query_and_user(&q, &user, pool.clone())?,
             allowed_langs: user.allowed_langs,
             blocks: Blocks {
-                user_blocks: postgres::select_user_blocks(user.id, pool.clone()),
-                domain_blocks: postgres::select_domain_blocks(user.id, pool.clone()),
+                blocking_users: postgres::select_blocking_users(user.id, pool.clone()),
+                blocked_users: postgres::select_blocked_users(user.id, pool.clone()),
+                blocked_domains: postgres::select_blocked_domains(user.id, pool.clone()),
             },
         })
     }
@@ -167,8 +168,9 @@ pub enum Scope {
 
 #[derive(Clone, Default, Debug, PartialEq)]
 pub struct Blocks {
-    pub domain_blocks: HashSet<String>,
-    pub user_blocks: HashSet<i64>,
+    pub blocked_domains: HashSet<String>,
+    pub blocked_users: HashSet<i64>,
+    pub blocking_users: HashSet<i64>,
 }
 
 #[derive(Clone, Debug, PartialEq)]
@@ -187,43 +189,3 @@ impl UserData {
         }
     }
 }
-
-// fn set_timeline_and_filter(self, q: Query, pool: PgPool) -> Result<Self, Rejection> {
-//         let (read_scope, f) = (self.scopes.clone(), self.allowed_langs.clone());
-//         use Scope::*;
-//         let (filter, target_timeline) = match q.stream.as_ref() {
-//             // Public endpoints:
-//             tl @ "public" | tl @ "public:local" if q.media => (f, format!("{}:media", tl)),
-//             tl @ "public:media" | tl @ "public:local:media" => (f, tl.to_string()),
-//             tl @ "public" | tl @ "public:local" => (f, tl.to_string()),
-
-//             // Hashtag endpoints:
-//             tl @ "hashtag" | tl @ "hashtag:local" => (f, format!("{}:{}", tl, q.hashtag)),
-//             // Private endpoints: User:
-//             "user" if self.logged_in && read_scope.contains(&Statuses) => {
-//                 (HashSet::new(), format!("{}", self.id))
-//             }
-//             "user:notification" if self.logged_in && read_scope.contains(&Notifications) => {
-//                 (HashSet::new(), format!("{}", self.id))
-//             }
-//             // List endpoint:
-//             "list" if self.owns_list(q.list, pool) && read_scope.contains(&Lists) => {
-//                 (HashSet::new(), format!("list:{}", q.list))
-//             }
-//             // Direct endpoint:
-//             "direct" if self.logged_in && read_scope.contains(&Statuses) => {
-//                 (HashSet::new(), "direct".to_string())
-//             }
-//             // Reject unathorized access attempts for private endpoints
-//             "user" | "user:notification" | "direct" | "list" => {
-//                 return Err(warp::reject::custom("Error: Missing access token"))
-//             }
-//             // Other endpoints don't exist:
-//             _ => return Err(warp::reject::custom("Error: Nonexistent endpoint")),
-//         };
-//         Ok(Self {
-//             target_timeline,
-//             allowed_langs: filter,
-//             ..self
-//         })
-//     }

src/parse_client_request/user/postgres.rs

@@ -127,7 +127,18 @@ LIMIT 1",
 ///
 /// **NOTE**: because we check this when the user connects, it will not include any blocks
 /// the user adds until they refresh/reconnect.
-pub fn select_user_blocks(user_id: i64, pg_pool: PgPool) -> HashSet<i64> {
+pub fn select_blocked_users(user_id: i64, pg_pool: PgPool) -> HashSet<i64> {
+    //     "
+    // SELECT
+    //    1
+    //    FROM blocks
+    //    WHERE (account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 2)}))
+    //    OR (account_id = $2 AND target_account_id = $1)
+    // UNION SELECT
+    //    1
+    //    FROM mutes
+    //    WHERE account_id = $1 AND target_account_id IN (${placeholders(targetAccountIds, 2)})`
+    // , [req.accountId, unpackedPayload.account.id].concat(targetAccountIds)),`"
     pg_pool
         .0
         .get()
@@ -147,12 +158,33 @@ UNION SELECT target_account_id
         .map(|row| row.get(0))
         .collect()
 }
+/// Query Postgres for everyone who has blocked the user
+///
+/// **NOTE**: because we check this when the user connects, it will not include any blocks
+/// the user adds until they refresh/reconnect.
+pub fn select_blocking_users(user_id: i64, pg_pool: PgPool) -> HashSet<i64> {
+    pg_pool
+        .0
+        .get()
+        .unwrap()
+        .query(
+            "
+SELECT account_id
+  FROM blocks
+  WHERE target_account_id = $1",
+            &[&user_id],
+        )
+        .expect("Hard-coded query will return Some([0 or more rows])")
+        .iter()
+        .map(|row| row.get(0))
+        .collect()
+}
 
 /// Query Postgres for all current domain blocks
 ///
 /// **NOTE**: because we check this when the user connects, it will not include any blocks
 /// the user adds until they refresh/reconnect.
-pub fn select_domain_blocks(user_id: i64, pg_pool: PgPool) -> HashSet<String> {
+pub fn select_blocked_domains(user_id: i64, pg_pool: PgPool) -> HashSet<String> {
     pg_pool
         .0
         .get()

src/redis_to_client_stream/client_agent.rs

@@ -100,8 +100,10 @@ impl futures::stream::Stream for ClientAgent {
             log::warn!("Polling the Receiver took: {:?}", start_time.elapsed());
         };
 
-        let (allowed_langs, blocks) = (&self.subscription.allowed_langs, &self.subscription.blocks);
-        let (blocked_users, blocked_domains) = (&blocks.user_blocks, &blocks.domain_blocks);
+        let allowed_langs = &self.subscription.allowed_langs;
+        let blocked_users = &self.subscription.blocks.blocked_users;
+        let blocking_users = &self.subscription.blocks.blocking_users;
+        let blocked_domains = &self.subscription.blocks.blocked_domains;
         let (send, block) = (|msg| Ok(Ready(Some(msg))), Ok(NotReady));
         use Message::*;
         match result {
@@ -109,6 +111,7 @@ impl futures::stream::Stream for ClientAgent {
                 Update(status) if status.language_not_allowed(allowed_langs) => block,
                 Update(status) if status.involves_blocked_user(blocked_users) => block,
                 Update(status) if status.from_blocked_domain(blocked_domains) => block,
+                Update(status) if status.from_blocking_user(blocking_users) => block,
                 Update(status) => send(Update(status)),
                 Notification(notification) => send(Notification(notification)),
                 Conversation(notification) => send(Conversation(notification)),

src/redis_to_client_stream/message.rs

@@ -84,8 +84,25 @@ impl Status {
             None => false, // None means the user is on the local instance, which can't be blocked
         }
     }
+    /// Returns `true` if the Status is from an account that has blocked the current user.
+    pub fn from_blocking_user(&self, blocking_users: &HashSet<i64>) -> bool {
+        let toot = self.0.clone();
+        const ALLOW: bool = false;
+        const REJECT: bool = true;
 
-    /// Returns `true` if the User's list of blocked users includes a user involved in this toot.
+        let author = toot["account"]["id"]
+            .str_to_i64()
+            .unwrap_or_else(|_| log_fatal!("Could not process `account.id` in {:?}", toot));
+
+        if blocking_users.contains(&author) {
+            REJECT
+        } else {
+            ALLOW
+        }
+    }
+
+    /// Returns `true` if the User's list of blocked and muted users includes a user
+    /// involved in this toot.
     ///
     /// A user is involved if they:
     ///  * Wrote this toot