chore: cleanup

parakeet/src/entity_cache.rs

···
 
     /// Get multiple profiles with caching
     ///
-    /// This is the recommended way to get multiple profiles, ensuring each one
-    /// uses the cache individually
+    /// This is the recommended way to get multiple profiles, using batch hydration
+    /// for all cache misses to avoid N+1 queries
     pub async fn get_or_hydrate_batch(
         &self,
         dids: Vec<String>,
···
         id_cache: &Arc<IdCache>,
         hydrator: &StatefulHydrator<'_>,
     ) -> Vec<ProfileViewDetailed> {
-        let mut profiles = Vec::with_capacity(dids.len());
+        let mut cached_profiles = Vec::new();
+        let mut uncached_requests = Vec::new();
+        let mut did_to_actor_id = std::collections::HashMap::new();
 
+        // First pass: check cache for all DIDs
         for did in dids {
             // Get actor_id for each DID
             if let Ok(actor_id) = id_cache_helpers::get_actor_id_or_fetch(pool, id_cache, &did).await {
-                if let Some(profile) = self.get_or_hydrate(actor_id, did, hydrator).await {
-                    profiles.push(profile);
+                did_to_actor_id.insert(did.clone(), actor_id);
+
+                // Check cache
+                if let Some(profile) = self.cache.get(&actor_id).await {
+                    tracing::debug!(actor_id, "Profile cache hit in batch");
+                    cached_profiles.push(profile);
+                } else {
+                    // Track cache misses for batch hydration
+                    uncached_requests.push(did);
                 }
             }
         }
 
-        profiles
+        // Batch hydrate all uncached profiles at once
+        if !uncached_requests.is_empty() {
+            let cache_hit_count = cached_profiles.len();
+            let miss_count = uncached_requests.len();
+            let hit_rate = if cache_hit_count + miss_count > 0 {
+                (cache_hit_count as f64 / (cache_hit_count + miss_count) as f64) * 100.0
+            } else {
+                0.0
+            };
+
+            tracing::debug!(
+                "Profile cache batch: {} hits, {} misses ({:.1}% hit rate), hydrating misses",
+                cache_hit_count, miss_count, hit_rate
+            );
+
+            // Build a map of actor_id to DID for the uncached requests
+            let uncached_actor_ids: Vec<(i32, String)> = uncached_requests
+                .iter()
+                .filter_map(|did| {
+                    did_to_actor_id.get(did).map(|&actor_id| (actor_id, did.clone()))
+                })
+                .collect();
+
+            // Use the batch hydration method for all misses at once (passing actor_ids)
+            let hydrated = hydrator.hydrate_profiles_detailed_by_id(uncached_actor_ids.clone()).await;
+
+            // Store in cache and collect results
+            for (actor_id, _did) in uncached_actor_ids {
+                if let Some(profile) = hydrated.get(&actor_id) {
+                    // Cache the hydrated profile
+                    self.cache.insert(actor_id, profile.clone()).await;
+                    cached_profiles.push(profile.clone());
+                }
+            }
+        }
+
+        cached_profiles
     }
 
     pub async fn invalidate(&self, actor_id: i32) {

parakeet/src/hydration/mod.rs

···
         labels
     }
 
-    /// Get labels for multiple URIs using actor_ids (optimized version - avoids decompressing actors table)
+    /// Get labels for multiple URIs
     ///
-    /// This resolves labeler DIDs → actor_ids via IdCache first, avoiding the actors table join.
+    /// For now, fallback to the regular load_many function until we properly extract labels from loaded data
     async fn get_label_many_by_actor_ids(
         &self,
         uris: &[String],
     ) -> HashMap<String, Vec<parakeet_db::models::Label>> {
-        // Resolve labeler DIDs → actor_ids via IdCache
-        let labeler_dids: Vec<String> = self
-            .accept_labelers
-            .iter()
-            .map(|v| v.labeler.clone())
-            .collect();
-
-        let mut labeler_actor_ids = Vec::new();
-        for did in &labeler_dids {
-            if let Some(cached) = self.loaders.label.id_cache().get_actor_id(did).await {
-                labeler_actor_ids.push(cached.actor_id);
-            }
-        }
-
-        if labeler_actor_ids.is_empty() {
-            return HashMap::new();
-        }
-
+        // TODO: Extract labels from already-loaded post/actor data instead of querying separately
+        // Posts and actors already have labels field loaded
         self.loaders
             .label
-            .load_many_by_actor_ids(uris, &labeler_actor_ids)
+            .load_many(uris, &self.accept_labelers)
             .await
     }
 
-    /// Get profile labels using actor_ids (optimized version)
+    /// Get profile labels
     async fn get_profile_label_many_by_actor_ids(
         &self,
         uris: &[String],
···
                 .map(|did| format!("at://{did}/app.bsky.actor.profile/self")),
         );
 
+        // TODO: Extract labels from already-loaded actor data instead of querying separately
         let mut labels = self.get_label_many_by_actor_ids(&uris_full).await;
 
         for did in uris {

parakeet/src/hydration/posts/mod.rs

···
                 })
                 .collect::<HashMap<String, ProfileViewBasic>>()
         };
