Drop retweets whose original author handle matches a muted keyword

home-mixer/candidate_hydrators/mod.rs

@@ -15,6 +15,7 @@ pub mod media_info_hydrator;
 pub mod mutual_follow_jaccard_hydrator;
 pub mod quote_hydrator;
 pub mod quoted_post_text_hydrator;
+pub mod retweeted_author_screen_name_hydrator;
 pub mod semantic_id_hydrator;
 pub mod subscription_hydrator;
 pub mod topic_feedback_context_hydrator;

home-mixer/candidate_hydrators/retweeted_author_screen_name_hydrator.rs

@@ -0,0 +1,69 @@
+use crate::clients::gizmoduck_client::GizmoduckClient;
+use crate::models::candidate::PostCandidate;
+use crate::models::query::ScoredPostsQuery;
+use std::collections::HashSet;
+use std::sync::Arc;
+use tonic::async_trait;
+use xai_candidate_pipeline::hydrator::Hydrator;
+
+pub struct RetweetedAuthorScreenNameHydrator {
+    pub gizmoduck_client: Arc<dyn GizmoduckClient + Send + Sync>,
+}
+
+impl RetweetedAuthorScreenNameHydrator {
+    pub fn new(gizmoduck_client: Arc<dyn GizmoduckClient + Send + Sync>) -> Self {
+        Self { gizmoduck_client }
+    }
+}
+
+#[async_trait]
+impl Hydrator<ScoredPostsQuery, PostCandidate> for RetweetedAuthorScreenNameHydrator {
+    async fn hydrate(
+        &self,
+        _query: &ScoredPostsQuery,
+        candidates: &[PostCandidate],
+    ) -> Vec<Result<PostCandidate, String>> {
+        let retweeted_user_ids: Vec<i64> = candidates
+            .iter()
+            .filter_map(|c| c.retweeted_user_id)
+            .filter(|&id| id != 0)
+            .map(|id| id as i64)
+            .collect::<HashSet<_>>()
+            .into_iter()
+            .collect();
+
+        let users = if retweeted_user_ids.is_empty() {
+            Default::default()
+        } else {
+            self.gizmoduck_client.get_users(retweeted_user_ids).await
+        };
+
+        candidates
+            .iter()
+            .map(|candidate| {
+                let retweet_user = candidate
+                    .retweeted_user_id
+                    .filter(|&id| id != 0)
+                    .and_then(|id| users.get(&(id as i64)));
+                match retweet_user {
+                    Some(Err(err)) => Err(err.to_string()),
+                    Some(Ok(Some(user))) => Ok(PostCandidate {
+                        retweeted_screen_name: user
+                            .user
+                            .as_ref()
+                            .map(|u| u.profile.screen_name.clone()),
+                        ..Default::default()
+                    }),
+                    Some(Ok(None)) | None => Ok(PostCandidate {
+                        retweeted_screen_name: None,
+                        ..Default::default()
+                    }),
+                }
+            })
+            .collect()
+    }
+
+    fn update(&self, candidate: &mut PostCandidate, hydrated: PostCandidate) {
+        candidate.retweeted_screen_name = hydrated.retweeted_screen_name;
+    }
+}

home-mixer/candidate_pipeline/reverse_chron_posts_pipeline.rs

@@ -3,8 +3,10 @@ use crate::candidate_hydrators::conversation_gap_ancestor_hydrator::Conversation
 use crate::candidate_hydrators::core_data_candidate_hydrator::CoreDataCandidateHydrator;
 use crate::candidate_hydrators::following_blocked_by_hydrator::FollowingBlockedByHydrator;
 use crate::candidate_hydrators::quoted_post_text_hydrator::QuotedPostTextHydrator;
+use crate::candidate_hydrators::retweeted_author_screen_name_hydrator::RetweetedAuthorScreenNameHydrator;
 use crate::candidate_hydrators::tweet_type_metrics_hydrator::TweetTypeMetricsHydrator;
 use crate::candidate_hydrators::vf_following_candidate_hydrator::VFFollowingCandidateHydrator;
+use crate::clients::gizmoduck_client::{GizmoduckClient, MockGizmoduckClient, ProdGizmoduckClient};
 use crate::clients::night_owl_client::{MockNightOwlClient, NightOwlClient, ProdNightOwlClient};
 use crate::clients::s2s::{S2S_CHAIN_PATH, S2S_CRT_PATH, S2S_KEY_PATH};
 use crate::clients::tweet_entity_service_client::{MockTESClient, ProdTESClient, TESClient};
