fix(appview): populate handle from PDS ingest + backfill race-safely

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

@@ -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::<Vec<_>>()
             });
         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<u64> {
+    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())
+}

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<String>,
     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)