From 4cd9b7922a6eca70aa2e841aa378039b7593f997 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Sat, 15 Nov 2025 18:17:04 -0500 Subject: [PATCH] feat: repost stubs --- consumer/src/db/operations/feed/helpers.rs | 15 +++++- consumer/src/db/operations/feed/like.rs | 59 +++------------------- consumer/src/db/operations/feed/repost.rs | 55 +++----------------- 3 files changed, 28 insertions(+), 101 deletions(-) diff --git a/consumer/src/db/operations/feed/helpers.rs b/consumer/src/db/operations/feed/helpers.rs index aefcdb38..18cad765 100644 --- a/consumer/src/db/operations/feed/helpers.rs +++ b/consumer/src/db/operations/feed/helpers.rs @@ -320,7 +320,7 @@ pub(super) async fn get_list_id(conn: &C, at_uri: &str) -> Res /// Uses advisory locks to prevent race conditions. /// /// IMPORTANT: Actor must already exist before calling this function. -pub(super) async fn get_repost_id( +pub(crate) async fn get_repost_id( conn: &C, actor_id: i32, rkey: &str, @@ -367,5 +367,18 @@ pub(super) async fn get_repost_id( let repost_id: i64 = row.get(0); let was_created: bool = row.get(1); + // Defensive check: Verify the repost actually exists after creation + // This catches race conditions where the CTE returns an ID but the row isn't visible + let verify_row = conn + .query_opt("SELECT 1 FROM reposts WHERE id = $1", &[&repost_id]) + .await?; + + if verify_row.is_none() { + return Err(eyre::eyre!( + "Repost stub creation returned id={} but row not found (race condition detected)", + repost_id + )); + } + Ok((repost_id, was_created)) } diff --git a/consumer/src/db/operations/feed/like.rs b/consumer/src/db/operations/feed/like.rs index eb0d848e..fc220a02 100644 --- a/consumer/src/db/operations/feed/like.rs +++ b/consumer/src/db/operations/feed/like.rs @@ -13,9 +13,8 @@ //! Likes also support a "via" field for quote posts. use crate::Result; -use super::helpers::{get_actor_id, get_feedgen_id, get_post_id, get_repost_id}; +use super::helpers::{get_actor_id, get_feedgen_id, get_post_id}; use crate::types::records::AppBskyFeedLike; -use crate::utils::strongref_to_parts; use deadpool_postgres::GenericClient; use ipld_core::cid::Cid; @@ -23,25 +22,26 @@ use ipld_core::cid::Cid; /// /// Likes can target posts, feed generators, or labelers. The function: /// 1. Resolves the subject ID based on collection type, creating stubs if needed -/// 2. Resolves the via post ID if present (for quote posts) -/// 3. Inserts the like with advisory locking to prevent race conditions +/// 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. +/// +/// NOTE: via_repost_id is pre-resolved in the reference extraction phase +/// (workers.rs) to avoid FK constraint violations. pub async fn like_insert( conn: &C, rkey: i64, actor_id: i32, cid: Cid, rec: AppBskyFeedLike, - source: crate::database_writer::EventSource, + via_repost_id: Option, + _source: crate::database_writer::EventSource, ) -> Result { let total_start = std::time::Instant::now(); - - let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref()); let subject_uri = &rec.subject.uri; let subject_cid_str = rec.subject.cid.to_string(); let cid_bytes = cid.to_bytes(); @@ -110,50 +110,6 @@ pub async fn like_insert( }; let subject_ms = subject_start.elapsed().as_millis(); - // Resolve via reference if present - // Via can be a repost (most common case for "liked via repost") - let via_start = std::time::Instant::now(); - let via_repost_id = if let (Some(via_uri), Some(via_cid)) = (&via_uri, &via_cid) { - let via_collection = via_uri - .strip_prefix("at://") - .and_then(|s| s.split('/').nth(1)) - .ok_or_else(|| eyre::eyre!("Invalid via URI: {}", via_uri))?; - - let via_did = parakeet_db::at_uri_util::extract_did(via_uri) - .ok_or_else(|| eyre::eyre!("Invalid via URI: missing DID in {}", via_uri))?; - let via_rkey = parakeet_db::at_uri_util::extract_rkey(via_uri) - .ok_or_else(|| eyre::eyre!("Invalid via URI: missing rkey in {}", via_uri))?; - let (via_actor_id, _, _) = get_actor_id(conn, via_did).await?; - - match via_collection { - "app.bsky.feed.repost" => { - // Via is a repost - this is the main case for "liked via repost" - let (repost_id, was_created) = get_repost_id(conn, via_actor_id, via_rkey, via_cid).await?; - - // If we created a new stub, enqueue it for fetching - // (only if not during backfill) - if was_created && source != crate::database_writer::EventSource::Backfill { - let via_uri_owned = via_uri.to_string(); - crate::db::fetch_queue::enqueue(conn, &via_uri_owned).await?; - } - - Some(repost_id) - } - _ => { - // Unexpected via collection - log warning and skip - tracing::warn!( - via_uri = %via_uri, - via_collection = %via_collection, - "Unexpected via collection in like (expected repost)" - ); - None - } - } - } else { - None - }; - let via_ms = via_start.elapsed().as_millis(); - // Insert like with resolved IDs // Note: created_at is derived from TID rkey let insert_start = std::time::Instant::now(); @@ -201,7 +157,6 @@ pub async fn like_insert( total_ms = total_ms, lock_ms = lock_ms, subject_ms = subject_ms, - via_ms = via_ms, insert_ms = insert_ms, "Like insert timing breakdown" ); diff --git a/consumer/src/db/operations/feed/repost.rs b/consumer/src/db/operations/feed/repost.rs index ea300666..0aaa4b46 100644 --- a/consumer/src/db/operations/feed/repost.rs +++ b/consumer/src/db/operations/feed/repost.rs @@ -8,9 +8,8 @@ //! when new post stubs are created. use crate::Result; -use super::helpers::{get_actor_id, get_post_id, get_repost_id}; +use super::helpers::{get_actor_id, get_post_id}; use crate::types::records::AppBskyFeedRepost; -use crate::utils::strongref_to_parts; use deadpool_postgres::GenericClient; use ipld_core::cid::Cid; @@ -19,21 +18,23 @@ 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 -/// 2. Resolves the via post ID if present (for quote posts) -/// 3. Inserts the repost with advisory locking to prevent race conditions -/// 4. Enqueues the subject post for fetching if a new stub was created +/// 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. +/// +/// NOTE: via_repost_id is pre-resolved in the reference extraction phase +/// (workers.rs) to avoid FK constraint violations. pub async fn repost_insert( conn: &C, rkey: i64, actor_id: i32, cid: Cid, rec: AppBskyFeedRepost, + via_repost_id: Option, source: crate::database_writer::EventSource, ) -> Result { - let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref()); let subject_uri = &rec.subject.uri; let subject_cid_str = rec.subject.cid.to_string(); let cid_bytes = cid.to_bytes(); @@ -55,48 +56,6 @@ pub async fn repost_insert( let (post_id, subject_was_created) = get_post_id(conn, subject_actor_id, subject_rkey, &subject_cid_str).await?; - // Resolve via reference if present - // Via can be a repost (for reposts-of-reposts) - let via_repost_id = if let (Some(via_uri), Some(via_cid)) = (&via_uri, &via_cid) { - let via_collection = via_uri - .strip_prefix("at://") - .and_then(|s| s.split('/').nth(1)) - .ok_or_else(|| eyre::eyre!("Invalid via URI: {}", via_uri))?; - - let via_did = parakeet_db::at_uri_util::extract_did(via_uri) - .ok_or_else(|| eyre::eyre!("Invalid via URI: missing DID in {}", via_uri))?; - let via_rkey = parakeet_db::at_uri_util::extract_rkey(via_uri) - .ok_or_else(|| eyre::eyre!("Invalid via URI: missing rkey in {}", via_uri))?; - let (via_actor_id, _, _) = get_actor_id(conn, via_did).await?; - - match via_collection { - "app.bsky.feed.repost" => { - // Via is a repost - repost of a repost case - let (repost_id, was_created) = get_repost_id(conn, via_actor_id, via_rkey, via_cid).await?; - - // If we created a new stub, enqueue it for fetching - // (only if not during backfill) - if was_created && source != crate::database_writer::EventSource::Backfill { - let via_uri_owned = via_uri.to_string(); - crate::db::fetch_queue::enqueue(conn, &via_uri_owned).await?; - } - - Some(repost_id) - } - _ => { - // Unexpected via collection - log warning and skip - tracing::warn!( - via_uri = %via_uri, - via_collection = %via_collection, - "Unexpected via collection in repost (expected repost)" - ); - None - } - } - } else { - None - }; - // Insert or upgrade repost // If a stub exists, upgrade it to complete status and fill in the data // If it's a new repost, insert with status='complete' -- 2.54.0