@@ -56,6 +58,7 @@ impl ReverseChronPostsPipeline {
             xai_vf_client,
             vf_safety_labels_client,
             socialgraph_client,
+            gizmoduck_client,
         ) = tokio::join!(
             async {
                 Arc::new(
@@ -112,6 +115,17 @@ impl ReverseChronPostsPipeline {
                     .expect("Failed to create flock SocialGraphClient"),
                 ) as Arc<dyn SocialGraphClientOps>
             },
+            async {
+                Arc::new(
+                    ProdGizmoduckClient::new(
+                        None,
+                        datacenter,
+                        Some("home-mixer.prod".to_string()),
+                    )
+                    .await
+                    .expect("Failed to create Gizmoduck client"),
+                ) as Arc<dyn GizmoduckClient + Send + Sync>
+            },
         );
 
         Self::build(
@@ -121,6 +135,7 @@ impl ReverseChronPostsPipeline {
             xai_vf_client,
             vf_safety_labels_client,
             socialgraph_client,
+            gizmoduck_client,
         )
         .await
     }
@@ -133,6 +148,7 @@ impl ReverseChronPostsPipeline {
             Arc::new(MockVfClient) as Arc<dyn VfClient + Send + Sync>,
             Arc::new(MockTweetSafetyLabelClient) as Arc<dyn TweetSafetyLabelClient>,
             Arc::new(MockSocialGraphClient) as Arc<dyn SocialGraphClientOps>,
+            Arc::new(MockGizmoduckClient::default()) as Arc<dyn GizmoduckClient + Send + Sync>,
         )
         .await
     }
