feat: denormalize strategy; labelers

consumer/src/db/labels.rs

···
         return Ok(0); // Actor doesn't exist yet
     };
 
-    // drop any label defs not currently in the list
-    let _ = conn
-        .execute(
-            "DELETE FROM labeler_defs WHERE labeler_actor_id=$1 AND NOT label_identifier = any($2)",
-            &[&labeler_actor_id, &rec.policies.label_values],
-        )
-        .await?;
-
+    // Build labeler_defs array from label values and definitions
+    // Maps label_identifier -> definition
     let definitions = rec
         .policies
         .label_value_definitions
···
         .map(|def| (def.identifier.clone(), def))
         .collect::<HashMap<String, &LabelValueDefinition>>();
 
+    // Build arrays of values for each composite field
+    let mut label_identifiers = Vec::new();
+    let mut severities = Vec::new();
+    let mut blurs_vals = Vec::new();
+    let mut default_settings = Vec::new();
+    let mut adult_onlys = Vec::new();
+    let mut locales_vals = Vec::new();
+
     for label in &rec.policies.label_values {
         let definition = definitions.get(label);
 
-        let severity = definition.map(|v| v.severity.to_string());
-        let blurs = definition.map(|v| v.blurs.to_string());
-        let default_setting = definition
-            .and_then(|v| v.default_setting)
-            .map(|v| v.to_string());
-        let adult_only = definition.and_then(|v| v.adult_only).unwrap_or_default();
-        let locales = definition.and_then(|v| serde_json::to_value(&v.locales).ok());
+        label_identifiers.push(label.clone());
+        severities.push(definition.map(|v| v.severity.to_string()));
+        blurs_vals.push(definition.map(|v| v.blurs.to_string()));
+        default_settings.push(definition.and_then(|v| v.default_setting).map(|v| v.to_string()));
+        adult_onlys.push(definition.and_then(|v| v.adult_only).unwrap_or_default());
+        locales_vals.push(definition.and_then(|v| serde_json::to_value(&v.locales).ok()));
+    }
 
-        let _ = conn
-            .execute(
-                include_str!("sql/label_defs_upsert.sql"),
-                &[
-                    &labeler_actor_id,
-                    &label,
-                    &severity,
-                    &blurs,
-                    &default_setting,
-                    &adult_only,
-                    &locales,
-                ],
-            )
-            .await?;
-    }
+    // Update labeler_defs array on actors table using composite type constructor
+    // ROW(...) constructs the composite type, ARRAY[...] builds the array
+    conn.execute(
+        "UPDATE actors
+         SET labeler_defs = (
+             SELECT ARRAY_AGG(
+                 ROW(
+                     label_identifier,
+                     severity::text::label_severity,
+                     blurs::text::label_blurs,
+                     default_setting::text::label_default_setting,
+                     adult_only,
+                     locales,
+                     NOW()
+                 )::labeler_def_record
+                 ORDER BY idx
+             )
+             FROM unnest($2::text[], $3::text[], $4::text[], $5::text[], $6::boolean[], $7::jsonb[])
+                 WITH ORDINALITY AS t(label_identifier, severity, blurs, default_setting, adult_only, locales, idx)
+         )
+         WHERE id = $1",
+        &[
+            &labeler_actor_id,
+            &label_identifiers,
+            &severities,
+            &blurs_vals,
+            &default_settings,
+            &adult_onlys,
+            &locales_vals,
+        ],
+    )
+    .await?;
 
     Ok(0)
 }

consumer/src/db/operations/labeler.rs

···
 ///
 /// This function:
 /// 1. Gets/creates actor_id for the DID
-/// 2. Inserts stub labeler if not found (status='stub')
+/// 2. Sets labeler_cid and labeler_status='stub' on actors table if not already set
 /// 3. Returns actor_id
 ///
 /// Uses advisory locks to prevent concurrent transactions from racing on the same labeler URI.
 ///