+        // Extract labels directly from the loaded posts instead of querying separately
         let labels_future = async {
             let start = std::time::Instant::now();
-            let result = self.get_label_many_by_actor_ids(&post_uris).await;
-            let elapsed = start.elapsed().as_secs_f64() * 1000.0;
-            tracing::info!("  → Labels: {:.1} ms ({} posts)", elapsed, post_uri_count);
-            if elapsed > 10.0 {
-                tracing::warn!("  → Slow labels: {:.1} ms", elapsed);
+
+            // Build a map of URI -> labels from the posts we already loaded
+            let mut labels_map: HashMap<String, Vec<parakeet_db::models::Label>> = HashMap::new();
+
+            // Get allowed labeler actor IDs
+            let labeler_dids: Vec<String> = self
+                .accept_labelers
+                .iter()
+                .map(|v| v.labeler.clone())
+                .collect();
+
+            let mut labeler_actor_ids = Vec::new();
+            for did in &labeler_dids {
+                if let Some(cached) = self.loaders.label.id_cache().get_actor_id(did).await {
+                    labeler_actor_ids.push(cached.actor_id);
+                }
             }
-            result
+
+            // Extract labels from each post
+            for (uri, (post, _, _)) in &posts_with_stats {
+                if let Some(post_labels) = &post.post.labels {
+                    let mut labels_for_uri = Vec::new();
+
+                    for label_opt in post_labels {
+                        if let Some(label) = label_opt {
+                            // Only include labels from allowed labelers
+                            if labeler_actor_ids.contains(&label.labeler_actor_id) {
+                                // Resolve labeler DID
+                                if let Some(actor_data) = self.loaders.label.id_cache().get_actor_data(label.labeler_actor_id).await {
+                                    labels_for_uri.push(parakeet_db::models::Label {
+                                        labeler_actor_id: label.labeler_actor_id,
+                                        label: label.label.clone(),
+                                        uri: uri.clone(),
+                                        self_label: false, // Post labels are not self-labels
+                                        cid: None, // Not stored in denormalized structure
+                                        negated: label.negated,
+                                        expires: label.expires,
+                                        sig: None, // Not stored in denormalized structure
+                                        created_at: label.created_at,
+                                        labeler: actor_data.did,
+                                    });
+                                }
+                            }
+                        }
+                    }
+
+                    if !labels_for_uri.is_empty() {
+                        labels_map.insert(uri.clone(), labels_for_uri);
+                    }
+                }
+            }
+
+            let elapsed = start.elapsed().as_secs_f64() * 1000.0;
+            tracing::info!("  → Labels extracted from posts: {:.1} ms ({} posts)", elapsed, post_uri_count);
+
+            labels_map
         };
         let viewer_future = async {
             let start = std::time::Instant::now();

parakeet/src/hydration/profile/mod.rs

···
     }
 
     pub async fn hydrate_profile_basic(&self, did: String) -> Option<ProfileViewBasic> {
+        // Convert DID to actor_id first
+        let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch(
+            self.loaders.profile_state.pool(),
+            self.loaders.profile_state.id_cache(),
+            &did,
+        ).await.ok()?;
+
         let viewer = self.get_profile_viewer_state(&did).await;
         let verif = self.loaders.verification.load(did.clone()).await;
         let stats = self.loaders.profile_stats.load(did.clone()).await;
-        let profile_info = self.loaders.profile.load(did.clone()).await?;
+        let profile_info = self.loaders.profile_by_id.load(actor_id).await?;
 
         // Extract and convert inline labels
         let labels = if let Some(ref label_records) = profile_info.8 {
···
     }
 
     pub async fn hydrate_profile(&self, did: String) -> Option<ProfileView> {
-        let labels = self.get_profile_label(&did).await;
+        // Convert DID to actor_id first
+        let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch(
+            self.loaders.profile_state.pool(),
+            self.loaders.profile_state.id_cache(),
+            &did,
+        ).await.ok()?;
+
+        let profile_info = self.loaders.profile_by_id.load(actor_id).await?;
+
+        // Extract labels from loaded profile
+        let labels = if let Some(ref label_records) = profile_info.8 {
+            self.convert_actor_labels(&did, label_records).await
+        } else {
+            vec![]
+        };
+
         let viewer = self.get_profile_viewer_state(&did).await;
         let verif = self.loaders.verification.load(did.clone()).await;
         let stats = self.loaders.profile_stats.load(did.clone()).await;
-        let profile_info = self.loaders.profile.load(did).await?;
 
         Some(build_profile(
             profile_info,
···
     }
 
     pub async fn hydrate_profiles(&self, dids: Vec<String>) -> HashMap<String, ProfileView> {
-        let labels = self.get_profile_label_many(&dids).await;
+        // Convert DIDs to actor_ids first
+        let actor_id_map = match crate::id_cache_helpers::get_actor_ids_or_fetch(
+            self.loaders.profile_state.pool(),
+            self.loaders.profile_state.id_cache(),
+            &dids,
+        ).await {
+            Ok(map) => map,
+            Err(_) => return HashMap::new(),
+        };
+
+        let actor_ids: Vec<i32> = actor_id_map.values().copied().collect();
+
+        let profiles = self.loaders.profile_by_id.load_many(actor_ids).await;
+
+        // Extract labels from loaded profiles
+        let mut labels: HashMap<String, Vec<parakeet_db::models::Label>> = HashMap::new();
+        for (did, _, _, _, _, _, _, _, label_records) in profiles.values() {
+            if let Some(records) = label_records {
+                let converted = self.convert_actor_labels(did, records).await;
+                if !converted.is_empty() {
+                    labels.insert(did.clone(), converted);
+                }
+            }
+        }
+
         let viewers = self.get_profile_viewer_states(&dids).await;
         let verif = self.loaders.verification.load_many(dids.clone()).await;
         let stats = self.loaders.profile_stats.load_many(dids.clone()).await;
-        let profiles = self.loaders.profile.load_many(dids).await;
+
+        // Convert actor_id-keyed results back to DID-keyed
+        let id_to_did: HashMap<i32, String> = actor_id_map.iter().map(|(did, id)| (*id, did.clone())).collect();
 
         profiles
             .into_iter()
-            .map(|(k, profile_info)| {
-                let labels = labels.get(&k).cloned().unwrap_or_default();
-                let verif = verif.get(&k).cloned();
-                let viewer = viewers.get(&k).cloned();
-                let stats = stats.get(&k).copied();
+            .filter_map(|(actor_id, profile_info)| {
+                let did = id_to_did.get(&actor_id)?;
+                let labels = labels.get(did).cloned().unwrap_or_default();
+                let verif = verif.get(did).cloned();
+                let viewer = viewers.get(did).cloned();
+                let stats = stats.get(did).copied();
 
                 let v = build_profile(profile_info, stats, labels, verif, viewer, &self.cdn);
-                (k, v)
+                Some((did.clone(), v))
             })
             .collect()
     }
···
         note = "Use ProfileCache::get_or_hydrate() to ensure caching. Direct hydration bypasses the cache."
     )]
     pub async fn hydrate_profile_detailed(&self, did: String) -> Option<ProfileViewDetailed> {
-        let labels = self.get_profile_label(&did).await;
+        // Convert DID to actor_id first
+        let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch(
+            self.loaders.profile_state.pool(),
+            self.loaders.profile_state.id_cache(),
+            &did,
+        ).await.ok()?;
+
+        let profile_info = self.loaders.profile_by_id.load(actor_id).await?;
+
+        // Extract labels from loaded profile
+        let labels = if let Some(ref label_records) = profile_info.8 {
+            self.convert_actor_labels(&did, label_records).await
+        } else {
+            vec![]
+        };
+
         let viewer = self.get_profile_viewer_state(&did).await;
         let verif = self.loaders.verification.load(did.clone()).await;
         let stats = self.loaders.profile_stats.load(did.clone()).await;
-        let profile_info = self.loaders.profile.load(did).await?;
 
         Some(build_detailed(
             profile_info,
···
         &self,
         dids: Vec<String>,
     ) -> HashMap<String, ProfileViewDetailed> {
-        let labels = self.get_profile_label_many(&dids).await;
+        // Convert DIDs to actor_ids first
+        let actor_id_map = match crate::id_cache_helpers::get_actor_ids_or_fetch(
+            self.loaders.profile_state.pool(),
+            self.loaders.profile_state.id_cache(),
+            &dids,
+        ).await {
+            Ok(map) => map,
+            Err(_) => return HashMap::new(),
+        };
+
+        let actor_ids: Vec<i32> = actor_id_map.values().copied().collect();
+
+        let profiles = self.loaders.profile_by_id.load_many(actor_ids).await;
+
+        // Extract labels from loaded profiles
+        let mut labels: HashMap<String, Vec<parakeet_db::models::Label>> = HashMap::new();
+        for (did, _, _, _, _, _, _, _, label_records) in profiles.values() {
+            if let Some(records) = label_records {
+                let converted = self.convert_actor_labels(did, records).await;
+                if !converted.is_empty() {
+                    labels.insert(did.clone(), converted);
+                }
+            }
+        }
+
         let viewers = self.get_profile_viewer_states(&dids).await;
         let verif = self.loaders.verification.load_many(dids.clone()).await;
         let stats = self.loaders.profile_stats.load_many(dids.clone()).await;
-        let profiles = self.loaders.profile.load_many(dids).await;
+
+        // Convert actor_id-keyed results back to DID-keyed
+        let id_to_did: HashMap<i32, String> = actor_id_map.iter().map(|(did, id)| (*id, did.clone())).collect();
 
         profiles
             .into_iter()
-            .map(|(k, profile_info)| {
-                let labels = labels.get(&k).cloned().unwrap_or_default();
-                let verif = verif.get(&k).cloned();
-                let viewer = viewers.get(&k).cloned();
-                let stats = stats.get(&k).copied();
+            .filter_map(|(actor_id, profile_info)| {
+                let did = id_to_did.get(&actor_id)?;
+                let labels = labels.get(did).cloned().unwrap_or_default();
+                let verif = verif.get(did).cloned();
+                let viewer = viewers.get(did).cloned();
+                let stats = stats.get(did).copied();
 
                 let v = build_detailed(profile_info, stats, labels, verif, viewer, &self.cdn);
-                (k, v)
+                Some((did.clone(), v))
+            })
+            .collect()
+    }
+
+    /// Optimized version that takes actor_ids directly, avoiding DID lookups
+    pub async fn hydrate_profiles_detailed_by_id(
+        &self,
+        actor_ids_with_dids: Vec<(i32, String)>,
+    ) -> HashMap<i32, ProfileViewDetailed> {
+        // Extract just the actor_ids for queries
+        let actor_ids: Vec<i32> = actor_ids_with_dids.iter().map(|(id, _)| *id).collect();
+        let dids: Vec<String> = actor_ids_with_dids.iter().map(|(_, did)| did.clone()).collect();
+
+        // Build a map of actor_id to DID for later lookups
+        let id_to_did: std::collections::HashMap<i32, String> = actor_ids_with_dids.into_iter().collect();
+
+        // Load data using actor_ids where possible
+        let profiles = self.loaders.profile_by_id.load_many(actor_ids.clone()).await;
+        let stats = self.loaders.profile_stats_by_id.load_many(&actor_ids).await;
+
+        // Extract labels directly from loaded profiles
+        let mut labels: HashMap<String, Vec<parakeet_db::models::Label>> = HashMap::new();
+        for (actor_id, profile_info) in &profiles {
+            if let Some(did) = id_to_did.get(actor_id) {
+                if let Some(ref label_records) = profile_info.8 {
+                    let converted_labels = self.convert_actor_labels(did, label_records).await;
+                    if !converted_labels.is_empty() {
+                        labels.insert(did.clone(), converted_labels);
+                    }
+                }
+            }
+        }
+        let viewers = self.get_profile_viewer_states(&dids).await;
+        let verif = self.loaders.verification.load_many(dids).await;
+
+        profiles
+            .into_iter()
+            .map(|(actor_id, profile_info)| {
+                let did = id_to_did.get(&actor_id).cloned().unwrap_or_default();
+                let labels = labels.get(&did).cloned().unwrap_or_default();
+                let verif = verif.get(&did).cloned();
+                let viewer = viewers.get(&did).cloned();
+                let stats = stats.get(&actor_id).copied();
+
+                let v = build_detailed(profile_info, stats, labels, verif, viewer, &self.cdn);
+                (actor_id, v)
             })
             .collect()
     }

parakeet/src/loaders/labeler.rs

···
             .into_group_map_by(|v| v.uri.clone())
     }
 
-    /// Load labels by URIs using actor_ids (optimized version - avoids decompressing actors table)
-    ///
-    /// This is an optimized version that resolves labeler DIDs → actor_ids via IdCache first,
-    /// then queries by actor_ids directly, avoiding the 2-30ms penalty from joining the
-    /// compressed actors table.
-    ///
-    /// Expected performance: 2-30ms → 0.5-2ms (5-60x faster)
-    pub async fn load_many_by_actor_ids(
-        &self,
-        uris: &[String],
-        labeler_actor_ids: &[i32],
-    ) -> HashMap<String, Vec<models::Label>> {
-        let mut conn = self.0.get().await.unwrap();
-
-        if labeler_actor_ids.is_empty() || uris.is_empty() {
-            return HashMap::new();
-        }
-
-        let uri_refs: Vec<&str> = uris.iter().map(|s| s.as_str()).collect();
-
-        #[derive(diesel::QueryableByName)]
-        struct LabelRowById {
-            #[diesel(sql_type = diesel::sql_types::Integer)]
-            labeler_actor_id: i32,
-            #[diesel(sql_type = diesel::sql_types::Text)]
-            label: String,
-            #[diesel(sql_type = diesel::sql_types::Text)]
-            uri: String,
-            #[diesel(sql_type = diesel::sql_types::Bool)]
-            self_label: bool,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-            cid: Option<Vec<u8>>,
-            #[diesel(sql_type = diesel::sql_types::Bool)]
-            negated: bool,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Timestamptz>)]
-            expires: Option<chrono::DateTime<chrono::Utc>>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-            sig: Option<Vec<u8>>,
-            #[diesel(sql_type = diesel::sql_types::Timestamptz)]
-            created_at: chrono::DateTime<chrono::Utc>,
-        }
-
-        let labels: Vec<LabelRowById> = diesel_async::RunQueryDsl::load(
-            diesel::sql_query(include_str!("../sql/labels_by_actor_ids.sql"))
-                .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(&uri_refs)
-                .bind::<diesel::sql_types::Array<diesel::sql_types::Integer>, _>(labeler_actor_ids),
-            &mut conn,
-        )
-        .await
-        .unwrap_or_else(|e| {
-            tracing::error!("label load by actor_ids failed: {e}");
-            vec![]
-        });
-
-        // Collect unique labeler actor_ids to resolve
-        let labeler_ids: std::collections::HashSet<i32> = labels
-            .iter()
-            .map(|row| row.labeler_actor_id)
-            .collect();
-
-        // Batch resolve labeler actor_ids → DIDs using IdCache
-        let labeler_ids_vec: Vec<i32> = labeler_ids.into_iter().collect();
-        let id_to_actor_data = self.1.get_actor_data_many(&labeler_ids_vec).await;
-
-        // Build result with resolved labeler DIDs
-        labels
-            .into_iter()
-            .filter_map(|row| {
-                // Resolve labeler DID from IdCache
-                let labeler_did = match id_to_actor_data.get(&row.labeler_actor_id) {
-                    Some(data) => data.did.clone(),
-                    None => {
-                        tracing::warn!("label: missing labeler DID for actor_id {}", row.labeler_actor_id);
-                        return None;
-                    }
-                };
-
-                Some(models::Label {
-                    labeler_actor_id: row.labeler_actor_id,
-                    label: row.label.clone(),
-                    uri: row.uri.clone(),
-                    self_label: row.self_label,
-                    cid: row.cid,
-                    negated: row.negated,
-                    expires: row.expires,
-                    sig: row.sig,
-                    created_at: row.created_at,
-                    labeler: labeler_did,
-                })
-            })
-            .into_group_map_by(|v| v.uri.clone())
-    }
 
     /// Get the IdCache for DID → actor_id resolution
     pub fn id_cache(&self) -> &parakeet_db::id_cache::IdCache {

parakeet/src/loaders/mod.rs

···
 pub use misc::{EnrichedStarterPack, EnrichedVerification, StarterPackKey, StarterPackLoader, StarterPackLoaderRet, VerificationLoader};
 pub use post::{EnrichedThreadgate, HydratedPost, PostLoader, PostLoaderRet, PostStateLoader, PostWithComputed};
 pub use profile::{
-    EnrichedStatus, HandleLoader, Profile, ProfileByIdLoader, ProfileLoader, ProfileLoaderRet, ProfileStateLoader, ProfileStatsLoader, ProfileStatsByIdLoader,
+    EnrichedStatus, HandleLoader, Profile, ProfileByIdLoader, ProfileLoaderRet, ProfileStateLoader, ProfileStatsLoader, ProfileStatsByIdLoader,
 };
 
 // Re-export query builder functions (for testing)
···
 pub use list::build_lists_batch_query;
 pub use misc::{build_starterpack_feeds_query, build_starterpacks_batch_query, build_verifications_batch_query};
 pub use post::{build_posts_batch_query, build_posts_by_natural_keys_batch_query};
-pub use profile::{build_profiles_batch_query, build_profiles_by_id_batch_query, build_statuses_batch_query};
+pub use profile::build_profiles_by_id_batch_query;
 
 type CachingLoader<K, V, L> = Loader<K, V, L, PrefixedLoaderCache<K, V>>;
 
···
     pub like_state: LikeRecordLoader,
     pub posts: CachingLoader<String, PostLoaderRet, PostLoader>,
     pub post_state: PostStateLoader,
-    pub profile: CachingLoader<String, ProfileLoaderRet, ProfileLoader>,
     pub profile_by_id: CachingLoader<i32, ProfileLoaderRet, ProfileByIdLoader>,
     pub profile_stats: CachingLoader<String, parakeet_db::models::ProfileStats, ProfileStatsLoader>,
     pub profile_stats_by_id: ProfileStatsByIdLoader,
···
 
             // Occasionally changed: Profile metadata can be updated
             // 1 hour TTL, 50k capacity for profiles
-            profile: new_plc_loader(ProfileLoader(pool.clone()), "profile:", 3600, 50_000),
             profile_by_id: new_plc_loader(ProfileByIdLoader(pool.clone(), id_cache.clone()), "profile_id:", 3600, 50_000),
             // 1 hour TTL, 10k capacity for feeds/lists/etc
             feedgen: new_plc_loader(FeedGenLoader(pool.clone()), "feedgen:", 3600, 10_000),

parakeet/src/loaders/profile.rs

···
     pub thumb_cid: Option<Vec<u8>>,
 }
 
-/// Build SQL query for batch loading profiles with actor metadata
-///
-/// This function is public for testing purposes.
-///
-/// SCHEMA CHANGE: profiles, chat_decls, and notif_decl tables have been consolidated into actors table
-/// All data now comes from actors.profile_*, actors.chat_*, actors.notif_decl_* columns
-pub fn build_profiles_batch_query() -> &'static str {
-    "SELECT
-        a.did,
-        a.handle,
-        a.account_created_at,
-        a.sync_state,
-        a.id as actor_id,
-        a.profile_cid as cid,
-        a.profile_avatar_cid as avatar_cid,
-        a.profile_banner_cid as banner_cid,
-        a.profile_display_name as display_name,
-        a.profile_description as description,
-        a.profile_pinned_post_rkey as pinned_post_rkey,
-        a.profile_joined_sp_id as joined_sp_id,
-        a.profile_pronouns as pronouns,
-        a.profile_website as website,
-        a.chat_allow_incoming as allow_incoming,
-        CASE WHEN a.labeler_cid IS NOT NULL THEN a.id ELSE NULL END as labeler_actor_id,
-        a.notif_decl_allow_subscriptions as allow_subscriptions,
-        a.labels
-     FROM actors a
-     WHERE a.did = ANY($1)
-       AND a.status = 'active'::actor_status"
-}
-
-/// Build SQL query for batch loading statuses with embed URI reconstruction
-///
-/// This function is public for testing purposes.
-///
-/// SCHEMA CHANGE: statuses table has been consolidated into actors table
-/// All data now comes from actors.status_* columns
-pub fn build_statuses_batch_query() -> &'static str {
-    "SELECT
-        a.id as actor_id,
-        a.status_cid as cid,
-        a.status_created_at as created_at,
-        a.status_type as status,
-        a.status_duration as duration,
-        a.status_embed_post_actor_id as embed_post_actor_id,
-        a.status_embed_post_rkey as embed_post_rkey,
-        a.status_thumb_mime_type as thumb_mime_type,
-        a.status_thumb_cid as thumb_cid,
-        a.did,
-        (SELECT 'at://' || emb_a.did || '/app.bsky.feed.post/' || i64_to_tid(emb_p.rkey)
-         FROM posts emb_p
-         INNER JOIN actors emb_a ON emb_p.actor_id = emb_a.id
-         WHERE emb_p.actor_id = a.status_embed_post_actor_id AND emb_p.rkey = a.status_embed_post_rkey) as embed_uri
-     FROM actors a
-     WHERE a.did = ANY($1)
-       AND a.status_cid IS NOT NULL"
-}
 
 // Enriched Status with reconstructed fields
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
···
     }
 }
 
+/// ProfileLoaderRet is the return type for both ProfileByIdLoader
+///
+/// Tuple format:
+/// - String: did - always present
+/// - Option<String>: handle - from actors table
+/// - Option<DateTime>: account_created_at - from actors table
+/// - Option<Profile>: profile - optional (may not exist)
+/// - Option<ChatAllowIncoming>: chat declaration
+/// - bool: is_labeler
+/// - Option<EnrichedStatus>: status
+/// - Option<ProfileAllowSubscriptions>: notification subscription settings
+/// - Option<Vec<ActorLabelRecord>>: labels - from actors table
+pub type ProfileLoaderRet = (
+    String,                                      // did - always present (from actors table)
+    Option<String>,                              // handle - from actors table
+    Option<chrono::DateTime<chrono::Utc>>,       // account_created_at - from actors table
+    Option<Profile>,                             // profile - optional (may not exist)
+    Option<ChatAllowIncoming>,
+    bool,
+    Option<EnrichedStatus>,
+    Option<ProfileAllowSubscriptions>,
+    Option<Vec<parakeet_db::composite_types::ActorLabelRecord>>, // labels - from actors table
+);
+
 pub struct HandleLoader(pub(super) Pool<AsyncPgConnection>);
 impl BatchFn<String, String> for HandleLoader {
     async fn load(&mut self, keys: &[String]) -> HashMap<String, String> {
···
     }
 }
 
-pub struct ProfileLoader(pub(super) Pool<AsyncPgConnection>);
-pub type ProfileLoaderRet = (
-    String,                                      // did - always present (from actors table)
-    Option<String>,                              // handle - from actors table
-    Option<chrono::DateTime<chrono::Utc>>,       // account_created_at - from actors table
-    Option<Profile>,                             // profile - optional (may not exist)
-    Option<ChatAllowIncoming>,
-    bool,
-    Option<EnrichedStatus>,
-    Option<ProfileAllowSubscriptions>,
-    Option<Vec<parakeet_db::composite_types::ActorLabelRecord>>, // labels - from actors table
-);
-impl BatchFn<String, ProfileLoaderRet> for ProfileLoader {
-    async fn load(&mut self, keys: &[String]) -> HashMap<String, ProfileLoaderRet> {
-        let overall_start = std::time::Instant::now();
-        let mut conn = self.0.get().await.unwrap();
-
-        // Load basic actor/profile data with raw SQL to avoid Diesel DSL join complexity
-        let dids: Vec<&str> = keys.iter().map(|s| s.as_str()).collect();
-
-        #[derive(diesel::QueryableByName)]
-        #[allow(dead_code, reason = "Diesel QueryableByName requires all SQL columns even if unused")]
-        struct ActorRow {
-            #[diesel(sql_type = diesel::sql_types::Text)]
-            did: String,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-            handle: Option<String>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Timestamptz>)]
-            account_created_at: Option<chrono::DateTime<chrono::Utc>>,
-            #[diesel(sql_type = parakeet_db::schema::sql_types::ActorSyncState)]
-            sync_state: parakeet_db::types::ActorSyncState,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            actor_id: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-            cid: Option<Vec<u8>>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-            avatar_cid: Option<Vec<u8>>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-            banner_cid: Option<Vec<u8>>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-            display_name: Option<String>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-            description: Option<String>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-            pinned_post_rkey: Option<i64>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-            joined_sp_id: Option<i64>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-            pronouns: Option<String>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-            website: Option<String>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::ChatAllowIncoming>)]
-            allow_incoming: Option<parakeet_db::types::ChatAllowIncoming>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            labeler_actor_id: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::NotifAllowSubscriptions>)]
-            allow_subscriptions: Option<parakeet_db::types::NotifAllowSubscriptions>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<parakeet_db::schema::sql_types::ActorLabel>>)]
-            labels: Option<Vec<parakeet_db::composite_types::ActorLabelRecord>>,
-        }
-
-        let profile_query_start = std::time::Instant::now();
-        let res: Result<Vec<ActorRow>, _> = diesel_async::RunQueryDsl::load(
-            diesel::sql_query(build_profiles_batch_query())
-                .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(&dids),
-            &mut conn,
-        )
-        .await;
-        let profile_query_time = profile_query_start.elapsed().as_secs_f64() * 1000.0;
-
-        match res {
-            Ok(res) => {
-                let profile_count = res.len();
-
-                // Load enriched statuses separately with raw SQL
-                let status_dids: Vec<String> = res.iter().map(|row| row.did.clone()).collect();
-                let status_query_start = std::time::Instant::now();
-                let status_map = if !status_dids.is_empty() {
-                    let status_dids_refs: Vec<&str> = status_dids.iter().map(|s| s.as_str()).collect();
-
-                    #[derive(diesel::QueryableByName)]
-                    struct StatusRow {
-                        #[diesel(sql_type = diesel::sql_types::Integer)]
-                        actor_id: i32,
-                        #[diesel(sql_type = diesel::sql_types::Binary)]
-                        cid: Vec<u8>,
-                        #[diesel(sql_type = diesel::sql_types::Timestamptz)]
-                        created_at: chrono::DateTime<chrono::Utc>,
-                        #[diesel(sql_type = parakeet_db::schema::sql_types::StatusType)]
-                        status: parakeet_db::types::StatusType,
-                        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-                        duration: Option<i32>,
-                        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-                        embed_post_actor_id: Option<i32>,
-                        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-                        embed_post_rkey: Option<i64>,
-                        #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::ImageMimeType>)]
-                        thumb_mime_type: Option<parakeet_db::types::ImageMimeType>,
-                        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-                        thumb_cid: Option<Vec<u8>>,
-                        #[diesel(sql_type = diesel::sql_types::Text)]
-                        did: String,
-                        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-                        embed_uri: Option<String>,
-                    }
-
-                    let statuses: Vec<StatusRow> = diesel_async::RunQueryDsl::load(
-                        diesel::sql_query(build_statuses_batch_query())
-                            .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(&status_dids_refs),
-                        &mut conn,
-                    )
-                    .await
-                    .unwrap_or_default();
-
-                    statuses
-                        .into_iter()
-                        .map(|row| {
-                            // Reconstruct the status record
-                            let record = build_status_record(
-                                &row.status,
-                                &row.duration,
-                                &row.embed_uri,
-                                &None,  // embed_title - not currently loaded
-                                &None,  // embed_description - not currently loaded
-                                &row.thumb_mime_type,
-                                &row.thumb_cid,
-                                &row.created_at,
-                            );
-
-                            let enriched = EnrichedStatus {
-                                status: Status {
-                                    actor_id: row.actor_id,
-                                    cid: row.cid,
-                                    created_at: row.created_at,
-                                    status: row.status,
-                                    duration: row.duration,
-                                    embed_post_actor_id: row.embed_post_actor_id,
-                                    embed_post_rkey: row.embed_post_rkey,
-                                    thumb_mime_type: row.thumb_mime_type,
-                                    thumb_cid: row.thumb_cid,
-                                },
-                                did: row.did.clone(),
-                                created_at: row.created_at,
-                                record,
-                                embed_uri: row.embed_uri,
-                                embed_title: None,
-                                embed_description: None,
-                            };
-                            (row.did, enriched)
-                        })
-                        .collect::<HashMap<_, _>>()
-                } else {
-                    HashMap::new()
-                };
-                let status_query_time = status_query_start.elapsed().as_secs_f64() * 1000.0;
-                let status_count = status_map.len();
-
-                let results: HashMap<String, ProfileLoaderRet> = HashMap::from_iter(res.into_iter().map(
-                    |row| {
-                        // Construct Profile if actor_id is present (cid can be None for empty profiles)
-                        let profile = row.actor_id.map(|actor_id| Profile {
-                            actor_id,
-                            cid: row.cid,
-                            avatar_cid: row.avatar_cid,
-                            banner_cid: row.banner_cid,
-                            display_name: row.display_name,
-                            description: row.description,
-                            pinned_post_rkey: row.pinned_post_rkey,
-                            joined_sp_id: row.joined_sp_id,
-                            pronouns: row.pronouns,
-                            website: row.website,
-                        });
-
-                        let chat_decl = row.allow_incoming.and_then(|v| ChatAllowIncoming::from_str(&v.to_string()).ok());
-                        let notif_decl = row.allow_subscriptions.and_then(|v| ProfileAllowSubscriptions::from_str(&v.to_string()).ok());
-                        let is_labeler = row.labeler_actor_id.is_some();
-                        let status = status_map.get(&row.did).cloned();
-
-                        let val = (row.did.clone(), row.handle, row.account_created_at, profile, chat_decl, is_labeler, status, notif_decl, row.labels);
-
-                        (row.did, val)
-                    },
-                ));
-
-                let overall_time = overall_start.elapsed().as_secs_f64() * 1000.0;
-
-                if overall_time > 15.0 || profile_query_time > 10.0 || status_query_time > 5.0 {
-                    tracing::info!(
-                        "    → ProfileLoader: {:.1}ms total ({} profiles, {} statuses) | profile_query: {:.1}ms, status_query: {:.1}ms",
-                        overall_time, profile_count, status_count, profile_query_time, status_query_time
-                    );
-                }
-
-                results
-            }
-            Err(e) => {
-                tracing::error!("profile load failed: {e}");
-                HashMap::new()
-            }
-        }
-    }
-}
 
 /// Build SQL query for batch loading profiles by actor_id (uses consolidated actors table)
 ///
