From 59a3cb02ddef6cea38bae091fdbee852550898ee Mon Sep 17 00:00:00 2001 From: tomdebone Date: Sat, 18 Jul 2026 17:56:52 +0200 Subject: [PATCH] feat(appview): profile cache + Jetstream indexing + denormalised counts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the AppView-side half of the profile feature so non-local-PDS authors also get their profile metadata indexed (the Jetstream identity event stream only carries the handle, not display name / bio / avatar). The PDS-push path was already wired by the previous commit; this lands the Jetstream path. Migration 0005: * `profiles` table keyed by DID with display_name / description / avatar_cid / banner_cid plus denormalised post_count / follower_count / following_count. Backfilled from the posts table on apply. * `posts.avatar_cid` column — populated from the profiles cache at `upsert_post` time so the PostCard can render an avatar inline without a per-row PDS round trip. Migration 0007 (clean-up): the original 0005 also created a `LOWER(handle)` index that no query uses; this drops it idempotently so dev DBs that already applied 0005 converge. Indexer (`crates/appview/src/indexer.rs`): * New `app.bsky.actor.profile` arm in `apply_commit` calls `upsert_profile` on create, DELETEs the row on delete. Handle is looked up from `posts` (the Jetstream commit envelope doesn't carry it). * `upsert_post` signature is now `&mut PostRow` so it can fill `row.avatar_cid` from the profiles cache; the ON CONFLICT clause uses `COALESCE(EXCLUDED, posts)` so re-indexing doesn't overwrite an already-known avatar. * `upsert_profile` writes display_name / description / avatar_cid / banner_cid + the denormalised counts. * `blob_link_of` helper accepts both `{ $type, ref.$link }` and legacy flat `{ $link }` blob-ref shapes. Ingest (`crates/appview/src/ingest.rs`): * `app.bsky.actor.profile` create/delete arms in the PDS-push path. The handle-fallback previously did `SELECT handle FROM users WHERE did = $1` — but the AppView has no `users` table (it's PDS-owned state). Replaced with a simple use-what-the-PDS- sent approach; the handle_sync worker fills the column later. Routes (`crates/appview/src/routes.rs`): * `resolve_profile` reads the denormalised profile fields from the cache. When no profile row exists the `post_count` fallback uses a live `SELECT COUNT(*)` instead of `posts.len()`, so prolific authors without a profile row report the real count rather than the 50-post slice cap. Tests (DB-gated, run when DATABASE_URL_APPVIEW is set): * `blob_link_of_modern_shape` / `_legacy_flat_link` / `_missing_field`. * `upsert_profile_round_trip` — insert + replace semantics. * `apply_commit_indexes_profile_create` — end-to-end Jetstream arm + delete. --- crates/appview/src/indexer.rs | 426 +++++++++++++++++- crates/appview/src/ingest.rs | 37 +- crates/appview/src/routes.rs | 46 ++ crates/appview/src/routes/types.rs | 27 ++ migrations/appview/0005_profiles.sql | 55 +++ .../0007_drop_unused_profile_handle_idx.sql | 12 + 6 files changed, 585 insertions(+), 18 deletions(-) create mode 100644 migrations/appview/0005_profiles.sql create mode 100644 migrations/appview/0007_drop_unused_profile_handle_idx.sql diff --git a/crates/appview/src/indexer.rs b/crates/appview/src/indexer.rs index 5e62e99..69263be 100644 --- a/crates/appview/src/indexer.rs +++ b/crates/appview/src/indexer.rs @@ -189,7 +189,7 @@ impl Type for EmbedColumn { } #[derive(Debug)] -pub struct PostRow { + pub struct PostRow { pub uri: String, pub did: String, pub handle: String, @@ -202,6 +202,11 @@ pub struct PostRow { pub embed: Option, pub langs: Option>, pub created_at: chrono::DateTime, + /// Resolved author avatar CID from the `profiles` cache. Populated + /// at `upsert_post` time so the PostCard can render an avatar + /// without a per-row PDS round trip. NULL for users whose profile + /// hasn't been pushed yet. + pub avatar_cid: Option, } impl PostRow { @@ -276,30 +281,57 @@ impl PostRow { embed, langs, created_at, + // Avatar CID is populated later by the upsert path via a + // `SELECT avatar_cid FROM profiles WHERE did = $1` lookup, + // so a freshly indexed post starts at None. (The lookup + // happens in `upsert_post_with_avatar` below.) + avatar_cid: None, } } } /// Insert or update a post row keyed by URI. Idempotent. /// +/// `row.avatar_cid` is filled in-place with the current profile-avatar +/// CID for the row's author (from the `profiles` cache, NULL if the +/// profile hasn't been pushed yet). The ON CONFLICT clause uses +/// `COALESCE(EXCLUDED, posts)` so a backfill on a re-indexed post +/// won't overwrite an avatar we already had. +/// /// IMPORTANT: `indexed_at` is NOT touched on conflict. We deliberately /// preserve the original insert time so the `(indexed_at, uri)` keyset /// pagination order is stable across Jetstream replays / PDS re-syncs. -pub async fn upsert_post(db: &PgPool, row: &PostRow) -> Result<()> { +pub async fn upsert_post(db: &PgPool, row: &mut PostRow) -> Result<()> { + if row.avatar_cid.is_none() { + // Look up the latest avatar CID for this author from the + // profiles cache (populated by the PDS push path on profile + // updates). NULL if the profile hasn't been ingested yet — + // the post will display as the initial-letter avatar until + // the user uploads one. + row.avatar_cid = sqlx::query_scalar::<_, Option>( + "SELECT avatar_cid FROM profiles WHERE did = $1", + ) + .bind(&row.did) + .fetch_optional(db) + .await? + .flatten(); + } sqlx::query( r#"INSERT INTO posts (uri, did, handle, rkey, collection, text, cid, - parent_uri, root_uri, embed, langs, created_at) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) + parent_uri, root_uri, embed, langs, created_at, + avatar_cid) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT (uri) DO UPDATE SET - text = EXCLUDED.text, - cid = EXCLUDED.cid, - handle = COALESCE(NULLIF(EXCLUDED.handle,''), posts.handle), - parent_uri = EXCLUDED.parent_uri, - root_uri = EXCLUDED.root_uri, - embed = EXCLUDED.embed, - langs = EXCLUDED.langs, - created_at = EXCLUDED.created_at"#, + text = EXCLUDED.text, + cid = EXCLUDED.cid, + handle = COALESCE(NULLIF(EXCLUDED.handle,''), posts.handle), + parent_uri = EXCLUDED.parent_uri, + root_uri = EXCLUDED.root_uri, + embed = EXCLUDED.embed, + langs = EXCLUDED.langs, + created_at = EXCLUDED.created_at, + avatar_cid = COALESCE(EXCLUDED.avatar_cid, posts.avatar_cid)"#, ) .bind(&row.uri) .bind(&row.did) @@ -313,6 +345,7 @@ pub async fn upsert_post(db: &PgPool, row: &PostRow) -> Result<()> { .bind(EmbedColumn(row.embed.clone())) .bind(&row.langs) .bind(row.created_at) + .bind(&row.avatar_cid) .execute(db) .await?; Ok(()) @@ -592,7 +625,7 @@ pub async fn apply_commit( // handle — leave it empty so the upsert // COALESCE guard preserves the row's existing // (or backfilled-from-identity) handle. - let row = PostRow::from_record( + let mut row = PostRow::from_record( &ev.did, &rkey, &collection, @@ -600,7 +633,7 @@ pub async fn apply_commit( &record, None, ); - upsert_post(db, &row).await?; + upsert_post(db, &mut row).await?; applied = true; } else if op.action == "delete" { let rkey = op @@ -720,6 +753,42 @@ pub async fn apply_commit( } applied = true; } + "app.bsky.actor.profile" => { + // Jetstream carries profile records as plain + // commit ops (no separate collection). We treat + // any rkey — usually `self`, but spec allows + // rkey-rotation — as the user's authoritative + // profile and upsert into the `profiles` cache. + // + // The Jetstream `commit` envelope doesn't carry + // the handle; we look it up from the `posts` + // table (backfilled there by the `identity` + // event stream). Empty is fine — the next + // handle_sync pass will populate it. + if op.action == "create" { + let record = match op.record.clone() { + Some(r) if !r.is_null() => r, + _ => continue, + }; + let handle: String = sqlx::query_scalar( + "SELECT handle FROM posts \ + WHERE did = $1 AND handle <> '' \ + ORDER BY indexed_at DESC LIMIT 1", + ) + .bind(&ev.did) + .fetch_optional(db) + .await? + .unwrap_or_default(); + upsert_profile(db, &ev.did, &handle, &record).await?; + applied = true; + } else if op.action == "delete" { + sqlx::query("DELETE FROM profiles WHERE did = $1") + .bind(&ev.did) + .execute(db) + .await?; + applied = true; + } + } _ => { // Unrecognised collection — ignore (may happen when Jetstream // sends something we didn't subscribe to). @@ -1012,7 +1081,7 @@ mod tests { ] } }); - let row = PostRow::from_record( + let mut row = PostRow::from_record( "did:plc:embed", "embedkey", "app.twi.post", @@ -1020,7 +1089,7 @@ mod tests { &record, None, ); - upsert_post(&db, &row).await.unwrap(); + upsert_post(&db, &mut row).await.unwrap(); let embed: serde_json::Value = sqlx::query_scalar( "SELECT embed FROM posts WHERE uri = $1", @@ -1139,3 +1208,328 @@ pub async fn backfill_handle( .await?; Ok(res.rows_affected()) } + +/// Upsert a `profiles` row for `did`. The caller has just ingested +/// the profile record body (decoded CBOR), and the denormalised +/// counts are computed here (a single `SELECT COUNT(*)` over each +/// side-table — cheap with the existing PK indexes on `posts.did` and +/// `follows.{follower,subject}_did`). +pub async fn upsert_profile( + db: &PgPool, + did: &str, + handle: &str, + record: &Value, +) -> Result<()> { + let display_name = record + .get("displayName") + .and_then(|v| v.as_str()) + .map(str::to_string); + let description = record + .get("description") + .and_then(|v| v.as_str()) + .map(str::to_string); + let avatar_cid = blob_link_of(record, "avatar"); + let banner_cid = blob_link_of(record, "banner"); + + // Denormalised counts. Cheap with the existing PKs. + let post_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM posts WHERE did = $1 \ + AND collection IN ('app.twi.post','app.bsky.feed.post')", + ) + .bind(did) + .fetch_one(db) + .await?; + let follower_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM follows WHERE subject_did = $1", + ) + .bind(did) + .fetch_one(db) + .await?; + let following_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM follows WHERE follower_did = $1", + ) + .bind(did) + .fetch_one(db) + .await?; + + sqlx::query( + r#"INSERT INTO profiles + (did, handle, display_name, description, + avatar_cid, banner_cid, + post_count, follower_count, following_count, indexed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now()) + ON CONFLICT (did) DO UPDATE SET + handle = EXCLUDED.handle, + display_name = EXCLUDED.display_name, + description = EXCLUDED.description, + avatar_cid = EXCLUDED.avatar_cid, + banner_cid = EXCLUDED.banner_cid, + post_count = EXCLUDED.post_count, + follower_count= EXCLUDED.follower_count, + following_count= EXCLUDED.following_count, + indexed_at = now()"#, + ) + .bind(did) + .bind(handle) + .bind(display_name) + .bind(description) + .bind(avatar_cid) + .bind(banner_cid) + .bind(post_count) + .bind(follower_count) + .bind(following_count) + .execute(db) + .await?; + Ok(()) +} + +/// Pull a blob-ref `$link` out of a profile record's field. +/// Accepts both the modern shape +/// (`{ $type: "blob", ref: { $link: "..." } }`) +/// and the legacy shape (`{ $link: "..." }`) for robustness. +fn blob_link_of(record: &Value, field: &str) -> Option { + let v = record.get(field)?; + // Try `ref.$link` first, then flat `$link`. + if let Some(link) = v.get("ref").and_then(|r| r.get("$link")).and_then(|s| s.as_str()) { + return Some(link.to_string()); + } + v.get("$link").and_then(|s| s.as_str()).map(str::to_string) +} + +#[cfg(test)] +mod profile_tests { + use super::*; + use serde_json::json; + + /// Open the appview DB used by integration tests, running + /// migrations first. Returns `None` when no DB is reachable so + /// the test can `eprintln!` and bail (no panic). + async fn try_test_db() -> Option { + let url = std::env::var("DATABASE_URL_APPVIEW").ok()?; + match timeout( + Duration::from_secs(2), + sqlx::PgPool::connect(&url), + ) + .await + { + Ok(Ok(pool)) => match sqlx::migrate!("../../migrations/appview").run(&pool).await { + Ok(()) => Some(pool), + Err(_) => None, + }, + _ => None, + } + } + + #[test] + fn blob_link_of_modern_shape() { + let rec = json!({ + "avatar": { + "$type": "blob", + "ref": { "$link": "bafyavatar" }, + "mimeType": "image/png", + "size": 1234 + } + }); + assert_eq!(blob_link_of(&rec, "avatar").as_deref(), Some("bafyavatar")); + } + + #[test] + fn blob_link_of_legacy_flat_link() { + let rec = json!({ "banner": { "$link": "bafybanner" } }); + assert_eq!(blob_link_of(&rec, "banner").as_deref(), Some("bafybanner")); + } + + #[test] + fn blob_link_of_missing_field() { + let rec = json!({ "displayName": "x" }); + assert_eq!(blob_link_of(&rec, "avatar"), None); + } + + #[tokio::test] + async fn upsert_profile_round_trip() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + // Use a unique DID per test run so we don't collide with the + // migration backfill (which seeded a row for every distinct + // DID in `posts`). + let did = format!( + "did:plc:profile_test_{}", + uuid::Uuid::new_v4().simple() + ); + let handle = format!("user.{}.test", uuid::Uuid::new_v4().simple()); + + // Seed a couple of posts so post_count is non-zero. + for rkey in &["p1", "p2"] { + sqlx::query( + r#"INSERT INTO posts + (uri, did, handle, rkey, collection, text, cid, + parent_uri, root_uri, langs, created_at) + VALUES ($1,$2,$3,$4,'app.twi.post','seed','bafy',NULL,NULL,NULL, now()) + ON CONFLICT (uri) DO NOTHING"#, + ) + .bind(format!("at://{did}/app.twi.post/{rkey}")) + .bind(&did) + .bind(&handle) + .bind(rkey) + .execute(&db) + .await + .unwrap(); + } + + let record = json!({ + "displayName": "Alice", + "description": "tester", + "avatar": { "$type": "blob", "ref": { "$link": "bafyavatar" } }, + "banner": { "$type": "blob", "ref": { "$link": "bafybanner" } } + }); + upsert_profile(&db, &did, &handle, &record).await.unwrap(); + + let row: ( + String, // handle + Option, // display_name + Option, // description + Option, // avatar_cid + Option, // banner_cid + i64, // post_count + i64, // follower_count + i64, // following_count + ) = sqlx::query_as( + "SELECT handle, display_name, description, avatar_cid, banner_cid, \ + post_count, follower_count, following_count \ + FROM profiles WHERE did = $1", + ) + .bind(&did) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(row.0, handle); + assert_eq!(row.1.as_deref(), Some("Alice")); + assert_eq!(row.2.as_deref(), Some("tester")); + assert_eq!(row.3.as_deref(), Some("bafyavatar")); + assert_eq!(row.4.as_deref(), Some("bafybanner")); + assert_eq!(row.5, 2, "post_count must reflect seeded posts"); + + // Update: change display name, drop banner — verify replace + // semantics (NULL fields overwrite, not coalesce). + let record2 = json!({ "displayName": "Alice 2" }); + upsert_profile(&db, &did, &handle, &record2).await.unwrap(); + let name: Option = sqlx::query_scalar( + "SELECT display_name FROM profiles WHERE did = $1", + ) + .bind(&did) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(name.as_deref(), Some("Alice 2")); + + // Cleanup. + let _ = sqlx::query("DELETE FROM profiles WHERE did = $1") + .bind(&did) + .execute(&db) + .await; + let _ = sqlx::query("DELETE FROM posts WHERE did = $1") + .bind(&did) + .execute(&db) + .await; + } + + /// `apply_commit` must dispatch an `app.bsky.actor.profile` + /// create op into the `profiles` cache (this is the path Jetstream + /// uses for third-party PDS authors). + #[tokio::test] + async fn apply_commit_indexes_profile_create() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let did = format!( + "did:plc:profile_commit_{}", + uuid::Uuid::new_v4().simple() + ); + let _ = sqlx::query("DELETE FROM profiles WHERE did = $1") + .bind(&did) + .execute(&db) + .await; + + // Seed a post so the indexer can find a known handle. + sqlx::query( + r#"INSERT INTO posts + (uri, did, handle, rkey, collection, text, cid, + parent_uri, root_uri, langs, created_at) + VALUES ($1,$2,$3,'seed','app.twi.post','hi','bafy',NULL,NULL,NULL, now()) + ON CONFLICT (uri) DO NOTHING"#, + ) + .bind(format!("at://{did}/app.twi.post/seed")) + .bind(&did) + .bind("alice.test") + .execute(&db) + .await + .unwrap(); + + let commit = json!({ + "operation": "create", + "collection": "app.bsky.actor.profile", + "rkey": "self", + "cid": "bafyprofilecid", + "record": { + "displayName": "Alice", + "description": "hello", + "avatar": { "$type": "blob", "ref": { "$link": "bafyavatar" } } + } + }); + let ev = JetstreamEvent { + did: did.clone(), + time_us: 1_700_000_000_000_000, + kind: "commit".into(), + commit: Some(commit), + identity: None, + account: None, + }; + let applied = apply_commit(&db, &ev).await.unwrap(); + assert!(applied); + + let row: (Option, Option, Option) = sqlx::query_as( + "SELECT display_name, description, avatar_cid \ + FROM profiles WHERE did = $1", + ) + .bind(&did) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(row.0.as_deref(), Some("Alice")); + assert_eq!(row.1.as_deref(), Some("hello")); + assert_eq!(row.2.as_deref(), Some("bafyavatar")); + + // Delete op should wipe the row. + let del = json!({ + "operation": "delete", + "collection": "app.bsky.actor.profile", + "rkey": "self" + }); + let ev_del = JetstreamEvent { + did: did.clone(), + time_us: 1_700_000_001_000_000, + kind: "commit".into(), + commit: Some(del), + identity: None, + account: None, + }; + apply_commit(&db, &ev_del).await.unwrap(); + let remaining: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM profiles WHERE did = $1", + ) + .bind(&did) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(remaining, 0, "delete op must remove profile row"); + + // Cleanup. + let _ = sqlx::query("DELETE FROM posts WHERE did = $1") + .bind(&did) + .execute(&db) + .await; + } +} diff --git a/crates/appview/src/ingest.rs b/crates/appview/src/ingest.rs index 7c388ef..612781d 100644 --- a/crates/appview/src/ingest.rs +++ b/crates/appview/src/ingest.rs @@ -128,7 +128,7 @@ async fn apply( .clone() .unwrap_or_else(|| serde_json::json!({"text": "", "createdAt": chrono::Utc::now().to_rfc3339()})); let cid = req.cid.clone().unwrap_or_default(); - let row = indexer::PostRow::from_record( + let mut row = indexer::PostRow::from_record( &req.did, &req.rkey, &req.collection, @@ -136,7 +136,7 @@ async fn apply( &record, req.handle.as_deref(), ); - indexer::upsert_post(&state.db, &row).await.map_err(db_err)?; + indexer::upsert_post(&state.db, &mut row).await.map_err(db_err)?; Ok(true) } ("app.twi.post", "delete") | ("app.bsky.feed.post", "delete") => { @@ -219,6 +219,39 @@ async fn apply( .map_err(db_err)?; Ok(true) } + ("app.bsky.actor.profile", "create") if req.rkey == "self" => { + // Profile record push from the PDS — populate the + // `profiles` cache so the ProfileView-Page and PostCard + // avatar get the new display name / bio / avatar / banner + // without waiting for the next handle-sync pass. + let record = match &req.record { + Some(r) if !r.is_null() => r.clone(), + _ => return Ok(false), + }; + // Use the handle the PDS provided when present. We + // deliberately do NOT fall back to a DB lookup here: + // the AppView has no `users` table — the PDS owns that + // state. If the PDS omits the handle, we write an empty + // string and the `handle_sync` worker (or a subsequent + // Jetstream `identity` event) will fill it in. + let handle = req + .handle + .clone() + .filter(|h| !h.is_empty()) + .unwrap_or_default(); + indexer::upsert_profile(&state.db, &req.did, &handle, &record) + .await + .map_err(db_err)?; + Ok(true) + } + ("app.bsky.actor.profile", "delete") if req.rkey == "self" => { + sqlx::query("DELETE FROM profiles WHERE did = $1") + .bind(&req.did) + .execute(&state.db) + .await + .map_err(db_err)?; + Ok(true) + } (coll, action) => { // Unrecognised collection/action — return ok=false so the PDS // doesn't retry. Future collections should be added above. diff --git a/crates/appview/src/routes.rs b/crates/appview/src/routes.rs index d14154a..2c32b71 100644 --- a/crates/appview/src/routes.rs +++ b/crates/appview/src/routes.rs @@ -372,6 +372,11 @@ async fn resolve_profile( posts: Vec::new(), followers: 0, following: 0, + display_name: None, + description: None, + avatar_cid: None, + banner_cid: None, + post_count: 0, })); }; @@ -427,12 +432,51 @@ async fn resolve_profile( .await .map_err(db_err)?; + // Look up the denormalised profile metadata for this DID. May be + // None if the user has no profile record yet (a brand-new account, + // or a Jetstream-only author whose PDS we don't know about). In + // that case we fall back to a live `SELECT COUNT(*)` over `posts` + // so the `post_count` field reflects the true number of posts + // rather than the size of the (LIMIT-50'd) slice we just returned + // — otherwise a prolific author with no profile row would report + // `post_count: 50` no matter how many posts they actually have. + let profile_row: Option<(Option, Option, Option, Option, i64)> = + sqlx::query_as( + "SELECT display_name, description, avatar_cid, banner_cid, post_count + FROM profiles + WHERE did = $1", + ) + .bind(&target_did) + .fetch_optional(&state.db) + .await + .map_err(db_err)?; + let (display_name, description, avatar_cid, banner_cid, post_count) = match profile_row { + Some(row) => row, + None => { + let real_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*)::BIGINT FROM posts \ + WHERE did = $1 \ + AND collection IN ('app.twi.post','app.bsky.feed.post')", + ) + .bind(&target_did) + .fetch_one(&state.db) + .await + .map_err(db_err)?; + (None, None, None, None, real_count) + } + }; + Ok(Json(ProfileResponse { did: target_did, handle: display_handle, posts, followers, following, + display_name, + description, + avatar_cid, + banner_cid, + post_count, })) } @@ -830,6 +874,7 @@ mod tests { created_at: Utc::now(), like_count: 0, repost_count: 0, + avatar_cid: None, }, PostRow { uri: "at://x/app.twi.post/2".into(), @@ -846,6 +891,7 @@ mod tests { created_at: Utc::now(), like_count: 0, repost_count: 0, + avatar_cid: None, }, ]; decorate_handles(&mut rows); diff --git a/crates/appview/src/routes/types.rs b/crates/appview/src/routes/types.rs index 048da6d..ce9f62e 100644 --- a/crates/appview/src/routes/types.rs +++ b/crates/appview/src/routes/types.rs @@ -54,6 +54,12 @@ pub struct PostRow { pub like_count: i64, #[serde(default)] pub repost_count: i64, + /// Resolved author-avatar CID from the `profiles` cache. NULL + /// until the user has pushed a profile through the PDS path. The + /// PostCard uses this to render an + /// inline without a per-row PDS round trip. + #[serde(skip_serializing_if = "Option::is_none")] + pub avatar_cid: Option, } /// Raw `FromRow` impl — we read `embed` as the helper newtype then @@ -76,6 +82,7 @@ impl<'r> FromRow<'r, sqlx::postgres::PgRow> for PostRow { created_at: row.try_get("created_at")?, like_count: row.try_get::("like_count").unwrap_or(0), repost_count: row.try_get::("repost_count").unwrap_or(0), + avatar_cid: row.try_get::, _>("avatar_cid").ok().flatten(), }) } } @@ -100,6 +107,8 @@ pub struct PostRowWithIndexed { pub indexed_at: DateTime, pub like_count: i64, pub repost_count: i64, + /// Resolved author-avatar CID from the `profiles` cache. + pub avatar_cid: Option, } impl<'r> FromRow<'r, sqlx::postgres::PgRow> for PostRowWithIndexed { @@ -121,6 +130,7 @@ impl<'r> FromRow<'r, sqlx::postgres::PgRow> for PostRowWithIndexed { indexed_at: row.try_get("indexed_at")?, like_count: row.try_get::("like_count").unwrap_or(0), repost_count: row.try_get::("repost_count").unwrap_or(0), + avatar_cid: row.try_get::, _>("avatar_cid").ok().flatten(), }) } } @@ -142,6 +152,7 @@ impl From for PostRow { created_at: r.created_at, like_count: r.like_count, repost_count: r.repost_count, + avatar_cid: r.avatar_cid, } } } @@ -162,6 +173,22 @@ pub struct ProfileResponse { pub posts: Vec, pub followers: i64, pub following: i64, + /// Denormalised profile metadata from the `profiles` cache. + /// Optional — populated when the user has a profile record + /// pushed to the AppView (PDS write or Jetstream `identity` event). + #[serde(skip_serializing_if = "Option::is_none")] + pub display_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub description: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub avatar_cid: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub banner_cid: Option, + /// Denormalised count of the user's posts (computed by + /// `upsert_profile` from the `posts` table). Lets the + /// ProfileView-Page render without an extra `COUNT(*)`. + #[serde(default)] + pub post_count: i64, } /// `GET /api/search` response. `q` echoes the search string so the diff --git a/migrations/appview/0005_profiles.sql b/migrations/appview/0005_profiles.sql new file mode 100644 index 0000000..3ec7fdc --- /dev/null +++ b/migrations/appview/0005_profiles.sql @@ -0,0 +1,55 @@ +-- AppView database schema 0005: profile metadata + per-post avatar refs. +-- +-- Why +-- The Profile-View-Page and PostCard both render a user avatar. +-- Fetching the live `app.bsky.actor.profile/self` record from every +-- user's PDS on every render doesn't scale, and isn't always reachable +-- (e.g. a did:web: user whose PDS is offline). We cache the +-- denormalised profile fields the UI shows keyed by did, indexed by +-- handle so the `/api/profile/` lookup is index-driven. +-- +-- Source of truth: the user's own PDS. The PDS pushes profile records +-- via the existing `/internal/ingest-commit` path; this migration +-- adds the `(collection, action, rkey) == ('app.bsky.actor.profile', +-- 'create', 'self')` arm to the AppView's indexer to populate this +-- table. +-- +-- All fields nullable: a profile record can omit displayName, +-- description, avatar, banner independently. +-- +-- post_count / follower_count / following_count are denormalised +-- counts populated only when the row is created/replaced; the +-- Profile-View-Page reads them here so it doesn't have to issue a +-- separate COUNT(*) over posts/follows. +-- +-- The avatar_cid on posts is the resolved profile-avatar blob ref +-- (or NULL) for the post's author. The AppView fills it in at +-- upsert_post-time from the profiles table; the PostCard reads it +-- to inline an without a per-row +-- PDS round-trip. + +CREATE TABLE profiles ( + did TEXT PRIMARY KEY, + handle TEXT NOT NULL, + display_name TEXT, + description TEXT, + avatar_cid TEXT, + banner_cid TEXT, + post_count BIGINT NOT NULL DEFAULT 0, + follower_count BIGINT NOT NULL DEFAULT 0, + following_count BIGINT NOT NULL DEFAULT 0, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX profiles_indexed_at_idx ON profiles (indexed_at DESC); + +-- Backfill: seed a profile row for every handle we've already +-- resolved through the posts.did → posts.handle mapping. The display +-- fields stay NULL — they need the live PDS profile record. +INSERT INTO profiles (did, handle) +SELECT DISTINCT ON (did) did, handle + FROM posts + WHERE handle <> '' + ORDER BY did, indexed_at DESC +ON CONFLICT (did) DO NOTHING; + +ALTER TABLE posts ADD COLUMN IF NOT EXISTS avatar_cid TEXT; diff --git a/migrations/appview/0007_drop_unused_profile_handle_idx.sql b/migrations/appview/0007_drop_unused_profile_handle_idx.sql new file mode 100644 index 0000000..6bd7204 --- /dev/null +++ b/migrations/appview/0007_drop_unused_profile_handle_idx.sql @@ -0,0 +1,12 @@ +-- AppView database schema 0007: drop unused `profiles_handle_idx`. +-- +-- The original 0005 migration created a `LOWER(handle)` index on +-- `profiles`, anticipating handle-based lookups. In practice every +-- caller derives a DID first (via `posts.handle` or the handle-sync +-- worker) and then queries `profiles` by PK — so the index is dead +-- weight in storage and write-amplification cost. +-- +-- This migration drops it idempotently (`IF EXISTS`) so dev DBs that +-- already applied 0005 also converge. New installs no longer create +-- the index (0005 was edited when this was discovered). +DROP INDEX IF EXISTS profiles_handle_idx;