-/// Labelers have a 1:1 relationship with actors (actor_id is the PK),
-/// so the actor_id serves as both the labeler identifier and return value.
+/// Labelers are now denormalized into the actors table with labeler_* columns.
 pub async fn ensure_labeler_stub<C: GenericClient>(
     conn: &C,
     did: &str,
···
     // Get/create actor_id (discard allowlist status and was_created, not needed for labelers)
     let actor_id = crate::db::actor::ensure_actor_id(conn, did, None, None, chrono::Utc::now()).await?;
 
-    // Use CTE to SELECT first, then conditionally INSERT only if not found
-    // This prevents unnecessary sequence consumption when labeler stub already exists
-    // Still uses ON CONFLICT for race condition safety between concurrent transactions
+    // Set labeler_cid and labeler_status on actors table if not already set
+    // Only update if labeler_cid is NULL (stub not yet created)
     conn.execute(
-        "WITH existing AS (
-             SELECT actor_id FROM labelers WHERE actor_id = $1
-         )
-         INSERT INTO labelers (actor_id, cid, created_at, status)
-         SELECT $1, $2, NOW(), 'stub'::labeler_status
-         WHERE NOT EXISTS (SELECT 1 FROM existing)
-         ON CONFLICT (actor_id) DO NOTHING",
+        "UPDATE actors
+         SET labeler_cid = $2,
+             labeler_created_at = NOW(),
+             labeler_status = 'stub'::labeler_status,
+             labeler_like_count = 0
+         WHERE id = $1 AND labeler_cid IS NULL",
         &[&actor_id, &cid_digest],
     )
     .await?;
