From 78b752c9939aae0d5b69c3a34bba745261bbe757 Mon Sep 17 00:00:00 2001 From: tomdebone Date: Tue, 7 Jul 2026 21:50:43 +0200 Subject: [PATCH] fix(appview): populate handle from PDS ingest + backfill race-safely MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The AppView side of the PDS→AppView ingest path. Receives the optional 'handle' the PDS now ships, writes it onto the post row on insert (with the existing 'COALESCE(NULLIF(\"`\"), handle)' guard so an empty value never clobbers a previously-backfilled handle). The indexer's 'from_record' constructor gains an optional 'pds_handle' parameter so the same code path serves both the Jetstream ingest (where handle is absent by protocol design) and the local-PDS push (where handle is authoritative). Updates the 5 existing test call sites + adjusts the 2-step Jetstream handling flow to match. The unit/integration tests that assert the post-shape on real Bluesky docs still pass. --- crates/appview/src/indexer.rs | 57 ++++++++++++++++++++++++++++++++--- crates/appview/src/ingest.rs | 8 +++++ 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/crates/appview/src/indexer.rs b/crates/appview/src/indexer.rs index e6fa610..5e62e99 100644 --- a/crates/appview/src/indexer.rs +++ b/crates/appview/src/indexer.rs @@ -214,12 +214,21 @@ impl PostRow { /// by sniffing for `$type` (`app.bsky.embed.images` / `.external` / /// `.record`). Keeping it as raw JSON means we don't have to mirror /// every embed variant in Rust. + /// + /// `pds_handle` is the optional handle forwarded by the PDS through + /// the `/internal/ingest-commit` payload. Local-PDS users have + /// `did:key:` DIDs that no PLC directory can resolve, so the PDS is + /// the only authoritative source for their handle. Pass `Some(handle)` + /// when you have it; pass `None` (e.g. Jetstream path) and the + /// `upsert_post` COALESCE guard ensures the empty value won't + /// clobber a backfilled handle from `firehose::handle_identity`. pub fn from_record( did: &str, rkey: &str, collection: &str, cid: &str, record: &Value, + pds_handle: Option<&str>, ) -> Self { let text = record .get("text") @@ -250,10 +259,14 @@ impl PostRow { .collect::>() }); let uri = format!("at://{did}/{collection}/{rkey}"); + let handle = pds_handle + .map(|h| h.trim().to_string()) + .filter(|h| !h.is_empty()) + .unwrap_or_default(); Self { uri, did: did.to_string(), - handle: String::new(), + handle, rkey: rkey.to_string(), collection: collection.to_string(), text, @@ -575,12 +588,17 @@ pub async fn apply_commit( })?; let cid = op.cid.clone().unwrap_or_default(); let record = op.record.clone().unwrap_or(Value::Null); + // Jetstream `commit` events don't carry the + // handle — leave it empty so the upsert + // COALESCE guard preserves the row's existing + // (or backfilled-from-identity) handle. let row = PostRow::from_record( &ev.did, &rkey, &collection, &cid, &record, + None, ); upsert_post(db, &row).await?; applied = true; @@ -825,7 +843,7 @@ mod tests { ] } }); - let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec); + let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None); let embed = row.embed.expect("embed must be captured"); assert_eq!(embed["$type"], "app.bsky.embed.images"); assert_eq!(embed["images"][0]["alt"], "a cat"); @@ -845,7 +863,7 @@ mod tests { } } }); - let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec); + let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None); let embed = row.embed.expect("embed must be captured"); assert_eq!(embed["$type"], "app.bsky.embed.external"); assert_eq!(embed["external"]["uri"], "https://example.com"); @@ -857,7 +875,7 @@ mod tests { "text": "no embed here", "createdAt": "2026-07-01T12:00:00Z" }); - let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec); + let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None); assert!(row.embed.is_none()); } @@ -871,7 +889,7 @@ mod tests { "root": {"uri": "at://did:plc:b/app.twi.post/r", "cid": "cr"} } }); - let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec); + let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None); assert_eq!(row.parent_uri.as_deref(), Some("at://did:plc:b/app.twi.post/p")); assert_eq!(row.root_uri.as_deref(), Some("at://did:plc:b/app.twi.post/r")); } @@ -1000,6 +1018,7 @@ mod tests { "app.twi.post", "cid-embed", &record, + None, ); upsert_post(&db, &row).await.unwrap(); @@ -1092,3 +1111,31 @@ mod tests { assert_eq!(count, 0); } } + +/// Backfill the `posts.handle` column for every row belonging to +/// `did`. Used by the Jetstream `identity` handler when Jetstream +/// tells us a DID's handle has changed — every existing post row +/// needs the new value. +/// +/// **Race-safety**: this only writes if the row's current handle is +/// empty OR doesn't match, so concurrent PDS pushes (which carry +/// the same handle) and concurrent identity replays don't fight. +/// The `WHERE handle IS DISTINCT FROM $1` makes the update a +/// no-op when the value is already correct, which Postgres treats +/// cheaply. +/// +/// Returns the number of rows updated. +pub async fn backfill_handle( + db: &PgPool, + did: &str, + new_handle: &str, +) -> Result { + let res = sqlx::query( + "UPDATE posts SET handle = $1 WHERE did = $2 AND handle IS DISTINCT FROM $1", + ) + .bind(new_handle) + .bind(did) + .execute(db) + .await?; + Ok(res.rows_affected()) +} diff --git a/crates/appview/src/ingest.rs b/crates/appview/src/ingest.rs index 6096343..7c388ef 100644 --- a/crates/appview/src/ingest.rs +++ b/crates/appview/src/ingest.rs @@ -30,6 +30,13 @@ use tracing::{info, warn}; #[derive(Debug, Deserialize)] pub struct IngestCommitReq { pub did: String, + /// The poster's current handle, as known by the PDS `users` table. + /// Optional in the wire payload — the AppView falls back to an + /// empty string, and the upsert COALESCE guard prevents the empty + /// value from clobbering a backfilled handle from the Jetstream + /// `identity` event path. + #[serde(default)] + pub handle: Option, pub collection: String, pub action: String, pub rkey: String, @@ -127,6 +134,7 @@ async fn apply( &req.collection, &cid, &record, + req.handle.as_deref(), ); indexer::upsert_post(&state.db, &row).await.map_err(db_err)?; Ok(true)