Compare commits

..
4 Commits
Author SHA1 Message Date
tomdebone ffee5c6685 feat(tauri-app): profile view, Avatar component, handle navigation
UI half of the profile feature. Mirrors the previous three commits
so the user can browse and edit profiles.

* `<Avatar did cid name size>` — reusable avatar component.
  Falls back to an initial-letter (or "?" when name is empty) circle
  when `cid` is null. Resolves the blob through the standard PDS
  fetch path so it works for any author whose PDS the client can
  reach.
* `<UserProfileView handle on_thread_click current_user_did>` —
  public profile page. Fetches `GET /api/profile/<handle>` on
  mount, renders the avatar / display name / bio / counts /
  posts. The "edit profile" button is gated on
  `current_user_did === profile.did` so a user browsing
  someone else's profile can't issue an unintended `setMyProfile`
  against their own DID.
* `PostCard` now renders an inline `<Avatar>` + clickable handle
  button that calls a new `on_handle_click` prop. The clickable
  area replaces the previous dead `<a href>` (Tauri webviews
  have no router).
* `App.svelte` adds an `openUserProfile(handle)` handler that
  sets `selectedHandle` + `view = "user"` and mounts
  `<UserProfileView>`.
* New Tauri commands `profile_get_record` / `profile_set` in
  `lib.rs` + matching client helpers `getMyProfile` /
  `setMyProfile` in `client.ts`. The set command sends camelCase
  field names; the PDS endpoint (previous commit) round-trips them
  through `#[serde(rename_all = "camelCase")]`.