···
 
 pub async fn labeler_delete<C: GenericClient>(conn: &C, actor_id: i32) -> Result<u64> {
     // Labeler records always use rkey "self", so no rkey parameter needed
+    // Now sets labeler_* columns to NULL instead of deleting from separate table
 
     conn.execute(
-        "DELETE FROM labelers
-         WHERE actor_id = $1",
+        "UPDATE actors
+         SET labeler_cid = NULL,
+             labeler_created_at = NULL,
+             labeler_reasons = NULL,
+             labeler_subject_types = NULL,
+             labeler_subject_collections = NULL,
+             labeler_status = NULL,
+             labeler_like_count = NULL,
+             labeler_defs = NULL
+         WHERE id = $1",
         &[&actor_id],
     )
     .await
···
 ///
 /// This is called after bulk inserting labeler_likes to update the aggregate counts.
 /// Uses a single UPDATE statement with aggregation for efficiency.
+/// Now updates actors.labeler_like_count instead of labelers.like_count
 pub async fn increment_labeler_like_counts<C: GenericClient>(
     conn: &C,
     labeler_actor_ids: &[i32],
···
     }
 
     conn.execute(
-        "UPDATE labelers
-         SET like_count = like_count + counts.count
+        "UPDATE actors
+         SET labeler_like_count = COALESCE(labeler_like_count, 0) + counts.count
          FROM (
              SELECT actor_id, COUNT(*) as count
              FROM unnest($1::int[]) as actor_id
              GROUP BY actor_id
          ) AS counts
-         WHERE labelers.actor_id = counts.actor_id",
+         WHERE actors.id = counts.actor_id",
         &[&labeler_actor_ids],
     )
     .await
···
 /// Decrement like_count for a single labeler
 ///
 /// This is called when deleting a labeler_like record.
+/// Now updates actors.labeler_like_count instead of labelers.like_count
 pub async fn decrement_labeler_like_count<C: GenericClient>(
     conn: &C,
     labeler_actor_id: i32,
 ) -> Result<u64> {
     conn.execute(
-        "UPDATE labelers
-         SET like_count = GREATEST(like_count - 1, 0)
-         WHERE actor_id = $1",
+        "UPDATE actors
+         SET labeler_like_count = GREATEST(COALESCE(labeler_like_count, 0) - 1, 0)
+         WHERE id = $1",
         &[&labeler_actor_id],
     )
     .await

consumer/src/db/sql/label_service_upsert.sql

···
--- Insert/update labeler service with self-contained schema (no records table)
+-- Insert/update labeler service on actors table (denormalized)
 -- Parameters: $1=actor_id, $2=cid(bytea), $3=reasons, $4=subject_types, $5=subject_collections
 -- NOTE: actor_id is provided by dispatcher after ensuring actor exists
-INSERT INTO labelers (actor_id, cid, reasons, subject_types, subject_collections)
-SELECT
-    $1,  -- actor_id (provided by dispatcher)
-    $2::bytea,  -- cid (embedded)
-    $3::text[]::reason_type[],
-    $4::text[]::subject_type[],
-    $5
-ON CONFLICT (actor_id) DO UPDATE SET
-    cid=EXCLUDED.cid,
-    reasons=EXCLUDED.reasons,
-    subject_types=EXCLUDED.subject_types,
-    subject_collections=EXCLUDED.subject_collections,
-    status='complete'::labeler_status
+-- Sets labeler_* columns on actors table instead of separate labelers table
+UPDATE actors
+SET
+    labeler_cid = $2::bytea,
+    labeler_created_at = COALESCE(labeler_created_at, NOW()),
+    labeler_reasons = $3::text[]::reason_type[],
+    labeler_subject_types = $4::text[]::subject_type[],
+    labeler_subject_collections = $5,
+    labeler_status = 'complete'::labeler_status,
+    labeler_like_count = COALESCE(labeler_like_count, 0)
+WHERE id = $1

migrations/2025-12-08-004705_denormalize_labelers/down.sql

···
+-- Revert Phase 2: Restore labelers tables from actors
+
+-- Recreate labelers table
+CREATE TABLE labelers (
+  actor_id INTEGER PRIMARY KEY,
+  cid BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  reasons reason_type[],
+  subject_types subject_type[],
+  subject_collections TEXT[],
+  status labeler_status NOT NULL DEFAULT 'complete',
+  like_count INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE INDEX idx_labelers_like_count_desc ON labelers(like_count DESC);
+
+-- Recreate labeler_defs table
+CREATE TABLE labeler_defs (
+  labeler_actor_id INTEGER NOT NULL,
+  label_identifier TEXT NOT NULL,
+  severity label_severity,
+  blurs label_blurs,
+  default_setting label_default_setting,
+  adult_only BOOLEAN NOT NULL DEFAULT FALSE,
+  locales JSONB,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (labeler_actor_id, label_identifier)
+);
+
+-- Recreate labeler_likes table
+CREATE TABLE labeler_likes (
+  actor_id INTEGER NOT NULL,
+  rkey BIGINT NOT NULL,
+  labeler_actor_id INTEGER NOT NULL,
+  PRIMARY KEY (actor_id, rkey)
+);
+
+CREATE INDEX idx_labeler_likes_labeler ON labeler_likes(labeler_actor_id);
+CREATE INDEX idx_labeler_likes_rkey ON labeler_likes(rkey);
+
+-- Restore data from actors table
+INSERT INTO labelers (actor_id, cid, created_at, reasons, subject_types, subject_collections, status, like_count)
+SELECT id, labeler_cid, labeler_created_at, labeler_reasons, labeler_subject_types, labeler_subject_collections, labeler_status, labeler_like_count
+FROM actors
+WHERE labeler_cid IS NOT NULL;
+
+-- Restore labeler_defs from array
+INSERT INTO labeler_defs (labeler_actor_id, label_identifier, severity, blurs, default_setting, adult_only, locales, created_at)
+SELECT
+  a.id,
+  (def).label_identifier,
+  (def).severity,
+  (def).blurs,
+  (def).default_setting,
+  (def).adult_only,
+  (def).locales,
+  (def).created_at
+FROM actors a, unnest(a.labeler_defs) AS def
+WHERE a.labeler_defs IS NOT NULL;
+
+-- Drop labeler columns from actors
+ALTER TABLE actors
+  DROP COLUMN labeler_defs,
+  DROP COLUMN labeler_like_count,
+  DROP COLUMN labeler_status,
+  DROP COLUMN labeler_subject_collections,
+  DROP COLUMN labeler_subject_types,
+  DROP COLUMN labeler_reasons,
+  DROP COLUMN labeler_created_at,
+  DROP COLUMN labeler_cid;

migrations/2025-12-08-004705_denormalize_labelers/up.sql

···
+-- Phase 2: Denormalize labelers into actors table
+-- Move labeler data from separate tables into actors table with composite type arrays
+
+-- Add labeler columns to actors (flatten 1:1 relationship)
+ALTER TABLE actors
+  ADD COLUMN labeler_cid BYTEA,
+  ADD COLUMN labeler_created_at TIMESTAMPTZ,
+  ADD COLUMN labeler_reasons reason_type[],
+  ADD COLUMN labeler_subject_types subject_type[],
+  ADD COLUMN labeler_subject_collections TEXT[],
+  ADD COLUMN labeler_status labeler_status,
+  ADD COLUMN labeler_like_count INTEGER DEFAULT 0;
+
+-- Add labeler_defs array (composite type)
+ALTER TABLE actors
+  ADD COLUMN labeler_defs labeler_def_record[];
+
+-- Backfill labeler data from labelers table
+UPDATE actors a
+SET
+  labeler_cid = l.cid,
+  labeler_created_at = l.created_at,
+  labeler_reasons = l.reasons,
+  labeler_subject_types = l.subject_types,
+  labeler_subject_collections = l.subject_collections,
+  labeler_status = l.status,
+  labeler_like_count = l.like_count
+FROM labelers l
+WHERE a.id = l.actor_id;
+
+-- Backfill labeler_defs array (using composite type)
+UPDATE actors a
+SET labeler_defs = (
+  SELECT ARRAY_AGG(
+    ROW(
+      ld.label_identifier,
+      ld.severity,
+      ld.blurs,
+      ld.default_setting,
+      ld.adult_only,
+      ld.locales,
+      ld.created_at
+    )::labeler_def_record
+    ORDER BY ld.created_at
+  )
+  FROM labeler_defs ld
+  WHERE ld.labeler_actor_id = a.id
+);
+
+-- Drop old tables (labeler_likes will be handled later if needed for arrays)
+DROP TABLE labeler_defs;
+DROP TABLE labeler_likes;
+DROP TABLE labelers;

parakeet-db/src/composite_types.rs

···
 
 use crate::schema::sql_types::{
     PostExtEmbed, PostVideoEmbed, PostImageEmbed, PostFacetEmbed, PostVideoCaption,
-    FollowRecord, MuteRecord, BlockRecord, BookmarkRecord, ThreadMuteRecord,
-    ListMuteRecord, ListBlockRecord, LabelerDefRecord, PostLabel, ActorLabel,
+    LabelerDefRecord,
+    // Note: Other composite types will be added when used in Phase 3-7:
+    // FollowRecord, MuteRecord, BlockRecord, BookmarkRecord, ThreadMuteRecord,
+    // ListMuteRecord, ListBlockRecord, PostLabel, ActorLabel,
 };
+
+// Placeholder SQL types for composite types not yet used in tables
+// These will be auto-generated by diesel once the columns are added
+#[allow(dead_code)]
+mod placeholder_sql_types {
+    use diesel::query_builder::QueryId;
+    use diesel::sql_types::SqlType;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "follow_record"))]
+    pub struct FollowRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "mute_record"))]
+    pub struct MuteRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "block_record"))]
+    pub struct BlockRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "bookmark_record"))]
+    pub struct BookmarkRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "thread_mute_record"))]
+    pub struct ThreadMuteRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "list_mute_record"))]
+    pub struct ListMuteRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "list_block_record"))]
+    pub struct ListBlockRecord;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "post_label"))]
+    pub struct PostLabel;
+
+    #[derive(QueryId, SqlType)]
+    #[diesel(postgres_type(name = "actor_label"))]
+    pub struct ActorLabel;
+}
+
+use placeholder_sql_types::*;
 use crate::types::{
     ImageMimeType, VideoMimeType, FacetType, CaptionMimeType, LanguageCode,
     LabelSeverity, LabelBlurs, LabelDefaultSetting,

parakeet-db/src/models.rs

···
 //
 // =============================================================================
 
+use crate::composite_types::LabelerDef;
 use crate::tid_util::{decode_tid, encode_tid, TidError};
 use crate::types::*;
 use chrono::prelude::*;
···
     pub lists_count: Option<i16>,
     pub feeds_count: Option<i16>,
     pub starterpacks_count: Option<i16>,
+    // Labeler fields (from labelers table, denormalized - NULL for non-labelers)
+    pub labeler_cid: Option<Vec<u8>>,
+    pub labeler_created_at: Option<DateTime<Utc>>,
+    pub labeler_reasons: Option<Vec<Option<ReasonType>>>,
+    pub labeler_subject_types: Option<Vec<Option<SubjectType>>>,
+    pub labeler_subject_collections: Option<Vec<Option<String>>>,
+    pub labeler_status: Option<LabelerStatus>,
+    pub labeler_like_count: Option<i32>,
+    pub labeler_defs: Option<Vec<Option<LabelerDef>>>,
 }
 
 // AllowlistEntry model removed - allowlist table dropped in favor of actors.sync_state