···
 
 pub struct ProfileStateLoader(pub(super) Pool<AsyncPgConnection>, pub(super) std::sync::Arc<parakeet_db::id_cache::IdCache>);
 impl ProfileStateLoader {
+    /// Get the connection pool
+    pub fn pool(&self) -> &Pool<AsyncPgConnection> {
+        &self.0
+    }
+
     /// Get single profile state (DID-based interface, internally uses actor_ids)
     pub async fn get(&self, viewer_did: &str, subject_did: &str) -> Option<db::ProfileStateRet> {
         let results = self.get_many(viewer_did, &vec![subject_did.to_string()]).await;
···
     }
 
     /// Get the IdCache for DID → actor_id resolution
-    pub fn id_cache(&self) -> &parakeet_db::id_cache::IdCache {
+    pub fn id_cache(&self) -> &std::sync::Arc<parakeet_db::id_cache::IdCache> {
         &self.1
     }
 }

parakeet/src/sql/labels_by_actor_ids.sql

···
--- Phase 7: Labels denormalized into actors.labels[] and posts.labels[] arrays
--- Optimized labels query that uses actor_ids instead of DIDs
--- This avoids decompressing the actors table by using IdCache to resolve DIDs beforehand
---
--- Parameters:
---   $1: uris (text[]) - Array of URIs to fetch labels for
---   $2: labeler_actor_ids (integer[]) - Array of allowed labeler actor IDs
---
--- Performance: Maintained similar performance with denormalized structure
--- Avoids: Separate labels table join
---
--- Note: Caller must use IdCache to:
---   1. Resolve labeler DIDs → actor_ids before this query
---   2. Resolve labeler_actor_id → DID after this query
---
--- Note: The new composite types (actor_label, post_label) do not include:
---   - uri (reconstructed from actor/post)
---   - self_label (always false for post labels, always true for actor self-labels)
---   - cid (not stored in denormalized structure)
---   - sig (not stored in denormalized structure)
-
--- Extract labels from actors table
-SELECT
-    (label_record).labeler_actor_id as labeler_actor_id,
-    (label_record).label as label,
-    'at://' || a.did as uri,
-    true as self_label,  -- Actor labels are always self-labels
-    NULL::bytea as cid,  -- Not stored in denormalized structure
-    (label_record).negated as negated,
-    (label_record).expires as expires,
-    NULL::bytea as sig,  -- Not stored in denormalized structure
-    (label_record).created_at as created_at
-FROM actors a, UNNEST(a.labels) as label_record
-WHERE 'at://' || a.did = ANY($1)
-  AND (label_record).negated = false
-  AND ((label_record).labeler_actor_id = a.id OR (label_record).labeler_actor_id = ANY($2))
-
-UNION ALL
-
--- Extract labels from posts table
-SELECT
-    (label_record).labeler_actor_id as labeler_actor_id,
-    (label_record).label as label,
-    'at://' || a.did || '/app.bsky.feed.post/' || parakeet_db.i64_to_tid(p.rkey) as uri,
-    false as self_label,  -- Post labels are not self-labels
-    NULL::bytea as cid,  -- Not stored in denormalized structure
-    (label_record).negated as negated,
-    (label_record).expires as expires,
-    NULL::bytea as sig,  -- Not stored in denormalized structure
-    (label_record).created_at as created_at
-FROM posts p
-INNER JOIN actors a ON p.actor_id = a.id, UNNEST(p.labels) as label_record
-WHERE 'at://' || a.did || '/app.bsky.feed.post/' || parakeet_db.i64_to_tid(p.rkey) = ANY($1)
-  AND (label_record).negated = false
-  AND (label_record).labeler_actor_id = ANY($2)
-
-ORDER BY created_at

parakeet/src/xrpc/app_bsky/actor.rs

···
 use axum_extra::extract::Query as ExtraQuery;
 use lexica::app_bsky::actor::ProfileViewDetailed;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 
 #[derive(Debug, Deserialize)]
 pub struct ActorQuery {
···
         (None, None)
     };
     let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_did, maybe_actor_id).await;
