From 6309f5918a6dc488113742f7b3e50167fc9954dc Mon Sep 17 00:00:00 2001 From: SuspiciousIceCream Date: Sun, 16 Aug 2026 12:13:57 +0800 Subject: [PATCH] Refactor imports and FollowingPipelineDeps struct Refactor imports for better organization and readability. Update the FollowingPipelineDeps struct to group dependencies together, simplifying the build process. --- .../following_candidate_pipeline.rs | 305 +++++++++++++----- 1 file changed, 220 insertions(+), 85 deletions(-) diff --git a/home-mixer/candidate_pipeline/following_candidate_pipeline.rs b/home-mixer/candidate_pipeline/following_candidate_pipeline.rs index c1efd927..1c30780f 100644 --- a/home-mixer/candidate_pipeline/following_candidate_pipeline.rs +++ b/home-mixer/candidate_pipeline/following_candidate_pipeline.rs @@ -1,15 +1,33 @@ use crate::candidate_pipeline::reverse_chron_posts_pipeline::ReverseChronPostsPipeline; -use crate::clients::ad_index_client::{AdIndexClient, MockAdIndexClient, ProdAdIndexClient}; +use crate::clients::ad_index_client::{ + AdIndexClient, + MockAdIndexClient, + ProdAdIndexClient, +}; use crate::clients::past_request_timestamps_client::{ - MockPastRequestTimestampsClient, PastRequestTimestampsClient, ProdPastRequestTimestampsClient, + 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::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, + MockServedHistoryClient, + ProdServedHistoryClient, + ServedHistoryClient, }; use crate::clients::who_to_follow_client::{ - MockWhoToFollowClient, ProdWhoToFollowClient, WhoToFollowClient, + MockWhoToFollowClient, + ProdWhoToFollowClient, + WhoToFollowClient, }; use crate::filters::invalid_conversation_module_filter::InvalidConversationModuleFilter; use crate::models::query::ScoredPostsQuery; @@ -33,12 +51,16 @@ use crate::sources::reverse_chron_posts_source::ReverseChronPostsSource; use crate::sources::who_to_follow_source::WhoToFollowSource; use std::sync::Arc; use tonic::async_trait; + use xai_candidate_pipeline::candidate_pipeline::CandidatePipeline; use xai_candidate_pipeline::component_library::clients::kafka_publisher_client::{ - KafkaPublisherClient, MockKafkaPublisherClient, + KafkaPublisherClient, + MockKafkaPublisherClient, }; use xai_candidate_pipeline::component_library::clients::{ - MockSocialGraphClient, SocialGraphClient, SocialGraphClientOps, + MockSocialGraphClient, + SocialGraphClient, + SocialGraphClientOps, }; use xai_candidate_pipeline::filter::Filter; use xai_candidate_pipeline::hydrator::Hydrator; @@ -59,6 +81,26 @@ pub struct FollowingCandidatePipeline { side_effects: Arc>>>, } +/// All dependencies required to construct the following candidate pipeline. +/// +/// Keeping these together makes `build()` easier to read and reduces the +/// chance of accidentally wiring a dependency into the wrong component. +struct FollowingPipelineDeps { + ad_index_client: Arc, + served_history_client: Arc, + past_request_timestamps_client: Arc, + socialgraph_client: Arc, + reverse_chron_pipeline: Arc, + who_to_follow_client: Arc, + prompts_client: Arc, + + ads_injection_logging: AdsInjectionLoggingSideEffect, + served_ad_history: ServedAdHistoryCacheSideEffect, + publish_seen_ids: PublishSeenIdsToKafkaSideEffect, + served_candidates: ServedCandidatesKafkaSideEffect, + client_events: ClientEventsKafkaSideEffect, +} + impl FollowingCandidatePipeline { pub async fn new(datacenter: &str) -> Self { let ( @@ -130,12 +172,12 @@ impl FollowingCandidatePipeline { ClientEventsKafkaSideEffect::prod(), ); - Self::build( + Self::build(FollowingPipelineDeps { ad_index_client, served_history_client, past_request_timestamps_client, socialgraph_client, - Arc::new(reverse_chron_pipeline), + reverse_chron_pipeline: Arc::new(reverse_chron_pipeline), who_to_follow_client, prompts_client, ads_injection_logging, @@ -143,66 +185,102 @@ impl FollowingCandidatePipeline { publish_seen_ids, served_candidates, client_events, - ) + }) } - #[allow(clippy::too_many_arguments)] - fn build( - ad_index_client: Arc, - served_history_client: Arc, - past_request_timestamps_client: Arc, - socialgraph_client: Arc, - reverse_chron_pipeline: Arc, - who_to_follow_client: Arc, - prompts_client: Arc, - ads_injection_logging: AdsInjectionLoggingSideEffect, - served_ad_history: ServedAdHistoryCacheSideEffect, - publish_seen_ids: PublishSeenIdsToKafkaSideEffect, - served_candidates: ServedCandidatesKafkaSideEffect, - client_events: ClientEventsKafkaSideEffect, - ) -> Self { - let query_hydrators: Vec>> = vec![ - Box::new(ServedHistoryQueryHydrator::from_client(Arc::clone( - &served_history_client, - ))), - Box::new(PastRequestTimestampsQueryHydrator::new(Arc::clone( - &past_request_timestamps_client, - ))), - Box::new(FollowedUserIdsQueryHydrator { socialgraph_client }), + fn build(deps: FollowingPipelineDeps) -> Self { + let FollowingPipelineDeps { + ad_index_client, + served_history_client, + past_request_timestamps_client, + socialgraph_client, + reverse_chron_pipeline, + who_to_follow_client, + prompts_client, + ads_injection_logging, + served_ad_history, + publish_seen_ids, + served_candidates, + client_events, + } = deps; + + // Query hydration happens before candidate generation. + let query_hydrators = vec![ + Box::new( + ServedHistoryQueryHydrator::from_client( + Arc::clone(&served_history_client), + ), + ) + as Box>, + Box::new( + PastRequestTimestampsQueryHydrator::new( + Arc::clone(&past_request_timestamps_client), + ), + ), + Box::new(FollowedUserIdsQueryHydrator { + socialgraph_client, + }), ]; - let sources: Vec>> = vec![ - Box::new(ReverseChronPostsSource::new(reverse_chron_pipeline)), - Box::new(AdsSource { ad_index_client }), + // Candidate sources are intentionally kept in the same order as the + // existing implementation. + let sources = vec![ + Box::new(ReverseChronPostsSource::new(reverse_chron_pipeline)) + as Box>, + Box::new(AdsSource { + ad_index_client, + }), Box::new(WhoToFollowSource { who_to_follow_client, }), - Box::new(PromptsSource { prompts_client }), + Box::new(PromptsSource { + prompts_client, + }), + ]; + + // No candidate hydrators are currently configured. + let hydrators = Vec::new(); + + // Preserve the existing filtering behavior. + let filters = vec![ + Box::new(InvalidConversationModuleFilter) + as Box>, ]; - let hydrators: Vec>> = vec![]; - let filters: Vec>> = - vec![Box::new(InvalidConversationModuleFilter)]; let selector = FollowingBlenderSelector::new(); - let side_effects: Arc>>> = - Arc::new(vec![ - Box::new(ads_injection_logging), - Box::new(served_ad_history), - Box::new(publish_seen_ids), - Box::new(served_candidates), - Box::new(client_events), - Box::new(ResponseStatsSideEffect), - Box::new(UpdatePastRequestTimestampsSideEffect::new( - past_request_timestamps_client, - )), - Box::new(UpdateServedHistorySideEffect::new(Arc::clone( - &served_history_client, - ))), - Box::new(TruncateServedHistorySideEffect::new(served_history_client)), - ]); + // No post-selection filters are currently configured. + let post_selection_filters = Vec::new(); - let post_selection_filters: Vec>> = vec![]; + // Keep side-effect ordering unchanged. + // + // The clients are cloned only where they are genuinely shared by + // multiple components. + let side_effects: Arc< + Vec>>, + > = Arc::new(vec![ + Box::new(ads_injection_logging), + Box::new(served_ad_history), + Box::new(publish_seen_ids), + Box::new(served_candidates), + Box::new(client_events), + Box::new(ResponseStatsSideEffect), + Box::new( + UpdatePastRequestTimestampsSideEffect::new( + past_request_timestamps_client, + ), + ), + Box::new( + UpdateServedHistorySideEffect::new( + Arc::clone(&served_history_client), + ), + ), + Box::new( + TruncateServedHistorySideEffect::new( + served_history_client, + ), + ), + ]); Self { query_hydrators, @@ -216,24 +294,61 @@ impl FollowingCandidatePipeline { } pub async fn mock() -> Self { - let ad_index_client: Arc = Arc::new(MockAdIndexClient); - let served_history_client: Arc = Arc::new(MockServedHistoryClient); - let past_request_timestamps_client: Arc = + let ad_index_client: Arc = + Arc::new(MockAdIndexClient); + + let served_history_client: Arc = + Arc::new(MockServedHistoryClient); + + let past_request_timestamps_client: + Arc = Arc::new(MockPastRequestTimestampsClient); - let socialgraph_client: Arc = Arc::new(MockSocialGraphClient); - let reverse_chron_pipeline = Arc::new(ReverseChronPostsPipeline::mock().await); - let who_to_follow_client: Arc = + + let socialgraph_client: Arc = + Arc::new(MockSocialGraphClient); + + let reverse_chron_pipeline = + Arc::new(ReverseChronPostsPipeline::mock().await); + + let who_to_follow_client: + Arc = Arc::new(MockWhoToFollowClient); - let prompts_client: Arc = Arc::new(MockPromptsClient); - let mock_kafka = Arc::new(MockKafkaPublisherClient) as Arc; - let ads_injection_logging = AdsInjectionLoggingSideEffect::new(Arc::clone(&mock_kafka)); - let served_ad_history = ServedAdHistoryCacheSideEffect::new(Arc::new( - xai_ad_index_history::InMemoryUserAdHistoryStore::default(), - )); - let publish_seen_ids = PublishSeenIdsToKafkaSideEffect::new(Arc::clone(&mock_kafka)); - let served_candidates = ServedCandidatesKafkaSideEffect::new(Arc::clone(&mock_kafka)); - let client_events = ClientEventsKafkaSideEffect::new(Arc::clone(&mock_kafka)); - Self::build( + + let prompts_client: Arc = + Arc::new(MockPromptsClient); + + let mock_kafka = + Arc::new(MockKafkaPublisherClient) + as Arc; + + let ads_injection_logging = + AdsInjectionLoggingSideEffect::new( + Arc::clone(&mock_kafka), + ); + + let served_ad_history = + ServedAdHistoryCacheSideEffect::new( + Arc::new( + xai_ad_index_history::InMemoryUserAdHistoryStore::default(), + ), + ); + + let publish_seen_ids = + PublishSeenIdsToKafkaSideEffect::new( + Arc::clone(&mock_kafka), + ); + + let served_candidates = + ServedCandidatesKafkaSideEffect::new( + Arc::clone(&mock_kafka), + ); + + let client_events = + ClientEventsKafkaSideEffect::new( + Arc::clone(&mock_kafka), + ); + + Self::build(FollowingPipelineDeps { ad_index_client, served_history_client, past_request_timestamps_client, @@ -246,45 +361,65 @@ impl FollowingCandidatePipeline { publish_seen_ids, served_candidates, client_events, - ) + }) } } #[async_trait] -impl CandidatePipeline for FollowingCandidatePipeline { - fn query_hydrators(&self) -> &[Box>] { +impl CandidatePipeline + for FollowingCandidatePipeline +{ + fn query_hydrators( + &self, + ) -> &[Box>] { &self.query_hydrators } - fn sources(&self) -> &[Box>] { + fn sources( + &self, + ) -> &[Box>] { &self.sources } - fn hydrators(&self) -> &[Box>] { + fn hydrators( + &self, + ) -> &[Box>] { &self.hydrators } - fn filters(&self) -> &[Box>] { + fn filters( + &self, + ) -> &[Box>] { &self.filters } - fn scorers(&self) -> &[Box>] { + fn scorers( + &self, + ) -> &[Box>] { &[] } - fn selector(&self) -> &dyn Selector { + fn selector( + &self, + ) -> &dyn Selector { &self.selector } - fn post_selection_hydrators(&self) -> &[Box>] { + fn post_selection_hydrators( + &self, + ) -> &[Box>] { &[] } - fn post_selection_filters(&self) -> &[Box>] { + fn post_selection_filters( + &self, + ) -> &[Box>] { &self.post_selection_filters } - fn side_effects(&self) -> Arc>>> { + fn side_effects( + &self, + ) -> Arc>>> { Arc::clone(&self.side_effects) }