···
                          // Note: created_at derived from TID rkey via created_at() method
 }
 
-// Labeler Likes (rare, ~0.01% of likes)
-#[derive(Clone, Debug, Queryable, Selectable, Identifiable)]
-#[diesel(table_name = crate::schema::labeler_likes)]
-#[diesel(primary_key(actor_id, rkey))]
-#[diesel(check_for_backend(diesel::pg::Pg))]
-pub struct LabelerLike {
-    pub actor_id: i32,           // PK: FK to actors (liker)
-    pub rkey: i64,               // PK: TID as INT8
-    pub labeler_actor_id: i32,   // FK to actors (labeler service)
-                                 // Note: created_at derived from TID rkey via created_at() method
-}
+// Note: Labeler Likes table dropped - labeler data moved to actors table
+// The labeler_like_count is maintained on actors.labeler_like_count
+// Individual like records (labeler_likes table) were dropped for simplicity
 
 #[derive(Clone, Debug, Queryable, Selectable, Identifiable)]
 #[diesel(table_name = crate::schema::reposts)]
···
 // MODERATION & LABELS
 // =============================================================================
 
-#[derive(Clone, Debug, Serialize, Deserialize, Queryable, Selectable, Identifiable)]
-#[diesel(table_name = crate::schema::labelers)]
-#[diesel(primary_key(actor_id))]
-#[diesel(check_for_backend(diesel::pg::Pg))]
-pub struct Labeler {
-    pub actor_id: i32,             // PK: FK to actors (no rkey, always 'self')
-    pub cid: Vec<u8>,              // 32-byte CID digest
-    pub created_at: DateTime<Utc>, // From AT Protocol record
-    pub reasons: Option<array_helpers::ReasonTypeArray>, // ENUM array: spam | violation | misleading | etc.
-    pub subject_types: Option<array_helpers::SubjectTypeArray>, // ENUM array: account | record | chat
-    pub subject_collections: Option<array_helpers::TextArray>, // Collection names (as TEXT since generic)
-    pub status: LabelerStatus,                                 // ENUM: complete | stub | deleted
-    pub like_count: i32, // Aggregated count of likes (maintained by database_writer)
-}
-
-#[derive(Clone, Debug, Serialize, Deserialize, Queryable, Selectable)]
-#[diesel(table_name = crate::schema::labeler_defs)]
-#[diesel(primary_key(labeler_actor_id, label_identifier))]
-#[diesel(check_for_backend(diesel::pg::Pg))]
-pub struct LabelerDef {
-    pub labeler_actor_id: i32,                        // PK: FK to actors
-    pub label_identifier: String,                     // PK: Label name
-    pub severity: Option<LabelSeverity>,              // ENUM: inform | alert | none
-    pub blurs: Option<LabelBlurs>,                    // ENUM: content | media | none
-    pub default_setting: Option<LabelDefaultSetting>, // ENUM: ignore | warn | hide
-    pub adult_only: bool,
-    pub locales: Option<serde_json::Value>,
-    pub created_at: DateTime<Utc>,
-}
+// Note: Labeler and LabelerDef structs removed
+// Labeler data is now stored directly on actors table with labeler_* columns
+// LabelerDef is now a composite type in composite_types.rs, stored as labeler_defs array on actors
 
 #[derive(Clone, Debug)]
 pub struct Label {
···
 
 impl_tid_rkey!(Post);
 impl_tid_rkey!(FeedgenLike);
-impl_tid_rkey!(LabelerLike);
+// impl_tid_rkey!(LabelerLike); // Removed - labeler_likes table dropped
 impl_tid_rkey!(Repost);
 impl_tid_rkey!(Follow);
 impl_tid_rkey!(Block);
···
 
 impl_tid_created_at!(Post);
 impl_tid_created_at!(FeedgenLike);
-impl_tid_created_at!(LabelerLike);
+// impl_tid_created_at!(LabelerLike); // Removed - labeler_likes table dropped
 impl_tid_created_at!(Repost);
 impl_tid_created_at!(Follow);
 impl_tid_created_at!(Block);

parakeet-db/src/schema.rs

···
     pub struct LabelSeverity;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "labeler_def_record"))]