-    let mut profiles_map = hyd.hydrate_profiles_detailed(dids.clone()).await;
+
+    // Use the cache's batch method instead of direct hydration
+    let profiles_vec = state.profile_cache
+        .get_or_hydrate_batch(dids.clone(), &state.pool, &state.id_cache, &hyd)
+        .await;
+
+    // Convert Vec to HashMap for compatibility with existing code
+    let mut profiles_map: std::collections::HashMap<String, ProfileViewDetailed> = profiles_vec
+        .into_iter()
+        .map(|profile| (profile.did.clone(), profile))
+        .collect();
 
     // Maintain search result order (hydration returns HashMap)
     let actors: Vec<ProfileViewDetailed> = dids
···
         (None, None)
     };
     let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_did, maybe_actor_id).await;
-    let mut profiles_map = hyd.hydrate_profiles_detailed(dids.clone()).await;
+
+    // Use the cache's batch method instead of direct hydration
+    let profiles_vec = state.profile_cache
+        .get_or_hydrate_batch(dids.clone(), &state.pool, &state.id_cache, &hyd)
+        .await;
+
+    // Convert Vec to HashMap for compatibility with existing code
+    let mut profiles_map: std::collections::HashMap<String, ProfileViewDetailed> = profiles_vec
+        .into_iter()
+        .map(|profile| (profile.did.clone(), profile))
+        .collect();
 
     // Maintain priority order
     let actors: Vec<ProfileViewDetailed> = dids