@@ -144,6 +160,7 @@ impl ReverseChronPostsPipeline {
         xai_vf_client: Arc<dyn VfClient + Send + Sync>,
         vf_safety_labels_client: Arc<dyn TweetSafetyLabelClient>,
         socialgraph_client: Arc<dyn SocialGraphClientOps>,
+        gizmoduck_client: Arc<dyn GizmoduckClient + Send + Sync>,
     ) -> Self {
         let sources: Vec<Box<dyn Source<ScoredPostsQuery, PostCandidate>>> =
             vec![Box::new(FollowingNightOwlSource {
@@ -156,6 +173,7 @@ impl ReverseChronPostsPipeline {
                 &tes_client,
             ))),
             Box::new(QuotedPostTextHydrator::new(tes_client)),
+            Box::new(RetweetedAuthorScreenNameHydrator::new(gizmoduck_client)),
         ];
 
         let filters: Vec<Box<dyn Filter<ScoredPostsQuery, PostCandidate>>> = vec![

home-mixer/filters/following_viewer_muted_keyword_filter.rs

@@ -59,7 +59,94 @@ fn candidate_matches(
 ) -> bool {
     std::iter::once(candidate.tweet_text.as_str())
         .chain(candidate.quoted_tweet_text.as_deref())
+        .chain(candidate.retweeted_screen_name.as_deref())
         .chain(candidate.ancestor_texts.values().map(String::as_str))
         .filter(|text| !text.is_empty())
         .any(|text| matcher.matches(&tokenizer.tokenize(text)))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::models::user_features::UserFeatures;
+
+    fn query(muted_keywords: Vec<String>) -> ScoredPostsQuery {
+        ScoredPostsQuery {
+            user_features: UserFeatures {
+                muted_keywords,
+                ..Default::default()
+            },
+            ..Default::default()
+        }
+    }
+
+    fn text_candidate(tweet_id: u64, tweet_text: &str) -> PostCandidate {
+        PostCandidate {
+            tweet_id,
+            tweet_text: tweet_text.to_string(),
+            author_id: 12345,
+            ..Default::default()
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn drops_retweet_when_original_author_handle_matches_muted_keyword() {
+        let filter = FollowingViewerMutedKeywordFilter::new();
+        let retweet = PostCandidate {
+            tweet_id: 1,
+            tweet_text: "ordinary news".to_string(),
+            retweeted_tweet_id: Some(88),
+            retweeted_user_id: Some(99),
+            retweeted_screen_name: Some("spamaccount".to_string()),
+            author_id: 12345,
+            ..Default::default()
+        };
+
+        let result = filter.filter(
+            &query(vec!["spamaccount".to_string()]),
+            vec![retweet, text_candidate(2, "ordinary news")],
+        );
+
+        assert_eq!(result.kept.len(), 1);
+        assert_eq!(result.kept[0].tweet_id, 2);
+        assert_eq!(result.removed.len(), 1);
+        assert_eq!(result.removed[0].tweet_id, 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn keeps_retweet_when_original_author_handle_does_not_match() {
+        let filter = FollowingViewerMutedKeywordFilter::new();
+        let retweet = PostCandidate {
+            tweet_id: 1,
+            tweet_text: "ordinary news".to_string(),
+            retweeted_tweet_id: Some(88),
+            retweeted_user_id: Some(99),
+            retweeted_screen_name: Some("normaluser".to_string()),
+            author_id: 12345,
+            ..Default::default()
+        };
+
+        let result = filter.filter(&query(vec!["spamaccount".to_string()]), vec![retweet]);
+
+        assert_eq!(result.kept.len(), 1);
+        assert!(result.removed.is_empty());
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn still_drops_when_quoted_text_matches() {
+        let filter = FollowingViewerMutedKeywordFilter::new();
+        let quote = PostCandidate {
+            tweet_id: 1,
+            tweet_text: "sharing this".to_string(),
+            quoted_tweet_text: Some("this is spam content".to_string()),
+            retweeted_screen_name: Some("normaluser".to_string()),
+            author_id: 12345,
+            ..Default::default()
+        };
+
+        let result = filter.filter(&query(vec!["spam".to_string()]), vec![quote]);
+
+        assert_eq!(result.removed.len(), 1);
+        assert_eq!(result.removed[0].tweet_id, 1);
+    }
+}

home-mixer/filters/viewer_muted_keyword_filter.rs

@@ -43,8 +43,7 @@ impl Filter<ScoredPostsQuery, PostCandidate> for ViewerMutedKeywordFilter {
             let mut removed = Vec::new();
 
             for candidate in candidates {
-                let tweet_text_token_sequence = tokenizer.tokenize(&candidate.tweet_text);
-                if matcher.matches(&tweet_text_token_sequence) {
+                if candidate_matches(&candidate, &tokenizer, &matcher) {
                     removed.push(candidate);
                 } else {
                     kept.push(candidate);
@@ -56,6 +55,17 @@ impl Filter<ScoredPostsQuery, PostCandidate> for ViewerMutedKeywordFilter {
     }
 }
 
+fn candidate_matches(
+    candidate: &PostCandidate,
+    tokenizer: &TweetTokenizer,
+    matcher: &MatchTweetGroup,
+) -> bool {
+    std::iter::once(candidate.tweet_text.as_str())
+        .chain(candidate.retweeted_screen_name.as_deref())
+        .filter(|text| !text.is_empty())
+        .any(|text| matcher.matches(&tokenizer.tokenize(text)))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -315,4 +325,48 @@ mod tests {
         assert_eq!(result.kept[0].tweet_id, 4);
         assert_eq!(result.removed.len(), 3);
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn drops_retweet_when_original_author_handle_matches_muted_keyword() {
+        let filter = ViewerMutedKeywordFilter::new();
+        let query = create_test_query(vec!["spamaccount".to_string()]);
+
+        let retweet = PostCandidate {
+            tweet_id: 1,
+            tweet_text: "ordinary news".to_string(),
+            retweeted_tweet_id: Some(88),
+            retweeted_user_id: Some(99),
+            retweeted_screen_name: Some("spamaccount".to_string()),
+            author_id: 12345,
+            ..Default::default()
+        };
+
+        let result = filter.filter(&query, vec![retweet, create_test_candidate(2, "ordinary news")]);
+
+        assert_eq!(result.kept.len(), 1);
+        assert_eq!(result.kept[0].tweet_id, 2);
+        assert_eq!(result.removed.len(), 1);
+        assert_eq!(result.removed[0].tweet_id, 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn keeps_retweet_when_original_author_handle_does_not_match() {
+        let filter = ViewerMutedKeywordFilter::new();
+        let query = create_test_query(vec!["spamaccount".to_string()]);
+
+        let retweet = PostCandidate {
+            tweet_id: 1,
+            tweet_text: "ordinary news".to_string(),
+            retweeted_tweet_id: Some(88),
+            retweeted_user_id: Some(99),
+            retweeted_screen_name: Some("normaluser".to_string()),
+            author_id: 12345,
+            ..Default::default()
+        };
+
+        let result = filter.filter(&query, vec![retweet]);
+
+        assert_eq!(result.kept.len(), 1);
+        assert!(result.removed.is_empty());
+    }
 }