+1 -1

crates/consumer/src/backfill/mod.rs

reviewed

··· 368 368 posts: Vec<(String, Cid, records::AppBskyFeedPost)>, 369 369 reposts: Vec<(String, StrongRef, Option<StrongRef>, DateTime<Utc>)>, 370 370 blocks: Vec<(String, String, DateTime<Utc>)>, 371 371 - follows: Vec<(String, String, DateTime<Utc>)>, 371 371 + follows: Vec<(String, String, Option<StrongRef>, DateTime<Utc>)>, 372 372 list_items: Vec<(String, records::AppBskyGraphListItem)>, 373 373 verifications: Vec<(String, Cid, records::AppBskyGraphVerification)>, 374 374 threadgates: Vec<(String, Cid, records::AppBskyFeedThreadgate)>, // not COPY'd but needs to be kept until last.

+1 -1

crates/consumer/src/backfill/repo.rs

reviewed

··· 214 214 copies.push_record(&at_uri, cid); 215 215 copies 216 216 .follows 217 217 - .push((rkey.to_string(), rec.subject, rec.created_at)); 217 217 + .push((rkey.to_string(), rec.subject, rec.via, rec.created_at)); 218 218 } 219 219 RecordTypes::AppBskyGraphListItem(rec) => { 220 220 let split_aturi = rec.list.rsplitn(4, '/').collect::<Vec<_>>();

+16 -4

crates/consumer/src/db/copy.rs

reviewed

··· 307 307 .await 308 308 } 309 309 310 310 + const FOLLOW_TYPES: &[Type] = &[ 311 311 + Type::TEXT, 312 312 + Type::TEXT, 313 313 + Type::TEXT, 314 314 + Type::TEXT, 315 315 + Type::TEXT, 316 316 + Type::TIMESTAMP, 317 317 + ]; 318 318 + type FollowsRow = (String, String, Option<StrongRef>, DateTime<Utc>); 319 319 + 310 320 pub async fn copy_follows( 311 321 conn: &mut Transaction<'_>, 312 322 did: &str, 313 313 - data: Vec<SubjectRefRow>, 323 323 + data: Vec<FollowsRow>, 314 324 ) -> PgExecResult { 315 325 if data.is_empty() { 316 326 return Ok(0); ··· 323 333 .await?; 324 334 325 335 let writer = conn 326 326 - .copy_in("COPY follows_tmp (rkey, did, subject, created_at) FROM STDIN (FORMAT binary)") 336 336 + .copy_in("COPY follows_tmp (rkey, did, subject, via_uri, via_cid, created_at) FROM STDIN (FORMAT binary)") 327 337 .await?; 328 328 - let writer = BinaryCopyInWriter::new(writer, SUBJECT_TYPES); 338 338 + let writer = BinaryCopyInWriter::new(writer, FOLLOW_TYPES); 329 339 330 340 pin_mut!(writer); 331 341 332 342 for row in data { 343 343 + let (via_uri, via_cid) = strongref_to_parts(row.2.as_ref()); 344 344 + 333 345 let writer = writer.as_mut(); 334 346 writer 335 335 - .write(&[&row.0, &did, &row.1, &row.2.naive_utc()]) 347 347 + .write(&[&row.0, &did, &row.1, &via_uri, &via_cid, &row.3.naive_utc()]) 336 348 .await?; 337 349 } 338 350

+4 -2

crates/consumer/src/db/record.rs

reviewed

··· 142 142 repo: &str, 143 143 rec: AppBskyGraphFollow, 144 144 ) -> PgExecResult { 145 145 + let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref()); 146 146 + 145 147 conn.execute( 146 146 - "INSERT INTO follows (rkey, did, subject, created_at) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", 147 147 - &[&rkey, &repo, &rec.subject, &rec.created_at], 148 148 + "INSERT INTO follows (rkey, did, subject, via_uri, via_cid created_at) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING", 149 149 + &[&rkey, &repo, &rec.subject, &via_uri, &via_cid, &rec.created_at], 148 150 ).await 149 151 } 150 152

+1

crates/consumer/src/indexer/records.rs

reviewed

··· 276 276 pub struct AppBskyGraphFollow { 277 277 pub subject: String, 278 278 pub created_at: DateTime<Utc>, 279 279 + pub via: Option<StrongRef>, 279 280 } 280 281 281 282 #[derive(Debug, Deserialize, Serialize)]

+2

crates/parakeet-db/src/schema.rs

reviewed

··· 112 112 did -> Text, 113 113 subject -> Text, 114 114 created_at -> Timestamptz, 115 115 + via_uri -> Nullable<Text>, 116 116 + via_cid -> Nullable<Text>, 115 117 } 116 118 } 117 119

migrations/.diesel_lock

reviewed

This is a binary file and will not be displayed.

+1

migrations/2026-04-26-173028-0000_follow-via/down.sql

reviewed

··· 1 1 + alter table follows drop column via_uri, drop column via_cid;

+3

migrations/2026-04-26-173028-0000_follow-via/up.sql

reviewed

··· 1 1 + alter table follows 2 2 + add column via_uri text, 3 3 + add column via_cid text;