···
         .into_response());
     }
 
+    // Convert DIDs to actor_ids for efficient profile loading
+    let did_to_actor_id = crate::id_cache_helpers::get_actor_ids_or_fetch(
+        &state.pool,
+        &state.id_cache,
+        &all_dids,
+    ).await.unwrap_or_default();
+
+    let actor_ids: Vec<i32> = did_to_actor_id.values().copied().collect();
+
     // Load profiles to check quality (already sorted by follower count from SQL)
-    let profiles = state
+    let profiles_by_id = state
         .dataloaders
-        .profile
-        .load_many(all_dids.clone())
+        .profile_by_id
+        .load_many(actor_ids)
         .await;
+
+    // Convert back to DID-keyed map for compatibility
+    let profiles: HashMap<String, _> = profiles_by_id
+        .into_iter()
+        .filter_map(|(actor_id, profile)| {
+            did_to_actor_id.iter()
+                .find(|(_, id)| **id == actor_id)
+                .map(|(did, _)| (did.clone(), profile))
+        })
+        .collect();
 
     // Filter by quality, maintaining follower-count order from SQL query
     let ranked_dids: Vec<String> = all_dids

parakeet/src/xrpc/app_bsky/unspecced/mod.rs

···
         return Ok(Json(GetSuggestedUsersResponse { actors: Vec::new() }));
     }
 
