diff --git a/crates/pds-server/src/appview_push.rs b/crates/pds-server/src/appview_push.rs index a0089d0..f8e8ee2 100644 --- a/crates/pds-server/src/appview_push.rs +++ b/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 { + self.push( + did, + Some(handle), + "app.bsky.actor.profile", + "create", + "self", + None, + Some(record), + None, + ) + .await + } + async fn push( &self, did: &str, diff --git a/crates/pds-server/src/main.rs b/crates/pds-server/src/main.rs index bb86013..cf05462 100644 --- a/crates/pds-server/src/main.rs +++ b/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), diff --git a/crates/pds-server/src/routes/mod.rs b/crates/pds-server/src/routes/mod.rs index 0f12945..f76c011 100644 --- a/crates/pds-server/src/routes/mod.rs +++ b/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; diff --git a/crates/pds-server/src/routes/profile.rs b/crates/pds-server/src/routes/profile.rs new file mode 100644 index 0000000..3531e9a --- /dev/null +++ b/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, +} + +#[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, + /// Optional new bio / description. `null`/missing preserves. + pub description: Option, + /// CID of the avatar blob, already uploaded via uploadBlob. + /// `null`/missing preserves. + pub avatar_blob_cid: Option, + /// CID of the banner blob. `null`/missing preserves. + pub banner_blob_cid: Option, +} + +/// Read the authenticated user's profile record. +pub async fn get_profile( + State(state): State, + headers: axum::http::HeaderMap, +) -> Result, (StatusCode, axum::Json)> { + 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, + headers: axum::http::HeaderMap, + axum::Json(req): axum::Json, +) -> Result, (StatusCode, axum::Json)> { + 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)> { + 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, (StatusCode, axum::Json)> { + 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 = + 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, + 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)> { + 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); + } +} diff --git a/crates/pds-server/src/state.rs b/crates/pds-server/src/state.rs index 8d4ca1f..596b4be 100644 --- a/crates/pds-server/src/state.rs +++ b/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 diff --git a/lexicons/app/bsky/actor/profile.json b/lexicons/app/bsky/actor/profile.json new file mode 100644 index 0000000..ebe2142 --- /dev/null +++ b/lexicons/app/bsky/actor/profile.json @@ -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"] + } + } + } + } + } +}