diff --git a/consumer/src/backfill/mod.rs b/consumer/src/backfill/mod.rs index bef3cee0..150f9235 100644 --- a/consumer/src/backfill/mod.rs +++ b/consumer/src/backfill/mod.rs @@ -336,9 +336,9 @@ async fn check_pds_repo_status( #[derive(Debug, Default)] struct CopyStore { - likes: Vec<(String, records::StrongRef, DateTime)>, + likes: Vec<(String, records::StrongRef, Option, DateTime)>, posts: Vec<(String, Cid, records::AppBskyFeedPost)>, - reposts: Vec<(String, records::StrongRef, DateTime)>, + reposts: Vec<(String, records::StrongRef, Option, DateTime)>, blocks: Vec<(String, String, DateTime)>, follows: Vec<(String, String, DateTime)>, list_items: Vec<(String, records::AppBskyGraphListItem)>, diff --git a/consumer/src/backfill/repo.rs b/consumer/src/backfill/repo.rs index 7e7daf49..a887fcbe 100644 --- a/consumer/src/backfill/repo.rs +++ b/consumer/src/backfill/repo.rs @@ -131,7 +131,7 @@ async fn record_index( deltas.incr(&rec.subject.uri, AggregateType::Like).await; copies.push_record(&at_uri, cid); - copies.likes.push((at_uri, rec.subject, rec.created_at)); + copies.likes.push((at_uri, rec.subject, rec.via, rec.created_at)); } RecordTypes::AppBskyFeedPost(rec) => { let maybe_reply = rec.reply.as_ref().map(|v| v.parent.uri.clone()); @@ -167,7 +167,7 @@ async fn record_index( deltas.incr(&rec.subject.uri, AggregateType::Repost).await; copies.push_record(&at_uri, cid); - copies.reposts.push((at_uri, rec.subject, rec.created_at)); + copies.reposts.push((at_uri, rec.subject, rec.via, rec.created_at)); } RecordTypes::AppBskyGraphBlock(rec) => { copies.push_record(&at_uri, cid); diff --git a/consumer/src/db/copy.rs b/consumer/src/db/copy.rs index d3a943b1..82ae0827 100644 --- a/consumer/src/db/copy.rs +++ b/consumer/src/db/copy.rs @@ -14,9 +14,11 @@ const STRONGREF_TYPES: &[Type] = &[ Type::TEXT, Type::TEXT, Type::TEXT, + Type::TEXT, + Type::TEXT, Type::TIMESTAMP, ]; -type StrongRefRow = (String, records::StrongRef, DateTime); +type StrongRefRow = (String, records::StrongRef, Option, DateTime); // SubjectRefs are used in both blocks and follows const SUBJECT_TYPES: &[Type] = &[Type::TEXT, Type::TEXT, Type::TEXT, Type::TIMESTAMP]; @@ -39,7 +41,7 @@ pub async fn copy_likes( let writer = conn .copy_in( - "COPY likes_tmp (at_uri, did, subject, subject_cid, created_at) FROM STDIN (FORMAT binary)", + "COPY likes_tmp (at_uri, did, subject, subject_cid, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)", ) .await?; let writer = BinaryCopyInWriter::new(writer, STRONGREF_TYPES); @@ -47,6 +49,8 @@ pub async fn copy_likes( pin_mut!(writer); for row in data { + let (via_uri, via_cid) = strongref_to_parts(row.2.as_ref()); + let writer = writer.as_mut(); writer .write(&[ @@ -54,7 +58,9 @@ pub async fn copy_likes( &did, &row.1.uri, &row.1.cid.to_string(), - &row.2.naive_utc(), + &via_uri, + &via_cid, + &row.3.naive_utc(), ]) .await?; } @@ -82,7 +88,7 @@ pub async fn copy_reposts( let writer = conn .copy_in( - "COPY reposts_tmp (at_uri, did, post, post_cid, created_at) FROM STDIN (FORMAT binary)", + "COPY reposts_tmp (at_uri, did, post, post_cid, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)", ) .await?; let writer = BinaryCopyInWriter::new(writer, STRONGREF_TYPES); @@ -90,6 +96,8 @@ pub async fn copy_reposts( pin_mut!(writer); for row in data { + let (via_uri, via_cid) = strongref_to_parts(row.2.as_ref()); + let writer = writer.as_mut(); writer .write(&[ @@ -97,7 +105,9 @@ pub async fn copy_reposts( &did, &row.1.uri, &row.1.cid.to_string(), - &row.2.naive_utc(), + &via_uri, + &via_cid, + &row.3.naive_utc(), ]) .await?; } diff --git a/consumer/src/db/record.rs b/consumer/src/db/record.rs index 23d8749e..7bfcc580 100644 --- a/consumer/src/db/record.rs +++ b/consumer/src/db/record.rs @@ -156,9 +156,11 @@ pub async fn like_insert( 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, created_at) VALUES ($1, $2, $3, $4, $5)", - &[&at_uri, &repo, &rec.subject.uri, &rec.subject.cid.to_string(), &rec.created_at] + "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] ).await } @@ -523,13 +525,17 @@ pub async fn repost_insert( 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, created_at) VALUES ($1, $2, $3, $4, $5)", + "INSERT INTO reposts (at_uri, did, post, post_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, ], ) diff --git a/consumer/src/indexer/records.rs b/consumer/src/indexer/records.rs index a6b0c10e..8fc6d41c 100644 --- a/consumer/src/indexer/records.rs +++ b/consumer/src/indexer/records.rs @@ -206,6 +206,7 @@ pub struct AppBskyFeedGenerator { pub struct AppBskyFeedLike { pub subject: StrongRef, pub created_at: DateTime, + pub via: Option, } #[derive(Debug, Deserialize, Serialize)] @@ -260,6 +261,7 @@ impl PostgateEmbeddingRules { pub struct AppBskyFeedRepost { pub subject: StrongRef, pub created_at: DateTime, + pub via: Option, } #[derive(Debug, Deserialize, Serialize)] diff --git a/migrations/2025-06-18-200415_like-repost-via/down.sql b/migrations/2025-06-18-200415_like-repost-via/down.sql new file mode 100644 index 00000000..3f277f37 --- /dev/null +++ b/migrations/2025-06-18-200415_like-repost-via/down.sql @@ -0,0 +1,2 @@ +alter table likes drop column via_uri, drop column via_cid; +alter table reposts drop column via_uri, drop column via_cid; \ No newline at end of file diff --git a/migrations/2025-06-18-200415_like-repost-via/up.sql b/migrations/2025-06-18-200415_like-repost-via/up.sql new file mode 100644 index 00000000..f88daf43 --- /dev/null +++ b/migrations/2025-06-18-200415_like-repost-via/up.sql @@ -0,0 +1,7 @@ +alter table likes + add column via_uri text, + add column via_cid text; + +alter table reposts + add column via_uri text, + add column via_cid text; \ No newline at end of file diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index cd858b60..8583c128 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -125,6 +125,8 @@ diesel::table! { subject_cid -> Text, created_at -> Timestamptz, indexed_at -> Timestamp, + via_uri -> Nullable, + via_cid -> Nullable, } } @@ -283,6 +285,8 @@ diesel::table! { post_cid -> Text, created_at -> Timestamptz, indexed_at -> Timestamp, + via_uri -> Nullable, + via_cid -> Nullable, } }