+    // Convert DIDs to actor_ids for profile loading
+    let actor_id_map = match crate::id_cache_helpers::get_actor_ids_or_fetch(
+        &state.pool,
+        &state.id_cache,
+        &all_dids,
+    ).await {
+        Ok(map) => map,
+        Err(_) => return Ok(Json(GetSuggestedUsersResponse { actors: Vec::new() })),
+    };
+
+    let actor_ids: Vec<i32> = actor_id_map.values().copied().collect();
+
     // Load profiles to check quality (already sorted by follower count from SQL)
-    let profiles = state.dataloaders.profile.load_many(all_dids.clone()).await;
+    let profiles_by_id = state.dataloaders.profile_by_id.load_many(actor_ids).await;
+
+    // Convert actor_id-keyed profiles back to DID-keyed for easier lookup
+    let id_to_did: std::collections::HashMap<i32, String> = actor_id_map.iter().map(|(did, id)| (*id, did.clone())).collect();
+    let profiles: std::collections::HashMap<String, _> = profiles_by_id
+        .into_iter()
+        .filter_map(|(actor_id, profile_info)| {
+            id_to_did.get(&actor_id).map(|did| (did.clone(), profile_info))
+        })
+        .collect();
 
     // Filter by quality, maintaining follower-count order from SQL query
     let ranked_dids: Vec<String> = all_dids

