From f36b61d86bea574510cb4a92908ff7f058a062b5 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Sun, 7 Dec 2025 20:02:21 -0500 Subject: [PATCH] feat: denormalize strategy; labelers --- consumer/src/db/labels.rs | 79 ++++++---- consumer/src/db/operations/labeler.rs | 50 ++++--- consumer/src/db/sql/label_service_upsert.sql | 26 ++-- .../down.sql | 70 +++++++++ .../up.sql | 53 +++++++ parakeet-db/src/composite_types.rs | 52 ++++++- parakeet-db/src/models.rs | 60 +++----- parakeet-db/src/schema.rs | 79 ++++------ parakeet/src/hydration/labeler.rs | 12 +- parakeet/src/loaders/labeler.rs | 135 ++++++------------ 10 files changed, 362 insertions(+), 254 deletions(-) create mode 100644 migrations/2025-12-08-004705_denormalize_labelers/down.sql create mode 100644 migrations/2025-12-08-004705_denormalize_labelers/up.sql diff --git a/consumer/src/db/labels.rs b/consumer/src/db/labels.rs index cd6e6c0d..698c95c2 100644 --- a/consumer/src/db/labels.rs +++ b/consumer/src/db/labels.rs @@ -17,14 +17,8 @@ pub async fn maintain_label_defs( 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 @@ -32,33 +26,58 @@ pub async fn maintain_label_defs( .map(|def| (def.identifier.clone(), def)) .collect::>(); + // 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()); - - let _ = conn - .execute( - include_str!("sql/label_defs_upsert.sql"), - &[ - &labeler_actor_id, - &label, - &severity, - &blurs, - &default_setting, - &adult_only, - &locales, - ], - ) - .await?; + 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())); } + // 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) } diff --git a/consumer/src/db/operations/labeler.rs b/consumer/src/db/operations/labeler.rs index d82bb1ad..6d7ebe60 100644 --- a/consumer/src/db/operations/labeler.rs +++ b/consumer/src/db/operations/labeler.rs @@ -8,13 +8,12 @@ use ipld_core::cid::Cid; /// /// 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( conn: &C, did: &str, @@ -36,17 +35,15 @@ pub async fn ensure_labeler_stub( // 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?; @@ -108,10 +105,19 @@ pub async fn labeler_upsert( pub async fn labeler_delete(conn: &C, actor_id: i32) -> Result { // 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 @@ -122,6 +128,7 @@ pub async fn labeler_delete(conn: &C, actor_id: i32) -> Result /// /// 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( conn: &C, labeler_actor_ids: &[i32], @@ -131,14 +138,14 @@ pub async fn increment_labeler_like_counts( } 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 @@ -148,14 +155,15 @@ pub async fn increment_labeler_like_counts( /// 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( conn: &C, labeler_actor_id: i32, ) -> Result { 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 diff --git a/consumer/src/db/sql/label_service_upsert.sql b/consumer/src/db/sql/label_service_upsert.sql index 41b0cd76..2eb2fabe 100644 --- a/consumer/src/db/sql/label_service_upsert.sql +++ b/consumer/src/db/sql/label_service_upsert.sql @@ -1,16 +1,14 @@ --- 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 diff --git a/migrations/2025-12-08-004705_denormalize_labelers/down.sql b/migrations/2025-12-08-004705_denormalize_labelers/down.sql new file mode 100644 index 00000000..73366a6e --- /dev/null +++ b/migrations/2025-12-08-004705_denormalize_labelers/down.sql @@ -0,0 +1,70 @@ +-- 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; diff --git a/migrations/2025-12-08-004705_denormalize_labelers/up.sql b/migrations/2025-12-08-004705_denormalize_labelers/up.sql new file mode 100644 index 00000000..ef3d13ff --- /dev/null +++ b/migrations/2025-12-08-004705_denormalize_labelers/up.sql @@ -0,0 +1,53 @@ +-- 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; diff --git a/parakeet-db/src/composite_types.rs b/parakeet-db/src/composite_types.rs index a169479a..12cb12e5 100644 --- a/parakeet-db/src/composite_types.rs +++ b/parakeet-db/src/composite_types.rs @@ -25,9 +25,57 @@ 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, diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index f344efb1..7dcb9a9e 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -34,6 +34,7 @@ // // ============================================================================= +use crate::composite_types::LabelerDef; use crate::tid_util::{decode_tid, encode_tid, TidError}; use crate::types::*; use chrono::prelude::*; @@ -151,6 +152,15 @@ pub struct Actor { pub lists_count: Option, pub feeds_count: Option, pub starterpacks_count: Option, + // Labeler fields (from labelers table, denormalized - NULL for non-labelers) + pub labeler_cid: Option>, + pub labeler_created_at: Option>, + pub labeler_reasons: Option>>, + pub labeler_subject_types: Option>>, + pub labeler_subject_collections: Option>>, + pub labeler_status: Option, + pub labeler_like_count: Option, + pub labeler_defs: Option>>, } // AllowlistEntry model removed - allowlist table dropped in favor of actors.sync_state @@ -414,17 +424,9 @@ pub struct FeedgenLike { // 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)] @@ -643,35 +645,9 @@ pub struct ThreadMute { // 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, // 32-byte CID digest - pub created_at: DateTime, // From AT Protocol record - pub reasons: Option, // ENUM array: spam | violation | misleading | etc. - pub subject_types: Option, // ENUM array: account | record | chat - pub subject_collections: Option, // 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, // ENUM: inform | alert | none - pub blurs: Option, // ENUM: content | media | none - pub default_setting: Option, // ENUM: ignore | warn | hide - pub adult_only: bool, - pub locales: Option, - pub created_at: DateTime, -} +// 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 { @@ -825,7 +801,7 @@ pub mod array_helpers { 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); @@ -844,7 +820,7 @@ impl_tid_rkey!(Verification); 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); diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 51752215..9b947043 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -49,6 +49,10 @@ pub mod sql_types { #[diesel(postgres_type(name = "label_severity"))] 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; @@ -130,10 +134,6 @@ pub mod sql_types { #[diesel(postgres_type(name = "list_block_record"))] 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; @@ -204,6 +204,10 @@ diesel::table! { 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, @@ -246,6 +250,14 @@ diesel::table! { lists_count -> Nullable, feeds_count -> Nullable, starterpacks_count -> Nullable, + labeler_cid -> Nullable, + labeler_created_at -> Nullable, + labeler_reasons -> Nullable>>, + labeler_subject_types -> Nullable>>, + labeler_subject_collections -> Nullable>>, + labeler_status -> Nullable, + labeler_like_count -> Nullable, + labeler_defs -> Nullable>>, } } @@ -367,50 +379,6 @@ diesel::table! { } } -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, - blurs -> Nullable, - default_setting -> Nullable, - adult_only -> Bool, - locales -> Nullable, - 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>>, - subject_types -> Nullable>>, - subject_collections -> Nullable>>, - status -> LabelerStatus, - like_count -> Int4, - } -} - diesel::table! { labels (labeler_actor_id, label, uri) { labeler_actor_id -> Int4, @@ -514,6 +482,17 @@ diesel::table! { } } +diesel::table! { + post_likes (actor_id, rkey) { + actor_id -> Int4, + rkey -> Int8, + post_actor_id -> Int4, + post_rkey -> Int8, + via_repost_actor_id -> Nullable, + via_repost_rkey -> Nullable, + } +} + diesel::table! { use diesel::sql_types::*; use super::sql_types::LanguageCode; @@ -686,9 +665,6 @@ diesel::allow_tables_to_appear_in_same_query!( follows, handle_resolution_queue, jetstream_cursors, - labeler_defs, - labeler_likes, - labelers, labels, list_blocks, list_items, @@ -697,6 +673,7 @@ diesel::allow_tables_to_appear_in_same_query!( mutes, notifications, post_aggregate_stats, + post_likes, posts, reposts, spatial_ref_sys, diff --git a/parakeet/src/hydration/labeler.rs b/parakeet/src/hydration/labeler.rs index c57f03c4..b87bc5ac 100644 --- a/parakeet/src/hydration/labeler.rs +++ b/parakeet/src/hydration/labeler.rs @@ -36,14 +36,15 @@ fn build_view( fn build_view_detailed( enriched: EnrichedLabeler, - defs: Vec, + defs: Vec, creator: ProfileView, labels: Vec, viewer: Option, likes: Option, ) -> 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() }); @@ -73,13 +74,14 @@ fn build_view_detailed( }) }) .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 { diff --git a/parakeet/src/loaders/labeler.rs b/parakeet/src/loaders/labeler.rs index 2b95feaa..3b6ff2f3 100644 --- a/parakeet/src/loaders/labeler.rs +++ b/parakeet/src/loaders/labeler.rs @@ -7,14 +7,14 @@ use itertools::Itertools as _; 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 ) } @@ -65,29 +65,34 @@ pub fn build_labels_many_query() -> &'static 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, pub created_at: chrono::DateTime, - pub like_count: i32, // Count of likes from labeler_likes table + pub reasons: Option>>, + pub subject_types: Option>>, + pub subject_collections: Option>>, + pub status: parakeet_db::types::LabelerStatus, + pub like_count: i32, } pub struct LabelServiceLoader(pub(super) Pool); -pub type LabelServiceLoaderRet = (EnrichedLabeler, Vec); +pub type LabelServiceLoaderRet = (EnrichedLabeler, Vec); impl BatchFn for LabelServiceLoader { async fn load(&mut self, keys: &[String]) -> HashMap { let mut conn = self.0.get().await.unwrap(); - // Load labelers using Diesel DSL (loads from labelers table directly) - let labelers: Vec = 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 = 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 @@ -96,89 +101,41 @@ impl BatchFn for LabelServiceLoader { vec![] }); - if labelers.is_empty() { + if actors.is_empty() { return HashMap::new(); } - // Get actor IDs for loading other data - let labeler_actor_ids: Vec = 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 = 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::>() - .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, - #[diesel(sql_type = Timestamptz)] - created_at: chrono::DateTime, - #[diesel(sql_type = Integer)] - like_count: i32, - } - - let records: Vec = diesel_async::RunQueryDsl::load( - diesel::sql_query(query), - &mut conn, - ) - .await - .unwrap_or_default(); - let record_by_actor: HashMap, chrono::DateTime, i32)> = - records.into_iter().map(|row| (row.actor_id, (row.cid, row.created_at, row.like_count))).collect(); - - // Load label definitions - let defs: Vec = 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> = HashMap::new(); - for def in defs { - defs_by_actor.entry(def.labeler_actor_id).or_default().push(def); - } - // Build result map: DID -> (EnrichedLabeler, Vec) - 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 = 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() } -- 2.51.2