Stop WhoToFollow from recommending muted or blocked accounts

home-mixer/candidate_pipeline/following_candidate_pipeline.rs

@@ -14,7 +14,9 @@ use crate::clients::who_to_follow_client::{
 use crate::filters::invalid_conversation_module_filter::InvalidConversationModuleFilter;
 use crate::models::query::ScoredPostsQuery;
 use crate::params;
+use crate::query_hydrators::blocked_user_ids_query_hydrator::BlockedUserIdsQueryHydrator;
 use crate::query_hydrators::followed_user_ids_query_hydrator::FollowedUserIdsQueryHydrator;
+use crate::query_hydrators::muted_user_ids_query_hydrator::MutedUserIdsQueryHydrator;
 use crate::query_hydrators::past_request_timestamps_query_hydrator::PastRequestTimestampsQueryHydrator;
 use crate::query_hydrators::served_history_query_hydrator::ServedHistoryQueryHydrator;
 use crate::selectors::FollowingBlenderSelector;
@@ -168,6 +170,12 @@ impl FollowingCandidatePipeline {
             Box::new(PastRequestTimestampsQueryHydrator::new(Arc::clone(
                 &past_request_timestamps_client,
             ))),
+            Box::new(BlockedUserIdsQueryHydrator {
+                socialgraph_client: socialgraph_client.clone(),
+            }),
+            Box::new(MutedUserIdsQueryHydrator {
+                socialgraph_client: socialgraph_client.clone(),
+            }),
             Box::new(FollowedUserIdsQueryHydrator { socialgraph_client }),
         ];
 

home-mixer/candidate_pipeline/for_you_candidate_pipeline.rs

@@ -3,6 +3,7 @@ use crate::clients::past_request_timestamps_client::{
     MockPastRequestTimestampsClient, PastRequestTimestampsClient, ProdPastRequestTimestampsClient,
 };
 use crate::clients::prompts_client::{MockPromptsClient, ProdPromptsClient, PromptsClient};
+use crate::clients::s2s::{S2S_CHAIN_PATH, S2S_CRT_PATH, S2S_KEY_PATH};
 use crate::clients::served_history_client::{
     MockServedHistoryClient, ProdServedHistoryClient, ServedHistoryClient,
 };
@@ -14,6 +15,8 @@ use crate::filters::ad_adjacent_served_filter::AdAdjacentServedFilter;
 use crate::filters::push_to_home_dedup_filter::PushToHomeDedupFilter;
 use crate::models::query::ScoredPostsQuery;
 use crate::params;
+use crate::query_hydrators::blocked_user_ids_query_hydrator::BlockedUserIdsQueryHydrator;
+use crate::query_hydrators::muted_user_ids_query_hydrator::MutedUserIdsQueryHydrator;
 use crate::query_hydrators::past_request_timestamps_query_hydrator::PastRequestTimestampsQueryHydrator;
 use crate::query_hydrators::served_history_query_hydrator::ServedHistoryQueryHydrator;
 use crate::scored_posts_server::ScoredPostsServer;
@@ -41,8 +44,8 @@ use xai_candidate_pipeline::component_library::clients::kafka_publisher_client::
     KafkaPublisherClient, MockKafkaPublisherClient,
 };
 use xai_candidate_pipeline::component_library::clients::{
-    MockReplyMixerClient, MockStratoClient, ProdReplyMixerClient, ProdStratoClient,
-    ReplyMixerClient, StratoClient,
+    MockReplyMixerClient, MockSocialGraphClient, MockStratoClient, ProdReplyMixerClient,
+    ProdStratoClient, ReplyMixerClient, SocialGraphClient, SocialGraphClientOps, StratoClient,
 };
 use xai_candidate_pipeline::filter::Filter;
 use xai_candidate_pipeline::hydrator::Hydrator;