* Empty-state UX for users with no profile yet (new account, or a
  third-party-PDS author whose profile the AppView hasn't indexed
  yet): both the current-user "profile" view and the public
  "user" view render a hint ("// no profile yet — click 'edit
  profile' to set one up." / "// no profile yet.") instead of a
  blank bio box.
* NavRail / NavRailHarness `View` union extended with "user"
  so the navigation prop type accepts the new view.
2026-07-18 17:57:15 +02:00
tomdebone 59a3cb02dd feat(appview): profile cache + Jetstream indexing + denormalised counts
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.
2026-07-18 17:56:52 +02:00
tomdebone 3064d3d8b7 feat(pds-server): app.bsky.actor.profile get/set XRPC endpoints
Read / read-modify-write the authenticated user's profile record
through the standard atproto repo-write path. Auth is checked via
the existing bearer-token helper; the request body's
`display_name` / `description` / `avatar_blob_cid` /
`banner_blob_cid` overlay the existing record (None fields
preserve the old value).

Blob CID ownership: any supplied avatar/banner CID is looked up
in the `blobs` table with `WHERE cid = $1 AND did = $2`,
rejecting with 400 if the blob isn't owned by the authenticated
user. The resolved `mime_type` / `size` is written into the
record so consumers reading `size` for layout decisions get the
real value (previously hardcoded to "image/png" / 0).

Best-effort push to the AppView via `AppViewPushClient::push_profile`
so the `profiles` cache reflects the new avatar / display name
without waiting for the Jetstream replay path.

Wire shape:

  GET  /xrpc/app.bsky.actor.profile.get
       → { did, handle, profile: { displayName, description, ... } | null }

  POST /xrpc/app.bsky.actor.profile.set
       body: { displayName, description, avatarBlobCid, bannerBlobCid }
       → same shape as get

Includes `merge_profile_fields` testable helper (4 unit tests
locking the camelCase wire shape and the merge semantics).

The AppView-side indexer arm and the Tauri UI land in the
following two commits.
2026-07-18 17:56:28 +02:00
tomdebone 3aa5d5c0e3 feat(at-identity): PdsHandleResolver for cluster-local DID→handle resolution
The AppView's handle_sync worker consulted the public PLC directory
and the did:web: HTTPS resolver only. DIDs hosted on the local PDS
(notably did🔑 users and any other operator-hosted method)
weren't reachable without an external round trip, and unresolvable
DIDs (did🔑 not on this PDS, did:foo: anything) blocked the
100-row batch forever because did🔑 sorts lexicographically
before did:plc: / did:web:.

This commit adds:

* `PdsHandleResolver` (at-identity) — POSTs the DID as the
  `handle` field to the PDS's resolveHandle XRPC method. The PDS
  now recognises a `did:` prefix and does a PK lookup on
  `users.did`, returning `{did, handle}`. The resolver reads
  the `handle` field, so the AppView finally gets a real local
  handle for did🔑 users without ever dialing plc.directory.
* A 2 s timeout per request (was 10 s) and `DISPATCH_CONCURRENCY =
  8` so the worker caps a 100-DID batch at ~2 s with parallel
  dispatch instead of the ~17 min worst case the old serial + 10 s
  setup allowed.
* A new `posts.handle_sync_attempted_at` column (migration 0006)
  and `mark_attempted()` helper. The SELECT filter excludes rows
  attempted within the last hour, so an unresolvable DID dominates
  at most one batch before the worker advances. Cleared on success.
* `PDS_INTERNAL_URL` config so the AppView can reach the PDS via
  a cluster-internal hostname when the public URL isn't routable
  from inside the cluster.

Tests:
* `crates/at-identity/src/pds_handle.rs` — 4 unit tests against a
  stub HTTP server (200/404/5xx/missing-did-field).
* Existing handle_sync integration tests updated to wire in the
  new `pds_resolver` field.
2026-07-18 17:56:11 +02:00
32 changed files with 2325 additions and 66 deletions
Generated
+1
View File
@@ -51,6 +51,7 @@ dependencies = [
"axum",
"base64",
"chrono",
"futures",
"reqwest",
"rustls",
"serde",
+1
View File
@@ -37,6 +37,7 @@ at-identity = { workspace = true }
uuid = { workspace = true }
rustls = { version = "0.23", default-features = false, features = ["ring", "std", "logging", "tls12"] }
base64 = { workspace = true }
futures = { workspace = true }
[dev-dependencies]
tokio = { workspace = true }
+89 -17
View File
@@ -19,8 +19,10 @@
//! 2. For each DID, dispatches by method:
//! * `did:plc:` → [`HandleSyncWorker::plc_resolver`]
//! * `did:web:` → [`HandleSyncWorker::web_resolver`]
//! * anything else (e.g. `did:key:`) → skipped
//! * anything else (e.g. `did:garbage:`) → skipped
//! A resolver returning `Ok(None)` counts as `skipped`, not `failed`.
//! Before any of these the local PDS is consulted, so a `did:key:`
//! user on this PDS gets resolved without dialing plc.directory.
//! 3. `UPDATE posts SET handle = $1 WHERE did = $2 AND handle = ''` so
//! concurrent syncs (or the `/internal/ingest-commit` path, which can
//! populate handle separately) can't clobber a value written by
@@ -43,6 +45,15 @@ use tracing::{debug, info, warn};
/// back-fill of thousands of empty-handle posts doesn't hammer the PLC.
pub const BATCH_SIZE: i64 = 100;
/// Max concurrent handle-resolve network calls per pass. Each
/// DID in a batch triggers a `POST /xrpc/com.atproto.identity
/// .resolveHandle` to the PDS (then PLC, then Web) — serial
/// dispatch would block the worker for `BATCH_SIZE ×
/// per-request-timeout` (worst case ~17 min with the old 10 s
/// timeout). Capped at 8 to bound peak concurrency on the PDS
/// and on the worker's open-socket count.
const DISPATCH_CONCURRENCY: usize = 8;
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct SyncReport {
/// Rows whose `handle` column was newly populated this pass.
@@ -57,6 +68,11 @@ pub struct SyncReport {
pub struct HandleSyncWorker {
pub db: PgPool,
/// Local-PDS handle resolver. Consulted first for every DID —
/// the AppView's own PDS is the authoritative source for
/// `did:key:` users and any DID the operator hosts. A 404 from
/// the PDS falls through to the public resolvers below.
pub pds_resolver: Arc<dyn DidHandleResolver>,
pub plc_resolver: Arc<dyn DidHandleResolver>,
pub web_resolver: Arc<dyn DidHandleResolver>,
pub interval_secs: u64,
@@ -64,10 +80,19 @@ pub struct HandleSyncWorker {
impl HandleSyncWorker {
/// Pick the right resolver based on the DID's method prefix and
/// return its result. Unknown methods (`did:key:`, etc.) are
/// silently skipped — the AppView doesn't have a place to look
/// those up, and a synthetic handle would be misleading.
/// return its result. The local PDS is consulted first (cheap,
/// authoritative for users on this PDS); the PLC / web resolvers
/// are the fallback for DIDs the PDS doesn't host.
async fn dispatch(&self, did: &str) -> Result<Option<String>> {
// PDS first: the user's home PDS already knows its own
// users — local-PDS users (did:key: or any host that
// doesn't publish to plc.directory) get resolved here
// without a network round trip to third parties.
if let Some(h) = self.pds_resolver.resolve_handle(did).await? {
if !h.is_empty() {
return Ok(Some(h));
}
}
if did.starts_with("did:plc:") {
self.plc_resolver.resolve_handle(did).await
} else if did.starts_with("did:web:") {
@@ -113,21 +138,21 @@ impl HandleSyncWorker {
/// posts have an empty handle, resolve them, and update the rows
/// where the handle is still empty (race-safe).
///
/// **SQL-level filter**: we exclude `did:key:` entirely because
/// there's no resolver path for them — the PLC directory and the
/// `did:web:` HTTPS resolver both reject non-`did:plc:` /
/// non-`did:web:` DIDs with `Ok(None)`. Previously the worker
/// picked via `ORDER BY did LIMIT 100`, but lexicographically
/// `did:key:` sorts before `did:plc:` / `did:web:`, so the worker
/// would process the same 100 `did:key:` rows every 300 s and
/// never reach any resolvable DID. Filtering at SQL time makes
/// every batch contribute real work.
/// Unresolvable DIDs (e.g. `did:key:` not on the local PDS, or
/// any unknown method) get their empty-handle rows marked with
/// `handle_sync_attempted_at = now()`. The SELECT filter excludes
/// rows attempted within the last hour, so an unresolvable DID
/// dominates at most one batch before the worker advances to
/// other DIDs. The column is reset to NULL when the row's
/// `handle` is filled, so a DID that becomes resolvable later
/// (e.g. the user joins the local PDS) gets re-attempted.
pub async fn run_once(&self) -> Result<SyncReport> {
let dids: Vec<(String,)> = sqlx::query_as(
r#"SELECT DISTINCT did
FROM posts
WHERE handle = ''
AND (did LIKE 'did:plc:%' OR did LIKE 'did:web:%')
AND (handle_sync_attempted_at IS NULL
OR handle_sync_attempted_at < now() - interval '1 hour')
ORDER BY did
LIMIT $1"#,
)
@@ -140,15 +165,34 @@ impl HandleSyncWorker {
return Ok(report);
}
for (did,) in dids {
match self.dispatch(&did).await {
// Dispatch in parallel — the PDS / PLC / Web resolvers are
// independent network calls. Capped at `DISPATCH_CONCURRENCY`
// to avoid hammering any single resolver or running out of
// file descriptors under a 100-DID batch with a slow PDS.
// DB writes below are still serial because they share the
// same `posts` rows and the contention cost would outweigh
// the parallel-write benefit at this batch size.
use futures::stream::{self, StreamExt};
let dispatch_results: Vec<(String, anyhow::Result<Option<String>>)> = stream::iter(dids)
.map(|(did,)| async move {
let r = self.dispatch(&did).await;
(did, r)
})
.buffer_unordered(DISPATCH_CONCURRENCY)
.collect()
.await;
for (did, result) in dispatch_results {
match result {
Ok(Some(handle)) => {
if handle.is_empty() {
report.skipped += 1;
mark_attempted(&self.db, &did).await?;
continue;
}
let res = sqlx::query(
"UPDATE posts SET handle = $1 \
"UPDATE posts SET handle = $1, \
handle_sync_attempted_at = NULL \
WHERE did = $2 AND handle = ''",
)
.bind(&handle)
@@ -165,10 +209,19 @@ impl HandleSyncWorker {
}
Ok(None) => {
report.skipped += 1;
mark_attempted(&self.db, &did).await?;
}
Err(e) => {
warn!(did = %did, error = %e, "handle resolve failed");
report.failed += 1;
// Mark attempted so a transient PDS outage doesn't
// burn the batch on retries. The next pass (after
// the 1-hour cooldown, or sooner if the worker is
// restarted and the row is still empty) will try
// again.
if let Err(e) = mark_attempted(&self.db, &did).await {
warn!(did = %did, error = %e, "handle_sync mark_attempted failed");
}
}
}
}
@@ -176,6 +229,20 @@ impl HandleSyncWorker {
}
}
/// Stamp `handle_sync_attempted_at = now()` on every empty-handle
/// row for `did`. Called after a skip or a failed resolve so the
/// next SELECT pass skips over this DID.
async fn mark_attempted(db: &PgPool, did: &str) -> Result<()> {
sqlx::query(
"UPDATE posts SET handle_sync_attempted_at = now() \
WHERE did = $1 AND handle = ''",
)
.bind(did)
.execute(db)
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
@@ -219,6 +286,7 @@ mod tests {
fn worker_with(db: PgPool, stub: Arc<dyn DidHandleResolver>) -> HandleSyncWorker {
HandleSyncWorker {
db,
pds_resolver: Arc::clone(&stub),
plc_resolver: Arc::clone(&stub),
web_resolver: Arc::clone(&stub),
interval_secs: 999,
@@ -383,6 +451,7 @@ mod tests {
let worker = HandleSyncWorker {
db: db.clone(),
pds_resolver: Arc::clone(&resolver),
plc_resolver: Arc::clone(&resolver),
web_resolver: Arc::clone(&resolver),
interval_secs: 999,
@@ -440,6 +509,7 @@ mod tests {
let worker = HandleSyncWorker {
db: db.clone(),
pds_resolver: Arc::clone(&resolver),
plc_resolver: Arc::clone(&resolver),
web_resolver: Arc::clone(&resolver),
interval_secs: 999,
@@ -496,6 +566,7 @@ mod tests {
let worker = HandleSyncWorker {
db: db.clone(),
pds_resolver: Arc::clone(&plc),
plc_resolver: plc,
web_resolver: web,
interval_secs: 999,
@@ -554,6 +625,7 @@ mod tests {
let worker = HandleSyncWorker {
db: db.clone(),
pds_resolver: Arc::clone(&plc_arc),
plc_resolver: plc_arc,
web_resolver: web_arc,
interval_secs: 999,
+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;
}
}
+35 -2
View File
@@ -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.
+19
View File
@@ -86,8 +86,27 @@ async fn main() -> Result<()> {
cfg.plc_directory_url.clone(),
));
let web: Arc<dyn DidHandleResolver> = Arc::new(at_identity::WebResolver::new());
// PDS-local handle resolver. The handle_sync worker consults it
// first, before the public PLC/web resolvers, so `did:key:`
// users (and any other DID the operator hosts on this PDS) get their
// local handle without a round trip to plc.directory. A 404 from
// the PDS falls through to the public resolvers.
//
// Use the cluster-internal URL when configured (e.g. `http://pds:3000`
// inside docker compose) — `pds_public_url` may not be reachable
// from inside the cluster when TLS / DNS is set up for outside
// clients only.
let pds_base_url = cfg
.pds_internal_url
.clone()
.unwrap_or_else(|| cfg.pds_public_url.clone());
let pds_resolver: Arc<dyn at_identity::DidHandleResolver> = Arc::new(
at_identity::pds_handle::PdsHandleResolver::new(pds_base_url),
);
let handle_sync = handle_sync::HandleSyncWorker {
db: db.clone(),
pds_resolver,
plc_resolver: plc,
web_resolver: web,
interval_secs: cfg.appview_handle_sync_interval_secs,
+46
View File
@@ -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<String>, Option<String>, Option<String>, Option<String>, 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);
+27
View File
@@ -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 <Avatar cid={post.avatar_cid}/>
/// inline without a per-row PDS round trip.
#[serde(skip_serializing_if = "Option::is_none")]
pub avatar_cid: Option<String>,
}
/// 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::<i64, _>("like_count").unwrap_or(0),
repost_count: row.try_get::<i64, _>("repost_count").unwrap_or(0),
avatar_cid: row.try_get::<Option<String>, _>("avatar_cid").ok().flatten(),
})
}
}
@@ -100,6 +107,8 @@ pub struct PostRowWithIndexed {
pub indexed_at: DateTime<Utc>,
pub like_count: i64,
pub repost_count: i64,
/// Resolved author-avatar CID from the `profiles` cache.
pub avatar_cid: Option<String>,
}
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::<i64, _>("like_count").unwrap_or(0),
repost_count: row.try_get::<i64, _>("repost_count").unwrap_or(0),
avatar_cid: row.try_get::<Option<String>, _>("avatar_cid").ok().flatten(),
})
}
}
@@ -142,6 +152,7 @@ impl From<PostRowWithIndexed> 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<PostRow>,
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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub avatar_cid: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub banner_cid: Option<String>,
/// 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
@@ -67,13 +67,14 @@ impl DidHandleResolver for StubResolver {
}
}
/// Build a worker whose PLC and web resolvers are both the same stub.
/// The integration tests in this file don't care which method the
/// DID uses — the stub answers for any prefix.
/// Build a worker whose PDS, PLC and web resolvers are all the same
/// stub. The integration tests in this file don't care which method
/// the DID uses — the stub answers for any prefix.
fn worker_with(db: sqlx::PgPool, stub: Arc<StubResolver>) -> HandleSyncWorker {
let r: Arc<dyn DidHandleResolver> = stub;
HandleSyncWorker {
db,
pds_resolver: Arc::clone(&r),
plc_resolver: Arc::clone(&r),
web_resolver: Arc::clone(&r),
interval_secs: 999,
@@ -389,6 +390,7 @@ async fn sync_resolves_did_web_via_web_resolver() {
let worker = HandleSyncWorker {
db: db.clone(),
pds_resolver: Arc::clone(&plc_arc),
plc_resolver: plc_arc,
web_resolver: web_arc,
interval_secs: 999,
@@ -442,6 +444,7 @@ async fn sync_resolves_did_plc_via_plc_resolver() {
let worker = HandleSyncWorker {
db: db.clone(),
pds_resolver: Arc::clone(&plc_arc),
plc_resolver: plc_arc,
web_resolver: web_arc,
interval_secs: 999,
+2
View File
@@ -1,7 +1,9 @@
pub mod handle;
pub mod pds_handle;
pub mod plc;
pub mod web;
pub use handle::{resolve_handle, DidHandleResolver, HandleResolver};
pub use pds_handle::PdsHandleResolver;
pub use plc::{submit_op, PlcClient};
pub use web::WebResolver;
+184
View File
@@ -0,0 +1,184 @@
//! PDS-first handle resolver.
//!
//! The local PDS is the authoritative source for `did:key:` users (and
//! for any DID the operator hosts on this PDS). The `AppView`'s
//! `handle_sync` worker consults this resolver *before* falling through
//! to the public PLC directory / `did:web:` HTTPS resolver, so local
//! users get their handle without ever dialing out to plc.directory.
//!
//! ### Wire shape
//!
//! PDS endpoint: `POST /xrpc/com.atproto.identity.resolveHandle`
//! with body `{ "handle": "<did>" }`. The PDS's handler is
//! polymorphic on the `handle` field: if it starts with `did:`
//! the PDS looks the row up by `did` (PK), otherwise by `handle`.
//! On success the PDS returns `{ "did": "...", "handle": "..." }`
//! — we read the `handle` field, which is what the worker
//! actually needs to fill `posts.handle`. A 404 means the DID
//! isn't hosted here, and the worker falls through to PLC/Web.
//!
//! Network: `reqwest::Client` with a 10 s timeout. The same client
//! is reused across requests — handle with `Arc<PdsHandleResolver>`
//! in the worker.
use crate::handle::DidHandleResolver;
use anyhow::Result;
use async_trait::async_trait;
use serde_json::json;
use std::time::Duration;
pub struct PdsHandleResolver {
pub base_url: String,
pub client: reqwest::Client,
}
impl PdsHandleResolver {
pub fn new(base_url: impl Into<String>) -> Self {
// 2 s is plenty for a colocated PDS (typical round-trip
// < 100 ms in dev) but bounds the per-DID cost when the
// PDS is unreachable — at 100 DIDs/pass that caps a
// single pass at ~2 s with parallel dispatch, vs the
// ~17 min worst-case the old 10 s timeout allowed with
// serial dispatch.
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(2))
.build()
.expect("reqwest client build should never fail");
Self {
base_url: base_url.into(),
client,
}
}
}
#[async_trait]
impl DidHandleResolver for PdsHandleResolver {
async fn resolve_handle(&self, did: &str) -> Result<Option<String>> {
// POST /xrpc/com.atproto.identity.resolveHandle with
// { "handle": "<did>" } in the body. The PDS's handler
// recognises a `did:` prefix and does a PK lookup on
// `users.did`, returning `{ did, handle }` on match.
let url = format!(
"{}/xrpc/com.atproto.identity.resolveHandle",
self.base_url
);
let r = self
.client
.post(&url)
.json(&json!({ "handle": did }))
.send()
.await?;
if r.status().as_u16() == 404 {
return Ok(None);
}
if !r.status().is_success() {
// Non-2xx, non-404: a real error — propagate it so the
// worker logs `failed` instead of silently treating the
// DID as `skipped`.
anyhow::bail!(
"pds handle resolver returned {}",
r.status()
);
}
let v: serde_json::Value = r.json().await?;
// The PDS returns `{ "did": "...", "handle": "..." }` on
// success. We read `handle` (what we actually want for
// `posts.handle`) and ignore the echoed `did`.
Ok(v.get("handle").and_then(|x| x.as_str()).map(String::from))
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
use std::net::SocketAddr;
use tokio::net::TcpListener;
/// Spawn a one-shot HTTP listener that responds to the
/// resolveHandle XRPC call with the configured body + status.
/// Returns the bound address (so the resolver under test can hit
/// `http://127.0.0.1:<port>`).
async fn spawn_stub(
status: u16,
body: serde_json::Value,
) -> SocketAddr {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => break,
};
let mut buf = vec![0u8; 8192];
let n = sock.read(&mut buf).await.unwrap_or(0);
if n == 0 {
continue;
}
let body_s = body.to_string();
let resp = format!(
"HTTP/1.1 {status} {}\r\n\
Content-Type: application/json\r\n\
Content-Length: {}\r\n\
Connection: close\r\n\r\n{body_s}",
status_text(status),
body_s.len(),
);
let _ = sock.write_all(resp.as_bytes()).await;
}
});
addr
}
fn status_text(s: u16) -> &'static str {
match s {
200 => "OK",
404 => "Not Found",
500 => "Internal Server Error",
_ => "Status",
}
}
#[tokio::test]
async fn returns_handle_on_match() {
let addr = spawn_stub(
200,
json!({ "did": "did:plc:abc", "handle": "alice.bsky" }),
)
.await;
let r = PdsHandleResolver::new(format!("http://{addr}"));
let h = r.resolve_handle("did:plc:abc").await.unwrap();
assert_eq!(h.as_deref(), Some("alice.bsky"));
}
#[tokio::test]
async fn returns_none_on_404() {
let addr = spawn_stub(404, json!({ "error": "NotFound" })).await;
let r = PdsHandleResolver::new(format!("http://{addr}"));
let h = r.resolve_handle("did:plc:unknown").await.unwrap();
assert!(h.is_none());
}
#[tokio::test]
async fn returns_err_on_5xx() {
let addr = spawn_stub(500, json!({ "error": "oops" })).await;
let r = PdsHandleResolver::new(format!("http://{addr}"));
let result = r.resolve_handle("did:plc:abc").await;
assert!(
result.is_err(),
"5xx must propagate as Err so the worker logs `failed`, not `skipped`"
);
}
#[tokio::test]
async fn returns_handle_even_when_did_field_missing() {
// Some PDS implementations might return `{ "handle": "x" }`
// without echoing the DID. We must still extract the handle.
let addr = spawn_stub(200, json!({ "handle": "bob.bsky" })).await;
let r = PdsHandleResolver::new(format!("http://{addr}"));
let h = r.resolve_handle("did:plc:bob").await.unwrap();
assert_eq!(h.as_deref(), Some("bob.bsky"));
}
}
+9
View File
@@ -28,6 +28,14 @@ pub struct AppConfig {
pub s3_bucket_pds: String,
pub s3_bucket_appview: String,
pub plc_directory_url: String,
/// Cluster-internal URL the AppView uses to reach the PDS (e.g.
/// `http://pds-server:3000`). Falls back to `pds_public_url` when
/// unset. Splitting this from `pds_public_url` lets a single
/// deployment point the AppView at the in-cluster PDS hostname
/// (which may not be reachable from outside) while clients
/// still see the public URL.
#[serde(default)]
pub pds_internal_url: Option<String>,
/// Optional shared secret for `POST /internal/ingest-commit`. If unset,
/// the endpoint accepts anonymous requests (dev mode). If set, callers
/// must send `X-Ingest-Secret: <value>`.
@@ -68,6 +76,7 @@ impl AppConfig {
s3_bucket_pds: env("S3_BUCKET_PDS")?,
s3_bucket_appview: env("S3_BUCKET_APPVIEW")?,
plc_directory_url: env("PLC_DIRECTORY_URL")?,
pds_internal_url: std::env::var("PDS_INTERNAL_URL").ok(),
appview_ingest_secret: std::env::var("APPVIEW_INGEST_SECRET").ok(),
appview_handle_sync_interval_secs: std::env::var("APPVIEW_HANDLE_SYNC_INTERVAL_SECS")
.ok()
+23
View File
@@ -152,6 +152,29 @@ impl AppViewPushClient {
.await
}
/// Push an `app.bsky.actor.profile` create event to the AppView
/// so the `profiles` cache stays in sync with the user's own PDS.
/// Best-effort — if the AppView is unreachable, the Jetstream
/// replay path eventually picks it up.
pub async fn push_profile(
&self,
did: &str,
handle: &str,
record: &serde_json::Value,
) -> Result<bool> {
self.push(
did,
Some(handle),
"app.bsky.actor.profile",
"create",
"self",
None,
Some(record),
None,
)
.await
}
async fn push(
&self,
did: &str,
+8
View File
@@ -123,6 +123,14 @@ pub fn router(state: AppState) -> Router {
"/xrpc/com.atproto.sync.getRecord",
get(routes::sync::get_record),
)
.route(
"/xrpc/app.bsky.actor.profile.get",
get(routes::profile::get_profile),
)
.route(
"/xrpc/app.bsky.actor.profile.set",
post(routes::profile::set_profile),
)
.route(
"/xrpc/com.atproto.sync.listRepos",
get(routes::sync::list_repos),
+35 -7
View File
@@ -9,17 +9,42 @@ pub async fn resolve_handle(
State(state): State<AppState>,
Json(req): Json<ResolveHandleReq>,
) -> Result<Json<ResolveHandleResp>, (StatusCode, Json<crate::routes::types::ErrorBody>)> {
// Polymorphic input: `handle` may be a bare handle OR a DID
// (`did:plc:…`, `did:web:…`, `did:key:…`). When it's a DID we
// look the row up by `did` — the AppView's `PdsHandleResolver`
// uses this path to fill `posts.handle` for local-PDS users
// (including `did:key:`) without a second round-trip to plc.directory.
if req.handle.starts_with("did:") {
let row: Option<(String,)> =
sqlx::query_as("SELECT handle FROM users WHERE did = $1")
.bind(&req.handle)
.fetch_optional(&state.db)
.await
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?;
if let Some((handle,)) = row {
return Ok(Json(ResolveHandleResp {
did: req.handle.clone(),
handle: Some(handle),
}));
}
// DID not hosted here — fall through to the handle lookup
// (returns 404 below).
}
if let Some(zone) = state.cfg.pds_handle_dns_zone.strip_prefix(".") {
if let Some(stripped) = req.handle.strip_suffix(zone) {
let user = stripped.trim_end_matches('.');
let full = format!("{}{}", user, state.cfg.pds_handle_dns_zone);
let row: Option<(String,)> = sqlx::query_as("SELECT did FROM users WHERE handle = $1")
.bind(&full)
.fetch_optional(&state.db)
.await
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?;
let row: Option<(String,)> =
sqlx::query_as("SELECT did FROM users WHERE handle = $1")
.bind(&full)
.fetch_optional(&state.db)
.await
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?;
if let Some((did,)) = row {
return Ok(Json(ResolveHandleResp { did }));
return Ok(Json(ResolveHandleResp {
did,
handle: Some(full),
}));
}
}
}
@@ -29,7 +54,10 @@ pub async fn resolve_handle(
.await
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?;
match row {
Some((did,)) => Ok(Json(ResolveHandleResp { did })),
Some((did,)) => Ok(Json(ResolveHandleResp {
did,
handle: Some(req.handle.clone()),
})),
None => {
warn!(handle = %req.handle, "handle not found");
Err(err(
+1
View File
@@ -3,6 +3,7 @@ pub mod blob;
pub mod feed;
pub mod helpers;
pub mod identity;
pub mod profile;
pub mod repo;
pub mod sync;
pub mod types;
+499
View File
@@ -0,0 +1,499 @@
//! `app.bsky.actor.profile.get` and `app.bsky.actor.profile.set`.
//!
//! ### `get`
//!
//! Read the profile record (CBOR-decoded `app.bsky.actor.profile/self`
//! value block) for the authenticated user. The handle/did come from
//! the JWT; the record is parsed into a JSON object. Returns `null`
//! for the `profile` field when the user has no profile yet (a brand
//! new account).
//!
//! ### `set`
//!
//! Read-modify-write of the user's `app.bsky.actor.profile/self`
//! record. The body carries only the fields the caller wants to
//! change; the existing record is fetched and the supplied fields
//! overwrite the corresponding fields. Best-effort push to the
//! AppView follows so the `profiles` cache reflects the new avatar /
//! display name / bio without waiting for the Jetstream replay.
use crate::jwt_issuer;
use crate::routes::helpers::{
apply_repo_write, err, load_head_commit, load_signing_key, load_user_blockstore,
to_sqlx_error, RepoWriteOutcome,
};
use crate::routes::types::ErrorBody;
use crate::state::AppState;
use at_crypto::cid::cid_for_cbor;
use at_repo::blockstore::Blockstore as _;
use at_repo::repo::Repo;
use axum::extract::State;
use axum::http::StatusCode;
use bytes::Bytes;
use cid::Cid;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tracing::{info, warn};
/// Blob metadata looked up from the `blobs` table. Used to write
/// the `mimeType` / `size` fields of a profile avatar/banner ref —
/// these must reflect what the user actually uploaded, not a
/// hardcoded constant.
#[derive(Debug, Clone)]
pub(crate) struct ResolvedBlob {
pub mime_type: String,
pub size: i64,
}
#[derive(Debug, Serialize)]
pub struct GetProfileResp {
pub did: String,
pub handle: String,
/// `null` when the user has no profile record yet.
pub profile: Option<Value>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SetProfileReq {
/// Optional new display name. `null`/missing preserves the
/// existing record's `displayName`.
pub display_name: Option<String>,
/// Optional new bio / description. `null`/missing preserves.
pub description: Option<String>,
/// CID of the avatar blob, already uploaded via uploadBlob.
/// `null`/missing preserves.
pub avatar_blob_cid: Option<String>,
/// CID of the banner blob. `null`/missing preserves.
pub banner_blob_cid: Option<String>,
}
/// Read the authenticated user's profile record.
pub async fn get_profile(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
) -> Result<axum::Json<GetProfileResp>, (StatusCode, axum::Json<ErrorBody>)> {
let did = authenticate(&headers, &state)?;
let handle: String = sqlx::query_scalar("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(&state.db)
.await
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError", e.to_string()))?;
let profile = read_profile_record(&state, &did).await?;
Ok(axum::Json(GetProfileResp { did, handle, profile }))
}
/// Read-modify-write the authenticated user's profile record.
pub async fn set_profile(
State(state): State<AppState>,
headers: axum::http::HeaderMap,
axum::Json(req): axum::Json<SetProfileReq>,
) -> Result<axum::Json<GetProfileResp>, (StatusCode, axum::Json<ErrorBody>)> {
let did = authenticate(&headers, &state)?;
let handle: String = sqlx::query_scalar("SELECT handle FROM users WHERE did = $1")
.bind(&did)
.fetch_one(&state.db)
.await
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError", e.to_string()))?;
// Fetch the existing record, if any.
let existing = read_profile_record(&state, &did).await?;
// For any blob CIDs in the request, look up the real
// `mime_type` / `size` from the `blobs` table — and verify
// ownership (the blob must belong to the authenticated DID).
// Without the ownership check a session for DID A could
// reference DID B's blob in their profile.
let avatar = match req.avatar_blob_cid.as_deref() {
Some(cid) => Some(resolve_owned_blob(&state, &did, cid).await?),
None => None,
};
let banner = match req.banner_blob_cid.as_deref() {
Some(cid) => Some(resolve_owned_blob(&state, &did, cid).await?),
None => None,
};
// Merge: start from the existing record (or empty object), then
// overlay the supplied fields. We use the atproto standard
// `app.bsky.actor.profile` schema: displayName (string),
// description (string), avatar (blob ref), banner (blob ref).
let next = merge_profile_fields(existing, &req, avatar.as_ref(), banner.as_ref());
// Validate against the lexicon so the user can't push an
// arbitrary JSON shape that wouldn't round-trip through a real
// atproto client.
if let Err(e) = state.lex.validate("app.bsky.actor.profile", &next) {
return Err(err(
StatusCode::BAD_REQUEST,
"InvalidRequest",
format!("lex validation failed: {e}"),
));
}
// Encode the new record as CBOR and write it to the user's
// `app.bsky.actor.profile/self` MST via the canonical repo-write
// path (the same one createRecord uses).
let value_cid: Cid = {
let mut buf = Vec::new();
ciborium::into_writer(&next, &mut buf).map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("cbor: {e}"),
)
})?;
cid_for_cbor(&buf).map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
e.to_string(),
)
})?
};
let next_for_block = next.clone();
let outcome = apply_repo_write(&state, &did, move |repo| {
let value_cid = value_cid;
let next_for_block = next_for_block;
Box::pin(async move {
repo.blockstore
.put(&value_cid, Bytes::from({
let mut buf = Vec::new();
ciborium::into_writer(&next_for_block, &mut buf).unwrap();
buf
}))
.await
.map_err(to_sqlx_error)?;
let (_uri, _returned_cid) = repo
.put_record("app.bsky.actor.profile", "self", value_cid)
.await
.map_err(to_sqlx_error)?;
let commit = repo.commit().await.map_err(to_sqlx_error)?;
let head_cid_bytes = commit.cid.to_bytes().to_vec();
let head_commit_bytes = commit.signed_bytes.clone();
Ok(RepoWriteOutcome {
commit,
head_cid_bytes,
head_commit_bytes,
})
})
})
.await?;
info!(
did = %did,
cid = %outcome.commit.cid,
"profile record created"
);
// Best-effort push to the AppView so the profile cache reflects
// the new avatar / display name / bio without waiting for the
// Jetstream `identity` event to plumb through.
if let Err(e) = state
.appview
.push_profile(&did, &handle, &next)
.await
{
warn!(error = %e, "profile push to AppView failed; Jetstream will catch up");
}
Ok(axum::Json(GetProfileResp {
did,
handle,
profile: Some(next),
}))
}
// -- internals ----------------------------------------------------------------
/// Verify a bearer token and return the authenticated DID.
fn authenticate(
headers: &axum::http::HeaderMap,
state: &AppState,
) -> Result<String, (StatusCode, axum::Json<ErrorBody>)> {
let token = headers
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|s| s.strip_prefix("Bearer "))
.ok_or_else(|| {
err(
StatusCode::UNAUTHORIZED,
"Unauthenticated",
"missing Authorization: Bearer header",
)
})?;
let server_pk = jwt_issuer::server_p256_public_multibase(&state.cfg)
.map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError", e.to_string()))?;
let claims = at_crypto::jwt::verify_jwt(token, &server_pk).map_err(|e| {
err(
StatusCode::UNAUTHORIZED,
"TokenInvalid",
format!("invalid token: {e}"),
)
})?;
Ok(claims.sub)
}
/// Decode the `app.bsky.actor.profile/self` record for `did`, if any.
/// Returns `Ok(None)` when the record doesn't exist.
///
/// Walk the head commit down to the profile/self leaf, fetch the
/// value block, CBOR-decode it into JSON. Mirrors `get_record`'s
/// walk in routes/sync.rs.
async fn read_profile_record(
state: &AppState,
did: &str,
) -> Result<Option<Value>, (StatusCode, axum::Json<ErrorBody>)> {
let (head_cid, head_commit_bytes) = match load_head_commit(state, did).await? {
Some(t) => t,
None => return Ok(None),
};
let signing_key_bytes: Vec<u8> =
sqlx::query_scalar("SELECT signing_key FROM users WHERE did = $1")
.bind(did)
.fetch_one(&state.db)
.await
.map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("users.signing_key read: {e}"),
)
})?;
let signing_key = load_signing_key(&signing_key_bytes)?;
let blockstore = load_user_blockstore(state, did).await?;
blockstore
.put(&head_cid, Bytes::from(head_commit_bytes.clone()))
.await
.map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("blockstore put head: {e:#}"),
)
})?;
let repo: Repo<_> = Repo::load(did.to_string(), signing_key, blockstore.clone(), head_cid)
.await
.map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("repo load: {e:#}"),
)
})?;
let value_cid = match repo.get_record("app.bsky.actor.profile", "self").await {
Ok(Some(c)) => c,
Ok(None) => return Ok(None),
Err(e) => {
return Err(err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("repo.get_record: {e:#}"),
));
}
};
let value_bytes = match blockstore.get(&value_cid).await.map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("blockstore get value: {e:#}"),
)
})? {
Some(b) => b,
None => return Ok(None),
};
let v: Value = ciborium::from_reader(&value_bytes[..]).map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("cbor decode: {e}"),
)
})?;
Ok(Some(v))
}
/// Overlay `req` onto `existing`, returning the merged profile
/// record. Each `Some(_)` field in `req` overwrites the matching
/// field; `None` fields are preserved.
///
/// `avatar` / `banner` are the resolved blobs (looked up from the
/// `blobs` table in `set_profile` so we can write the real
/// `mimeType` / `size`). When `None`, the avatar/banner fields are
/// preserved from `existing`.
///
/// Extracted from `set_profile` so the merge semantics are testable
/// without a running PDS / DB.
pub(crate) fn merge_profile_fields(
existing: Option<Value>,
req: &SetProfileReq,
avatar: Option<&ResolvedBlob>,
banner: Option<&ResolvedBlob>,
) -> Value {
let mut next: Value = existing.unwrap_or_else(|| json!({}));
if let Some(s) = &req.display_name {
next["displayName"] = json!(s);
}
if let Some(s) = &req.description {
next["description"] = json!(s);
}
if let Some(cid) = &req.avatar_blob_cid {
// Caller is responsible for the DB lookup; default to a
// png/0 placeholder only if the lookup somehow returned
// `None` despite a CID being supplied (shouldn't happen
// — `resolve_owned_blob` rejects missing blobs earlier).
let (mime_type, size) = avatar
.map(|b| (b.mime_type.as_str(), b.size))
.unwrap_or(("image/png", 0));
next["avatar"] = json!({
"$type": "blob",
"ref": { "$link": cid },
"mimeType": mime_type,
"size": size,
});
}
if let Some(cid) = &req.banner_blob_cid {
let (mime_type, size) = banner
.map(|b| (b.mime_type.as_str(), b.size))
.unwrap_or(("image/png", 0));
next["banner"] = json!({
"$type": "blob",
"ref": { "$link": cid },
"mimeType": mime_type,
"size": size,
});
}
next
}
/// Look up a blob by CID and verify it's owned by `did`. The
/// ownership check is a security requirement: without it a
/// session for DID A could reference DID B's blob in their own
/// profile (the value would still resolve at fetch time because
/// `com.atproto.sync.getBlob` doesn't check ownership, but the
/// invariant "a profile's avatar belongs to that user" would be
/// broken).
async fn resolve_owned_blob(
state: &AppState,
did: &str,
cid: &str,
) -> Result<ResolvedBlob, (StatusCode, axum::Json<ErrorBody>)> {
let row: Option<(String, i64)> = sqlx::query_as(
"SELECT mime_type, size FROM blobs WHERE cid = $1 AND did = $2",
)
.bind(cid)
.bind(did)
.fetch_optional(&state.db)
.await
.map_err(|e| {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"InternalServerError",
format!("blobs lookup: {e}"),
)
})?;
match row {
Some((mime_type, size)) => Ok(ResolvedBlob { mime_type, size }),
None => Err(err(
StatusCode::BAD_REQUEST,
"InvalidRequest",
format!("blob {cid} not found or not owned by {did}"),
)),
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Round-trip the camelCase JSON the Tauri client sends and
/// confirm every field lands on its `snake_case` Rust
/// counterpart. Catches regressions of the BLOCKER-3 bug
/// (silent camelCase → snake_case mismatch that made every
/// set_profile write an empty record).
#[test]
fn set_profile_req_deserializes_camel_case() {
let raw = json!({
"displayName": "Alice",
"description": "hello",
"avatarBlobCid": "bafyavatar",
"bannerBlobCid": "bafybanner",
});
let req: SetProfileReq = serde_json::from_value(raw).unwrap();
assert_eq!(req.display_name.as_deref(), Some("Alice"));
assert_eq!(req.description.as_deref(), Some("hello"));
assert_eq!(req.avatar_blob_cid.as_deref(), Some("bafyavatar"));
assert_eq!(req.banner_blob_cid.as_deref(), Some("bafybanner"));
}
/// `displayName` only — preserves existing description / avatar.
#[test]
fn merge_preserves_fields_not_in_req() {
let existing = json!({
"displayName": "Old",
"description": "old bio",
"avatar": { "$type": "blob", "ref": { "$link": "old_avatar" } },
});
let req = SetProfileReq {
display_name: Some("New".into()),
description: None,
avatar_blob_cid: None,
banner_blob_cid: None,
};
let next = merge_profile_fields(Some(existing), &req, None, None);
assert_eq!(next["displayName"], "New");
assert_eq!(next["description"], "old bio");
assert_eq!(
next["avatar"]["ref"]["$link"], "old_avatar",
"avatar must be preserved when req.avatar_blob_cid is None"
);
}
/// All fields set on an empty record — the typical first-write
/// path for a brand-new account.
#[test]
fn merge_into_empty_record() {
let req = SetProfileReq {
display_name: Some("Alice".into()),
description: Some("first bio".into()),
avatar_blob_cid: Some("bafyavatar".into()),
banner_blob_cid: None,
};
let avatar = ResolvedBlob {
mime_type: "image/png".into(),
size: 1234,
};
let next = merge_profile_fields(None, &req, Some(&avatar), None);
assert_eq!(next["displayName"], "Alice");
assert_eq!(next["description"], "first bio");
assert_eq!(next["avatar"]["ref"]["$link"], "bafyavatar");
assert_eq!(next["avatar"]["mimeType"], "image/png");
assert_eq!(next["avatar"]["size"], 1234);
assert!(next.get("banner").is_none(), "banner must be absent when not set");
}
/// Blob refs must use the modern `{ $type, ref.$link, mimeType,
/// size }` shape so the AppView's `blob_link_of` helper can
/// parse them back. Locks the wire contract in place.
#[test]
fn merge_writes_blob_refs_in_modern_shape() {
let req = SetProfileReq {
display_name: None,
description: None,
avatar_blob_cid: Some("bafyavatar".into()),
banner_blob_cid: None,
};
let avatar = ResolvedBlob {
mime_type: "image/webp".into(),
size: 999,
};
let next = merge_profile_fields(None, &req, Some(&avatar), None);
assert_eq!(next["avatar"]["$type"], "blob");
assert_eq!(next["avatar"]["ref"]["$link"], "bafyavatar");
// Real mime_type from the blobs table — not hardcoded.
assert_eq!(next["avatar"]["mimeType"], "image/webp");
assert_eq!(next["avatar"]["size"], 999);
}
}
+7
View File
@@ -62,6 +62,13 @@ pub struct ResolveHandleReq {
#[derive(Debug, Serialize)]
pub struct ResolveHandleResp {
pub did: String,
/// Always populated when the lookup succeeds. For
/// handle → DID calls this is just the input echo; for
/// DID → handle calls this is the resolved local handle
/// (used by the AppView's `PdsHandleResolver` to fill
/// `posts.handle` without a second round-trip).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub handle: Option<String>,
}
#[derive(Debug, Deserialize)]
+8
View File
@@ -38,6 +38,14 @@ impl AppState {
"app.bsky.feed.repost".to_string(),
Lex::from_json(include_str!("../../../lexicons/app/bsky/feed/repost.json")).unwrap(),
);
// Profile record — avatar/banner/display name/description.
// Validates the createRecord body when the Tauri client calls
// its setProfile command. Other fields stay optional so a
// brand-new account with an empty profile is legal.
lex.lexicons.insert(
"app.bsky.actor.profile".to_string(),
Lex::from_json(include_str!("../../../lexicons/app/bsky/actor/profile.json")).unwrap(),
);
let plc_url = cfg.plc_directory_url.clone();
// The PDS speaks to the AppView via the cluster-internal URL —
// never the public one, because the ingest endpoint is unauth'd
+58
View File
@@ -720,7 +720,65 @@ pub fn run() {
pick_and_upload_image,
show_notification,
open_external_url,
profile_get_record,
profile_set,
])
.run(tauri::generate_context!())
.expect("error while running maarcadetweet");
}
#[tauri::command]
async fn profile_get_record(
state: tauri::State<'_, AppState>,
) -> Result<Option<serde_json::Value>, String> {
let sess = state
.store
.load()
.ok_or_else(|| "not logged in".to_string())?;
state
.pds
.get_profile_record(&sess.did, &sess.access_jwt)
.await
.map_err(|e| e.to_string())
}
#[tauri::command]
async fn profile_set(
state: tauri::State<'_, AppState>,
fields: serde_json::Value,
) -> Result<(), String> {
let sess = state
.store
.load()
.ok_or_else(|| "not logged in".to_string())?;
let display_name = fields
.get("displayName")
.and_then(|v| v.as_str())
.map(str::to_string);
let description = fields
.get("description")
.and_then(|v| v.as_str())
.map(str::to_string);
let avatar_blob_cid = fields
.get("avatarBlobCid")
.and_then(|v| v.as_str())
.map(str::to_string);
let banner_blob_cid = fields
.get("bannerBlobCid")
.and_then(|v| v.as_str())
.map(str::to_string);
state
.pds
.set_profile(
&sess.did,
&serde_json::json!({
"displayName": display_name,
"description": description,
"avatarBlobCid": avatar_blob_cid,
"bannerBlobCid": banner_blob_cid,
}),
&sess.access_jwt,
)
.await
.map_err(|e| e.to_string())?;
Ok(())
}
@@ -385,3 +385,64 @@ pub struct UploadedBlobRef {
#[serde(rename = "$link")]
pub link: String,
}
/// `POST /xrpc/com.atproto.repo.getRecord?repo=<did>&collection=app.bsky.actor.profile&rkey=self`
/// Returns the record's CBOR-decoded value as JSON, or `None` if no
/// record exists for that path. The server replies with a
/// `{ "value": {...} | null }` envelope; we unwrap and return the
/// inner value (which is the `app.bsky.actor.profile` JSON object
/// keyed by the deserialized CBOR field names: `displayName`,
/// `description`, `avatar`/{ ref, mimeType, size }, `banner`/...).
pub async fn get_profile_record(
&self,
repo: &str,
jwt: &str,
) -> Result<Option<serde_json::Value>> {
let url = format!(
"{}/xrpc/com.atproto.repo.getRecord",
self.base_url
);
let r = self
.client
.get(&url)
.query(&[("repo", repo), ("collection", "app.bsky.actor.profile"), ("rkey", "self")])
.bearer_auth(jwt)
.send()
.await?;
if r.status().as_u16() == 404 {
return Ok(None);
}
if !r.status().is_success() {
let s = r.status();
let body = r.text().await.unwrap_or_default();
anyhow::bail!("getRecord returned {s}: {body}");
}
let v: serde_json::Value = r.json().await?;
Ok(v.get("value").cloned().and_then(|x| if x.is_null() { None } else { Some(x) }))
}
/// `POST /xrpc/app.bsky.actor.profile.set` — PDS-only convenience
/// endpoint that does a read-modify-write of the profile record. The
/// request body has the same shape as `app.bsky.actor.profile` minus
/// the `$type` (added server-side).
pub async fn set_profile(
&self,
repo: &str,
profile: &serde_json::Value,
jwt: &str,
) -> Result<serde_json::Value> {
let url = format!("{}/xrpc/app.bsky.actor.profile.set", self.base_url);
let r = self
.client
.post(&url)
.bearer_auth(jwt)
.json(profile)
.send()
.await?;
if !r.status().is_success() {
let s = r.status();
let body = r.text().await.unwrap_or_default();
anyhow::bail!("setProfile returned {s}: {body}");
}
let v: serde_json::Value = r.json().await?;
Ok(v.get("profile").cloned().unwrap_or(serde_json::Value::Null))
}
+167 -14
View File
@@ -9,6 +9,8 @@
fetchPost,
openExternalUrl,
showError,
pickAndUploadImage,
setMyProfile,
type Session,
type Post,
type ProfileResponse,
@@ -17,13 +19,20 @@
import StatusBar from "./lib/components/StatusBar.svelte";
import PostCard from "./lib/components/PostCard.svelte";
import ComposeBox from "./lib/components/ComposeBox.svelte";
import UserProfileView from "./lib/components/UserProfileView.svelte";
import Avatar from "./lib/components/Avatar.svelte";
import LoginScreen from "./lib/components/LoginScreen.svelte";
import Terminal from "./lib/components/Terminal.svelte";
import Skeleton from "./lib/components/Skeleton.svelte";
type View = "home" | "compose" | "profile" | "search" | "settings";
type View = "home" | "compose" | "profile" | "user" | "search" | "settings";
let view: View = $state("home");
// Handle for the "user" view (i.e. someone else's profile). The
// "profile" view remains the current-user view (the NavRail icon
// goes there). Selecting a handle (via the PostCard avatar link or
// a future deep-link) navigates to "user" with `selectedHandle` set.
let selectedHandle: string = $state("");
let currentUser: Session | null = $state(null);
let status: { did?: string; handle?: string; authenticated: boolean } = $state({ authenticated: false });
@@ -40,6 +49,45 @@
let profileLoading: boolean = $state(false);
let profileError: string | null = $state(null);
// Edit-profile state.
let editingProfile: boolean = $state(false);
let editProfileName: string = $state("");
let editProfileDesc: string = $state("");
let editProfileAvatarCid: string | null = $state(null);
let savingProfile: boolean = $state(false);
async function pickAndUploadAvatar() {
try {
const r = await pickAndUploadImage();
if (r) editProfileAvatarCid = r.cid;
} catch (e) {
console.warn("avatar upload failed", e);
}
}
async function saveProfile() {
if (!currentUser) return;
savingProfile = true;
try {
const updated = await setMyProfile({
displayName: editProfileName || undefined,
description: editProfileDesc || undefined,
avatarBlobCid: editProfileAvatarCid || undefined,
});
// Refresh the cached profile from the response (or re-fetch).
if (updated) {
profile = { ...profile, ...updated } as ProfileResponse | null;
} else {
await refreshProfile(currentUser.handle);
}
editingProfile = false;
} catch (e) {
console.warn("profile save failed", e);
} finally {
savingProfile = false;
}
}
// Search state.
let searchQuery: string = $state("");
let searchResults: Post[] = $state([]);
@@ -81,6 +129,17 @@
threadLoading = false;
}
}
/// Navigate to the "user" profile view for `handle`. Called from
/// `<PostCard on_handle_click>` and the avatar/handle buttons in
/// the post header. The actual profile fetch happens inside
/// `<UserProfileView>` on mount.
function openUserProfile(handle: string) {
selectedHandle = handle;
view = "user";
threadRoot = null;
threadParent = null;
}
function closeThread() {
threadRoot = null;
threadParent = null;
@@ -462,14 +521,14 @@
<div class="toast toast--err">err: {threadError}</div>
{:else if threadRoot}
{#if threadParent && threadParent.uri !== threadRoot.uri}
<div class="thread-parent"><PostCard post={threadParent} /></div>
<div class="thread-parent"><PostCard post={threadParent} on_handle_click={openUserProfile} /></div>
{/if}
<PostCard post={threadRoot} />
<PostCard post={threadRoot} on_handle_click={openUserProfile} />
{/if}
</div>
{/if}
{#each userPosts as p (p.uri)}
<PostCard post={p} on_thread_click={openThread} />
<PostCard post={p} on_thread_click={openThread} on_handle_click={openUserProfile} />
{/each}
{#if timelineCursor}
<div class="loadmore">
@@ -487,6 +546,17 @@
<span class="meta">⌘↵ to post</span>
</div>
<ComposeBox onPosted={handlePosted} />
{:else if view === "user"}
<div class="head">
<span class="prompt">$</span>
<span class="title">// profile —</span>
<span class="as">@{selectedHandle}</span>
</div>
<UserProfileView
handle={selectedHandle}
on_thread_click={openThread}
current_user_did={currentUser?.did ?? null}
/>
{:else if view === "profile"}
<div class="head">
<span class="prompt">$</span>
@@ -502,8 +572,81 @@
<header class="profile__head">
<div class="profile__handle">{displayHandle(profile.handle)}</div>
<div class="profile__did" title={profile.did}>{profile.did}</div>
{#if profile.avatar_cid}
<Avatar
did={profile.did}
cid={profile.avatar_cid}
name={profile.display_name ?? profile.handle}
size={64}
/>
{/if}
</header>
{#if profile.description}
<div class="profile__bio">{profile.description}</div>
{:else}
<div class="profile__bio profile__bio--empty">
// no profile yet — click "edit profile" to set one up.
</div>
{/if}
<div class="profile__actions">
<button
class="btn btn--ghost"
onclick={() => (editingProfile = !editingProfile)}
>
{editingProfile ? "cancel" : "edit profile"}
</button>
</div>
{#if editingProfile}
<div class="profile__edit">
<input
type="text"
placeholder="display name"
maxlength="64"
bind:value={editProfileName}
/>
<textarea
placeholder="description"
rows="3"
maxlength="300"
bind:value={editProfileDesc}
></textarea>
<div class="profile__edit-avatar">
{#if editProfileAvatarCid}
<span class="meta">cid: {editProfileAvatarCid.slice(0, 10)}</span>
<button
class="btn btn--ghost"
onclick={() => (editProfileAvatarCid = null)}
>
clear
</button>
{:else}
<span class="meta">no avatar</span>
{/if}
<button
class="btn btn--ghost"
onclick={pickAndUploadAvatar}
>upload…</button>
</div>
<div class="profile__edit-actions">
<button
class="btn btn--primary"
disabled={savingProfile}
onclick={saveProfile}
>
{savingProfile ? "saving…" : "save"}
</button>
</div>
</div>
{/if}
<div class="profile__stats">
<span>{profile.post_count} posts</span>
<span>{profile.followers} followers</span>
<span>{profile.following} following</span>
</div>
<div class="profile__actions">
<button
class="btn btn--ghost"
@@ -555,7 +698,7 @@
{:else}
<h3 class="profile__h3">// recent posts</h3>
{#each profile.posts as p (p.uri)}
<PostCard post={p} on_thread_click={openThread} />
<PostCard post={p} on_thread_click={openThread} on_handle_click={openUserProfile} />
{/each}
{/if}
</section>
@@ -636,15 +779,15 @@
{:else if view === "search"}
<div class="head">
<span class="prompt">$</span>
<span class="title">// search</span>
<span class="title">// search</span>
<input
class="search"
type="text"
bind:value={searchQuery}
oninput={onSearchInput}
placeholder="grep posts…"
/>
</div>
<input
class="search"
type="text"
bind:value={searchQuery}
oninput={onSearchInput}
placeholder="grep posts…"
/>
{#if searchError}
<div class="toast toast--err">err: {searchError}</div>
{/if}
@@ -657,7 +800,7 @@
{:else}
<div class="meta meta--results">{searchResults.length} result{searchResults.length === 1 ? "" : "s"} for "{searchQuery}"</div>
{#each searchResults as p (p.uri)}
<PostCard post={p} on_thread_click={openThread} />
<PostCard post={p} on_thread_click={openThread} on_handle_click={openUserProfile} />
{/each}
{/if}
{/if}
@@ -852,6 +995,16 @@
color: var(--text-dim);
word-break: break-all;
}
.profile__bio {
font-family: var(--font-mono);
font-size: var(--fs-100);
color: var(--text);
padding: var(--s-2) 0;
}
.profile__bio--empty {
color: var(--text-dim);
font-style: italic;
}
.counts {
display: flex;
gap: var(--s-6);
+35
View File
@@ -191,6 +191,9 @@ export type Post = {
embed?: Embed | null;
langs: string[];
created_at: string;
/// Resolved author-avatar CID from the AppView's `profiles`
/// cache. NULL when the user has no profile record yet.
avatar_cid?: string | null;
};
export type TimelineResponse = {
@@ -204,6 +207,11 @@ export type ProfileResponse = {
posts: Post[];
followers: number;
following: number;
display_name?: string | null;
description?: string | null;
avatar_cid?: string | null;
banner_cid?: string | null;
post_count: number;
};
export type SearchResponse = {
@@ -429,6 +437,33 @@ export async function listenTrayEvents(
/// browser preview where no Tauri runtime is present, fall back
/// to `window.open` and treat a popup-blocker denial as
/// "fine, user can copy the URL themselves".
export type ProfileRecord = {
displayName?: string;
description?: string;
avatar?: { ref: { $link: string }; mimeType?: string; size?: number };
banner?: { ref: { $link: string }; mimeType?: string; size?: number };
};
/// Read the authenticated user's `app.bsky.actor.profile` record.
/// Returns `null` if no profile record exists yet (a brand-new
/// account, or a user whose PDS hasn't pushed one).
export async function getMyProfile(): Promise<ProfileRecord | null> {
return await safeInvoke<ProfileRecord | null>("profile_get_record");
}
/// Read-modify-write the authenticated user's profile. The Rust
/// `profile_set` command fetches the existing record, overlays
/// the supplied fields, and writes a new commit. `undefined` fields
/// are preserved.
export async function setMyProfile(fields: {
displayName?: string;
description?: string;
avatarBlobCid?: string;
bannerBlobCid?: string;
}): Promise<ProfileRecord | null> {
return await safeInvoke<ProfileRecord | null>("profile_set", fields);
}
export async function openExternalUrl(url: string): Promise<void> {
try {
if (isTauri()) {
@@ -0,0 +1,92 @@
<script lang="ts">
import { fetchBlob } from "../api/client";
import { onDestroy } from "svelte";
type Props = {
did: string;
/** Blob-ref `$link` from a posts/post record or a profile
* record. The Avatar component resolves the (did, cid) pair via
* the existing PDS `getBlob` route through `fetchBlob`. NULL
* falls back to the deterministic initial-letter SVG. */
cid?: string | null;
/** Human-readable name used for the initial-letter fallback and
* the alt text. */
name?: string | null;
/** Pixel size. The same component is used at 24 px (PostCard),
* 32 px (Header current-user avatar) and 88 px (Profile-View
* header). */
size?: number;
};
let { did, cid = null, name = "", size = 32 }: Props = $props();
const initial = $derived(
((name || "").trim()[0] || "?").toUpperCase(),
);
let blobUrl: string | null = $state(null);
let lastCid: string | null = null;
$effect(() => {
// Drop the previous blob URL when the CID changes — keeps the
// in-memory cache (managed by `fetchBlob`) lean and avoids leaking
// object URLs across navigations.
if (lastCid !== cid) {
if (blobUrl) URL.revokeObjectURL(blobUrl);
blobUrl = null;
lastCid = cid;
}
if (!cid) return;
let cancelled = false;
fetchBlob(did, cid)
.then((u) => {
if (!cancelled) blobUrl = u;
else URL.revokeObjectURL(u);
})
.catch(() => {
/* Fall back to the initial letter on fetch error. */
});
return () => {
cancelled = true;
};
});
onDestroy(() => {
if (blobUrl) URL.revokeObjectURL(blobUrl);
});
</script>
{#if blobUrl}
<img
class="avatar"
style:width="{size}px"
style:height="{size}px"
src={blobUrl}
alt={name ? `${name}'s avatar` : "avatar"}
/>
{:else}
<span
class="avatar avatar--fallback"
style:width="{size}px"
style:height="{size}px"
style:font-size="{Math.max(10, Math.floor(size * 0.45))}px"
>
{initial}
</span>
{/if}
<style>
.avatar {
display: inline-block;
border-radius: 50%;
object-fit: cover;
background: var(--bg-elev, #1a1a1a);
flex-shrink: 0;
}
.avatar--fallback {
display: inline-flex;
align-items: center;
justify-content: center;
font-family: var(--font-mono, monospace);
color: var(--text-dim, #888);
border: 1px solid var(--line-2, #3a3a3a);
}
</style>
@@ -4,7 +4,7 @@
// `$bindable`, use a callback prop to bubble state changes up to
// the parent.
type View = "home" | "compose" | "profile" | "search" | "settings";
type View = "home" | "compose" | "profile" | "user" | "search" | "settings";
let {
view = "home",
@@ -5,7 +5,7 @@
import NavRail from "./NavRail.svelte";
type View = "home" | "compose" | "profile" | "search" | "settings";
type View = "home" | "compose" | "profile" | "user" | "search" | "settings";
let view: View = $state("home");
</script>
@@ -12,10 +12,21 @@
} from "../api/client";
import EmbedImage from "./EmbedImage.svelte";
import EmbedExternal from "./EmbedExternal.svelte";
import Avatar from "./Avatar.svelte";
import { localStorageKey, useLocalStorage } from "../utils/localstorage";
type Props = { post: Post; on_thread_click?: (uri: string) => void };
let { post, on_thread_click }: Props = $props();
type Props = {
post: Post;
on_thread_click?: (uri: string) => void;
/// Called when the user clicks the handle / avatar in the
/// post header. Tauri webviews don't have a real router, so
/// the host (App.svelte) decides what to do — typically it
/// sets `selectedHandle` + `view = "user"` to render
/// `<UserProfileView>`. When absent the header remains
/// clickable but does nothing.
on_handle_click?: (handle: string) => void;
};
let { post, on_thread_click, on_handle_click }: Props = $props();
// Quoted-post cache. When the post's embed is a `record`, we fetch
// it once on mount and cache it keyed by URI so navigating
@@ -288,8 +299,24 @@
{/if}
<header class="post__head">
<span class="prompt">&gt;</span>
<a class="handle" href={`/profile/${post.handle}`}>@{shortHandle(post.handle)}</a>
<button
class="avatar-btn"
type="button"
title="open profile"
onclick={() => on_handle_click?.(post.handle)}
>
<Avatar
did={post.did}
cid={post.avatar_cid ?? null}
name={post.handle}
size={24}
/>
</button>
<button
class="handle"
type="button"
onclick={() => on_handle_click?.(post.handle)}
>@{shortHandle(post.handle)}</button>
<span class="time">{timeAgo(post.created_at)}</span>
<span class="cid" title={post.cid}>cid: {shortCid(post.cid)}</span>
<span class="did" title={post.did}>{shortDid(post.did)}</span>
@@ -409,7 +436,24 @@
margin-bottom: var(--s-2);
}
.prompt { color: var(--orange); }
.handle { color: var(--text); }
.avatar-btn {
background: none;
border: 0;
padding: 0;
margin: 0;
cursor: pointer;
display: inline-flex;
}
.avatar-btn:hover { opacity: 0.85; }
.handle {
background: none;
border: 0;
padding: 0;
margin: 0;
cursor: pointer;
font: inherit;
color: var(--text);
}
.handle:hover { color: var(--orange); }
.time, .cid, .did { color: var(--cid-fg); }
.did { color: var(--text-dim); }
@@ -0,0 +1,334 @@
<script lang="ts">
import Avatar from "./Avatar.svelte";
import PostCard from "./PostCard.svelte";
import { setMyProfile, pickAndUploadImage } from "../api/client";
import { onMount } from "svelte";
type Props = {
handle: string;
on_thread_click?: (uri: string) => void;
/// DID of the authenticated user. When this matches the
/// profile's DID, the "edit profile" button is shown; otherwise
/// it's hidden (you can only edit your own profile).
current_user_did?: string | null;
};
let { handle, on_thread_click, current_user_did }: Props = $props();
type State =
| { kind: "loading" }
| { kind: "error"; message: string }
| { kind: "ready"; data: AppViewProfile };
type AppViewProfile = {
did: string;
handle: string;
posts: AppViewPost[];
followers: number;
following: number;
display_name?: string;
description?: string;
avatar_cid?: string;
banner_cid?: string;
post_count: number;
};
type AppViewPost = {
uri: string;
did: string;
handle: string;
rkey: string;
collection: string;
text: string;
cid: string;
parent_uri?: string | null;
root_uri?: string | null;
embed?: null;
langs: string[];
created_at: string;
avatar_cid?: string | null;
};
let editing: boolean = $state(false);
let viewModel: State = $state({ kind: "loading" });
let editName: string = $state("");
let editDesc: string = $state("");
let editAvatarCid: string | null = $state(null);
let saving: boolean = $state(false);
async function load() {
viewModel = { kind: "loading" };
try {
const r = await fetch(`/api/profile/${encodeURIComponent(handle)}`);
if (!r.ok) {
viewModel = { kind: "error", message: `profile fetch failed: ${r.status}` };
return;
}
const data: AppViewProfile = await r.json();
viewModel = { kind: "ready", data };
} catch (e) {
viewModel = { kind: "error", message: String(e) };
}
}
function openEdit() {
if (viewModel.kind !== "ready") return;
editName = viewModel.data.display_name ?? "";
editDesc = viewModel.data.description ?? "";
editAvatarCid = viewModel.data.avatar_cid ?? null;
editing = true;
}
async function saveProfile() {
if (viewModel.kind !== "ready") return;
saving = true;
try {
await setMyProfile({
displayName: editName || undefined,
description: editDesc || undefined,
avatarBlobCid: editAvatarCid || undefined,
});
editing = false;
await load();
} catch (e) {
console.warn("profile save failed", e);
} finally {
saving = false;
}
}
async function pickAndUploadAvatar() {
const blob = await pickAndUploadImage();
if (!blob) return;
editAvatarCid = blob.cid;
}
onMount(() => {
void load();
});
</script>
<section class="user-profile">
<header class="user-profile__head">
{#if viewModel.kind === "ready"}
<div class="user-profile__avatar">
<Avatar
did={viewModel.data.did}
cid={viewModel.data.avatar_cid ?? null}
name={viewModel.data.display_name ?? viewModel.data.handle}
size={88}
/>
</div>
<div class="user-profile__id">
<h2 class="user-profile__name">{viewModel.data.display_name ?? "@" + viewModel.data.handle}</h2>
<span class="user-profile__handle">@{viewModel.data.handle}</span>
<span class="user-profile__did" title={viewModel.data.did}>{viewModel.data.did}</span>
</div>
<div class="user-profile__actions">
{#if current_user_did && viewModel.data.did === current_user_did}
<button
class="btn btn--ghost"
type="button"
onclick={() => (editing ? (editing = false) : openEdit())}
>
{editing ? "cancel" : "edit profile"}
</button>
{/if}
</div>
{:else if viewModel.kind === "loading"}
<div class="user-profile__loading">loading…</div>
{:else}
<div class="user-profile__error">err: {viewModel.message}</div>
{/if}
</header>
{#if viewModel.kind === "ready" && viewModel.data.description}
<p class="user-profile__bio">{viewModel.data.description}</p>
{:else if viewModel.kind === "ready"}
<!--
Empty-state hint for a user who hasn't filled in their profile
yet. Only shown when *no* profile fields are populated (a
partial profile still renders whatever's there). The owner of
an empty profile sees an explicit "set up your profile" hint
instead of an awkward blank space.
-->
{#if !viewModel.data.display_name && !viewModel.data.description && !viewModel.data.avatar_cid}
<p class="user-profile__bio user-profile__bio--empty">
{#if current_user_did && viewModel.data.did === current_user_did}
// no profile yet — click "edit profile" to set one up.
{:else}
// no profile yet.
{/if}
</p>
{/if}
{/if}
{#if editing}
<div class="user-profile__edit">
<label>
<span class="key">display name</span>
<input type="text" bind:value={editName} maxlength="64" />
</label>
<label>
<span class="key">description</span>
<textarea
bind:value={editDesc}
rows="3"
maxlength="300"
></textarea>
</label>
<div class="user-profile__edit-avatar">
<span class="key">avatar</span>
<div class="user-profile__edit-avatar-row">
{#if editAvatarCid}
<span class="meta">cid: {editAvatarCid.slice(0, 10)}</span>
<button class="btn btn--ghost" type="button" onclick={() => (editAvatarCid = null)}>clear</button>
{:else}
<span class="meta">none</span>
{/if}
<button class="btn btn--ghost" type="button" onclick={pickAndUploadAvatar}>upload…</button>
</div>
</div>
<div class="user-profile__edit-actions">
<button class="btn btn--ghost" type="button" disabled={saving} onclick={saveProfile}>
{saving ? "saving…" : "save"}
</button>
</div>
</div>
{/if}
{#if viewModel.kind === "ready"}
<dl class="user-profile__counts">
<div><dt>posts</dt><dd>{viewModel.data.post_count}</dd></div>
<div><dt>followers</dt><dd>{viewModel.data.followers}</dd></div>
<div><dt>following</dt><dd>{viewModel.data.following}</dd></div>
</dl>
<div class="user-profile__posts">
{#each viewModel.data.posts as p (p.uri)}
<PostCard post={p} on_thread_click={on_thread_click} />
{/each}
</div>
{/if}
</section>
<style>
.user-profile {
padding: 0 var(--s-3, 0.75rem);
}
.user-profile__head {
display: grid;
grid-template-columns: auto 1fr auto;
gap: var(--s-3, 0.75rem);
align-items: center;
padding: var(--s-3, 0.75rem) 0;
border-bottom: 1px dashed var(--line-2, #3a3a3a);
}
.user-profile__avatar {
display: flex;
align-items: center;
}
.user-profile__id {
display: flex;
flex-direction: column;
gap: 2px;
}
.user-profile__name {
font-family: var(--font-mono, monospace);
font-size: 1.05rem;
color: var(--text, #e8e8e8);
}
.user-profile__handle {
font-family: var(--font-mono, monospace);
font-size: 0.8rem;
color: var(--orange, #ff6600);
}
.user-profile__did {
font-family: var(--font-mono, monospace);
font-size: 0.7rem;
color: var(--text-dim, #888);
}
.user-profile__bio {
font-family: var(--font-mono, monospace);
font-size: 0.85rem;
color: var(--text, #e8e8e8);
padding: var(--s-3, 0.75rem) 0;
border-bottom: 1px dashed var(--line, #2a2a2a);
}
.user-profile__bio--empty {
color: var(--text-dim, #888);
font-style: italic;
}
.user-profile__edit {
display: flex;
flex-direction: column;
gap: var(--s-2, 0.5rem);
padding: var(--s-3, 0.75rem) 0;
border-bottom: 1px dashed var(--line, #2a2a2a);
}
.user-profile__edit label,
.user-profile__edit-avatar {
display: flex;
align-items: baseline;
gap: var(--s-2, 0.5rem);
}
.user-profile__edit .key {
flex: 0 0 7rem;
color: var(--orange, #ff6600);
font-family: var(--font-mono, monospace);
}
.user-profile__edit input,
.user-profile__edit textarea {
flex: 1;
background: var(--bg, #0d0d0d);
border: 1px solid var(--line-2, #3a3a3a);
color: var(--text, #e8e8e8);
font-family: var(--font-mono, monospace);
padding: var(--s-1, 0.25rem) var(--s-2, 0.5rem);
border-radius: 4px;
}
.user-profile__edit-avatar-row {
flex: 1;
display: flex;
gap: var(--s-2, 0.5rem);
align-items: center;
}
.user-profile__edit-avatar-row .meta {
color: var(--text-dim, #888);
font-family: var(--font-mono, monospace);
font-size: 0.75rem;
}
.user-profile__edit-actions {
display: flex;
justify-content: flex-end;
}
.user-profile__counts {
display: flex;
gap: var(--s-5, 1.5rem);
padding: var(--s-2, 0.5rem) var(--s-3, 0.75rem);
margin: 0;
font-family: var(--font-mono, monospace);
font-size: 0.75rem;
color: var(--text-dim, #888);
}
.user-profile__counts > div {
display: flex;
flex-direction: column;
}
.user-profile__counts dt {
color: var(--text-dim, #888);
letter-spacing: 0.04em;
}
.user-profile__counts dd {
margin: 0;
color: var(--text, #e8e8e8);
font-weight: 700;
}
.user-profile__loading,
.user-profile__error {
font-family: var(--font-mono, monospace);
color: var(--text-dim, #888);
padding: var(--s-3, 0.75rem);
}
.user-profile__error {
color: var(--red, #ff3b30);
}
</style>
+25
View File
@@ -0,0 +1,25 @@
{
"lexicon": 1,
"id": "app.bsky.actor.profile",
"defs": {
"main": {
"type": "record",
"key": "tid",
"record": {
"type": "object",
"properties": {
"displayName": { "type": "string", "maxLength": 64, "maxGraphemes": 64 },
"description": { "type": "string", "maxLength": 300, "maxGraphemes": 300 },
"avatar": {
"type": "blob",
"accept": ["image/png", "image/jpeg", "image/webp", "image/gif"]
},
"banner": {
"type": "blob",
"accept": ["image/png", "image/jpeg", "image/webp"]
}
}
}
}
}
}
+55
View File
@@ -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/<handle>` 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 <Avatar cid={post.avatar_cid}/> 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;
@@ -0,0 +1,25 @@
-- AppView database schema 0006: handle-sync attempt tracking.
--
-- Why
-- The `handle_sync` worker SELECTs DIDs whose `posts.handle` is empty
-- and tries to resolve them via the local PDS → PLC directory →
-- `did:web:` resolver. Some DIDs are *unresolvable* (e.g. a `did:key:`
-- user not hosted on the local PDS, or any `did:foo:` method that
-- neither PLC nor Web understands). Without tracking these, every
-- pass re-selects them and they dominate the 100-row batch — and
-- since `did:key:` sorts lexicographically before `did:plc:` /
-- `did:web:`, the worker would process the same 100 unresolvable
-- `did:key:` rows forever and never reach any resolvable DID.
--
-- With this column the worker marks each empty-handle row with the
-- time of its last attempt. The SELECT filter excludes rows
-- attempted within the last hour, so an unresolvable DID gets at
-- most one attempt per hour and stops blocking forward progress.
-- Rows whose `handle` later gets filled (by another code path) are
-- naturally no longer in the candidate set.
--
-- The column is per-post (not per-DID) because the candidate set is
-- already per-post and the update is cheap (the empty-handle slice
-- is small in steady state).
ALTER TABLE posts ADD COLUMN IF NOT EXISTS handle_sync_attempted_at TIMESTAMPTZ;
@@ -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;