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.

crates/pds-server/src/appview_push.rs

@@ -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,

crates/pds-server/src/main.rs

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

crates/pds-server/src/routes/mod.rs

@@ -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;

crates/pds-server/src/routes/profile.rs

@@ -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);
+    }
+}

crates/pds-server/src/state.rs

@@ -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