diff --git a/consumer/src/db/bulk_resolve/mod.rs b/consumer/src/db/bulk_resolve/mod.rs index 4263b5d1..7bc83454 100644 --- a/consumer/src/db/bulk_resolve/mod.rs +++ b/consumer/src/db/bulk_resolve/mod.rs @@ -64,14 +64,19 @@ pub async fn resolve_actor_dids_bulk( Ok(result) } -/// Resolve multiple post AT-URIs to post_ids in a single query +/// Resolve multiple post AT-URIs to natural keys (actor_id, rkey) /// -/// Returns a HashMap of AT-URI → post_id for all found posts. -/// Does NOT create stubs (use create_post_stubs_bulk for that). +/// Returns a HashMap of AT-URI → (actor_id, rkey) for all URIs. +/// Does NOT check if posts exist - with natural keys, we can reference posts +/// that don't exist yet (no FK constraint). +/// +/// This function only ensures actors exist (creates stubs if needed), then +/// returns natural key pairs based on parsing the URIs. /// /// # Performance /// -/// Single query joining actors and posts using `WHERE (a.did, p.rkey) = ANY(...)`. +/// Much faster than the old approach - only needs to resolve actor DIDs, +/// not query the posts table. pub async fn resolve_post_uris_bulk( conn: &C, at_uris: &[&str], @@ -81,8 +86,8 @@ pub async fn resolve_post_uris_bulk( } // Parse URIs to extract (did, rkey) pairs - let mut did_rkey_pairs: Vec<(String, i64)> = Vec::with_capacity(at_uris.len()); let mut uri_to_did_rkey: HashMap = HashMap::new(); + let mut dids_set: std::collections::HashSet = std::collections::HashSet::new(); for uri in at_uris { let did = parakeet_db::at_uri_util::extract_did(uri) @@ -92,41 +97,34 @@ pub async fn resolve_post_uris_bulk( let rkey_i64 = parakeet_db::models::tid_to_i64(rkey) .map_err(|e| eyre::eyre!("Invalid TID in AT URI {}: {}", uri, e))?; - did_rkey_pairs.push((did.to_string(), rkey_i64)); uri_to_did_rkey.insert(uri.to_string(), (did.to_string(), rkey_i64)); + dids_set.insert(did.to_string()); } - // PostgreSQL doesn't support direct ANY with tuple arrays, so we use UNNEST - // to create a temporary table of (did, rkey) pairs and join against it - let dids: Vec = did_rkey_pairs.iter().map(|(d, _)| d.clone()).collect(); - let rkeys: Vec = did_rkey_pairs.iter().map(|(_, r)| *r).collect(); + // Resolve all actor DIDs to actor_ids (creates stubs for missing actors) + let dids_vec: Vec<&str> = dids_set.iter().map(|s| s.as_str()).collect(); + let did_to_actor_id = resolve_actor_dids_bulk(conn, &dids_vec).await?; - let rows = conn - .query( - "SELECT a.did, p.rkey, p.actor_id - FROM actors a - INNER JOIN posts p ON p.actor_id = a.id - WHERE (a.did, p.rkey) IN ( - SELECT UNNEST($1::text[]), UNNEST($2::bigint[]) - )", - &[&dids, &rkeys], - ) - .await?; + // Create actor stubs for any missing DIDs + let missing_dids: Vec<&str> = dids_vec + .iter() + .filter(|did| !did_to_actor_id.contains_key(**did)) + .copied() + .collect(); - // Build reverse mapping: (did, rkey) → (actor_id, rkey) - let mut did_rkey_to_natural_key: HashMap<(String, i64), (i32, i64)> = HashMap::new(); - for row in rows { - let did: String = row.get(0); - let rkey: i64 = row.get(1); - let actor_id: i32 = row.get(2); - did_rkey_to_natural_key.insert((did, rkey), (actor_id, rkey)); + let mut did_to_actor_id = did_to_actor_id; + if !missing_dids.is_empty() { + let created = create_actor_stubs_bulk(conn, &missing_dids).await?; + did_to_actor_id.extend(created); } - // Map back to URIs + // Map URIs to natural keys let mut result = HashMap::new(); for (uri, (did, rkey)) in uri_to_did_rkey { - if let Some(&natural_key) = did_rkey_to_natural_key.get(&(did, rkey)) { - result.insert(uri, natural_key); + if let Some(&actor_id) = did_to_actor_id.get(&did) { + result.insert(uri, (actor_id, rkey)); + } else { + return Err(eyre::eyre!("Failed to resolve actor DID {} for URI {}", did, uri)); } } @@ -172,126 +170,9 @@ pub async fn create_actor_stubs_bulk( Ok(result) } -/// Create post stubs for missing AT-URIs in bulk using UNNEST -/// -/// This creates post records with status='stub' and content=NULL -/// for all URIs that don't already exist. -/// -/// Returns a HashMap of AT-URI → post_id for newly created stubs. -/// -/// # Input Format -/// -/// Each tuple is (at_uri, cid_str) where: -/// - at_uri: Full AT URI (at://did/collection/rkey) -/// - cid_str: CID string (will be parsed to get digest) -/// -/// # Performance -/// -/// Single INSERT with UNNEST instead of N individual inserts. -/// Uses ON CONFLICT DO NOTHING for idempotency. -pub async fn create_post_stubs_bulk( - conn: &C, - uri_cid_pairs: &[(&str, &str)], -) -> Result> { - if uri_cid_pairs.is_empty() { - return Ok(HashMap::new()); - } - - // Parse all URIs and CIDs upfront - let mut actor_ids_vec: Vec = Vec::with_capacity(uri_cid_pairs.len()); - let mut rkeys_vec: Vec = Vec::with_capacity(uri_cid_pairs.len()); - let mut cid_digests: Vec> = Vec::with_capacity(uri_cid_pairs.len()); - let mut uris_vec: Vec = Vec::with_capacity(uri_cid_pairs.len()); - - // First, collect all DIDs to resolve in bulk - let dids: Vec<&str> = uri_cid_pairs - .iter() - .map(|(uri, _)| { - parakeet_db::at_uri_util::extract_did(uri) - .ok_or_else(|| eyre::eyre!("Invalid AT URI: missing DID in {}", uri)) - }) - .collect::>>()?; - - // Resolve all actor_ids in one query - let did_to_actor = resolve_actor_dids_bulk(conn, &dids).await?; - - // If any actors are missing, create them - let missing_dids: Vec<&str> = dids - .iter() - .filter(|&&did| !did_to_actor.contains_key(did)) - .copied() - .collect(); - - let mut did_to_actor = did_to_actor; - if !missing_dids.is_empty() { - let created = create_actor_stubs_bulk(conn, &missing_dids).await?; - did_to_actor.extend(created); - } - - // Now process all URI/CID pairs - for (uri, cid_str) in uri_cid_pairs { - let did = parakeet_db::at_uri_util::extract_did(uri) - .ok_or_else(|| eyre::eyre!("Invalid AT URI: missing DID in {}", uri))?; - let rkey = parakeet_db::at_uri_util::extract_rkey(uri) - .ok_or_else(|| eyre::eyre!("Invalid AT URI: missing rkey in {}", uri))?; - let rkey_i64 = parakeet_db::models::tid_to_i64(rkey) - .map_err(|e| eyre::eyre!("Invalid TID in AT URI {}: {}", uri, e))?; - - let actor_id = *did_to_actor - .get(did) - .ok_or_else(|| eyre::eyre!("Actor not found for DID {} (should have been created)", did))?; - - // Parse CID - let cid = ipld_core::cid::Cid::try_from(*cid_str) - .map_err(|e| eyre::eyre!("Invalid CID format {}: {}", cid_str, e))?; - let cid_bytes = cid.to_bytes(); - let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) - .ok_or_else(|| eyre::eyre!("CID must be valid AT Protocol CID"))?; - - actor_ids_vec.push(actor_id); - rkeys_vec.push(rkey_i64); - cid_digests.push(cid_digest.to_vec()); - uris_vec.push(uri.to_string()); - } - - // Bulk insert post stubs - let rows = conn - .query( - "WITH stub_data AS ( - SELECT - UNNEST($1::int[]) as actor_id, - UNNEST($2::bigint[]) as rkey, - UNNEST($3::bytea[]) as cid - ) - INSERT INTO posts (actor_id, rkey, cid, content, status) - SELECT actor_id, rkey, cid, NULL, 'stub'::post_status - FROM stub_data - ON CONFLICT (actor_id, rkey) DO NOTHING - RETURNING actor_id, rkey", - &[&actor_ids_vec, &rkeys_vec, &cid_digests], - ) - .await?; - - // Build mapping: (actor_id, rkey) → (actor_id, rkey) - let mut actor_rkey_set: std::collections::HashSet<(i32, i64)> = std::collections::HashSet::new(); - for row in rows { - let actor_id: i32 = row.get(0); - let rkey: i64 = row.get(1); - actor_rkey_set.insert((actor_id, rkey)); - } - - // Map back to URIs (using our pre-parsed data) - let mut result = HashMap::new(); - for (i, uri) in uris_vec.iter().enumerate() { - let actor_id = actor_ids_vec[i]; - let rkey = rkeys_vec[i]; - if actor_rkey_set.contains(&(actor_id, rkey)) { - result.insert(uri.clone(), (actor_id, rkey)); - } - } - - Ok(result) -} +// NOTE: create_post_stubs_bulk() removed - with natural keys, we don't need to create +// post stubs anymore. Posts can be referenced by (actor_id, rkey) even if they don't +// exist yet, since there's no FK constraint. /// Helper: Find URIs that are missing from the resolved map pub fn find_missing_uris<'a>( @@ -317,6 +198,11 @@ pub fn find_missing_uris<'a>( /// - uris_with_cids: List of (at_uri, cid_str) pairs /// /// Returns a HashMap mapping every input URI to (actor_id, rkey). +/// Resolve post URIs with CIDs to natural keys +/// +/// This is a convenience wrapper around resolve_post_uris_bulk that accepts +/// URIs with CIDs. The CIDs are currently ignored (we don't create stubs anymore), +/// but keeping this function for backward compatibility with callers. pub async fn resolve_and_ensure_posts_bulk( conn: &C, uris_with_cids: &[(&str, &str)], @@ -326,24 +212,7 @@ pub async fn resolve_and_ensure_posts_bulk( } let uris: Vec<&str> = uris_with_cids.iter().map(|(uri, _)| *uri).collect(); - - // Resolve existing posts - let mut resolved = resolve_post_uris_bulk(conn, &uris).await?; - - // Find missing posts - let missing: Vec<(&str, &str)> = uris_with_cids - .iter() - .filter(|&&(uri, _)| !resolved.contains_key(uri)) - .copied() - .collect(); - - // Create stubs for missing posts - if !missing.is_empty() { - let created = create_post_stubs_bulk(conn, &missing).await?; - resolved.extend(created); - } - - Ok(resolved) + resolve_post_uris_bulk(conn, &uris).await } /// Resolve multiple feedgen AT-URIs to feedgen_ids in a single query @@ -449,14 +318,19 @@ pub async fn resolve_labeler_dids_bulk( Ok(result) } -/// Resolve multiple repost AT-URIs to repost_ids in a single query +/// Resolve multiple repost AT-URIs to natural keys (actor_id, rkey) /// -/// Returns a HashMap of AT-URI → repost_id for all found reposts. -/// Does NOT create stubs (use create_repost_stubs_bulk for missing reposts). +/// Returns a HashMap of AT-URI → (actor_id, rkey) for all URIs. +/// Does NOT check if reposts exist - with natural keys, we can reference reposts +/// that don't exist yet (no FK constraint). +/// +/// This function only ensures actors exist (creates stubs if needed), then +/// returns natural key pairs based on parsing the URIs. /// /// # Performance /// -/// Single query joining actors and reposts using `WHERE (a.did, r.rkey) = ANY(...)`. +/// Much faster than the old approach - only needs to resolve actor DIDs, +/// not query the reposts table. pub async fn resolve_repost_uris_bulk( conn: &C, at_uris: &[&str], @@ -466,8 +340,8 @@ pub async fn resolve_repost_uris_bulk( } // Parse URIs to extract (did, rkey) pairs - let mut did_rkey_pairs: Vec<(String, i64)> = Vec::with_capacity(at_uris.len()); let mut uri_to_did_rkey: HashMap = HashMap::new(); + let mut dids_set: std::collections::HashSet = std::collections::HashSet::new(); for uri in at_uris { let did = parakeet_db::at_uri_util::extract_did(uri) @@ -477,167 +351,50 @@ pub async fn resolve_repost_uris_bulk( let rkey_i64 = parakeet_db::models::tid_to_i64(rkey) .map_err(|e| eyre::eyre!("Invalid TID in AT URI {}: {}", uri, e))?; - did_rkey_pairs.push((did.to_string(), rkey_i64)); uri_to_did_rkey.insert(uri.to_string(), (did.to_string(), rkey_i64)); + dids_set.insert(did.to_string()); } - // Use UNNEST to create temporary table and join - let dids: Vec = did_rkey_pairs.iter().map(|(d, _)| d.clone()).collect(); - let rkeys: Vec = did_rkey_pairs.iter().map(|(_, r)| *r).collect(); - - let rows = queries::resolve_repost_uris(conn, &dids, &rkeys).await?; - - // Build reverse mapping: (did, rkey) → (actor_id, rkey) - let mut did_rkey_to_natural_key: HashMap<(String, i64), (i32, i64)> = HashMap::new(); - for (did, rkey, actor_id) in rows { - did_rkey_to_natural_key.insert((did.clone(), rkey), (actor_id, rkey)); - } - - // Map back to URIs - let mut result = HashMap::new(); - for (uri, (did, rkey)) in uri_to_did_rkey { - if let Some(&natural_key) = did_rkey_to_natural_key.get(&(did, rkey)) { - result.insert(uri, natural_key); - } - } - - Ok(result) -} - -/// Create repost stubs for missing AT-URIs in bulk using UNNEST -/// -/// This creates repost records with status='stub' for all URIs that don't already exist. -/// Each repost stub needs: -/// - actor_id (resolved from repost author DID) -/// - rkey (from repost URI) -/// - post_id (resolved from the repost's subject URI) -/// - cid (from the repost CID) -/// -/// Returns a HashMap of AT-URI → repost_id for newly created stubs. -/// -/// # Input Format -/// -/// Each tuple is (repost_at_uri, repost_cid_str, subject_post_uri, subject_post_cid_str) where: -/// - repost_at_uri: Full AT URI of the repost (at://did/app.bsky.feed.repost/rkey) -/// - repost_cid_str: CID string for the repost record -/// - subject_post_uri: AT URI of the post being reposted -/// - subject_post_cid_str: CID string of the post being reposted -/// -/// # Performance -/// -/// Single INSERT with UNNEST instead of N individual inserts. -/// Uses ON CONFLICT DO NOTHING for idempotency. -pub async fn create_repost_stubs_bulk( - conn: &C, - repost_data: &[(&str, &str, &str, &str)], -) -> Result> { - if repost_data.is_empty() { - return Ok(HashMap::new()); - } - - // Parse all URIs and CIDs upfront - let mut actor_ids_vec: Vec = Vec::with_capacity(repost_data.len()); - let mut rkeys_vec: Vec = Vec::with_capacity(repost_data.len()); - let mut post_actor_ids_vec: Vec = Vec::with_capacity(repost_data.len()); - let mut post_rkeys_vec: Vec = Vec::with_capacity(repost_data.len()); - let mut cid_digests: Vec> = Vec::with_capacity(repost_data.len()); - let mut uris_vec: Vec = Vec::with_capacity(repost_data.len()); - - // First, collect all repost author DIDs and subject post URIs - let repost_dids: Vec<&str> = repost_data - .iter() - .map(|(repost_uri, _, _, _)| { - parakeet_db::at_uri_util::extract_did(repost_uri) - .ok_or_else(|| eyre::eyre!("Invalid repost URI: missing DID in {}", repost_uri)) - }) - .collect::>>()?; + // Resolve all actor DIDs to actor_ids (creates stubs for missing actors) + let dids_vec: Vec<&str> = dids_set.iter().map(|s| s.as_str()).collect(); + let did_to_actor_id = resolve_actor_dids_bulk(conn, &dids_vec).await?; - let subject_post_pairs: Vec<(&str, &str)> = repost_data + // Create actor stubs for any missing DIDs + let missing_dids: Vec<&str> = dids_vec .iter() - .map(|(_, _, subject_uri, subject_cid)| (*subject_uri, *subject_cid)) - .collect(); - - // Resolve all repost authors (create actor stubs if needed) - let mut did_to_actor = resolve_actor_dids_bulk(conn, &repost_dids).await?; - let missing_dids: Vec<&str> = repost_dids - .iter() - .filter(|&&did| !did_to_actor.contains_key(did)) + .filter(|did| !did_to_actor_id.contains_key(**did)) .copied() .collect(); + + let mut did_to_actor_id = did_to_actor_id; if !missing_dids.is_empty() { let created = create_actor_stubs_bulk(conn, &missing_dids).await?; - did_to_actor.extend(created); - } - - // Resolve all subject posts (create post stubs if needed) - let post_uri_to_natural_key = resolve_and_ensure_posts_bulk(conn, &subject_post_pairs).await?; - - // Now process all repost data - for (repost_uri, repost_cid_str, subject_uri, _subject_cid_str) in repost_data { - let did = parakeet_db::at_uri_util::extract_did(repost_uri) - .ok_or_else(|| eyre::eyre!("Invalid repost URI: missing DID in {}", repost_uri))?; - let rkey = parakeet_db::at_uri_util::extract_rkey(repost_uri) - .ok_or_else(|| eyre::eyre!("Invalid repost URI: missing rkey in {}", repost_uri))?; - let rkey_i64 = parakeet_db::models::tid_to_i64(rkey) - .map_err(|e| eyre::eyre!("Invalid TID in repost URI {}: {}", repost_uri, e))?; - - let actor_id = *did_to_actor - .get(did) - .ok_or_else(|| eyre::eyre!("Actor not found for DID {} (should have been created)", did))?; - - let (post_actor_id, post_rkey) = *post_uri_to_natural_key - .get(*subject_uri) - .ok_or_else(|| eyre::eyre!("Post not found for URI {} (should have been created)", subject_uri))?; - - // Parse repost CID - let cid = ipld_core::cid::Cid::try_from(*repost_cid_str) - .map_err(|e| eyre::eyre!("Invalid repost CID format {}: {}", repost_cid_str, e))?; - let cid_bytes = cid.to_bytes(); - let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) - .ok_or_else(|| eyre::eyre!("Repost CID must be valid AT Protocol CID"))?; - - actor_ids_vec.push(actor_id); - rkeys_vec.push(rkey_i64); - post_actor_ids_vec.push(post_actor_id); - post_rkeys_vec.push(post_rkey); - cid_digests.push(cid_digest.to_vec()); - uris_vec.push(repost_uri.to_string()); - } - - // Bulk insert repost stubs - let rows = queries::create_repost_stubs(conn, &actor_ids_vec, &rkeys_vec, &post_actor_ids_vec, &post_rkeys_vec, &cid_digests).await?; - - // Build mapping: (actor_id, rkey) → () - let mut actor_rkey_set: std::collections::HashSet<(i32, i64)> = std::collections::HashSet::new(); - for (actor_id, rkey) in rows { - actor_rkey_set.insert((actor_id, rkey)); + did_to_actor_id.extend(created); } - // Map back to URIs (using our pre-parsed data) + // Map URIs to natural keys let mut result = HashMap::new(); - for (i, uri) in uris_vec.iter().enumerate() { - let actor_id = actor_ids_vec[i]; - let rkey = rkeys_vec[i]; - if actor_rkey_set.contains(&(actor_id, rkey)) { - result.insert(uri.clone(), (actor_id, rkey)); + for (uri, (did, rkey)) in uri_to_did_rkey { + if let Some(&actor_id) = did_to_actor_id.get(&did) { + result.insert(uri, (actor_id, rkey)); + } else { + return Err(eyre::eyre!("Failed to resolve actor DID {} for URI {}", did, uri)); } } Ok(result) } -/// Resolve and ensure all repost URIs exist (resolve + create stubs for missing) -/// -/// This is a convenience function that: -/// 1. Resolves all existing reposts -/// 2. Creates stubs for missing reposts -/// 3. Returns complete URI → (actor_id, rkey) mapping -/// -/// # Input Format -/// -/// - repost_data: List of (repost_at_uri, repost_cid_str, subject_post_uri, subject_post_cid_str) +// NOTE: create_repost_stubs_bulk() removed - with natural keys, we don't need to create +// repost stubs anymore. Reposts can be referenced by (actor_id, rkey) even if they don't +// exist yet, since there's no FK constraint. + +/// Resolve repost URIs to natural keys /// -/// Returns a HashMap mapping every input repost URI to (actor_id, rkey). +/// This is a convenience wrapper around resolve_repost_uris_bulk that accepts +/// reposts with extra CID/post data. The CIDs and post URIs are currently ignored +/// (we don't create stubs anymore), but keeping this function for backward +/// compatibility with callers. pub async fn resolve_and_ensure_reposts_bulk( conn: &C, repost_data: &[(&str, &str, &str, &str)], @@ -647,24 +404,7 @@ pub async fn resolve_and_ensure_reposts_bulk( } let uris: Vec<&str> = repost_data.iter().map(|(uri, _, _, _)| *uri).collect(); - - // Resolve existing reposts - let mut resolved = resolve_repost_uris_bulk(conn, &uris).await?; - - // Find missing reposts - let missing: Vec<(&str, &str, &str, &str)> = repost_data - .iter() - .filter(|&&(uri, _, _, _)| !resolved.contains_key(uri)) - .copied() - .collect(); - - // Create stubs for missing reposts - if !missing.is_empty() { - let created = create_repost_stubs_bulk(conn, &missing).await?; - resolved.extend(created); - } - - Ok(resolved) + resolve_repost_uris_bulk(conn, &uris).await } /// Bulk resolve link URIs to uri_ids diff --git a/consumer/src/db/bulk_resolve/queries.rs b/consumer/src/db/bulk_resolve/queries.rs index 9c42c689..6891c996 100644 --- a/consumer/src/db/bulk_resolve/queries.rs +++ b/consumer/src/db/bulk_resolve/queries.rs @@ -37,42 +37,6 @@ pub async fn resolve_repost_uris( Ok(result) } -/// SQL query to create repost stubs in bulk -/// -/// Returns rows of (actor_id, rkey) for newly created stubs. -pub async fn create_repost_stubs( - conn: &C, - actor_ids: &[i32], - rkeys: &[i64], - post_actor_ids: &[i32], - post_rkeys: &[i64], - cid_digests: &[Vec], -) -> QueryResult> { - let rows = conn - .query( - "WITH stub_data AS ( - SELECT - UNNEST($1::int[]) as actor_id, - UNNEST($2::bigint[]) as rkey, - UNNEST($3::int[]) as post_actor_id, - UNNEST($4::bigint[]) as post_rkey, - UNNEST($5::bytea[]) as cid - ) - INSERT INTO reposts (actor_id, rkey, post_actor_id, post_rkey, cid, status) - SELECT actor_id, rkey, post_actor_id, post_rkey, cid, 'stub'::repost_status - FROM stub_data - ON CONFLICT (actor_id, rkey) DO NOTHING - RETURNING actor_id, rkey", - &[&actor_ids, &rkeys, &post_actor_ids, &post_rkeys, &cid_digests], - ) - .await?; - - let mut result = Vec::new(); - for row in rows { - let actor_id: i32 = row.get(0); - let rkey: i64 = row.get(1); - result.push((actor_id, rkey)); - } - - Ok(result) -} +// NOTE: create_repost_stubs() removed - with natural keys, we don't need to create +// repost stubs anymore. Reposts can be referenced by (actor_id, rkey) even if they don't +// exist yet, since there's no FK constraint. diff --git a/consumer/src/db/operations/feed/helpers.rs b/consumer/src/db/operations/feed/helpers.rs index eed54958..6f6ce2d8 100644 --- a/consumer/src/db/operations/feed/helpers.rs +++ b/consumer/src/db/operations/feed/helpers.rs @@ -13,75 +13,32 @@ use crate::Result; use deadpool_postgres::GenericClient; use eyre::{Context as _, OptionExt as _}; -/// Get a post ID by actor_id and rkey, creating a stub if necessary +/// Get a post natural key by actor_id and rkey /// -/// This function is called when we need a post ID for posts that may not exist yet -/// (e.g., parent/root posts in a reply). If the post doesn't exist, we create a stub with: -/// - status = 'stub' -/// - content = '' (empty) -/// - The real actor_id, rkey, and CID from the reference +/// With natural keys, we don't need to create post stubs anymore since: +/// - Posts use (actor_id, rkey) composite natural keys +/// - No FK constraint requires posts to exist before being referenced +/// - We can reference posts by their natural key even if they don't exist yet /// -/// Uses advisory locks to prevent concurrent transactions from racing on the same post. -/// The lock ensures only one transaction at a time can create/access a specific post, -/// while other posts can still be processed in parallel. +/// This function simply validates and converts the rkey to i64 format. /// -/// Returns ((actor_id, rkey), was_created) where was_created indicates if a new stub was created. +/// Returns ((actor_id, rkey), was_created=false) for consistency with previous API. +/// was_created is always false since we no longer create stubs. /// /// IMPORTANT: Actor must already exist before calling this function. /// Use get_actor_id to create the actor first if needed. pub(super) async fn get_post_id( - conn: &C, + _conn: &C, actor_id: i32, rkey: &str, - cid_str: &str, + _cid_str: &str, ) -> Result<((i32, i64), bool)> { - // Convert rkey (TID string) to INT8 for database lookup + // Convert rkey (TID string) to INT8 let rkey_i64 = parakeet_db::models::tid_to_i64(rkey) .wrap_err_with(|| format!("Invalid TID encoding in rkey: {}", rkey))?; - // Acquire advisory lock using actor_id and rkey to prevent concurrent access races - let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("posts", actor_id, rkey_i64); - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) - .await?; - - // Parse the CID string to get the digest - let cid = ipld_core::cid::Cid::try_from(cid_str) - .wrap_err_with(|| format!("Invalid CID format: {}", cid_str))?; - let cid_bytes = cid.to_bytes(); - let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) - .ok_or_eyre("CID must be valid AT Protocol CID")?; - - // Decode timestamp from TID (bits 1-53 contain microseconds since epoch) - // Note: Timestamp is derivable from rkey at query time via tid_timestamp(rkey) - // We don't store it explicitly in the posts table - - // Use CTE to SELECT first, then conditionally INSERT only if not found - // Still uses ON CONFLICT for race condition safety between concurrent transactions - let row = conn - .query_one( - "WITH existing AS ( - SELECT actor_id, rkey FROM posts WHERE actor_id = $1 AND rkey = $2 - ), - inserted AS ( - INSERT INTO posts (actor_id, rkey, cid, content, status) - SELECT $1, $2, $3, NULL, 'stub'::post_status - WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (actor_id, rkey) DO UPDATE SET actor_id = EXCLUDED.actor_id - RETURNING actor_id, rkey - ) - SELECT - COALESCE((SELECT actor_id FROM existing), (SELECT actor_id FROM inserted)) as actor_id, - COALESCE((SELECT rkey FROM existing), (SELECT rkey FROM inserted)) as rkey, - (SELECT actor_id FROM existing) IS NULL as was_created", - &[&actor_id, &rkey_i64, &cid_digest], - ) - .await?; - - let returned_actor_id: i32 = row.get(0); - let returned_rkey: i64 = row.get(1); - let was_created: bool = row.get(2); - - Ok(((returned_actor_id, returned_rkey), was_created)) + // Return natural key directly - no stub creation needed + Ok(((actor_id, rkey_i64), false)) } /// Get an actor ID by DID, creating the actor if it doesn't exist @@ -277,65 +234,29 @@ pub(super) async fn get_list_id(conn: &C, at_uri: &str) -> Res Ok((row.get(0), row.get(1))) } -/// Get a repost natural key by actor_id and rkey, creating a stub if it doesn't exist +/// Get a repost natural key by actor_id and rkey /// -/// This function: -/// 1. Looks up existing repost by (actor_id, rkey) -/// 2. If not found, creates a stub repost with status='stub' -/// 3. Returns ((actor_id, rkey), was_created) where was_created indicates if a new stub was created +/// With natural keys, we don't need to create repost stubs anymore since: +/// - Reposts use (actor_id, rkey) composite natural keys +/// - No FK constraint requires reposts to exist before being referenced +/// - We can reference reposts by their natural key even if they don't exist yet /// -/// Uses advisory locks to prevent race conditions. +/// This function simply validates and converts the rkey to i64 format. +/// +/// Returns ((actor_id, rkey), was_created=false) for consistency with previous API. +/// was_created is always false since we no longer create stubs. /// /// IMPORTANT: Actor must already exist before calling this function. pub(crate) async fn get_repost_id( - conn: &C, + _conn: &C, actor_id: i32, rkey: &str, - cid_str: &str, + _cid_str: &str, ) -> Result<((i32, i64), bool)> { - // Convert rkey (TID string) to INT8 for database lookup + // Convert rkey (TID string) to INT8 let rkey_i64 = parakeet_db::models::tid_to_i64(rkey) .wrap_err_with(|| format!("Invalid TID encoding in rkey: {}", rkey))?; - // Parse the CID string to get the digest - let cid = ipld_core::cid::Cid::try_from(cid_str) - .wrap_err_with(|| format!("Invalid CID format: {}", cid_str))?; - let cid_bytes = cid.to_bytes(); - let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) - .ok_or_eyre("CID must be valid AT Protocol CID")?; - - // Acquire advisory lock using actor_id and rkey to prevent concurrent access races - let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("reposts", actor_id, rkey_i64); - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) - .await?; - - // Use CTE to SELECT first, then conditionally INSERT only if not found - // This prevents unnecessary repost stub creation - // Still uses ON CONFLICT for race condition safety between concurrent transactions - // Note: CID is from the via field's StrongRef (real CID from referencing record) - let row = conn - .query_one( - "WITH existing AS ( - SELECT actor_id, rkey FROM reposts WHERE actor_id = $1 AND rkey = $2 - ), - inserted AS ( - INSERT INTO reposts (actor_id, rkey, cid, post_actor_id, post_rkey, via_repost_actor_id, via_repost_rkey, status) - SELECT $1, $2, $3, NULL, NULL, NULL, NULL, 'stub'::repost_status - WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (actor_id, rkey) DO UPDATE SET actor_id = EXCLUDED.actor_id - RETURNING actor_id, rkey - ) - SELECT - COALESCE((SELECT actor_id FROM existing), (SELECT actor_id FROM inserted)) as actor_id, - COALESCE((SELECT rkey FROM existing), (SELECT rkey FROM inserted)) as rkey, - (SELECT actor_id FROM existing) IS NULL as was_created", - &[&actor_id, &rkey_i64, &cid_digest], - ) - .await?; - - let returned_actor_id: i32 = row.get(0); - let returned_rkey: i64 = row.get(1); - let was_created: bool = row.get(2); - - Ok(((returned_actor_id, returned_rkey), was_created)) + // Return natural key directly - no stub creation needed + Ok(((actor_id, rkey_i64), false)) } diff --git a/consumer/src/db/operations/feed/like.rs b/consumer/src/db/operations/feed/like.rs index 5a2e26af..12ad0c1a 100644 --- a/consumer/src/db/operations/feed/like.rs +++ b/consumer/src/db/operations/feed/like.rs @@ -2,13 +2,13 @@ //! //! This module handles like database operations: //! - Creating likes for posts, feed generators, and labelers -//! - Resolving subject IDs and creating stubs as needed +//! - Resolving subject natural keys (no stub creation for posts with natural keys) //! - Deleting likes and returning subject URIs //! //! Likes can target multiple subject types: -//! - Posts (app.bsky.feed.post) -//! - Feed generators (app.bsky.feed.generator) -//! - Labelers (app.bsky.labeler.service) +//! - Posts (app.bsky.feed.post) - uses natural keys, no stub creation +//! - Feed generators (app.bsky.feed.generator) - uses synthetic IDs, creates stubs +//! - Labelers (app.bsky.labeler.service) - uses synthetic IDs, creates stubs //! //! Likes also support a "via" field for quote posts. @@ -20,14 +20,15 @@ use deadpool_postgres::GenericClient; /// Insert a like record /// /// Likes can target posts, feed generators, or labelers. The function: -/// 1. Resolves the subject ID based on collection type, creating stubs if needed +/// 1. Resolves the subject natural key/ID based on collection type +/// - Posts: No stub creation (natural keys allow referencing non-existent posts) +/// - Feedgens/Labelers: Creates stubs if needed (synthetic IDs require existence) /// 2. Inserts the like with advisory locking to prevent race conditions /// /// Performance tracking: Logs timing breakdown for slow inserts (>100ms). /// /// IMPORTANT: Likes do NOT trigger fetch queue enqueueing. Only reposts and quotes -/// should trigger fetches. The stub creation maintains referential integrity for -/// foreign key constraints, but the actual record fetch happens only when needed. +/// should trigger fetches. /// /// NOTE: via_repost_id is already resolved in the reference extraction phase /// (workers.rs extract_and_resolve_references) before this function is called. @@ -70,9 +71,10 @@ pub async fn like_insert( _ => "post", // Fallback }; - // Resolve subject, creating stub if needed - // Track whether stub was created so we only enqueue newly created stubs - // For posts, we get natural keys (actor_id, rkey); for other types we get single IDs + // Resolve subject natural key/ID + // For posts: No stub creation (natural keys allow referencing non-existent posts) + // For feedgens/labelers: Creates stubs if needed (synthetic IDs require existence) + // Track whether stub was created (always false for posts) let subject_start = std::time::Instant::now(); let (subject_post_key, subject_id, _subject_was_created): (Option<(i32, i64)>, Option, bool) = match subject_collection { "app.bsky.feed.post" => { @@ -190,8 +192,6 @@ pub async fn like_insert( // Note: We do NOT enqueue the subject for fetching when a like is created. // Likes should not trigger post fetches - only reposts and quotes should fetch posts. - // The stub creation above maintains referential integrity for foreign key constraints. - // If the post is reposted or quoted later, it will be fetched then. let total_ms = total_start.elapsed().as_millis(); diff --git a/consumer/src/db/operations/feed/repost.rs b/consumer/src/db/operations/feed/repost.rs index 3d8197dc..9baa51f2 100644 --- a/consumer/src/db/operations/feed/repost.rs +++ b/consumer/src/db/operations/feed/repost.rs @@ -1,11 +1,11 @@ //! Repost operations //! //! This module handles repost database operations: -//! - Creating reposts with post resolution and stub creation +//! - Creating reposts with post natural key resolution (no stub creation) //! - Deleting reposts and returning post URIs //! -//! Unlike likes, reposts only target posts and trigger fetch queue enqueueing -//! when new post stubs are created. +//! Reposts only target posts which use natural keys (actor_id, rkey). +//! No stub creation is needed since posts can be referenced even if they don't exist yet. use crate::Result; use super::helpers::{get_actor_id, get_post_id}; @@ -17,12 +17,11 @@ use ipld_core::cid::Cid; /// /// Reposts can target posts and support a "via" field for quote posts. /// The function: -/// 1. Resolves the subject post ID, creating stub if needed +/// 1. Resolves the subject post natural key (no stub creation needed) /// 2. Inserts the repost with advisory locking to prevent race conditions -/// 3. Enqueues the subject post for fetching if a new stub was created /// -/// IMPORTANT: Unlike likes, reposts DO trigger fetch queue enqueueing when -/// a new post stub is created. This ensures reposted content is fetched. +/// With natural keys, we no longer create post stubs or enqueue posts for fetching. +/// Posts can be referenced by (actor_id, rkey) even if they don't exist yet. /// /// NOTE: via_repost_id is already resolved in the reference extraction phase /// (workers.rs extract_and_resolve_references) before this function is called. @@ -34,7 +33,7 @@ pub async fn repost_insert( cid: Cid, rec: AppBskyFeedRepost, via_repost_key: Option<(i32, i64)>, - source: crate::database_writer::EventSource, + _source: crate::database_writer::EventSource, ) -> Result { let cid_bytes = cid.to_bytes(); let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes) @@ -51,14 +50,13 @@ pub async fn repost_insert( // Note: via_repost_id is already resolved in the reference extraction phase (workers.rs) // This ensures the FK constraint is satisfied before we reach this point - // Resolve post natural key, creating stub if needed - // Track whether stub was created so we only enqueue newly created stubs + // Resolve post natural key (no stub creation with natural keys) let subject_did = parakeet_db::at_uri_util::extract_did(subject_uri) .ok_or_else(|| eyre::eyre!("Invalid subject URI: missing DID in {}", subject_uri))?; let subject_rkey = parakeet_db::at_uri_util::extract_rkey(subject_uri) .ok_or_else(|| eyre::eyre!("Invalid subject URI: missing rkey in {}", subject_uri))?; let (subject_actor_id, _, _) = get_actor_id(conn, subject_did).await?; - let ((post_actor_id, post_rkey), subject_was_created) = + let ((post_actor_id, post_rkey), _subject_was_created) = get_post_id(conn, subject_actor_id, subject_rkey, &subject_cid_str).await?; // Insert or upgrade repost @@ -86,14 +84,8 @@ pub async fn repost_insert( ) .await?; - // Enqueue to fetch queue (fire-and-forget) - // These are non-critical queue operations that shouldn't block the insert - // Skip enqueue during backfill - // Only enqueue subject if we created a new stub (not if it already existed) - if rows > 0 && source != crate::database_writer::EventSource::Backfill && subject_was_created { - let subject_uri_owned = subject_uri.to_string(); - crate::db::fetch_queue::enqueue(conn, &subject_uri_owned).await?; - } + // Note: We no longer enqueue posts for fetching when reposts are created. + // With natural keys, we don't create post stubs, so there's nothing to enqueue. Ok(rows) } diff --git a/consumer/tests/bulk_resolve_queries_test.rs b/consumer/tests/bulk_resolve_queries_test.rs index 8c7bde4c..af188f38 100644 --- a/consumer/tests/bulk_resolve_queries_test.rs +++ b/consumer/tests/bulk_resolve_queries_test.rs @@ -53,27 +53,7 @@ async fn test_resolve_repost_uris_with_data_query() -> eyre::Result<()> { Ok(()) } -#[tokio::test] -async fn test_create_repost_stubs_query() -> eyre::Result<()> { - let pool = test_pool(); - let conn = pool.get().await.wrap_err("Failed to get connection")?; - - // Test with empty arrays (valid SQL, should return no results) - let actor_ids: Vec = vec![]; - let rkeys: Vec = vec![]; - let post_actor_ids: Vec = vec![]; - let post_rkeys: Vec = vec![]; - let cid_digests: Vec> = vec![]; - - let result = bulk_resolve::queries::create_repost_stubs(&conn, &actor_ids, &rkeys, &post_actor_ids, &post_rkeys, &cid_digests).await; - - assert!( - result.is_ok(), - "create_repost_stubs query should be valid SQL: {:?}", - result.err() - ); - Ok(()) -} +// NOTE: test_create_repost_stubs_query removed - we no longer create repost stubs with natural keys // ============================================================================ // Actor DID Resolution Tests @@ -284,11 +264,13 @@ async fn test_resolve_post_uris_bulk_missing() -> eyre::Result<()> { let mut conn = pool.get().await.wrap_err("Failed to get connection")?; let txn = conn.transaction().await?; - // Try to resolve a post that doesn't exist + // With natural keys, we can resolve URIs even if actors/posts don't exist + // The function creates actor stubs if needed and returns natural keys let uris = vec!["at://did:plc:missing/app.bsky.feed.post/3l6kdoqxe7k2a"]; let result = bulk_resolve::resolve_post_uris_bulk(&txn, &uris).await?; - assert_eq!(result.len(), 0, "Missing post should not be in result"); + assert_eq!(result.len(), 1, "Should return natural key even if post doesn't exist"); + assert!(result.contains_key("at://did:plc:missing/app.bsky.feed.post/3l6kdoqxe7k2a")); txn.rollback().await?; Ok(()) @@ -298,42 +280,7 @@ async fn test_resolve_post_uris_bulk_missing() -> eyre::Result<()> { // Post Stub Creation Tests // ============================================================================ -#[tokio::test] -async fn test_create_post_stubs_bulk_empty() -> eyre::Result<()> { - let pool = test_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - let txn = conn.transaction().await?; - - let uri_cid_pairs: Vec<(&str, &str)> = vec![]; - let result = bulk_resolve::create_post_stubs_bulk(&txn, &uri_cid_pairs).await?; - - assert_eq!(result.len(), 0, "Empty input should return empty map"); - txn.rollback().await?; - Ok(()) -} - -#[tokio::test] -async fn test_create_post_stubs_bulk_new() -> eyre::Result<()> { - let pool = test_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - let txn = conn.transaction().await?; - - // Ensure actor exists first - ensure_actor_id(&txn, "did:plc:author", Some(&ActorStatus::Active), None, Utc::now()).await?; - - // Create post stubs with valid CID - let cid_str = test_cid().to_string(); - let uri_cid_pairs = vec![ - ("at://did:plc:author/app.bsky.feed.post/3l6kdoqxe7k2a", cid_str.as_str()), - ]; - let result = bulk_resolve::create_post_stubs_bulk(&txn, &uri_cid_pairs).await?; - - assert_eq!(result.len(), 1); - assert!(result.contains_key("at://did:plc:author/app.bsky.feed.post/3l6kdoqxe7k2a")); - - txn.rollback().await?; - Ok(()) -} +// NOTE: test_create_post_stubs_bulk tests removed - we no longer create post stubs with natural keys // ============================================================================ // Resolve and Ensure Posts Tests diff --git a/migrations/2025-11-19-011902_drop_natural_key_fk_constraints/down.sql b/migrations/2025-11-19-011902_drop_natural_key_fk_constraints/down.sql new file mode 100644 index 00000000..6f3cf69d --- /dev/null +++ b/migrations/2025-11-19-011902_drop_natural_key_fk_constraints/down.sql @@ -0,0 +1,14 @@ +-- This file should undo anything in `up.sql` +-- Re-add FK constraints (though this may fail if there are orphaned references) + +-- Re-add FK constraint on reposts -> posts +ALTER TABLE reposts ADD CONSTRAINT reposts_post_fkey + FOREIGN KEY (post_actor_id, post_rkey) + REFERENCES posts(actor_id, rkey) + ON DELETE CASCADE; + +-- Re-add self-referencing FK constraint on reposts -> reposts (via_repost) +ALTER TABLE reposts ADD CONSTRAINT reposts_via_repost_fkey + FOREIGN KEY (via_repost_actor_id, via_repost_rkey) + REFERENCES reposts(actor_id, rkey) + ON DELETE CASCADE; diff --git a/migrations/2025-11-19-011902_drop_natural_key_fk_constraints/up.sql b/migrations/2025-11-19-011902_drop_natural_key_fk_constraints/up.sql new file mode 100644 index 00000000..ee2f026e --- /dev/null +++ b/migrations/2025-11-19-011902_drop_natural_key_fk_constraints/up.sql @@ -0,0 +1,14 @@ +-- Drop FK constraints on natural key references +-- With natural keys (actor_id, rkey), we don't need FK constraints since: +-- 1. We can reference posts/reposts by their natural key even if they don't exist yet +-- 2. TimescaleDB hypertables don't support compound FK constraints efficiently +-- 3. This eliminates the need for stub creation during indexing + +-- Drop FK constraint on reposts -> posts +ALTER TABLE reposts DROP CONSTRAINT IF EXISTS reposts_post_fkey; + +-- Drop self-referencing FK constraint on reposts -> reposts (via_repost) +ALTER TABLE reposts DROP CONSTRAINT IF EXISTS reposts_via_repost_fkey; + +-- Note: post_likes already has no FK constraint on (post_actor_id, post_rkey) +-- Note: We keep actor_id FK constraints since actors still use synthetic IDs