From 1a188ea34b17c6888a080ab3dc9b7e986855617b Mon Sep 17 00:00:00 2001 From: Mia Date: Sat, 5 Jul 2025 19:31:29 +0100 Subject: [PATCH] feat!: break at_uri into did/rkey for likes,reposts,follows,blocks --- consumer/src/backfill/mod.rs | 19 ++++-- consumer/src/backfill/repo.rs | 12 ++-- consumer/src/db/copy.rs | 8 +-- consumer/src/db/record.rs | 63 ++++++++++++------- consumer/src/indexer/mod.rs | 19 +++--- .../up.sql | 12 ++-- .../up.sql | 12 ++-- parakeet-db/src/models.rs | 8 +-- parakeet-db/src/schema.rs | 16 ++--- 9 files changed, 102 insertions(+), 67 deletions(-) diff --git a/consumer/src/backfill/mod.rs b/consumer/src/backfill/mod.rs index 5563ce3a..8be7a1a9 100644 --- a/consumer/src/backfill/mod.rs +++ b/consumer/src/backfill/mod.rs @@ -282,12 +282,12 @@ async fn handle_backfill_rows( let items: Vec = serde_json::from_value(row.data).unwrap(); for item in items { + let Some((_, rkey)) = item.at_uri.rsplit_once("/") else { + return Ok(()); + }; + match item.inner { BackfillItemInner::Create(record) | BackfillItemInner::Update(record) => { - let Some((_, rkey)) = item.at_uri.rsplit_once("/") else { - return Ok(()); - }; - let Some(cid) = item.cid else { continue; }; @@ -295,8 +295,15 @@ async fn handle_backfill_rows( indexer::index_op(conn, deltas, repo, cid, record, &item.at_uri, rkey).await? } BackfillItemInner::Delete => { - indexer::index_op_delete(conn, deltas, repo, item.collection, &item.at_uri) - .await? + indexer::index_op_delete( + conn, + deltas, + repo, + item.collection, + &item.at_uri, + rkey, + ) + .await? } } } diff --git a/consumer/src/backfill/repo.rs b/consumer/src/backfill/repo.rs index 4833062c..d0e92069 100644 --- a/consumer/src/backfill/repo.rs +++ b/consumer/src/backfill/repo.rs @@ -133,7 +133,7 @@ async fn record_index( copies.push_record(&at_uri, cid); copies .likes - .push((at_uri, rec.subject, rec.via, rec.created_at)); + .push((rkey.to_string(), rec.subject, rec.via, rec.created_at)); } RecordTypes::AppBskyFeedPost(rec) => { let maybe_reply = rec.reply.as_ref().map(|v| v.parent.uri.clone()); @@ -171,18 +171,22 @@ async fn record_index( copies.push_record(&at_uri, cid); copies .reposts - .push((at_uri, rec.subject, rec.via, rec.created_at)); + .push((rkey.to_string(), rec.subject, rec.via, rec.created_at)); } RecordTypes::AppBskyGraphBlock(rec) => { copies.push_record(&at_uri, cid); - copies.blocks.push((at_uri, rec.subject, rec.created_at)); + copies + .blocks + .push((rkey.to_string(), rec.subject, rec.created_at)); } RecordTypes::AppBskyGraphFollow(rec) => { deltas.incr(did, AggregateType::Follow).await; deltas.incr(&rec.subject, AggregateType::Follower).await; copies.push_record(&at_uri, cid); - copies.follows.push((at_uri, rec.subject, rec.created_at)); + copies + .follows + .push((rkey.to_string(), rec.subject, rec.created_at)); } RecordTypes::AppBskyGraphListItem(rec) => { let split_aturi = rec.list.rsplitn(4, '/').collect::>(); diff --git a/consumer/src/db/copy.rs b/consumer/src/db/copy.rs index 0ad91e45..6dd4d057 100644 --- a/consumer/src/db/copy.rs +++ b/consumer/src/db/copy.rs @@ -46,7 +46,7 @@ pub async fn copy_likes( let writer = conn .copy_in( - "COPY likes_tmp (at_uri, did, subject, subject_cid, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)", + "COPY likes_tmp (rkey, did, subject, subject_cid, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)", ) .await?; let writer = BinaryCopyInWriter::new(writer, STRONGREF_TYPES); @@ -93,7 +93,7 @@ pub async fn copy_reposts( let writer = conn .copy_in( - "COPY reposts_tmp (at_uri, did, post, post_cid, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)", + "COPY reposts_tmp (rkey, did, post, post_cid, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)", ) .await?; let writer = BinaryCopyInWriter::new(writer, STRONGREF_TYPES); @@ -213,7 +213,7 @@ pub async fn copy_blocks( .await?; let writer = conn - .copy_in("COPY blocks_tmp (at_uri, did, subject, created_at) FROM STDIN (FORMAT binary)") + .copy_in("COPY blocks_tmp (rkey, did, subject, created_at) FROM STDIN (FORMAT binary)") .await?; let writer = BinaryCopyInWriter::new(writer, SUBJECT_TYPES); @@ -292,7 +292,7 @@ pub async fn copy_follows( .await?; let writer = conn - .copy_in("COPY follows_tmp (at_uri, did, subject, created_at) FROM STDIN (FORMAT binary)") + .copy_in("COPY follows_tmp (rkey, did, subject, created_at) FROM STDIN (FORMAT binary)") .await?; let writer = BinaryCopyInWriter::new(writer, SUBJECT_TYPES); diff --git a/consumer/src/db/record.rs b/consumer/src/db/record.rs index 137366bf..29d629d0 100644 --- a/consumer/src/db/record.rs +++ b/consumer/src/db/record.rs @@ -24,19 +24,22 @@ pub async fn record_delete(conn: &mut C, at_uri: &str) -> PgEx pub async fn block_insert( conn: &mut C, - at_uri: &str, + rkey: &str, repo: &str, rec: AppBskyGraphBlock, ) -> PgExecResult { conn.execute( - "INSERT INTO blocks (at_uri, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", - &[&at_uri, &repo, &rec.subject, &rec.created_at], + "INSERT INTO blocks (rkey, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + &[&rkey, &repo, &rec.subject, &rec.created_at], ).await } -pub async fn block_delete(conn: &mut C, at_uri: &str) -> PgExecResult { - conn.execute("DELETE FROM blocks WHERE at_uri=$1", &[&at_uri]) - .await +pub async fn block_delete(conn: &mut C, rkey: &str, repo: &str) -> PgExecResult { + conn.execute( + "DELETE FROM blocks WHERE rkey=$1 AND did=$2", + &[&rkey, &repo], + ) + .await } pub async fn chat_decl_upsert( @@ -94,21 +97,25 @@ pub async fn feedgen_delete(conn: &mut C, at_uri: &str) -> PgE pub async fn follow_insert( conn: &mut C, - at_uri: &str, + rkey: &str, repo: &str, rec: AppBskyGraphFollow, ) -> PgExecResult { conn.execute( - "INSERT INTO follows (at_uri, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", - &[&at_uri, &repo, &rec.subject, &rec.created_at], + "INSERT INTO follows (rkey, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + &[&rkey, &repo, &rec.subject, &rec.created_at], ).await } -pub async fn follow_delete(conn: &mut C, at_uri: &str) -> PgOptResult { +pub async fn follow_delete( + conn: &mut C, + rkey: &str, + repo: &str, +) -> PgOptResult { let res = conn .query_opt( - "DELETE FROM follows WHERE at_uri=$1 RETURNING subject", - &[&at_uri], + "DELETE FROM follows WHERE rkey=$1 AND did=$2 RETURNING subject", + &[&rkey, &repo], ) .await?; @@ -153,23 +160,27 @@ pub async fn labeler_delete(conn: &mut C, repo: &str) -> PgExe pub async fn like_insert( conn: &mut C, - at_uri: &str, + rkey: &str, repo: &str, rec: AppBskyFeedLike, ) -> PgExecResult { let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref()); conn.execute( - "INSERT INTO likes (at_uri, did, subject, subject_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)", - &[&at_uri, &repo, &rec.subject.uri, &rec.subject.cid.to_string(), &via_uri, &via_cid, &rec.created_at] + "INSERT INTO likes (rkey, did, subject, subject_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)", + &[&rkey, &repo, &rec.subject.uri, &rec.subject.cid.to_string(), &via_uri, &via_cid, &rec.created_at] ).await } -pub async fn like_delete(conn: &mut C, at_uri: &str) -> PgOptResult { +pub async fn like_delete( + conn: &mut C, + rkey: &str, + repo: &str, +) -> PgOptResult { let res = conn .query_opt( - "DELETE FROM likes WHERE at_uri=$1 RETURNING subject", - &[&at_uri], + "DELETE FROM likes WHERE rkey=$1 AND did=$2 RETURNING subject", + &[&rkey, &repo], ) .await?; @@ -523,16 +534,16 @@ pub async fn profile_delete(conn: &mut C, repo: &str) -> PgExe pub async fn repost_insert( conn: &mut C, - at_uri: &str, + rkey: &str, repo: &str, rec: AppBskyFeedRepost, ) -> PgExecResult { let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref()); conn.execute( - "INSERT INTO reposts (at_uri, did, post, post_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)", + "INSERT INTO reposts (rkey, did, post, post_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)", &[ - &at_uri, + &rkey, &repo, &rec.subject.uri, &rec.subject.cid.to_string(), @@ -544,11 +555,15 @@ pub async fn repost_insert( .await } -pub async fn repost_delete(conn: &mut C, at_uri: &str) -> PgOptResult { +pub async fn repost_delete( + conn: &mut C, + rkey: &str, + repo: &str, +) -> PgOptResult { let res = conn .query_opt( - "DELETE FROM reposts WHERE at_uri=$1 RETURNING post", - &[&at_uri], + "DELETE FROM reposts WHERE rkey=$1 AND repo=$2 RETURNING post", + &[&rkey, &repo], ) .await?; diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index a062d962..0f81d643 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -477,7 +477,7 @@ async fn process_op( index_op(conn, deltas, repo, cid, decoded, &full_path, rkey).await?; } else if op.action == "delete" { - index_op_delete(conn, deltas, repo, collection, &full_path).await?; + index_op_delete(conn, deltas, repo, collection, &full_path, rkey).await?; } else { tracing::warn!("op contained invalid action {}", op.action); } @@ -539,7 +539,7 @@ pub async fn index_op( } RecordTypes::AppBskyFeedLike(record) => { let subject = record.subject.uri.clone(); - let count = db::like_insert(conn, at_uri, repo, record).await?; + let count = db::like_insert(conn, rkey, repo, record).await?; deltas .add_delta(&subject, AggregateType::Like, count as i32) @@ -608,7 +608,7 @@ pub async fn index_op( deltas .incr(&record.subject.uri, AggregateType::Repost) .await; - db::repost_insert(conn, at_uri, repo, record).await?; + db::repost_insert(conn, rkey, repo, record).await?; } RecordTypes::AppBskyFeedThreadgate(record) => { let split_aturi = record.post.rsplitn(4, '/').collect::>(); @@ -620,11 +620,11 @@ pub async fn index_op( db::threadgate_upsert(conn, at_uri, cid, record).await?; } RecordTypes::AppBskyGraphBlock(record) => { - db::block_insert(conn, at_uri, repo, record).await?; + db::block_insert(conn, rkey, repo, record).await?; } RecordTypes::AppBskyGraphFollow(record) => { let subject = record.subject.clone(); - let count = db::follow_insert(conn, at_uri, repo, record).await?; + let count = db::follow_insert(conn, rkey, repo, record).await?; deltas .add_delta(repo, AggregateType::Follow, count as i32) @@ -696,11 +696,12 @@ pub async fn index_op_delete( repo: &str, collection: CollectionType, at_uri: &str, + rkey: &str, ) -> Result<(), tokio_postgres::Error> { match collection { CollectionType::BskyProfile => db::profile_delete(conn, repo).await?, CollectionType::BskyStatus => db::status_delete(conn, repo).await?, - CollectionType::BskyBlock => db::block_delete(conn, at_uri).await?, + CollectionType::BskyBlock => db::block_delete(conn, rkey, repo).await?, CollectionType::BskyFeedGen => { let count = db::feedgen_delete(conn, at_uri).await?; deltas @@ -709,7 +710,7 @@ pub async fn index_op_delete( count } CollectionType::BskyFeedLike => { - if let Some(subject) = db::like_delete(conn, at_uri).await? { + if let Some(subject) = db::like_delete(conn, rkey, repo).await? { deltas.decr(&subject, AggregateType::Like).await; } 0 @@ -733,14 +734,14 @@ pub async fn index_op_delete( } CollectionType::BskyFeedPostgate => db::postgate_delete(conn, at_uri).await?, CollectionType::BskyFeedRepost => { - if let Some(subject) = db::repost_delete(conn, at_uri).await? { + if let Some(subject) = db::repost_delete(conn, rkey, repo).await? { deltas.decr(&subject, AggregateType::Repost).await; } 0 } CollectionType::BskyFeedThreadgate => db::threadgate_delete(conn, at_uri).await?, CollectionType::BskyFollow => { - if let Some(followee) = db::follow_delete(conn, at_uri).await? { + if let Some(followee) = db::follow_delete(conn, rkey, repo).await? { deltas.decr(&followee, AggregateType::Follower).await; deltas.decr(repo, AggregateType::Follow).await; } diff --git a/migrations/2025-01-29-213341_follows_and_blocks/up.sql b/migrations/2025-01-29-213341_follows_and_blocks/up.sql index ccd91458..c3fdd2e4 100644 --- a/migrations/2025-01-29-213341_follows_and_blocks/up.sql +++ b/migrations/2025-01-29-213341_follows_and_blocks/up.sql @@ -1,9 +1,11 @@ create table blocks ( - at_uri text primary key, + rkey text not null, did text not null references actors (did), subject text not null, - created_at timestamptz not null + created_at timestamptz not null, + + primary key (did, rkey) ); create index blocks_did_index on blocks (did); @@ -11,10 +13,12 @@ create index blocks_subject_index on blocks (subject); create table follows ( - at_uri text primary key, + rkey text not null, did text not null references actors (did), subject text not null, - created_at timestamptz not null + created_at timestamptz not null, + + primary key (did, rkey) ); create index follow_did_index on follows (did); diff --git a/migrations/2025-04-05-114428_likes_and_reposts/up.sql b/migrations/2025-04-05-114428_likes_and_reposts/up.sql index dfb7c604..b354c5f3 100644 --- a/migrations/2025-04-05-114428_likes_and_reposts/up.sql +++ b/migrations/2025-04-05-114428_likes_and_reposts/up.sql @@ -1,11 +1,13 @@ create table likes ( - at_uri text primary key, + rkey text not null, did text not null references actors (did), subject text not null, subject_cid text not null, created_at timestamptz not null, - indexed_at timestamp not null default now() + indexed_at timestamp not null default now(), + + primary key (did, rkey) ); create index likes_did_index on likes (did); @@ -13,12 +15,14 @@ create index likes_subject_index on likes (subject); create table reposts ( - at_uri text primary key, + rkey text not null, did text not null references actors (did), post text not null, post_cid text not null, created_at timestamptz not null, - indexed_at timestamp not null default now() + indexed_at timestamp not null default now(), + + primary key (did, rkey) ); create index reposts_did_index on reposts (did); diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index e0586721..621ce029 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -31,7 +31,7 @@ pub struct NewActor<'a> { #[diesel(table_name = crate::schema::blocks)] #[diesel(check_for_backend(diesel::pg::Pg))] pub struct NewBlock<'a> { - pub at_uri: &'a str, + pub rkey: &'a str, pub did: &'a str, pub subject: &'a str, pub created_at: NaiveDateTime, @@ -41,7 +41,7 @@ pub struct NewBlock<'a> { #[diesel(table_name = crate::schema::follows)] #[diesel(check_for_backend(diesel::pg::Pg))] pub struct NewFollow<'a> { - pub at_uri: &'a str, + pub rkey: &'a str, pub did: &'a str, pub subject: &'a str, pub created_at: NaiveDateTime, @@ -519,7 +519,7 @@ pub struct UpsertThreadgate<'a> { #[diesel(table_name = crate::schema::likes)] #[diesel(check_for_backend(diesel::pg::Pg))] pub struct NewLike<'a> { - pub at_uri: &'a str, + pub rkey: &'a str, pub did: &'a str, pub subject: &'a str, pub subject_cid: String, @@ -530,7 +530,7 @@ pub struct NewLike<'a> { #[diesel(table_name = crate::schema::reposts)] #[diesel(check_for_backend(diesel::pg::Pg))] pub struct NewRepost<'a> { - pub at_uri: &'a str, + pub rkey: &'a str, pub did: &'a str, pub post: &'a str, pub post_cid: String, diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 8583c128..a4219d5f 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -34,8 +34,8 @@ diesel::table! { } diesel::table! { - blocks (at_uri) { - at_uri -> Text, + blocks (did, rkey) { + rkey -> Text, did -> Text, subject -> Text, created_at -> Timestamptz, @@ -68,8 +68,8 @@ diesel::table! { } diesel::table! { - follows (at_uri) { - at_uri -> Text, + follows (did, rkey) { + rkey -> Text, did -> Text, subject -> Text, created_at -> Timestamptz, @@ -118,8 +118,8 @@ diesel::table! { } diesel::table! { - likes (at_uri) { - at_uri -> Text, + likes (did, rkey) { + rkey -> Text, did -> Text, subject -> Text, subject_cid -> Text, @@ -278,8 +278,8 @@ diesel::table! { } diesel::table! { - reposts (at_uri) { - at_uri -> Text, + reposts (did, rkey) { + rkey -> Text, did -> Text, post -> Text, post_cid -> Text, -- 2.51.2