parakeet/tests/loaders_test.rs

···
 // ============================================================================
 // PROFILE LOADER TESTS
 // ============================================================================
-// ProfileLoader uses complex SQL with:
-// - Queries consolidated actors table (profile_*, chat_*, notif_decl_* columns)
-// - LEFT JOIN to labelers table (still separate)
-// - Format string for DID list interpolation
-// - Separate status query with embed reconstruction
-
-#[tokio::test]
-async fn test_profile_loader_main_query_empty() -> eyre::Result<()> {
-    common::ensure_test_db_ready().await;
-    let pool = common::test_diesel_pool();
-    let _conn = pool.get().await.wrap_err("Failed to get connection")?;
-
-    // Simulate ProfileLoader's main query with empty DID list
-    let dids: Vec<String> = vec![];
-
-    if dids.is_empty() {
-        // Empty case - should handle gracefully
-        return Ok(());
-    }
-
-    // This would be the actual query, but it won't execute with empty DIDs
-    // Just verifying the logic handles it
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_profile_loader_main_query_structure() -> eyre::Result<()> {
-    common::ensure_test_db_ready().await;
-    let pool = common::test_diesel_pool();
-    let mut conn = pool.get().await.wrap_err("Failed to get connection")?;
-
-    let dids = ["did:plc:test1".to_string(), "did:plc:test2".to_string()];
-
-    // Build the query using the actual ProfileLoader query builder (no SQL duplication!)
-    let dids_refs: Vec<&str> = dids.iter().map(|s| s.as_str()).collect();
-
-    let query = parakeet::loaders::build_profiles_batch_query();
-
-    #[derive(diesel::QueryableByName)]
-    #[allow(dead_code)]
-    struct ActorRow {
-        #[diesel(sql_type = diesel::sql_types::Text)]
-        did: String,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        handle: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-        actor_id: Option<i32>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        cid: Option<Vec<u8>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Timestamptz>)]
-        created_at: Option<chrono::DateTime<chrono::Utc>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        avatar_cid: Option<Vec<u8>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        banner_cid: Option<Vec<u8>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        display_name: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        description: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        pinned_post_id: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        joined_sp_id: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        pronouns: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        website: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::ChatAllowIncoming>)]
-        allow_incoming: Option<parakeet_db::types::ChatAllowIncoming>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-        labeler_actor_id: Option<i32>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::NotifAllowSubscriptions>)]
-        allow_subscriptions: Option<parakeet_db::types::NotifAllowSubscriptions>,
-    }
-
-    diesel::sql_query(query)
-        .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(&dids_refs)
-        .load::<ActorRow>(&mut conn)
-        .await
-        .wrap_err("ProfileLoader main query SQL failed")?;
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_profile_loader_status_query_structure() -> eyre::Result<()> {
-    common::ensure_test_db_ready().await;
-    let pool = common::test_diesel_pool();
-    let mut conn = pool.get().await.wrap_err("Failed to get connection")?;
-
-    let dids = ["did:plc:test1".to_string()];
-    let dids_refs: Vec<&str> = dids.iter().map(|s| s.as_str()).collect();
-
-    // Use the actual status query builder from ProfileLoader (no SQL duplication!)
-    let status_query = parakeet::loaders::build_statuses_batch_query();
-
-    #[derive(diesel::QueryableByName)]
-    #[allow(dead_code)]
-    struct StatusRow {
-        #[diesel(sql_type = diesel::sql_types::Integer)]
-        actor_id: i32,
-        #[diesel(sql_type = diesel::sql_types::Binary)]
-        cid: Vec<u8>,
-        #[diesel(sql_type = diesel::sql_types::Timestamptz)]
-        created_at: chrono::DateTime<chrono::Utc>,
-        #[diesel(sql_type = parakeet_db::schema::sql_types::StatusType)]
-        status: parakeet_db::types::StatusType,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-        duration: Option<i32>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        embed_post_id: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::ImageMimeType>)]
-        thumb_mime_type: Option<parakeet_db::types::ImageMimeType>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        thumb_cid: Option<Vec<u8>>,
-        #[diesel(sql_type = diesel::sql_types::Text)]
-        did: String,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        embed_uri: Option<String>,
-    }
-
-    diesel::sql_query(status_query)
-        .bind::<diesel::sql_types::Array<diesel::sql_types::Text>, _>(&dids_refs)
-        .load::<StatusRow>(&mut conn)
-        .await
-        .wrap_err("ProfileLoader status query SQL failed")?;
-
-    Ok(())
-}
+// ProfileByIdLoader uses optimized SQL querying by actor_id (primary key)
+// instead of DID (btree index). Queries consolidated actors table with
+// profile_*, chat_*, notif_decl_*, and status_* columns.
 
 #[tokio::test]
 async fn test_profile_by_id_loader_query_structure() -> eyre::Result<()> {

parakeet/tests/sql/batch_loading_test.rs

···
 // NOTE: Threadgate loading test removed - threadgates are now loaded inline with posts
 // in build_posts_batch_query() to eliminate a separate database roundtrip
 
-/// Test batch loading profiles (from loaders/profile.rs)
-#[tokio::test]
-async fn test_batch_profiles_query() -> eyre::Result<()> {
-    common::ensure_test_db_ready().await;
-    let pool = common::test_diesel_pool();
-    let mut conn = pool.get().await.wrap_err("Failed to get connection")?;
-
-    // Use the ACTUAL query builder from the source code
-    let query = parakeet::loaders::build_profiles_batch_query();
-    let dids = vec!["did:plc:test1", "did:plc:test2"];
-
-    diesel_async::RunQueryDsl::execute(
-        diesel::sql_query(query)
-            .bind::<Array<diesel::sql_types::Text>, _>(&dids),
-        &mut conn,
-    )
-    .await
-    .wrap_err("Batch profile loading query failed")?;
-
-    Ok(())
-}
-
-/// Test batch loading statuses (from loaders/profile.rs)
+/// Test batch loading profiles by ID (from loaders/profile.rs)
+/// Note: The old DID-based profile/status loaders have been removed as they were
+/// inefficient (required btree index lookups). ProfileByIdLoader uses primary key lookups.
 #[tokio::test]
-async fn test_batch_statuses_query() -> eyre::Result<()> {
+async fn test_batch_profiles_by_id_query() -> eyre::Result<()> {
     common::ensure_test_db_ready().await;
     let pool = common::test_diesel_pool();
     let mut conn = pool.get().await.wrap_err("Failed to get connection")?;
 
     // Use the ACTUAL query builder from the source code
-    let query = parakeet::loaders::build_statuses_batch_query();
-    let dids = vec!["did:plc:test1"];
+    let query = parakeet::loaders::build_profiles_by_id_batch_query();
+    let actor_ids = vec![1i32, 2i32];
 
     diesel_async::RunQueryDsl::execute(
         diesel::sql_query(query)
-            .bind::<Array<diesel::sql_types::Text>, _>(&dids),
+            .bind::<Array<diesel::sql_types::Integer>, _>(&actor_ids),
         &mut conn,
     )
     .await
-    .wrap_err("Batch status loading query failed")?;
+    .wrap_err("Batch profile loading by ID query failed")?;
 
     Ok(())
 }