use crate::candidate_hydrators::ads_brand_safety_vf_hydrator::AdsBrandSafetyVfHydrator; use crate::candidate_hydrators::ai_trend_feedback_context_hydrator::AiTrendFeedbackContextHydrator; use crate::candidate_hydrators::ancestor_video_hydrator::AncestorVideoHydrator; use crate::candidate_hydrators::bidirectional_follow_hydrator::BidirectionalFollowHydrator; use crate::candidate_hydrators::blocked_by_hydrator::BlockedByHydrator; use crate::candidate_hydrators::core_data_candidate_hydrator::CoreDataCandidateHydrator; use crate::candidate_hydrators::engagement_counts_hydrator::EngagementCountsHydrator; use crate::candidate_hydrators::filtered_topics_hydrator::FilteredTopicsHydrator; use crate::candidate_hydrators::following_replied_users_hydrator::FollowingRepliedUsersHydrator; use crate::candidate_hydrators::gizmoduck_hydrator::GizmoduckCandidateHydrator; use crate::candidate_hydrators::in_network_candidate_hydrator::InNetworkCandidateHydrator; use crate::candidate_hydrators::language_code_hydrator::LanguageCodeHydrator; use crate::candidate_hydrators::media_info_hydrator::MediaInfoHydrator; use crate::candidate_hydrators::mutual_follow_jaccard_hydrator::MutualFollowJaccardHydrator; use crate::candidate_hydrators::new_user_soft_nsfw_label_hydrator::NewUserSoftNsfwLabelHydrator; use crate::candidate_hydrators::phoenix_reply_ancestors_hydrator::PhoenixReplyAncestorsHydrator; use crate::candidate_hydrators::post_engagers_hydrator::PostEngagersHydrator; use crate::candidate_hydrators::quote_hydrator::QuoteHydrator; use crate::candidate_hydrators::semantic_id_hydrator::SemanticIdHydrator; use crate::candidate_hydrators::subscription_hydrator::SubscriptionHydrator; use crate::candidate_hydrators::topic_feedback_context_hydrator::TopicFeedbackContextHydrator; use crate::candidate_hydrators::tweet_type_metrics_hydrator::TweetTypeMetricsHydrator; use crate::candidate_hydrators::vf_candidate_hydrator::VFCandidateHydrator; use crate::candidate_hydrators::video_embedding_hydrator::VideoEmbeddingHydrator; use crate::candidate_hydrators::video_media_hydrator::VideoMediaHydrator; use crate::clients::author_brand_safety_client::{ AuthorBrandSafetyClient, MockAuthorBrandSafetyClient, }; use crate::clients::engagement_counts_client::{ EngagementCountsClient, ProdEngagementCountsClient, }; use crate::clients::engagement_signals_client::{ EngagementSignalsClient, MockEngagementSignalsClient, ProdEngagementSignalsClient, }; use crate::clients::gizmoduck_client::{GizmoduckClient, MockGizmoduckClient, ProdGizmoduckClient}; use crate::clients::post_v8_search_embedding_client::{ ManhattanPostV8SearchEmbeddingClient, NoopPostV8SearchEmbeddingClient, PostV8SearchEmbeddingClient, }; use crate::clients::impressed_posts_client::ImpressedPostsClient; use crate::clients::popular_authors_store_client::{ ManhattanPopularAuthorsStore, ManhattanPopularPostsStore, }; use crate::clients::post_engagers_client::{ MockPostEngagersClient, PostEngagersClient, ProdPostEngagersClient, }; use crate::clients::recent_search_queries_store_client::{ MockRecentSearchQueriesStoreClient, ProdRecentSearchQueriesStoreClient, RecentSearchQueriesStoreClient, }; use crate::clients::s2s::{S2S_CHAIN_PATH, S2S_CRT_PATH, S2S_KEY_PATH}; use crate::clients::sid_retrieval_client::{ MockSidRetrievalClient, ProdSidRetrievalClient, SidRetrievalClient, }; use crate::clients::simclusters_ann_cache_client::{ MockSimClustersAnnCacheClient, ProdSimClustersAnnCacheClient, SimClustersAnnCacheClient, }; use crate::clients::simclusters_ann_client::{ MockSimClustersAnnClient, ProdSimClustersAnnClient, SimClustersAnnClient, }; use crate::clients::tweet_entity_service_client::{MockTESClient, ProdTESClient, TESClient}; use crate::clients::user_action_aggregation_client::{ MockUserActionAggregationClient, ProdUserActionAggregationClient, UserActionAggregationClient, }; use crate::clients::vm_ranker_client::{MockVMRankerClient, ProdVMRankerClient, VMRankerClient}; use crate::filters::age_filter::AgeFilter; use crate::filters::ancestor_video_filter::AncestorVideoFilter; use crate::filters::ancillary_vf_filter::AncillaryVFFilter; use crate::filters::author_socialgraph_filter::AuthorSocialgraphFilter; use crate::filters::core_data_hydration_filter::CoreDataHydrationFilter; use crate::filters::dedup_conversation_filter::DedupConversationFilter; use crate::filters::drop_duplicates_filter::DropDuplicatesFilter; use crate::filters::fair_share_ranking_filter::FairShareRankingFilter; use crate::filters::fav_holdout_filter::FavHoldoutFilter; use crate::filters::ineligible_subscription_filter::IneligibleSubscriptionFilter; use crate::filters::inventory_holdout_filter::InventoryHoldoutFilter; use crate::filters::new_user_min_engagement_filter::NewUserMinEngagementFilter; use crate::filters::new_user_soft_nsfw_filter::NewUserSoftNsfwFilter; use crate::filters::oon_nsfw_simclusters_filter::OONNsfwSimclustersFilter; use crate::filters::oon_retweet_reply_filter::OONRetweetReplyFilter; use crate::filters::previously_seen_posts_backup_filter::PreviouslySeenPostsBackupFilter; use crate::filters::previously_seen_posts_filter::PreviouslySeenPostsFilter; use crate::filters::previously_served_posts_filter::PreviouslyServedPostsFilter; use crate::filters::retweet_deduplication_filter::RetweetDeduplicationFilter; use crate::filters::self_tweet_filter::SelfTweetFilter; use crate::filters::topic_ids_filter::TopicIdsFilter; use crate::filters::vf_filter::VFFilter; use crate::filters::video_carousel_filter::VideoCarouselFilter; use crate::filters::video_filter::VideoFilter; use crate::filters::viewer_muted_keyword_filter::ViewerMutedKeywordFilter; use crate::filters::zero_fav_codivert_filter::ZeroFavCodivertFilter; use crate::models::candidate::PostCandidate; use crate::models::query::ScoredPostsQuery; use crate::params; use crate::query_hydrators::blocked_user_ids_query_hydrator::BlockedUserIdsQueryHydrator; use crate::query_hydrators::cached_posts_query_hydrator::CachedPostsQueryHydrator; use crate::query_hydrators::explicit_engagement_signals_query_hydrator::ExplicitEngagementSignalsQueryHydrator; use crate::query_hydrators::followed_grok_topics_query_hydrator::FollowedGrokTopicsQueryHydrator; use crate::query_hydrators::followed_starter_packs_query_hydrator::FollowedStarterPacksQueryHydrator; use crate::query_hydrators::followed_user_ids_query_hydrator::FollowedUserIdsQueryHydrator; use crate::query_hydrators::immersive_scoring_sequence_query_hydrator::ImmersiveScoringSequenceQueryHydrator; use crate::query_hydrators::implicit_engagement_signals_query_hydrator::ImplicitEngagementSignalsQueryHydrator; use crate::query_hydrators::impressed_posts_query_hydrator::ImpressedPostsQueryHydrator; use crate::query_hydrators::impression_bloom_filter_query_hydrator::ImpressionBloomFilterQueryHydrator; use crate::query_hydrators::ip_query_hydrator::IpQueryHydrator; use crate::query_hydrators::muted_user_ids_query_hydrator::MutedUserIdsQueryHydrator; use crate::query_hydrators::mutual_follow_query_hydrator::MutualFollowQueryHydrator; use crate::query_hydrators::recent_search_queries_query_hydrator::RecentSearchQueriesQueryHydrator; use crate::query_hydrators::retrieval_sequence_query_hydrator::RetrievalSequenceQueryHydrator; use crate::query_hydrators::scoring_sequence_query_hydrator::ScoringSequenceQueryHydrator; use crate::query_hydrators::subscribed_user_ids_query_hydrator::SubscribedUserIdsQueryHydrator; use crate::query_hydrators::user_demographics_query_hydrator::UserDemographicsQueryHydrator; use crate::query_hydrators::user_inferred_gender_query_hydrator::UserInferredGenderQueryHydrator; use crate::query_hydrators::user_installed_apps_query_hydrator::UserInstalledAppsQueryHydrator; use crate::scorers::immersive_video_scorer::ImmersiveVideoScorer; use crate::scorers::page_decode_rerank_scorer::PageDecodeRerankScorer; use crate::scorers::page_rl_exploration_scorer::PageRlExplorationScorer; use crate::scorers::phoenix_scorer::PhoenixScorer; use crate::scorers::vm_ranker::VMRanker; use crate::selectors::TopKScoreSelector; use crate::side_effects::author_served_metrics_side_effect::AuthorServedMetricsSideEffect; use crate::side_effects::debug_side_effect::DebugSideEffect; use crate::side_effects::mutual_follow_stats_side_effect::MutualFollowStatsSideEffect; use crate::side_effects::phoenix_experiments_side_effect::PhoenixExperimentsSideEffect; use crate::side_effects::phoenix_request_cache_side_effect::PhoenixRequestCacheSideEffect; use crate::side_effects::redis_post_candidate_cache_side_effect::RedisPostCandidateCacheSideEffect; use crate::side_effects::reranking_kafka_side_effect::RerankingKafkaSideEffect; use crate::side_effects::response_diversity_stats_side_effect::ResponseDiversityStatsSideEffect; use crate::side_effects::retrieval_candidates_kafka_side_effect::RetrievalCandidatesKafkaSideEffect; use crate::side_effects::scored_stats_side_effect::ScoredStatsSideEffect; use crate::side_effects::serendipity_stats_side_effect::SerendipityStatsSideEffect; use crate::sources::cached_posts_source::CachedPostsSource; use crate::sources::phoenix_moe_source::PhoenixMOESource; use crate::sources::phoenix_source::PhoenixSource; use crate::sources::phoenix_topics_source::PhoenixTopicsSource; use crate::sources::popular_posts_source::PopularPostsSource; use crate::sources::sid_source::SidSource; use crate::sources::simclusters_source::SimclustersSource; use crate::sources::thunder_source::ThunderSource; use crate::sources::tweet_mixer_source::TweetMixerSource; use crate::util::popular_authors::{ InMemoryPopularAuthorsStore, PopularAuthorsCache, PopularAuthorsStore, }; use crate::util::popular_posts::{InMemoryPopularPostsStore, PopularPostsCache, PopularPostsStore}; use xai_candidate_pipeline::component_library::clients::followed_grok_topics_store_client::{ FollowedGrokTopicsStoreClient, MockFollowedGrokTopicsStoreClient, ProdFollowedGrokTopicsStoreClient, }; use xai_candidate_pipeline::component_library::clients::followed_starter_packs_store_client::{ FollowedStarterPacksStoreClient, MockFollowedStarterPacksStoreClient, ProdFollowedStarterPacksStoreClient, }; use xai_candidate_pipeline::component_library::clients::gender_prediction_client::{ GenderPredictionGrpcClient, MockGenderPredictionGrpcClient, ProdGenderPredictionGrpcClient, }; use xai_candidate_pipeline::component_library::clients::kafka_publisher_client::{ KafkaCluster, KafkaPublisherClient, MockKafkaPublisherClient, ProdKafkaPublisherClient, LOGGED_SCORED_CANDIDATES_TOPIC, PHOENIX_SCORES_TOPIC, RERANKING_TOPIC, RETRIEVAL_CANDIDATES_TOPIC, }; use xai_candidate_pipeline::component_library::clients::media_info_cache_client::{ MediaInfoCacheClient, MockMediaInfoCacheClient, ProdMediaInfoCacheClient, }; use xai_candidate_pipeline::component_library::clients::user_demographics_client::{ MockUserDemographicsClient, ProdUserDemographicsClient, UserDemographicsClient, }; use xai_candidate_pipeline::component_library::clients::user_inferred_gender_store_client::{ MockUserInferredGenderStoreClient, ProdUserInferredGenderStoreClient, UserInferredGenderStoreClient, }; use xai_candidate_pipeline::component_library::clients::user_installed_apps_store_client::{ MockUserInstalledAppsStoreClient, ProdUserInstalledAppsStoreClient, UserInstalledAppsStoreClient, }; use xai_candidate_pipeline::component_library::clients::{MockSidClient, ProdSidClient, SidClient}; use xai_candidate_pipeline::component_library::clients::{ MockTweetMixerClient, ProdTweetMixerClient, TweetMixerClient, }; use std::sync::Arc; use std::time::Duration; use tonic::async_trait; use xai_candidate_pipeline::candidate_pipeline::CandidatePipeline; use xai_candidate_pipeline::component_library::clients::phoenix_prediction_client::{ MockPredictClient, PhoenixPredictionClient, }; use xai_candidate_pipeline::component_library::clients::phoenix_retrieval_client::{ MockRetrievalClient, PhoenixRetrievalClient, }; use xai_candidate_pipeline::component_library::clients::redis_client::{ MockRedisClient, RedisClient, }; use xai_candidate_pipeline::component_library::clients::{ ImpressionBloomFilterClient, MockImpressionBloomFilterClient, ProdImpressionBloomFilterClient, }; use xai_candidate_pipeline::component_library::clients::{ MockSocialGraphClient, SocialGraphClient, SocialGraphClientOps, }; use xai_candidate_pipeline::component_library::clients::{ MockStratoClient, ProdStratoClient, StratoClient, }; use xai_candidate_pipeline::component_library::clients::{ ProdThunderCapiClient, ThunderCapiClient, ThunderClient, }; use xai_candidate_pipeline::filter::Filter; use xai_candidate_pipeline::hydrator::Hydrator; use xai_candidate_pipeline::query_hydrator::QueryHydrator; use xai_candidate_pipeline::scorer::Scorer; use xai_candidate_pipeline::selector::Selector; use xai_candidate_pipeline::side_effect::SideEffect; use xai_candidate_pipeline::source::Source; use xai_feature_switches::FeatureSwitches; use xai_geo_ip::GeoIpLocationClient; use xai_redis_client::{XdsRedisClient, XdsRedisConfig}; use xai_visibility_filtering::tweet_safety_label::{ MockTweetSafetyLabelClient, TweetSafetyLabelClient, }; use xai_visibility_filtering::vf_client::{MockVfClient, StratoVfClient, VfClient}; use xai_x_rpc::wily_lookup_service::ShardCoordinate; pub struct PhoenixCandidatePipeline { query_hydrators: Vec>>, sources: Vec>>, hydrators: Vec>>, filters: Vec>>, scorers: Vec>>, selector: TopKScoreSelector, post_selection_hydrators: Vec>>, post_selection_filters: Vec>>, side_effects: Arc>>>, } impl PhoenixCandidatePipeline { pub(crate) async fn build_with_clients( user_action_aggregation_client: Arc, phoenix_client: Arc, phoenix_retrieval_client: Arc, thunder_client: Arc, thunder_capi_client: Option>, strato_client: Arc, tweet_mixer_client: Arc, simclusters_ann_client: Arc, simclusters_ann_cache_client: Arc, tes_client: Arc, media_info_cache_client: Arc, gizmoduck_client: Arc, strato_vf_client: Arc, xai_vf_client: Arc, redis_client: Arc, phoenix_kafka_client: Arc, logged_scored_candidates_kafka_client: Arc, reranking_kafka_client: Arc, retrieval_candidates_kafka_client: Arc, socialgraph_client: Arc, vm_ranker_client: Arc, vf_safety_labels_client: Arc, phoenix_request_cache_redis_atla_client: Arc, phoenix_request_cache_redis_pdxa_client: Arc, impression_bloom_filter_client: Arc, ip_client: Arc, user_demographics_client: Arc, user_inferred_gender_store_client: Arc, user_inferred_gender_grpc_client: Arc, impressed_posts_client: Arc, engagement_counts_client: Arc, post_engagers_client: Arc, followed_grok_topics_client: Arc, followed_starter_packs_client: Arc, user_installed_apps_client: Arc, recent_search_queries_client: Arc, engagement_signals_client: Arc, feature_switches: Arc, vm_ranker_xds: &super::VmRankerXdsConfig, sid_client: Arc, sid_retrieval_client: Arc, popular_authors: Arc, popular_posts: Arc, author_brand_safety_client: Arc, post_v8_search_embedding_client: Arc, ) -> PhoenixCandidatePipeline { let query_hydrators: Vec>> = vec![ Box::new(ScoringSequenceQueryHydrator::new( user_action_aggregation_client.clone(), )), Box::new(ImmersiveScoringSequenceQueryHydrator::new( user_action_aggregation_client.clone(), )), Box::new(RetrievalSequenceQueryHydrator::new( user_action_aggregation_client.clone(), )), Box::new(BlockedUserIdsQueryHydrator { socialgraph_client: socialgraph_client.clone(), }), Box::new(MutedUserIdsQueryHydrator { socialgraph_client: socialgraph_client.clone(), }), Box::new(FollowedUserIdsQueryHydrator { socialgraph_client: socialgraph_client.clone(), }), Box::new(SubscribedUserIdsQueryHydrator { socialgraph_client: socialgraph_client.clone(), }), Box::new(CachedPostsQueryHydrator { redis_client: redis_client.clone(), }), Box::new(MutualFollowQueryHydrator { strato_client: strato_client.clone(), }), Box::new(UserDemographicsQueryHydrator { client: user_demographics_client, }), Box::new(FollowedGrokTopicsQueryHydrator::new( followed_grok_topics_client, )), Box::new(FollowedStarterPacksQueryHydrator::new( followed_starter_packs_client, )), Box::new(UserInstalledAppsQueryHydrator::new( user_installed_apps_client, )), Box::new(RecentSearchQueriesQueryHydrator::new( recent_search_queries_client, )), Box::new(ExplicitEngagementSignalsQueryHydrator::new( engagement_signals_client.clone(), )), Box::new(ImplicitEngagementSignalsQueryHydrator::new( engagement_signals_client, )), Box::new(ImpressionBloomFilterQueryHydrator { client: impression_bloom_filter_client, }), Box::new(IpQueryHydrator { client: ip_client }), Box::new(UserInferredGenderQueryHydrator::new( user_inferred_gender_store_client, user_inferred_gender_grpc_client, )), ]; let _impressed_posts_hydrator = ImpressedPostsQueryHydrator { client: impressed_posts_client, }; let phoenix_source = Box::new(PhoenixSource { phoenix_retrieval_client: phoenix_retrieval_client.clone(), }); let phoenix_topics_source = Box::new(PhoenixTopicsSource { phoenix_retrieval_client: phoenix_retrieval_client.clone(), }); let phoenix_moe_source = Box::new(PhoenixMOESource { phoenix_retrieval_client, }); let popular_posts_source = Box::new(PopularPostsSource { popular_posts }); let thunder_source = Box::new(ThunderSource { thunder_client, thunder_capi_client, }); let tweet_mixer_source = Box::new(TweetMixerSource { tweet_mixer_client }); let core_data_hydrator = CoreDataCandidateHydrator::new(tes_client.clone()).await; let simclusters_source = Box::new(SimclustersSource::new( simclusters_ann_client, simclusters_ann_cache_client, core_data_hydrator.clone(), )); let sid_source = Box::new(SidSource::new( sid_retrieval_client, core_data_hydrator.clone(), )); let cached_posts_source = Box::new(CachedPostsSource); let sources: Vec>> = vec![ thunder_source, popular_posts_source, tweet_mixer_source, simclusters_source, phoenix_source, phoenix_topics_source, phoenix_moe_source, sid_source, cached_posts_source, ]; let hydrators: Vec>> = vec![ Box::new(InNetworkCandidateHydrator), Box::new(BidirectionalFollowHydrator { socialgraph_client: socialgraph_client.clone(), }), Box::new(core_data_hydrator), Box::new(PhoenixReplyAncestorsHydrator { tes_client: tes_client.clone(), }), Box::new( QuoteHydrator::new( tes_client.clone(), socialgraph_client.clone(), media_info_cache_client.clone(), ) .await, ), Box::new(MediaInfoHydrator::new(media_info_cache_client.clone()).await), Box::new(SubscriptionHydrator::new(tes_client.clone()).await), Box::new(GizmoduckCandidateHydrator::new(gizmoduck_client).await), Box::new(BlockedByHydrator::new(socialgraph_client).await), Box::new(FilteredTopicsHydrator { strato_client: strato_client.clone(), }), Box::new(LanguageCodeHydrator::new(tes_client.clone()).await), Box::new(EngagementCountsHydrator::new(engagement_counts_client).await), Box::new(PostEngagersHydrator::new(post_engagers_client)), Box::new(SemanticIdHydrator::new(sid_client)), Box::new(NewUserSoftNsfwLabelHydrator { client: vf_safety_labels_client.clone(), }), ]; let author_rules = Arc::new(xai_feature_switches::AuthorRulesEvaluator::new( feature_switches, )); let filters: Vec>> = vec![ Box::new(DropDuplicatesFilter), Box::new(ZeroFavCodivertFilter { author_rules: author_rules.clone(), }), Box::new(CoreDataHydrationFilter), Box::new(NewUserSoftNsfwFilter), Box::new(AgeFilter::new(Duration::from_secs(params::MAX_POST_AGE))), Box::new(SelfTweetFilter), Box::new(OONRetweetReplyFilter), Box::new(OONNsfwSimclustersFilter), Box::new(RetweetDeduplicationFilter), Box::new(IneligibleSubscriptionFilter), Box::new(PreviouslySeenPostsFilter), Box::new(PreviouslySeenPostsBackupFilter), Box::new(PreviouslyServedPostsFilter), Box::new(ViewerMutedKeywordFilter::new()), Box::new(AuthorSocialgraphFilter), Box::new(VideoFilter), Box::new(TopicIdsFilter), Box::new(NewUserMinEngagementFilter), Box::new(InventoryHoldoutFilter), Box::new(FavHoldoutFilter), Box::new(FairShareRankingFilter), ]; let phoenix_scorer = Box::new(PhoenixScorer { dispatch: crate::util::egress::PredictionDispatch { client: phoenix_client.clone(), max_retries_key: "rust_home_mixer_phoenix_xds_max_retries", }, }); let author_cold_start = crate::scorers::author_cold_start::AuthorColdStart { author_rules }; let xds_vm_ranker_client = super::build_vm_ranker_xds_client(vm_ranker_xds).await; let vm_ranker = Box::new(VMRanker { client: vm_ranker_client, xds_client: xds_vm_ranker_client, author_cold_start, }); let immersive_video_scorer = Box::new(ImmersiveVideoScorer { dispatch: crate::util::egress::PredictionDispatch { client: phoenix_client.clone(), max_retries_key: "rust_home_mixer_phoenix_xds_max_retries", }, }); let scorers: Vec>> = vec![ phoenix_scorer, immersive_video_scorer, vm_ranker, Box::new(PageDecodeRerankScorer), Box::new(PageRlExplorationScorer), ]; let selector = TopKScoreSelector; let post_selection_hydrators: Vec>> = vec![ Box::new( VFCandidateHydrator::new(strato_vf_client.clone(), xai_vf_client.clone()).await, ), Box::new(AdsBrandSafetyVfHydrator { client: vf_safety_labels_client, author_client: Some(author_brand_safety_client), }), Box::new(TweetTypeMetricsHydrator::new()), Box::new(FollowingRepliedUsersHydrator), Box::new(MutualFollowJaccardHydrator { strato_client: strato_client.clone(), }), Box::new(TopicFeedbackContextHydrator { strato_client: strato_client.clone(), }), Box::new(AiTrendFeedbackContextHydrator { strato_client: strato_client.clone(), }), Box::new(VideoMediaHydrator::new(tes_client.clone())), Box::new(VideoEmbeddingHydrator::new(post_v8_search_embedding_client)), Box::new(AncestorVideoHydrator { media_info_cache_client, }), ]; let post_selection_filters: Vec>> = vec![ Box::new(VFFilter), Box::new(AncillaryVFFilter), Box::new(AncestorVideoFilter), Box::new(DedupConversationFilter), Box::new(VideoCarouselFilter), ]; let side_effects: Arc>>> = Arc::new(vec![ Box::new(PhoenixExperimentsSideEffect::new( phoenix_client, phoenix_kafka_client, logged_scored_candidates_kafka_client, redis_client.clone(), user_action_aggregation_client, )), Box::new(RerankingKafkaSideEffect::new(reranking_kafka_client)), Box::new(RetrievalCandidatesKafkaSideEffect::new( retrieval_candidates_kafka_client, )), Box::new(RedisPostCandidateCacheSideEffect::new(redis_client)), Box::new(ScoredStatsSideEffect), Box::new(ResponseDiversityStatsSideEffect), Box::new(SerendipityStatsSideEffect), Box::new(AuthorServedMetricsSideEffect { popular_authors }), Box::new(MutualFollowStatsSideEffect), Box::new(DebugSideEffect), Box::new(PhoenixRequestCacheSideEffect::new( phoenix_request_cache_redis_atla_client, phoenix_request_cache_redis_pdxa_client, )), ]); PhoenixCandidatePipeline { query_hydrators, hydrators, filters, sources, scorers, selector, post_selection_hydrators, post_selection_filters, side_effects, } } pub async fn prod( shard_coordinate: Option, datacenter: &str, feature_switches: Arc, phoenix_xds: &super::PhoenixXdsConfig, vm_ranker_xds: &super::VmRankerXdsConfig, xai_vf_client: Arc, vf_safety_labels_client: Arc, ) -> PhoenixCandidatePipeline { let local_cache_eds = format!(""); let ( flock_socialgraph_client, user_action_aggregation_client, phoenix_client, phoenix_retrieval_client, thunder_client, strato_client, tweet_mixer_client, simclusters_ann_client, simclusters_ann_cache_client, tes_client, media_info_cache_client, gizmoduck_client, strato_vf_client, redis_client, (phoenix_request_cache_redis_atla_client, phoenix_request_cache_redis_pdxa_client), phoenix_kafka_client, logged_scored_candidates_kafka_client, reranking_kafka_client, retrieval_candidates_kafka_client, vm_ranker_client, impression_bloom_filter_client, ip_client, user_demographics_client, user_inferred_gender_store_client, user_inferred_gender_grpc_client, impressed_posts_client, followed_grok_topics_client, followed_starter_packs_client, user_installed_apps_client, recent_search_queries_client, engagement_signals_client, engagement_counts_client_impl, post_engagers_client, sid_client, thunder_capi_client, author_brand_safety_client, ) = tokio::join!( async { Arc::new( SocialGraphClient::new( datacenter, &S2S_CHAIN_PATH, &S2S_CRT_PATH, &S2S_KEY_PATH, ) .await .expect("Failed to create flock SocialGraphClient"), ) as Arc }, async { Arc::new( ProdUserActionAggregationClient::new() .await .expect("Failed to create User Action Aggregation client"), ) as Arc }, async { super::build_phoenix_xds_client(phoenix_xds) .await .expect("Failed to build Phoenix xDS ranking client") }, async { super::build_phoenix_xds_retrieval_client(phoenix_xds) .await .expect("Failed to build Phoenix xDS retrieval client") }, async { Arc::new(ThunderClient::new().await) }, async { Arc::new( ProdStratoClient::new(shard_coordinate, datacenter) .await .expect("Failed to create Strato client"), ) as Arc }, async { Arc::new( ProdTweetMixerClient::new(datacenter) .await .expect("Failed to create TweetMixer client"), ) as Arc }, async { Arc::new( ProdSimClustersAnnClient::new(datacenter) .await .expect("Failed to create SimClusters ANN client"), ) as Arc }, async { Arc::new( ProdSimClustersAnnCacheClient::new(datacenter) .await .expect("Failed to create SimClusters ANN cache client"), ) as Arc }, async { Arc::new( ProdTESClient::new(shard_coordinate, datacenter) .await .expect("Failed to create TES client"), ) as Arc }, async { Arc::new( ProdMediaInfoCacheClient::new(datacenter, "home-mixer") .await .expect("Failed to create MediaInfoCacheClient"), ) as Arc }, async { Arc::new( ProdGizmoduckClient::new( shard_coordinate, datacenter, Some("home-mixer.prod".to_string()), ) .await .expect("Failed to create Gizmoduck client"), ) as Arc }, async { Arc::new( StratoVfClient::new( S2S_CHAIN_PATH.clone(), S2S_CRT_PATH.clone(), S2S_KEY_PATH.clone(), "home-mixer.prod".to_string(), datacenter.to_string(), ) .await .expect("Failed to create VF client"), ) as Arc }, async { Arc::new( XdsRedisClient::new(XdsRedisConfig { eds_resource_name: local_cache_eds.clone(), }) .await .expect("Failed to create xDS Redis client for local cache"), ) as Arc }, super::shared_phoenix_request_cache_redis_clients(), async { Arc::new( ProdKafkaPublisherClient::new(PHOENIX_SCORES_TOPIC, KafkaCluster::Aiml).await, ) as Arc }, async { Arc::new( ProdKafkaPublisherClient::new( LOGGED_SCORED_CANDIDATES_TOPIC, KafkaCluster::Phoenix, ) .await, ) as Arc }, async { Arc::new( ProdKafkaPublisherClient::new(RERANKING_TOPIC, KafkaCluster::Phoenix).await, ) as Arc }, async { Arc::new( ProdKafkaPublisherClient::new( RETRIEVAL_CANDIDATES_TOPIC, KafkaCluster::Phoenix, ) .await, ) as Arc }, async { Arc::new( ProdVMRankerClient::new() .await .expect("Failed to create VMRanker client"), ) as Arc }, async { Arc::new( ProdImpressionBloomFilterClient::new(datacenter) .await .expect("Failed to create ImpressionBloomFilter client"), ) as Arc }, async { Arc::new( GeoIpLocationClient::new( &S2S_CHAIN_PATH, &S2S_CRT_PATH, &S2S_KEY_PATH, datacenter, ) .await .expect("Failed to create GeoIpLocationClient"), ) }, async { let s2s = xai_manhattan::s2s::S2sConfig { client_cert_path: S2S_CRT_PATH.clone(), client_key_path: S2S_KEY_PATH.clone(), ca_cert_path: S2S_CHAIN_PATH.clone(), }; Arc::new( ProdUserDemographicsClient::new(datacenter, s2s) .await .expect("Failed to create UserDemographics client"), ) as Arc }, async { let s2s = xai_manhattan::s2s::S2sConfig { client_cert_path: S2S_CRT_PATH.clone(), client_key_path: S2S_KEY_PATH.clone(), ca_cert_path: S2S_CHAIN_PATH.clone(), }; Arc::new( ProdUserInferredGenderStoreClient::new(datacenter, s2s) .await .expect("Failed to create UserInferredGenderStore client"), ) as Arc }, async { Arc::new( ProdGenderPredictionGrpcClient::new() .await .expect("Failed to create GenderPredictionGrpcClient"), ) as Arc }, async { Arc::new( crate::clients::impressed_posts_client::ProdImpressedPostsClient::new( datacenter, ) .await .expect("Failed to create ImpressedPosts client"), ) as Arc }, async { let s2s = xai_manhattan::s2s::S2sConfig { client_cert_path: S2S_CRT_PATH.clone(), client_key_path: S2S_KEY_PATH.clone(), ca_cert_path: S2S_CHAIN_PATH.clone(), }; Arc::new( ProdFollowedGrokTopicsStoreClient::new(datacenter, s2s) .await .expect("Failed to create FollowedGrokTopicsStore client"), ) as Arc }, async { let s2s = xai_manhattan::s2s::S2sConfig { client_cert_path: S2S_CRT_PATH.clone(), client_key_path: S2S_KEY_PATH.clone(), ca_cert_path: S2S_CHAIN_PATH.clone(), }; Arc::new( ProdFollowedStarterPacksStoreClient::new(datacenter, s2s) .await .expect("Failed to create FollowedStarterPacksStore client"), ) as Arc }, async { let s2s = xai_manhattan::s2s::S2sConfig { client_cert_path: S2S_CRT_PATH.clone(), client_key_path: S2S_KEY_PATH.clone(), ca_cert_path: S2S_CHAIN_PATH.clone(), }; Arc::new( ProdUserInstalledAppsStoreClient::new(datacenter, s2s) .await .expect("Failed to create UserInstalledAppsStore client"), ) as Arc }, async { match ProdRecentSearchQueriesStoreClient::new(datacenter).await { Ok(client) => Arc::new(client) as Arc, Err(e) => { tracing::warn!(error = %e, "RecentSearchQueriesStore client unavailable; hydration disabled"); Arc::new(MockRecentSearchQueriesStoreClient) as Arc } } }, async { let s2s = xai_manhattan::s2s::S2sConfig { client_cert_path: S2S_CRT_PATH.clone(), client_key_path: S2S_KEY_PATH.clone(), ca_cert_path: S2S_CHAIN_PATH.clone(), }; Arc::new( ProdEngagementSignalsClient::new(datacenter, s2s) .await .expect("Failed to create EngagementSignals client"), ) as Arc }, async { Arc::new( ProdEngagementCountsClient::new(datacenter) .await .expect("Failed to create EngagementCounts client"), ) }, async { Arc::new( ProdPostEngagersClient::new(datacenter) .await .expect("Failed to create PostEngagers client"), ) as Arc }, async { Arc::new(ProdSidClient::new()) as Arc }, async { match ProdThunderCapiClient::new(datacenter).await { Ok(c) => Some(Arc::new(c) as Arc), Err(e) => { tracing::warn!(error = %e, "ThunderCapiClient build failed; using proxy path"); None } } }, super::shared_author_brand_safety_client(datacenter), ); let engagement_counts_client: Arc = engagement_counts_client_impl; let popular_authors_store: Arc = match ManhattanPopularAuthorsStore::new(datacenter).await { Ok(store) => Arc::new(store), Err(e) => { tracing::warn!(error = %e, "popular authors store unavailable; using in-memory"); Arc::new(InMemoryPopularAuthorsStore::default()) } }; let popular_authors = Arc::new(PopularAuthorsCache::new(popular_authors_store)); let popular_posts_store: Arc = match ManhattanPopularPostsStore::new(datacenter).await { Ok(store) => Arc::new(store), Err(e) => { tracing::warn!(error = %e, "popular posts store unavailable; using in-memory"); Arc::new(InMemoryPopularPostsStore::default()) } }; let popular_posts = Arc::new(PopularPostsCache::new(popular_posts_store)); let post_v8_search_embedding_client: Arc = match ManhattanPostV8SearchEmbeddingClient::new(datacenter).await { Ok(client) => Arc::new(client), Err(err) => { tracing::error!( error = %err, "post v8_search Manhattan client unavailable; carousel embedding dedupe is a no-op" ); Arc::new(NoopPostV8SearchEmbeddingClient) } }; PhoenixCandidatePipeline::build_with_clients( user_action_aggregation_client, phoenix_client, phoenix_retrieval_client, thunder_client, thunder_capi_client, strato_client, tweet_mixer_client, simclusters_ann_client, simclusters_ann_cache_client, tes_client, media_info_cache_client, gizmoduck_client, strato_vf_client, xai_vf_client, redis_client, phoenix_kafka_client, logged_scored_candidates_kafka_client, reranking_kafka_client, retrieval_candidates_kafka_client, flock_socialgraph_client, vm_ranker_client, vf_safety_labels_client, phoenix_request_cache_redis_atla_client, phoenix_request_cache_redis_pdxa_client, impression_bloom_filter_client, ip_client, user_demographics_client, user_inferred_gender_store_client, user_inferred_gender_grpc_client, impressed_posts_client, engagement_counts_client, post_engagers_client, followed_grok_topics_client, followed_starter_packs_client, user_installed_apps_client, recent_search_queries_client, engagement_signals_client, feature_switches, vm_ranker_xds, sid_client, ProdSidRetrievalClient::new(vm_ranker_xds) as Arc, popular_authors, popular_posts, author_brand_safety_client, post_v8_search_embedding_client, ) .await } pub async fn mock() -> PhoenixCandidatePipeline { let user_action_aggregation_client = Arc::new(MockUserActionAggregationClient); let phoenix_client = Arc::new(MockPredictClient); let phoenix_retrieval_client: Arc = Arc::new(MockRetrievalClient); let thunder_client = Arc::new(ThunderClient::mock()); let strato_client = Arc::new(MockStratoClient::default()); let tweet_mixer_client: Arc = Arc::new(MockTweetMixerClient); let simclusters_ann_client: Arc = Arc::new(MockSimClustersAnnClient); let simclusters_ann_cache_client: Arc = Arc::new(MockSimClustersAnnCacheClient); let tes_client = Arc::new(MockTESClient::default()); let media_info_cache_client: Arc = Arc::new(MockMediaInfoCacheClient::default()); let gizmoduck_client = Arc::new(MockGizmoduckClient::default()); let strato_vf_client = Arc::new(MockVfClient); let xai_vf_client = Arc::new(MockVfClient); let redis_client = Arc::new(MockRedisClient::default()); let kafka_client: Arc = Arc::new(MockKafkaPublisherClient); let logged_scored_candidates_kafka_client: Arc = Arc::new(MockKafkaPublisherClient); let reranking_kafka_client: Arc = Arc::new(MockKafkaPublisherClient); let retrieval_candidates_kafka_client: Arc = Arc::new(MockKafkaPublisherClient); let mock_socialgraph: Arc = Arc::new(MockSocialGraphClient); let vm_ranker_client: Arc = Arc::new(MockVMRankerClient); let vf_safety_labels_client: Arc = Arc::new(MockTweetSafetyLabelClient); let phoenix_request_cache_redis_atla_client = Arc::new(MockRedisClient::default()); let phoenix_request_cache_redis_pdxa_client: Arc = Arc::new(MockRedisClient::default()); let impression_bloom_filter_client: Arc = Arc::new(MockImpressionBloomFilterClient::default()); let ip_client = Arc::new(GeoIpLocationClient::mock()); let user_demographics_client: Arc = Arc::new(MockUserDemographicsClient); let user_inferred_gender_store_client: Arc = Arc::new(MockUserInferredGenderStoreClient); let user_inferred_gender_grpc_client: Arc = Arc::new(MockGenderPredictionGrpcClient); let impressed_posts_client: Arc = Arc::new(crate::clients::impressed_posts_client::MockImpressedPostsClient::default()); let engagement_counts_client: Arc = Arc::new( crate::clients::engagement_counts_client::MockEngagementCountsClient::default(), ); let post_engagers_client: Arc = Arc::new(MockPostEngagersClient::default()); let followed_grok_topics_client: Arc = Arc::new(MockFollowedGrokTopicsStoreClient); let followed_starter_packs_client: Arc = Arc::new(MockFollowedStarterPacksStoreClient); let user_installed_apps_client: Arc = Arc::new(MockUserInstalledAppsStoreClient); let recent_search_queries_client: Arc = Arc::new(MockRecentSearchQueriesStoreClient); let engagement_signals_client: Arc = Arc::new(MockEngagementSignalsClient); let feature_switches = Arc::new(FeatureSwitches::new(vec![]).unwrap()); PhoenixCandidatePipeline::build_with_clients( user_action_aggregation_client, phoenix_client, phoenix_retrieval_client, thunder_client, None, strato_client, tweet_mixer_client, simclusters_ann_client, simclusters_ann_cache_client, tes_client, media_info_cache_client, gizmoduck_client, strato_vf_client, xai_vf_client, redis_client, kafka_client, logged_scored_candidates_kafka_client, reranking_kafka_client, retrieval_candidates_kafka_client, mock_socialgraph, vm_ranker_client, vf_safety_labels_client, phoenix_request_cache_redis_atla_client, phoenix_request_cache_redis_pdxa_client, impression_bloom_filter_client, ip_client, user_demographics_client, user_inferred_gender_store_client, user_inferred_gender_grpc_client, impressed_posts_client, engagement_counts_client, post_engagers_client, followed_grok_topics_client, followed_starter_packs_client, user_installed_apps_client, recent_search_queries_client, engagement_signals_client, feature_switches, &super::VmRankerXdsConfig::disabled_with_healthz(), Arc::new(MockSidClient) as Arc, Arc::new(MockSidRetrievalClient) as Arc, Arc::new(PopularAuthorsCache::new(Arc::new( InMemoryPopularAuthorsStore::default(), ))), Arc::new(PopularPostsCache::new(Arc::new( InMemoryPopularPostsStore::default(), ))), Arc::new(MockAuthorBrandSafetyClient::default()) as Arc, Arc::new(NoopPostV8SearchEmbeddingClient) as Arc, ) .await } } #[async_trait] impl CandidatePipeline for PhoenixCandidatePipeline { fn query_hydrators(&self) -> &[Box>] { &self.query_hydrators } fn sources(&self) -> &[Box>] { &self.sources } fn hydrators(&self) -> &[Box>] { &self.hydrators } fn filters(&self) -> &[Box>] { &self.filters } fn scorers(&self) -> &[Box>] { &self.scorers } fn selector(&self) -> &dyn Selector { &self.selector } fn post_selection_hydrators(&self) -> &[Box>] { &self.post_selection_hydrators } fn post_selection_filters(&self) -> &[Box>] { &self.post_selection_filters } fn side_effects(&self) -> Arc>>> { Arc::clone(&self.side_effects) } fn result_size(&self) -> usize { params::RESULT_SIZE } } #[cfg(test)] mod tests { use crate::candidate_pipeline::phoenix_candidate_pipeline::PhoenixCandidatePipeline; use crate::models::query::ScoredPostsQuery; use xai_candidate_pipeline::candidate_pipeline::CandidatePipeline; #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn test_candidate_pipeline() -> Result<(), Box> { xai_init_utils::init().rustls(); let pipeline = PhoenixCandidatePipeline::mock().await; let fs = xai_feature_switches::FeatureSwitches::new(vec![]).unwrap(); let mut results = fs.match_recipient(&xai_feature_switches::RecipientBuilder::new().build()); results.override_fs( "rust_home_mixer_enable_scoring_sequence_hydration".to_string(), "true", ); let pipeline_result = pipeline .execute(ScoredPostsQuery { user_id: 12, params: results.into(), ..Default::default() }) .await; let hydrated_query = pipeline_result.query; assert_eq!(hydrated_query.user_id, 12); assert!(hydrated_query.scoring_sequence.is_some()); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn video_carousel_filter_is_the_last_post_selection_filter() { xai_init_utils::init().rustls(); let pipeline = PhoenixCandidatePipeline::mock().await; let last = pipeline .post_selection_filters() .last() .map(|filter| filter.name()); assert_eq!(last, Some("VideoCarouselFilter")); } #[test] fn test_bulk_topic_detection() { let query = ScoredPostsQuery { topic_ids: vec![1, 2, 3, 4, 5, 6], ..Default::default() }; assert!(!query.is_bulk_topic_request()); let query = ScoredPostsQuery { topic_ids: vec![1, 2, 3, 4, 5, 6, 7], ..Default::default() }; assert!(query.is_bulk_topic_request()); let query = ScoredPostsQuery { topic_ids: vec![], ..Default::default() }; assert!(!query.is_bulk_topic_request()); } #[test] #[cfg(target_os = "macos")] fn test_feature_switches() { use std::env; use xai_feature_switches::{FeatureSwitches, RecipientBuilder}; let fs_local = format!( "/Users/{}/workspace/config/features/home-mixer/main/for_you.yml", env::var("USER").unwrap() ); let fs = FeatureSwitches::load_file(fs_local).expect("failed to load features.yml"); let recipient = RecipientBuilder::new().build(); let results = fs.match_recipient(&recipient); let results = results.get_i64("for_you_server_max_results"); assert_eq!(results, Some(35)); let fs_local = format!( "/Users/{}/workspace/config/features/home-mixer/main/home_mixer.yml", env::var("USER").unwrap() ); let fs = FeatureSwitches::load_file(fs_local).expect("failed to load features.yml"); let recipient = RecipientBuilder::new().build(); let results = fs.match_recipient(&recipient); let results = results.get_bool("home_mixer_enable_new_tweets_pill_avatars"); assert_eq!(results, Some(true)); let fs_local = format!( "/Users/{}/workspace/config/features/home-mixer/main/scored_tweets.yml", env::var("USER").unwrap() ); let fs = FeatureSwitches::load_file(fs_local).expect("failed to load features.yml"); let recipient = RecipientBuilder::new().build(); let results = fs.match_recipient(&recipient); let results = results.get_i64("scored_tweets_default_requested_max_results"); assert_eq!(results, Some(50)); let fs_local = format!( "/Users/{}/workspace/config/features/home-mixer/main/scored_video_tweets.yml", env::var("USER").unwrap() ); let fs = FeatureSwitches::load_file(fs_local).expect("failed to load features.yml"); let recipient = RecipientBuilder::new().build(); let results = fs.match_recipient(&recipient); let results = results.get_string("scored_video_tweets_in_network_earlybird_tensorflow_model"); assert_eq!(results, Some("timelines_unified_prod")); } }