+    pub struct LabelerDefRecord;
+
+    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "labeler_status"))]
     pub struct LabelerStatus;
 
···
     pub struct ListBlockRecord;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
-    #[diesel(postgres_type(name = "labeler_def_record"))]
-    pub struct LabelerDefRecord;
-
-    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "post_label"))]
     pub struct PostLabel;
 
···
     use super::sql_types::ImageMimeType;
     use super::sql_types::ChatAllowIncoming;
     use super::sql_types::NotifAllowSubscriptions;
+    use super::sql_types::ReasonType;
+    use super::sql_types::SubjectType;
+    use super::sql_types::LabelerStatus;
+    use super::sql_types::LabelerDefRecord;
 
     actors (id) {
         id -> Int4,
···
         lists_count -> Nullable<Int2>,
         feeds_count -> Nullable<Int2>,
         starterpacks_count -> Nullable<Int2>,
+        labeler_cid -> Nullable<Bytea>,
+        labeler_created_at -> Nullable<Timestamptz>,
+        labeler_reasons -> Nullable<Array<Nullable<ReasonType>>>,
+        labeler_subject_types -> Nullable<Array<Nullable<SubjectType>>>,
+        labeler_subject_collections -> Nullable<Array<Nullable<Text>>>,
+        labeler_status -> Nullable<LabelerStatus>,
+        labeler_like_count -> Nullable<Int4>,
+        labeler_defs -> Nullable<Array<Nullable<LabelerDefRecord>>>,
     }
 }
 
