From 49aa24c9737d7daba469a8d901378d2c13c41be2 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Sun, 7 Dec 2025 12:50:25 -0500 Subject: [PATCH] feat: denormalize gates; wip --- consumer/src/db/operations/feed/postgate.rs | 56 ++++++---- consumer/src/db/operations/feed/threadgate.rs | 105 ++++++------------ .../db/sql/postgate_upsert_denormalized.sql | 13 +++ .../db/sql/threadgate_upsert_denormalized.sql | 13 +++ 4 files changed, 96 insertions(+), 91 deletions(-) create mode 100644 consumer/src/db/sql/postgate_upsert_denormalized.sql create mode 100644 consumer/src/db/sql/threadgate_upsert_denormalized.sql diff --git a/consumer/src/db/operations/feed/postgate.rs b/consumer/src/db/operations/feed/postgate.rs index 9136f6ad..2065e900 100644 --- a/consumer/src/db/operations/feed/postgate.rs +++ b/consumer/src/db/operations/feed/postgate.rs @@ -13,45 +13,55 @@ use deadpool_postgres::GenericClient; use eyre::Context as _; use ipld_core::cid::Cid; -/// Insert or update a postgate record +/// Insert or update a postgate record (DENORMALIZED into posts table) /// -/// Postgates control embedding permissions for posts. They can specify: -/// - detached_embedding_uris: Specific AT URIs that are not allowed to embed this post -/// - embedding_rules: Global rules like "app.bsky.feed.postgate#disableRule" to block all embeds +/// Postgates control embedding permissions for posts. They are now stored directly +/// in the posts table as denormalized arrays: +/// - postgate_rules: Global rules like "app.bsky.feed.postgate#disableRule" to block all embeds +/// - postgate_detached_actor_ids/rkeys: Specific posts that are not allowed to embed this post /// -/// The postgate's created_at timestamp is derived from the TID rkey. -/// -/// OPTIMIZATION: Resolves all actor DIDs in bulk before calling SQL, +/// OPTIMIZATION: Resolves all detached embedding URIs in bulk before calling SQL, /// avoiding expensive URI parsing and JOINs in the database. +/// +/// NOTE: The postgate's own (actor_id, rkey, cid) are no longer stored separately. +/// We only store the postgate data in the post record itself. pub async fn postgate_upsert( conn: &C, actor_id: i32, rkey: i64, - cid: Cid, + _cid: Cid, // No longer stored - postgates are denormalized rec: &AppBskyFeedPostgate, ) -> Result { - let cid_bytes = cid.to_bytes(); - let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) - .expect("CID must be valid AT Protocol CID"); - let rules = rec .embedding_rules .iter() .map(|v| v.as_str().to_owned()) .collect::>(); - // Parse post URI to get post rkey + // Parse post URI to get post actor_id and rkey + let post_did = parakeet_db::at_uri_util::extract_did(&rec.post) + .ok_or_else(|| eyre::eyre!("Invalid post URI in postgate: missing DID in {}", rec.post))?; let post_rkey_str = parakeet_db::at_uri_util::extract_rkey(&rec.post) .ok_or_else(|| eyre::eyre!("Invalid post URI in postgate: {}", rec.post))?; let post_rkey = parakeet_db::models::tid_to_i64(post_rkey_str) .wrap_err_with(|| format!("Invalid TID in postgate post URI: {}", post_rkey_str))?; + // Look up post_actor_id from DID (should match actor_id in most cases) + let post_actor_id = conn + .query_one( + "SELECT id FROM actors WHERE did = $1", + &[&post_did], + ) + .await + .wrap_err_with(|| format!("Failed to find actor for DID: {}", post_did))? + .get::<_, i32>(0); + // OPTIMIZATION: Resolve detached embedding URIs to (actor_id, rkey) pairs in Rust let (detached_actor_ids, detached_rkeys): (Vec, Vec) = if !rec.detached_embedding_uris.is_empty() { let uri_refs: Vec<&str> = rec.detached_embedding_uris.iter().map(|s| s.as_str()).collect(); let resolved = crate::db::bulk_resolve::resolve_post_uris_bulk(conn, &uri_refs).await?; - // Build parallel arrays for SQL UNNEST + // Build parallel arrays for SQL let mut actor_ids = Vec::new(); let mut rkeys = Vec::new(); for uri in &rec.detached_embedding_uris { @@ -65,13 +75,11 @@ pub async fn postgate_upsert( (vec![], vec![]) }; - // Use optimized SQL with pre-resolved IDs (no JOINs!) + // Update posts table with denormalized postgate data conn.execute( - include_str!("../../sql/postgate_upsert.sql"), + include_str!("../../sql/postgate_upsert_denormalized.sql"), &[ - &actor_id, - &rkey, - &cid_digest, + &post_actor_id, &post_rkey, &rules, &detached_actor_ids, @@ -79,19 +87,23 @@ pub async fn postgate_upsert( ], ) .await - .wrap_err_with(|| format!("Failed to upsert postgate for actor_id:{} rkey:{}", actor_id, rkey)) + .wrap_err_with(|| format!("Failed to upsert postgate for post actor_id:{} rkey:{}", post_actor_id, post_rkey)) } -/// Delete a postgate record +/// Delete a postgate record (DENORMALIZED: clear posts table columns) /// /// Removes embedding controls for a post, allowing all embeds again. +/// Clears the denormalized postgate columns in the posts table. pub async fn postgate_delete( conn: &C, actor_id: i32, rkey: i64, ) -> Result { conn.execute( - "DELETE FROM postgates + "UPDATE posts + SET postgate_rules = NULL, + postgate_detached_actor_ids = NULL, + postgate_detached_rkeys = NULL WHERE actor_id = $1 AND rkey = $2", &[&actor_id, &rkey], diff --git a/consumer/src/db/operations/feed/threadgate.rs b/consumer/src/db/operations/feed/threadgate.rs index 639b5da5..c7ef6661 100644 --- a/consumer/src/db/operations/feed/threadgate.rs +++ b/consumer/src/db/operations/feed/threadgate.rs @@ -61,30 +61,25 @@ pub async fn threadgate_get( Ok(res) } -/// Insert or update a threadgate record +/// Insert or update a threadgate record (DENORMALIZED into posts table) /// -/// Threadgates control reply permissions for posts. They can specify: -/// - allow: Rules like mention, following, or list membership -/// - hidden_replies: Specific reply URIs to hide from the thread -/// - allowed_lists: Lists whose members can reply (extracted from allow rules) +/// Threadgates control reply permissions for posts. They are now stored directly +/// in the posts table as denormalized arrays: +/// - threadgate_allow: Rules like mention, following, or list membership +/// - threadgate_hidden_actor_ids/rkeys: Specific replies to hide from the thread /// -/// The threadgate's created_at timestamp is derived from the TID rkey. -/// -/// OPTIMIZATION: Resolves all actor DIDs and list IDs in bulk before calling SQL, +/// OPTIMIZATION: Resolves all hidden reply URIs in bulk before calling SQL, /// avoiding expensive URI parsing and JOINs in the database. +/// +/// NOTE: The threadgate's own (actor_id, rkey, cid) are no longer stored separately. +/// We only store the threadgate data in the post record itself. pub async fn threadgate_upsert( conn: &C, actor_id: i32, rkey: i64, - cid: Cid, + _cid: Cid, // No longer stored - threadgates are denormalized rec: AppBskyFeedThreadgate, ) -> Result { - let cid_bytes = cid.to_bytes(); - let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) - .expect("CID must be valid AT Protocol CID"); - - let record = serde_json::to_value(&rec).unwrap(); - // Extract allow rules as strings let allow = rec.allow.as_ref().map(|allow| { allow @@ -93,18 +88,32 @@ pub async fn threadgate_upsert( .collect::>() }); - // Parse post URI to get post rkey + // Parse post URI to get post actor_id and rkey + // Note: The post being protected is usually the same as the threadgate's actor, + // but we extract it from the URI to be safe + let post_did = parakeet_db::at_uri_util::extract_did(&rec.post) + .ok_or_else(|| eyre::eyre!("Invalid post URI in threadgate: missing DID in {}", rec.post))?; let post_rkey_str = parakeet_db::at_uri_util::extract_rkey(&rec.post) .ok_or_else(|| eyre::eyre!("Invalid post URI in threadgate: {}", rec.post))?; let post_rkey = parakeet_db::models::tid_to_i64(post_rkey_str) .wrap_err_with(|| format!("Invalid TID in threadgate post URI: {}", post_rkey_str))?; + // Look up post_actor_id from DID (should match actor_id in most cases) + let post_actor_id = conn + .query_one( + "SELECT id FROM actors WHERE did = $1", + &[&post_did], + ) + .await + .wrap_err_with(|| format!("Failed to find actor for DID: {}", post_did))? + .get::<_, i32>(0); + // OPTIMIZATION: Resolve hidden reply URIs to (actor_id, rkey) pairs in Rust let (hidden_actor_ids, hidden_rkeys): (Vec, Vec) = if !rec.hidden_replies.is_empty() { let uri_refs: Vec<&str> = rec.hidden_replies.iter().map(|s| s.as_str()).collect(); let resolved = crate::db::bulk_resolve::resolve_post_uris_bulk(conn, &uri_refs).await?; - // Build parallel arrays for SQL UNNEST + // Build parallel arrays for SQL let mut actor_ids = Vec::new(); let mut rkeys = Vec::new(); for uri in &rec.hidden_replies { @@ -118,77 +127,35 @@ pub async fn threadgate_upsert( (vec![], vec![]) }; - // OPTIMIZATION: Resolve allowed list URIs to list IDs in Rust - let list_ids: Vec = if let Some(ref allow_rules) = rec.allow { - let list_uris: Vec = allow_rules - .iter() - .filter_map(|rule| match rule { - ThreadgateRule::List { list } => Some(list.clone()), - _ => None, - }) - .collect(); - - if !list_uris.is_empty() { - // Parse list URIs to get (actor DID, rkey) pairs - let mut list_ids = Vec::new(); - for uri in &list_uris { - // Extract DID and rkey from list URI - let did = parakeet_db::at_uri_util::extract_did(uri) - .ok_or_else(|| eyre::eyre!("Invalid list URI: missing DID in {}", uri))?; - let rkey_str = parakeet_db::at_uri_util::extract_rkey(uri) - .ok_or_else(|| eyre::eyre!("Invalid list URI: missing rkey in {}", uri))?; - - // Look up list by (actor DID, rkey) - let list_row = conn - .query_opt( - "SELECT l.id FROM lists l - INNER JOIN actors a ON a.id = l.actor_id - WHERE a.did = $1 AND l.rkey = $2", - &[&did, &rkey_str], - ) - .await?; - - if let Some(row) = list_row { - list_ids.push(row.get(0)); - } - } - list_ids - } else { - vec![] - } - } else { - vec![] - }; - - // Use optimized SQL with pre-resolved IDs (no JOINs!) + // Update posts table with denormalized threadgate data conn.execute( - include_str!("../../sql/threadgate_upsert.sql"), + include_str!("../../sql/threadgate_upsert_denormalized.sql"), &[ - &actor_id, - &rkey, - &cid_digest, + &post_actor_id, &post_rkey, &allow, - &record, &hidden_actor_ids, &hidden_rkeys, - &list_ids, ], ) .await - .wrap_err_with(|| format!("Failed to upsert threadgate for actor_id:{} rkey:{}", actor_id, rkey)) + .wrap_err_with(|| format!("Failed to upsert threadgate for post actor_id:{} rkey:{}", post_actor_id, post_rkey)) } -/// Delete a threadgate record +/// Delete a threadgate record (DENORMALIZED: clear posts table columns) /// /// Removes reply controls for a post, allowing all replies again. +/// Clears the denormalized threadgate columns in the posts table. pub async fn threadgate_delete( conn: &C, actor_id: i32, rkey: i64, ) -> Result { conn.execute( - "DELETE FROM threadgates + "UPDATE posts + SET threadgate_allow = NULL, + threadgate_hidden_actor_ids = NULL, + threadgate_hidden_rkeys = NULL WHERE actor_id = $1 AND rkey = $2", &[&actor_id, &rkey], diff --git a/consumer/src/db/sql/postgate_upsert_denormalized.sql b/consumer/src/db/sql/postgate_upsert_denormalized.sql new file mode 100644 index 00000000..ddd3945a --- /dev/null +++ b/consumer/src/db/sql/postgate_upsert_denormalized.sql @@ -0,0 +1,13 @@ +-- Denormalized postgate upsert: Update posts table directly +-- Postgates control embedding permissions - now stored in posts table +-- Parameters: $1=actor_id, $2=post_rkey, $3=rules[], $4=detached_actor_ids[], $5=detached_rkeys[] +-- Note: We update the post record ($1, $2) with postgate data +-- The postgate's own (actor_id, rkey, cid) are no longer stored separately + +UPDATE posts +SET + postgate_rules = $3::text[]::postgate_rule[], + postgate_detached_actor_ids = $4::integer[], + postgate_detached_rkeys = $5::bigint[] +WHERE actor_id = $1 + AND rkey = $2 diff --git a/consumer/src/db/sql/threadgate_upsert_denormalized.sql b/consumer/src/db/sql/threadgate_upsert_denormalized.sql new file mode 100644 index 00000000..44cc9e11 --- /dev/null +++ b/consumer/src/db/sql/threadgate_upsert_denormalized.sql @@ -0,0 +1,13 @@ +-- Denormalized threadgate upsert: Update posts table directly +-- Threadgates control reply permissions - now stored in posts table +-- Parameters: $1=actor_id, $2=post_rkey, $3=allow[], $4=hidden_actor_ids[], $5=hidden_rkeys[] +-- Note: We update the post record ($1, $2) with threadgate data +-- The threadgate's own (actor_id, rkey, cid) are no longer stored separately + +UPDATE posts +SET + threadgate_allow = $3::text[]::threadgate_rule[], + threadgate_hidden_actor_ids = $4::integer[], + threadgate_hidden_rkeys = $5::bigint[] +WHERE actor_id = $1 + AND rkey = $2 -- 2.51.2