Actor preferences

src/api/actor/mod.rs

@@ -0,0 +1,5 @@
+mod preferences;
+mod profile;
+
+pub use preferences::{get_preferences, put_preferences};
+pub use profile::{get_profile, get_profiles};

src/api/actor/preferences.rs

@@ -0,0 +1,233 @@
+use crate::state::AppState;
+use axum::{
+    extract::State,
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+const APP_BSKY_NAMESPACE: &str = "app.bsky";
+
+#[derive(Serialize)]
+pub struct GetPreferencesOutput {
+    pub preferences: Vec<Value>,
+}
+
+pub async fn get_preferences(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+) -> Response {
+    let token = match crate::auth::extract_bearer_token_from_header(
+        headers.get("Authorization").and_then(|h| h.to_str().ok()),
+    ) {
+        Some(t) => t,
+        None => {
+            return (
+                StatusCode::UNAUTHORIZED,
+                Json(json!({"error": "AuthenticationRequired"})),
+            )
+                .into_response();
+        }
+    };
+
+    let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
+        Ok(user) => user,
+        Err(_) => {
+            return (
+                StatusCode::UNAUTHORIZED,
+                Json(json!({"error": "AuthenticationFailed"})),
+            )
+                .into_response();
+        }
+    };
+
+    let user_id: uuid::Uuid =
+        match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", auth_user.did)
+            .fetch_optional(&state.db)
+            .await
+        {
+            Ok(Some(id)) => id,
+            _ => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": "User not found"})),
+                )
+                    .into_response();
+            }
+        };
+
+    let prefs_result = sqlx::query!(
+        "SELECT name, value_json FROM account_preferences WHERE user_id = $1",
+        user_id
+    )
+    .fetch_all(&state.db)
+    .await;
+
+    let prefs = match prefs_result {
+        Ok(rows) => rows,
+        Err(_) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to fetch preferences"})),
+            )
+                .into_response();
+        }
+    };
+
+    let preferences: Vec<Value> = prefs
+        .into_iter()
+        .filter(|row| {
+            row.name == APP_BSKY_NAMESPACE || row.name.starts_with(&format!("{}.", APP_BSKY_NAMESPACE))
+        })
+        .filter_map(|row| {
+            if row.name == "app.bsky.actor.defs#declaredAgePref" {
+                return None;
+            }
+            serde_json::from_value(row.value_json).ok()
+        })
+        .collect();
+
+    (StatusCode::OK, Json(GetPreferencesOutput { preferences })).into_response()
+}
+
+#[derive(Deserialize)]
+pub struct PutPreferencesInput {
+    pub preferences: Vec<Value>,
+}
+
+pub async fn put_preferences(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    Json(input): Json<PutPreferencesInput>,
+) -> Response {
+    let token = match crate::auth::extract_bearer_token_from_header(
+        headers.get("Authorization").and_then(|h| h.to_str().ok()),
+    ) {
+        Some(t) => t,
+        None => {
+            return (
+                StatusCode::UNAUTHORIZED,
+                Json(json!({"error": "AuthenticationRequired"})),
+            )
+                .into_response();
+        }
+    };
+
+    let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
+        Ok(user) => user,
+        Err(_) => {
+            return (
+                StatusCode::UNAUTHORIZED,
+                Json(json!({"error": "AuthenticationFailed"})),
+            )
+                .into_response();
+        }
+    };
+
+    let user_id: uuid::Uuid =
+        match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", auth_user.did)
+            .fetch_optional(&state.db)
+            .await
+        {
+            Ok(Some(id)) => id,
+            _ => {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(json!({"error": "InternalError", "message": "User not found"})),
+                )
+                    .into_response();
+            }
+        };
+
+    for pref in &input.preferences {
+        let pref_type = match pref.get("$type").and_then(|t| t.as_str()) {
+            Some(t) => t,
+            None => {
+                return (
+                    StatusCode::BAD_REQUEST,
+                    Json(json!({"error": "InvalidRequest", "message": "Preference missing $type field"})),
+                )
+                    .into_response();
+            }
+        };
+
+        if !pref_type.starts_with(APP_BSKY_NAMESPACE) {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidRequest", "message": format!("Invalid preference namespace: {}", pref_type)})),
+            )
+                .into_response();
+        }
+
+        if pref_type == "app.bsky.actor.defs#declaredAgePref" {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidRequest", "message": "declaredAgePref is read-only"})),
+            )
+                .into_response();
+        }
+    }
+
+    let mut tx = match state.db.begin().await {
+        Ok(tx) => tx,
+        Err(_) => {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to start transaction"})),
+            )
+                .into_response();
+        }
+    };
+
+    let delete_result = sqlx::query!(
+        "DELETE FROM account_preferences WHERE user_id = $1 AND (name = $2 OR name LIKE $3)",
+        user_id,
+        APP_BSKY_NAMESPACE,
+        format!("{}.%", APP_BSKY_NAMESPACE)
+    )
+    .execute(&mut *tx)
+    .await;
+
+    if delete_result.is_err() {
+        let _ = tx.rollback().await;
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "Failed to clear preferences"})),
+        )
+            .into_response();
+    }
+
+    for pref in input.preferences {
+        let pref_type = pref.get("$type").and_then(|t| t.as_str()).unwrap();
+
+        let insert_result = sqlx::query!(
+            "INSERT INTO account_preferences (user_id, name, value_json) VALUES ($1, $2, $3)",
+            user_id,
+            pref_type,
+            pref
+        )
+        .execute(&mut *tx)
+        .await;
+
+        if insert_result.is_err() {
+            let _ = tx.rollback().await;
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError", "message": "Failed to save preference"})),
+            )
+                .into_response();
+        }
+    }
+
+    if let Err(_) = tx.commit().await {
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError", "message": "Failed to commit transaction"})),
+        )
+            .into_response();
+    }
+
+    StatusCode::OK.into_response()
+}