···
 }
 
 diesel::table! {
-    use diesel::sql_types::*;
-    use super::sql_types::LabelSeverity;
-    use super::sql_types::LabelBlurs;
-    use super::sql_types::LabelDefaultSetting;
-
-    labeler_defs (labeler_actor_id, label_identifier) {
-        labeler_actor_id -> Int4,
-        label_identifier -> Text,
-        severity -> Nullable<LabelSeverity>,
-        blurs -> Nullable<LabelBlurs>,
-        default_setting -> Nullable<LabelDefaultSetting>,
-        adult_only -> Bool,
-        locales -> Nullable<Jsonb>,
-        created_at -> Timestamptz,
-    }
-}
-
-diesel::table! {
-    labeler_likes (actor_id, rkey) {
-        actor_id -> Int4,
-        rkey -> Int8,
-        labeler_actor_id -> Int4,
-    }
-}
-
-diesel::table! {
-    use diesel::sql_types::*;
-    use super::sql_types::ReasonType;
-    use super::sql_types::SubjectType;
-    use super::sql_types::LabelerStatus;
-
-    labelers (actor_id) {
-        actor_id -> Int4,
-        cid -> Bytea,
-        created_at -> Timestamptz,
-        reasons -> Nullable<Array<Nullable<ReasonType>>>,
-        subject_types -> Nullable<Array<Nullable<SubjectType>>>,
-        subject_collections -> Nullable<Array<Nullable<Text>>>,
-        status -> LabelerStatus,
-        like_count -> Int4,
-    }
-}
-
-diesel::table! {
     labels (labeler_actor_id, label, uri) {
         labeler_actor_id -> Int4,
         label -> Text,
···
         stat_type -> PostStatType,
         value -> Int4,
         updated_at -> Timestamptz,
+    }
+}
+
+diesel::table! {
+    post_likes (actor_id, rkey) {
+        actor_id -> Int4,
+        rkey -> Int8,
+        post_actor_id -> Int4,
+        post_rkey -> Int8,
+        via_repost_actor_id -> Nullable<Int4>,
+        via_repost_rkey -> Nullable<Int8>,
     }
 }
 
···
     follows,
     handle_resolution_queue,
     jetstream_cursors,
-    labeler_defs,
-    labeler_likes,
-    labelers,
     labels,
     list_blocks,
     list_items,
···
     mutes,
     notifications,
     post_aggregate_stats,
+    post_likes,
     posts,
     reposts,
     spatial_ref_sys,

parakeet/src/hydration/labeler.rs