@@ -83,6 +86,7 @@ impl ForYouCandidatePipeline {
             tes_client,
             reply_mixer_client,
             strato_client,
+            socialgraph_client,
         ) = tokio::join!(
             async {
                 Arc::new(
@@ -145,6 +149,18 @@ impl ForYouCandidatePipeline {
                         .expect("Failed to create Strato client"),
                 ) as Arc<dyn StratoClient + Send + Sync>
             },
+            async {
+                Arc::new(
+                    SocialGraphClient::new(
+                        datacenter,
+                        &S2S_CHAIN_PATH,
+                        &S2S_CRT_PATH,
+                        &S2S_KEY_PATH,
+                    )
+                    .await
+                    .expect("Failed to create flock SocialGraphClient"),
+                ) as Arc<dyn SocialGraphClientOps>
+            },
         );
 
         Self::build(
@@ -162,6 +178,7 @@ impl ForYouCandidatePipeline {
             past_request_timestamps_client,
             tes_client,
             reply_mixer_client,
+            socialgraph_client,
         )
     }
 
@@ -181,6 +198,7 @@ impl ForYouCandidatePipeline {
         past_request_timestamps_client: Arc<dyn PastRequestTimestampsClient>,
         tes_client: Arc<dyn TESClient + Send + Sync>,
         reply_mixer_client: Arc<dyn ReplyMixerClient>,
+        socialgraph_client: Arc<dyn SocialGraphClientOps>,
     ) -> Self {
         let query_hydrators: Vec<Box<dyn QueryHydrator<ScoredPostsQuery>>> = vec![
             Box::new(ServedHistoryQueryHydrator::from_client(Arc::clone(
@@ -189,6 +207,12 @@ impl ForYouCandidatePipeline {
             Box::new(PastRequestTimestampsQueryHydrator::new(Arc::clone(
                 &past_request_timestamps_client,
             ))),
+            Box::new(BlockedUserIdsQueryHydrator {
+                socialgraph_client: socialgraph_client.clone(),
+            }),
+            Box::new(MutedUserIdsQueryHydrator {
+                socialgraph_client,
+            }),
         ];
 
         let sources: Vec<Box<dyn Source<ScoredPostsQuery, FeedItem>>> = vec![
@@ -263,6 +287,7 @@ impl ForYouCandidatePipeline {
         let reply_mixer_client: Arc<dyn ReplyMixerClient> = Arc::new(MockReplyMixerClient);
         let strato_client: Arc<dyn StratoClient + Send + Sync> =
             Arc::new(MockStratoClient::default());
+        let socialgraph_client: Arc<dyn SocialGraphClientOps> = Arc::new(MockSocialGraphClient);
         Self::build(
             scored_posts_server,
             strato_client,
@@ -278,6 +303,7 @@ impl ForYouCandidatePipeline {
             past_request_timestamps_client,
             tes_client,
             reply_mixer_client,
+            socialgraph_client,
         )
     }
 }

home-mixer/models/query.rs

@@ -113,6 +113,10 @@ pub struct ScoredPostsQuery {
     #[serde(skip)]
     pub served_history: Vec<ServedHistory>,
     pub who_to_follow_eligible: bool,
+    /// True only after a successful SGS mute-list read. Err leaves this false.
+    pub muted_user_ids_hydrated: bool,
+    /// True only after a successful SGS block-list read. Err leaves this false.
+    pub blocked_user_ids_hydrated: bool,
     pub feed_survey_eligible: bool,
     #[serde(serialize_with = "serialize_debug")]
     pub non_polling_timestamps: Option<NonPollingTimestamps>,
@@ -222,6 +226,8 @@ impl ScoredPostsQuery {
             request_context: String::new(),
             served_history: vec![],
             who_to_follow_eligible: false,
+            muted_user_ids_hydrated: false,
+            blocked_user_ids_hydrated: false,
             feed_survey_eligible: false,
             non_polling_timestamps: None,
             impressed_post_ids: Vec::new(),
@@ -242,6 +248,10 @@ impl ScoredPostsQuery {
     pub fn has_excluded_topics(&self) -> bool {
         !self.excluded_topic_ids.is_empty()
     }
+
+    pub fn mute_block_lists_ready(&self) -> bool {
+        self.muted_user_ids_hydrated && self.blocked_user_ids_hydrated
+    }
 }
 
 impl GetTwitterContextViewer for ScoredPostsQuery {

home-mixer/query_hydrators/blocked_user_ids_query_hydrator.rs

@@ -11,6 +11,10 @@ pub struct BlockedUserIdsQueryHydrator {
 
 #[async_trait]
 impl QueryHydrator<ScoredPostsQuery> for BlockedUserIdsQueryHydrator {
+    fn enable(&self, query: &ScoredPostsQuery) -> bool {
+        !query.blocked_user_ids_hydrated
+    }
+
     async fn hydrate(&self, query: &ScoredPostsQuery) -> Result<ScoredPostsQuery, String> {
         let blocked_user_ids = self
             .socialgraph_client
@@ -23,11 +27,41 @@ impl QueryHydrator<ScoredPostsQuery> for BlockedUserIdsQueryHydrator {
                 blocked_user_ids,
                 ..Default::default()
             },
+            blocked_user_ids_hydrated: true,
             ..Default::default()
         })
     }
 
     fn update(&self, query: &mut ScoredPostsQuery, hydrated: ScoredPostsQuery) {
         query.user_features.blocked_user_ids = hydrated.user_features.blocked_user_ids;
+        query.blocked_user_ids_hydrated = true;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use xai_candidate_pipeline::component_library::clients::MockSocialGraphClient;
+
+    #[test]
+    fn update_marks_list_ready() {
+        let hydrator = BlockedUserIdsQueryHydrator {
+            socialgraph_client: Arc::new(MockSocialGraphClient),
+        };
+        let mut query = ScoredPostsQuery::default();
+        let mut hydrated = ScoredPostsQuery::default();
+        hydrated.user_features.blocked_user_ids = vec![9];
+        hydrator.update(&mut query, hydrated);
+        assert_eq!(query.user_features.blocked_user_ids, vec![9]);
+        assert!(query.blocked_user_ids_hydrated);
+        assert!(!hydrator.enable(&query));
+    }
+
+    #[test]
+    fn enable_when_list_not_loaded() {
+        let hydrator = BlockedUserIdsQueryHydrator {
+            socialgraph_client: Arc::new(MockSocialGraphClient),
+        };
+        assert!(hydrator.enable(&ScoredPostsQuery::default()));
     }
 }

home-mixer/query_hydrators/muted_user_ids_query_hydrator.rs

@@ -11,6 +11,10 @@ pub struct MutedUserIdsQueryHydrator {
 
 #[async_trait]
 impl QueryHydrator<ScoredPostsQuery> for MutedUserIdsQueryHydrator {
+    fn enable(&self, query: &ScoredPostsQuery) -> bool {
+        !query.muted_user_ids_hydrated
+    }
+
     async fn hydrate(&self, query: &ScoredPostsQuery) -> Result<ScoredPostsQuery, String> {
         let muted_user_ids = self
             .socialgraph_client
@@ -23,11 +27,41 @@ impl QueryHydrator<ScoredPostsQuery> for MutedUserIdsQueryHydrator {
                 muted_user_ids,
                 ..Default::default()
             },
+            muted_user_ids_hydrated: true,
             ..Default::default()
         })
     }
 
     fn update(&self, query: &mut ScoredPostsQuery, hydrated: ScoredPostsQuery) {
         query.user_features.muted_user_ids = hydrated.user_features.muted_user_ids;
+        query.muted_user_ids_hydrated = true;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use xai_candidate_pipeline::component_library::clients::MockSocialGraphClient;
+
+    #[test]
+    fn update_marks_list_ready() {
+        let hydrator = MutedUserIdsQueryHydrator {
+            socialgraph_client: Arc::new(MockSocialGraphClient),
+        };
+        let mut query = ScoredPostsQuery::default();
+        let mut hydrated = ScoredPostsQuery::default();
+        hydrated.user_features.muted_user_ids = vec![7];
+        hydrator.update(&mut query, hydrated);
+        assert_eq!(query.user_features.muted_user_ids, vec![7]);
+        assert!(query.muted_user_ids_hydrated);
+        assert!(!hydrator.enable(&query));
+    }
+
+    #[test]
+    fn enable_when_list_not_loaded() {
+        let hydrator = MutedUserIdsQueryHydrator {
+            socialgraph_client: Arc::new(MockSocialGraphClient),
+        };
+        assert!(hydrator.enable(&ScoredPostsQuery::default()));
     }
 }