src/api/actor/profile.rs

@@ -0,0 +1,206 @@
+use crate::state::AppState;
+use axum::{
+    extract::{Query, State},
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    Json,
+};
+use jacquard_repo::storage::BlockStore;
+use reqwest::Client;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+use std::collections::HashMap;
+use tracing::{error, info};
+
+#[derive(Deserialize)]
+pub struct GetProfileParams {
+    pub actor: String,
+}
+
+#[derive(Deserialize)]
+pub struct GetProfilesParams {
+    pub actors: String,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct ProfileViewDetailed {
+    pub did: String,
+    pub handle: String,
+    #[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: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub banner: Option<String>,
+    #[serde(flatten)]
+    pub extra: HashMap<String, Value>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct GetProfilesOutput {
+    pub profiles: Vec<ProfileViewDetailed>,
+}
+
+async fn get_local_profile_record(state: &AppState, did: &str) -> Option<Value> {
+    let user_id: uuid::Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
+        .fetch_optional(&state.db)
+        .await
+        .ok()??;
+
+    let record_row = sqlx::query!(
+        "SELECT record_cid FROM records WHERE repo_id = $1 AND collection = 'app.bsky.actor.profile' AND rkey = 'self'",
+        user_id
+    )
+    .fetch_optional(&state.db)
+    .await
+    .ok()??;
+
+    let cid: cid::Cid = record_row.record_cid.parse().ok()?;
+    let block_bytes = state.block_store.get(&cid).await.ok()??;
+    serde_ipld_dagcbor::from_slice(&block_bytes).ok()
+}
+
+fn munge_profile_with_local(profile: &mut ProfileViewDetailed, local_record: &Value) {
+    if let Some(display_name) = local_record.get("displayName").and_then(|v| v.as_str()) {
+        profile.display_name = Some(display_name.to_string());
+    }
+    if let Some(description) = local_record.get("description").and_then(|v| v.as_str()) {
+        profile.description = Some(description.to_string());
+    }
+}
+
+async fn proxy_to_appview(
+    method: &str,
+    params: &HashMap<String, String>,
+    auth_header: Option<&str>,
+) -> Result<(StatusCode, Value), Response> {
+    let appview_url = match std::env::var("APPVIEW_URL") {
+        Ok(url) => url,
+        Err(_) => {
+            return Err(
+                (StatusCode::BAD_GATEWAY, Json(json!({"error": "UpstreamError", "message": "No upstream AppView configured"}))).into_response()
+            );
+        }
+    };
+
+    let target_url = format!("{}/xrpc/{}", appview_url, method);
+    info!("Proxying GET request to {}", target_url);
+
+    let client = Client::new();
+    let mut request_builder = client.get(&target_url).query(params);
+
+    if let Some(auth) = auth_header {
+        request_builder = request_builder.header("Authorization", auth);
+    }
+
+    match request_builder.send().await {
+        Ok(resp) => {
+            let status = StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
+            match resp.json::<Value>().await {
+                Ok(body) => Ok((status, body)),
+                Err(e) => {
+                    error!("Error parsing proxy response: {:?}", e);
+                    Err((StatusCode::BAD_GATEWAY, Json(json!({"error": "UpstreamError"}))).into_response())
+                }
+            }
+        }
+        Err(e) => {
+            error!("Error sending proxy request: {:?}", e);
+            if e.is_timeout() {
+                Err((StatusCode::GATEWAY_TIMEOUT, Json(json!({"error": "UpstreamTimeout"}))).into_response())
+            } else {
+                Err((StatusCode::BAD_GATEWAY, Json(json!({"error": "UpstreamError"}))).into_response())
+            }
+        }
+    }
+}
+
+pub async fn get_profile(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    Query(params): Query<GetProfileParams>,
+) -> Response {
+    let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok());
+
+    let auth_did = auth_header.and_then(|h| {
+        let token = crate::auth::extract_bearer_token_from_header(Some(h))?;
+        crate::auth::get_did_from_token(&token).ok()
+    });
+
+    let mut query_params = HashMap::new();
+    query_params.insert("actor".to_string(), params.actor.clone());
+
+    let (status, body) = match proxy_to_appview("app.bsky.actor.getProfile", &query_params, auth_header).await {
+        Ok(r) => r,
+        Err(e) => return e,
+    };
+
+    if !status.is_success() {
+        return (status, Json(body)).into_response();
+    }
+
+    let mut profile: ProfileViewDetailed = match serde_json::from_value(body) {
+        Ok(p) => p,
+        Err(_) => {
+            return (StatusCode::BAD_GATEWAY, Json(json!({"error": "UpstreamError", "message": "Invalid profile response"}))).into_response();
+        }
+    };
+
+    if let Some(ref did) = auth_did {
+        if profile.did == *did {
+            if let Some(local_record) = get_local_profile_record(&state, did).await {
+                munge_profile_with_local(&mut profile, &local_record);
+            }
+        }
+    }
+
+    (StatusCode::OK, Json(profile)).into_response()
+}
+
+pub async fn get_profiles(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    Query(params): Query<GetProfilesParams>,
+) -> Response {
+    let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok());
+
+    let auth_did = auth_header.and_then(|h| {
+        let token = crate::auth::extract_bearer_token_from_header(Some(h))?;
+        crate::auth::get_did_from_token(&token).ok()
+    });
+
+    let mut query_params = HashMap::new();
+    query_params.insert("actors".to_string(), params.actors.clone());
+
+    let (status, body) = match proxy_to_appview("app.bsky.actor.getProfiles", &query_params, auth_header).await {
+        Ok(r) => r,
+        Err(e) => return e,
+    };
+
+    if !status.is_success() {
+        return (status, Json(body)).into_response();
+    }
+
+    let mut output: GetProfilesOutput = match serde_json::from_value(body) {
+        Ok(p) => p,
+        Err(_) => {
+            return (StatusCode::BAD_GATEWAY, Json(json!({"error": "UpstreamError", "message": "Invalid profiles response"}))).into_response();
+        }
+    };
+
+    if let Some(ref did) = auth_did {
+        for profile in &mut output.profiles {
+            if profile.did == *did {
+                if let Some(local_record) = get_local_profile_record(&state, did).await {
+                    munge_profile_with_local(profile, &local_record);
+                }
+                break;
+            }
+        }
+    }
+
+    (StatusCode::OK, Json(output)).into_response()
+}