···
 
 fn build_view_detailed(
     enriched: EnrichedLabeler,
-    defs: Vec<models::LabelerDef>,
+    defs: Vec<parakeet_db::composite_types::LabelerDef>,
     creator: ProfileView,
     labels: Vec<models::Label>,
     viewer: Option<LabelerViewerState>,
     likes: Option<i32>,
 ) -> LabelerViewDetailed {
-    let reason_types = enriched.labeler.reasons.map(|v| {
+    let reason_types = enriched.reasons.map(|v| {
         v.iter()
+            .flatten()
             .filter_map(|v| ReasonType::from_str(&v.to_string()).ok())
             .collect()
     });
···
             })
         })
         .collect();
-    let subject_types = enriched.labeler.subject_types.map(|v| {
+    let subject_types = enriched.subject_types.map(|v| {
         v.iter()
+            .flatten()
             .filter_map(|v| SubjectType::from_str(&v.to_string()).ok())
             .collect()
     });
-    let subject_collections = enriched.labeler.subject_collections.map(|v| {
-        v.iter().map(|rt| rt.to_string()).collect()
+    let subject_collections = enriched.subject_collections.map(|v| {
+        v.iter().flatten().map(|rt| rt.to_string()).collect()
     });
 
     LabelerViewDetailed {

parakeet/src/loaders/labeler.rs

···
 use parakeet_db::{models, schema};
 use std::collections::HashMap;
 
-/// Build SQL query for loading labeler record metadata
+/// Build SQL query for loading labeler record metadata from actors table
 ///
 /// This function is public for testing purposes.
 pub fn build_labeler_records_query(actor_ids_str: &str) -> String {
     format!(
-        "SELECT actor_id, cid, created_at, like_count
-         FROM labelers
-         WHERE actor_id IN ({})",
+        "SELECT id as actor_id, labeler_cid as cid, labeler_created_at as created_at, labeler_like_count as like_count
+         FROM actors
+         WHERE id IN ({}) AND labeler_cid IS NOT NULL",
         actor_ids_str
     )
 }
···
          ORDER BY l.created_at"
 }
 
-// Enriched Labeler with reconstructed fields
+// Enriched Labeler with reconstructed fields from Actor
+// Note: Labeler data is now stored directly on actors table with labeler_* columns
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct EnrichedLabeler {
-    pub labeler: models::Labeler,
+    pub actor_id: i32,
     pub did: String,
     pub cid: Vec<u8>,
     pub created_at: chrono::DateTime<chrono::Utc>,
-    pub like_count: i32,     // Count of likes from labeler_likes table
+    pub reasons: Option<Vec<Option<parakeet_db::types::ReasonType>>>,
+    pub subject_types: Option<Vec<Option<parakeet_db::types::SubjectType>>>,
+    pub subject_collections: Option<Vec<Option<String>>>,
+    pub status: parakeet_db::types::LabelerStatus,
+    pub like_count: i32,
 }
 
 pub struct LabelServiceLoader(pub(super) Pool<AsyncPgConnection>);
-pub type LabelServiceLoaderRet = (EnrichedLabeler, Vec<models::LabelerDef>);
+pub type LabelServiceLoaderRet = (EnrichedLabeler, Vec<parakeet_db::composite_types::LabelerDef>);
 impl BatchFn<String, LabelServiceLoaderRet> for LabelServiceLoader {
     async fn load(&mut self, keys: &[String]) -> HashMap<String, LabelServiceLoaderRet> {
         let mut conn = self.0.get().await.unwrap();
 
-        // Load labelers using Diesel DSL (loads from labelers table directly)
-        let labelers: Vec<models::Labeler> = diesel_async::RunQueryDsl::load(
-            schema::labelers::table
-                .inner_join(schema::actors::table.on(schema::labelers::actor_id.eq(schema::actors::id)))
+        // Load labelers from actors table (actors with labeler_cid IS NOT NULL)
+        let actors: Vec<models::Actor> = diesel_async::RunQueryDsl::load(
+            schema::actors::table
                 .filter(schema::actors::did.eq_any(keys))
-                .filter(schema::labelers::status.eq(parakeet_db::types::LabelerStatus::Complete))
-                .select(models::Labeler::as_select()),
+                .filter(schema::actors::labeler_cid.is_not_null())
+                .filter(schema::actors::labeler_status.eq(parakeet_db::types::LabelerStatus::Complete))
+                .select(models::Actor::as_select()),
             &mut conn,
         )
         .await
···
             vec![]
         });
 
-        if labelers.is_empty() {
+        if actors.is_empty() {
             return HashMap::new();
         }
 
-        // Get actor IDs for loading other data
-        let labeler_actor_ids: Vec<i32> = labelers.iter().map(|l| l.actor_id).collect();
-
-        // Load DIDs for these labelers
-        let actors: Vec<(i32, String)> = diesel_async::RunQueryDsl::load(
-            schema::actors::table
-                .filter(schema::actors::id.eq_any(&labeler_actor_ids))
-                .select((schema::actors::id, schema::actors::did)),
-            &mut conn,
-        )
-        .await
-        .unwrap_or_default();
-        let did_by_actor: HashMap<i32, String> = actors.into_iter().collect();
-
-        // Load record metadata (cid, created_at) for these labelers
-        // Using raw SQL because RecordType doesn't implement Clone
-        use diesel::sql_types::{Binary, Integer, Timestamptz};
-
-        let actor_ids_str = labeler_actor_ids
-            .iter()
-            .map(|id| id.to_string())
-            .collect::<Vec<_>>()
-            .join(",");
-
-        let query = build_labeler_records_query(&actor_ids_str);
-
-        #[derive(diesel::QueryableByName)]
-        struct RecordRow {
-            #[diesel(sql_type = Integer)]
-            actor_id: i32,
-            #[diesel(sql_type = Binary)]
-            cid: Vec<u8>,
-            #[diesel(sql_type = Timestamptz)]
-            created_at: chrono::DateTime<chrono::Utc>,
-            #[diesel(sql_type = Integer)]
-            like_count: i32,
-        }
-
-        let records: Vec<RecordRow> = diesel_async::RunQueryDsl::load(
-            diesel::sql_query(query),
-            &mut conn,
-        )
-        .await
-        .unwrap_or_default();
-        let record_by_actor: HashMap<i32, (Vec<u8>, chrono::DateTime<chrono::Utc>, i32)> =
-            records.into_iter().map(|row| (row.actor_id, (row.cid, row.created_at, row.like_count))).collect();
-
-        // Load label definitions
-        let defs: Vec<models::LabelerDef> = diesel_async::RunQueryDsl::load(
-            schema::labeler_defs::table
-                .filter(schema::labeler_defs::labeler_actor_id.eq_any(&labeler_actor_ids)),
-            &mut conn,
-        )
-        .await
-        .unwrap_or_default();
-
-        // Group definitions by labeler_actor_id
-        let mut defs_by_actor: HashMap<i32, Vec<models::LabelerDef>> = HashMap::new();
-        for def in defs {
-            defs_by_actor.entry(def.labeler_actor_id).or_default().push(def);
-        }
-
         // Build result map: DID -> (EnrichedLabeler, Vec<LabelerDef>)
-        labelers
+        actors
             .into_iter()
-            .filter_map(|labeler| {
-                let actor_id = labeler.actor_id;
-                let did = did_by_actor.get(&actor_id)?.clone();
-                let (cid, created_at, like_count) = record_by_actor.get(&actor_id)?;
-                let defs = defs_by_actor.remove(&actor_id).unwrap_or_default();
+            .filter_map(|actor| {
+                // Extract labeler fields (all should be present if labeler_cid IS NOT NULL)
+                let cid = actor.labeler_cid?;
+                let created_at = actor.labeler_created_at?;
+                let status = actor.labeler_status?;
+                let like_count = actor.labeler_like_count.unwrap_or(0);
+
+                // Extract labeler_defs array and filter out NULLs
+                let defs: Vec<parakeet_db::composite_types::LabelerDef> = actor
+                    .labeler_defs
+                    .unwrap_or_default()
+                    .into_iter()
+                    .flatten()
+                    .collect();
 
                 let enriched = EnrichedLabeler {
-                    labeler,
-                    did: did.clone(),
-                    cid: cid.clone(),
-                    created_at: *created_at,
-                    like_count: *like_count,
+                    actor_id: actor.id,
+                    did: actor.did.clone(),
+                    cid,
+                    created_at,
+                    reasons: actor.labeler_reasons,
+                    subject_types: actor.labeler_subject_types,
+                    subject_collections: actor.labeler_subject_collections,
+                    status,
+                    like_count,
                 };
-                Some((did, (enriched, defs)))
+
+                Some((actor.did, (enriched, defs)))
             })
             .collect()
     }