feat: via for likes and reposts

consumer/src/backfill/mod.rs

···
 
 #[derive(Debug, Default)]
 struct CopyStore {
-    likes: Vec<(String, records::StrongRef, DateTime<Utc>)>,
+    likes: Vec<(String, records::StrongRef, Option<records::StrongRef>, DateTime<Utc>)>,
     posts: Vec<(String, Cid, records::AppBskyFeedPost)>,
-    reposts: Vec<(String, records::StrongRef, DateTime<Utc>)>,
+    reposts: Vec<(String, records::StrongRef, Option<records::StrongRef>, DateTime<Utc>)>,
     blocks: Vec<(String, String, DateTime<Utc>)>,
     follows: Vec<(String, String, DateTime<Utc>)>,
     list_items: Vec<(String, records::AppBskyGraphListItem)>,

consumer/src/backfill/repo.rs

···
             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());
···
             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);

consumer/src/db/copy.rs

···
     Type::TEXT,
     Type::TEXT,
     Type::TEXT,
+    Type::TEXT,
+    Type::TEXT,
     Type::TIMESTAMP,
 ];
-type StrongRefRow = (String, records::StrongRef, DateTime<Utc>);
+type StrongRefRow = (String, records::StrongRef, Option<records::StrongRef>, DateTime<Utc>);
 
 // SubjectRefs are used in both blocks and follows
 const SUBJECT_TYPES: &[Type] = &[Type::TEXT, Type::TEXT, Type::TEXT, Type::TIMESTAMP];
···
 
     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);
···
     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(&[
···
                 &did,
                 &row.1.uri,
                 &row.1.cid.to_string(),
-                &row.2.naive_utc(),
+                &via_uri,
+                &via_cid,
+                &row.3.naive_utc(),
             ])
             .await?;
     }
···
 
     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);
···
     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(&[
···
                 &did,
                 &row.1.uri,
                 &row.1.cid.to_string(),
-                &row.2.naive_utc(),
+                &via_uri,
+                &via_cid,
+                &row.3.naive_utc(),
             ])
             .await?;
     }

consumer/src/db/record.rs

···
     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
 }
 
···
     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,
         ],
     )

consumer/src/indexer/records.rs

···
 pub struct AppBskyFeedLike {
     pub subject: StrongRef,
     pub created_at: DateTime<Utc>,
+    pub via: Option<StrongRef>,
 }
 
 #[derive(Debug, Deserialize, Serialize)]
···
 pub struct AppBskyFeedRepost {
     pub subject: StrongRef,
     pub created_at: DateTime<Utc>,
+    pub via: Option<StrongRef>,
 }
 
 #[derive(Debug, Deserialize, Serialize)]

migrations/2025-06-18-200415_like-repost-via/down.sql

···
+alter table likes drop column via_uri, drop column via_cid;
+alter table reposts drop column via_uri, drop column via_cid;

migrations/2025-06-18-200415_like-repost-via/up.sql

···
+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;

parakeet-db/src/schema.rs

···
         subject_cid -> Text,
         created_at -> Timestamptz,
         indexed_at -> Timestamp,
+        via_uri -> Nullable<Text>,
+        via_cid -> Nullable<Text>,
     }
 }
 
···
         post_cid -> Text,
         created_at -> Timestamptz,
         indexed_at -> Timestamp,
+        via_uri -> Nullable<Text>,
+        via_cid -> Nullable<Text>,
     }
 }