src/api/identity/account.rs

@@ -415,16 +415,8 @@ pub async fn create_account(
     }
 
     let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
-    if let Err(e) = crate::notifications::enqueue_welcome_email(
-        &state.db,
-        user_id,
-        &input.email,
-        &input.handle,
-        &hostname,
-    )
-    .await
-    {
-        warn!("Failed to enqueue welcome email: {:?}", e);
+    if let Err(e) = crate::notifications::enqueue_welcome(&state.db, user_id, &hostname).await {
+        warn!("Failed to enqueue welcome notification: {:?}", e);
     }
 
     (

src/api/mod.rs

@@ -1,3 +1,4 @@
+pub mod actor;
 pub mod admin;
 pub mod feed;
 pub mod identity;

src/api/server/account_status.rs

@@ -247,11 +247,11 @@ pub async fn request_account_delete(
         }
     };
 
-    let user = match sqlx::query!("SELECT id, email, handle FROM users WHERE did = $1", did)
+    let user_id = match sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
         .fetch_optional(&state.db)
         .await
     {
-        Ok(Some(row)) => row,
+        Ok(Some(id)) => id,
         _ => {
             return (
                 StatusCode::INTERNAL_SERVER_ERROR,
@@ -260,9 +260,6 @@ pub async fn request_account_delete(
                 .into_response();
         }
     };
-    let user_id = user.id;
-    let email = user.email;
-    let handle = user.handle;
 
     let confirmation_token = Uuid::new_v4().to_string();
     let expires_at = Utc::now() + Duration::minutes(15);
@@ -286,15 +283,8 @@ pub async fn request_account_delete(
     }
 
     let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
-    if let Err(e) = crate::notifications::enqueue_account_deletion(
-        &state.db,
-        user_id,
-        &email,
-        &handle,
-        &confirmation_token,
-        &hostname,
-    )
-    .await
+    if let Err(e) =
+        crate::notifications::enqueue_account_deletion(&state.db, user_id, &confirmation_token, &hostname).await
     {
         warn!("Failed to enqueue account deletion notification: {:?}", e);
     }

src/api/server/password.rs

@@ -38,15 +38,12 @@ pub async fn request_password_reset(
             .into_response();
     }
 
-    let user = sqlx::query!(
-        "SELECT id, handle FROM users WHERE LOWER(email) = $1",
-        email
-    )
-    .fetch_optional(&state.db)
-    .await;
+    let user = sqlx::query!("SELECT id FROM users WHERE LOWER(email) = $1", email)
+        .fetch_optional(&state.db)
+        .await;
 
-    let (user_id, handle) = match user {
-        Ok(Some(row)) => (row.id, row.handle),
+    let user_id = match user {
+        Ok(Some(row)) => row.id,
         Ok(None) => {
             info!("Password reset requested for unknown email: {}", email);
             return (StatusCode::OK, Json(json!({}))).into_response();
@@ -83,15 +80,8 @@ pub async fn request_password_reset(
     }
 
     let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
-    if let Err(e) = crate::notifications::enqueue_password_reset(
-        &state.db,
-        user_id,
-        &email,
-        &handle,
-        &code,
-        &hostname,
-    )
-    .await
+    if let Err(e) =
+        crate::notifications::enqueue_password_reset(&state.db, user_id, &code, &hostname).await
     {
         warn!("Failed to enqueue password reset notification: {:?}", e);
     }