tomdebone/maarcadetweet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(appview): profile cache + Jetstream indexing + denormalised counts

Browse Source 
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.
This commit is contained in:
tomdebone
2026-07-18 17:56:52 +02:00
parent 3064d3d8b7
commit 59a3cb02dd
6 changed files with 585 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/appview/src/indexer.rs
+410 -16
View File
@@ -189,7 +189,7 @@ impl Type<Postgres> 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<Value>,
pub langs: Option<Vec<String>>,
pub created_at: chrono::DateTime<chrono::Utc>,
/// 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<String>,
}
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<String>>(
"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<String> {
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<PgPool> {
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<String>, // display_name
Option<String>, // description
Option<String>, // avatar_cid
Option<String>, // 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<String> = 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<String>, Option<String>, Option<String>) = 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;
}
}