Add more email-centric endpoints

.sqlx/query-1dfc53ab016cfc704e94aa2cfd9fec2d1f3591bb0e141231506dc76f9da30c4a.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE users SET password_reset_code = NULL, password_reset_code_expires_at = NULL WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "1dfc53ab016cfc704e94aa2cfd9fec2d1f3591bb0e141231506dc76f9da30c4a"
+}

.sqlx/query-303777d97e6ed344f8c699eae37b7b0c241c734a5b7726019c2a59ae277caee6.json

@@ -34,7 +34,8 @@
                 "email_verification",
                 "password_reset",
                 "email_update",
-                "account_deletion"
+                "account_deletion",
+                "admin_email"
               ]
             }
           }

.sqlx/query-40bd5b538224352dd8912e2c13a71b920ee3874f4acc12ebf4e6f62aae86c556.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, handle FROM users WHERE LOWER(email) = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "handle",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "40bd5b538224352dd8912e2c13a71b920ee3874f4acc12ebf4e6f62aae86c556"
+}

.sqlx/query-5d49bbf0307a0c642b0174d641de748fa648c97f8109255120e969c957ff95bf.json

@@ -34,7 +34,8 @@
                 "email_verification",
                 "password_reset",
                 "email_update",
-                "account_deletion"
+                "account_deletion",
+                "admin_email"
               ]
             }
           }

.sqlx/query-8786517e60ebcbc4150930ef766b14ee6766359ef9ca09d54116a40450a439b8.json

@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, password_reset_code, password_reset_code_expires_at FROM users WHERE password_reset_code = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "password_reset_code",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "password_reset_code_expires_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      true,
+      true
+    ]
+  },
+  "hash": "8786517e60ebcbc4150930ef766b14ee6766359ef9ca09d54116a40450a439b8"
+}

.sqlx/query-9387c8162414807caaf9380a57fc720b6d25cccfb54b9feda15c08560a7562cc.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE users SET password_reset_code = $1, password_reset_code_expires_at = $2 WHERE id = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Timestamptz",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "9387c8162414807caaf9380a57fc720b6d25cccfb54b9feda15c08560a7562cc"
+}

.sqlx/query-cb6f48aaba124c79308d20e66c23adb44d1196296b7f93fad19b2d17548ed3de.json

@@ -42,7 +42,8 @@
                 "email_verification",
                 "password_reset",
                 "email_update",
-                "account_deletion"
+                "account_deletion",
+                "admin_email"
               ]
             }
           }

.sqlx/query-d7259198aa28f202fbc5bb9466c8a16446b664532e1bc9eff6a783652265229b.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE users SET password_hash = $1, password_reset_code = NULL, password_reset_code_expires_at = NULL WHERE id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "d7259198aa28f202fbc5bb9466c8a16446b664532e1bc9eff6a783652265229b"
+}

.sqlx/query-e6a085193cbc5901c41e23c296ce3358bfd252e68502e5b8ccc9821d479d3c67.json

@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT id, email, handle FROM users WHERE did = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 1,
+        "name": "email",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "handle",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "e6a085193cbc5901c41e23c296ce3358bfd252e68502e5b8ccc9821d479d3c67"
+}

.sqlx/query-fe9d108977af562e9e0439e755749253e52d92031e27a71d18b21265b20a4535.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM sessions WHERE did = (SELECT did FROM users WHERE id = $1)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "fe9d108977af562e9e0439e755749253e52d92031e27a71d18b21265b20a4535"
+}

TODO.md

@@ -36,7 +36,7 @@ Lewis' corrected big boy todofile
     - [x] Implement `com.atproto.server.listAppPasswords`.
     - [x] Implement `com.atproto.server.requestAccountDelete`.
     - [ ] Implement `com.atproto.server.requestEmailConfirmation` / `requestEmailUpdate`.
-    - [ ] Implement `com.atproto.server.requestPasswordReset` / `resetPassword`.
+    - [x] Implement `com.atproto.server.requestPasswordReset` / `resetPassword`.
     - [ ] Implement `com.atproto.server.reserveSigningKey`.
     - [x] Implement `com.atproto.server.revokeAppPassword`.
     - [ ] Implement `com.atproto.server.updateEmail`.
@@ -97,7 +97,7 @@ Lewis' corrected big boy todofile
 - [x] Implement `com.atproto.admin.getAccountInfo` / `getAccountInfos`.
 - [x] Implement `com.atproto.admin.getInviteCodes`.
 - [x] Implement `com.atproto.admin.getSubjectStatus`.
-- [ ] Implement `com.atproto.admin.sendEmail`.
+- [x] Implement `com.atproto.admin.sendEmail`.
 - [x] Implement `com.atproto.admin.updateAccountEmail`.
 - [x] Implement `com.atproto.admin.updateAccountHandle`.
 - [x] Implement `com.atproto.admin.updateAccountPassword`.

migrations/202512212100_password_reset.sql

@@ -0,0 +1,4 @@
+ALTER TABLE users ADD COLUMN IF NOT EXISTS password_reset_code TEXT;
+ALTER TABLE users ADD COLUMN IF NOT EXISTS password_reset_code_expires_at TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS idx_users_password_reset_code ON users(password_reset_code) WHERE password_reset_code IS NOT NULL;

migrations/202512212200_admin_email_type.sql

@@ -0,0 +1 @@
+ALTER TYPE notification_type ADD VALUE IF NOT EXISTS 'admin_email';

src/api/admin/mod.rs

@@ -7,7 +7,7 @@ use axum::{
 };
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use tracing::error;
+use tracing::{error, warn};
 
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -1115,3 +1115,109 @@ pub async fn update_account_password(
         }
     }
 }
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SendEmailInput {
+    pub recipient_did: String,
+    pub sender_did: String,
+    pub content: String,
+    pub subject: Option<String>,
+    pub comment: Option<String>,
+}
+
+#[derive(Serialize)]
+pub struct SendEmailOutput {
+    pub sent: bool,
+}
+
+pub async fn send_email(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    Json(input): Json<SendEmailInput>,
+) -> Response {
+    let auth_header = headers.get("Authorization");
+    if auth_header.is_none() {
+        return (
+            StatusCode::UNAUTHORIZED,
+            Json(json!({"error": "AuthenticationRequired"})),
+        )
+            .into_response();
+    }
+
+    let recipient_did = input.recipient_did.trim();
+    let content = input.content.trim();
+
+    if recipient_did.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "recipientDid is required"})),
+        )
+            .into_response();
+    }
+
+    if content.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "content is required"})),
+        )
+            .into_response();
+    }
+
+    let user = sqlx::query!(
+        "SELECT id, email, handle FROM users WHERE did = $1",
+        recipient_did
+    )
+    .fetch_optional(&state.db)
+    .await;
+
+    let (user_id, email, handle) = match user {
+        Ok(Some(row)) => (row.id, row.email, row.handle),
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(json!({"error": "AccountNotFound", "message": "Recipient account not found"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error in send_email: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError"})),
+            )
+                .into_response();
+        }
+    };
+
+    let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string());
+    let subject = input
+        .subject
+        .clone()
+        .unwrap_or_else(|| format!("Message from {}", hostname));
+
+    let notification = crate::notifications::NewNotification::email(
+        user_id,
+        crate::notifications::NotificationType::AdminEmail,
+        email,
+        subject,
+        content.to_string(),
+    );
+
+    let result = crate::notifications::enqueue_notification(&state.db, notification).await;
+
+    match result {
+        Ok(_) => {
+            tracing::info!(
+                "Admin email queued for {} ({})",
+                handle,
+                recipient_did
+            );
+            (StatusCode::OK, Json(SendEmailOutput { sent: true })).into_response()
+        }
+        Err(e) => {
+            warn!("Failed to enqueue admin email: {:?}", e);
+            (StatusCode::OK, Json(SendEmailOutput { sent: false })).into_response()
+        }
+    }
+}

src/api/server/mod.rs

@@ -7,5 +7,6 @@ pub use meta::{describe_server, health};
 pub use session::{
     activate_account, check_account_status, create_app_password, create_session,
     deactivate_account, delete_session, get_service_auth, get_session, list_app_passwords,
-    refresh_session, request_account_delete, revoke_app_password,
+    refresh_session, request_account_delete, request_password_reset, reset_password,
+    revoke_app_password,
 };

src/api/server/session.rs

@@ -5,12 +5,13 @@ use axum::{
     http::StatusCode,
     response::{IntoResponse, Response},
 };
-use bcrypt::verify;
+use bcrypt::{hash, verify, DEFAULT_COST};
 use chrono::{Duration, Utc};
-use uuid::Uuid;
+use rand::Rng;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use tracing::{error, info, warn};
+use uuid::Uuid;
 
 #[derive(Deserialize)]
 pub struct GetServiceAuthParams {
@@ -1210,3 +1211,211 @@ pub async fn revoke_app_password(
         }
     }
 }
+
+fn generate_reset_code() -> String {
+    let mut rng = rand::thread_rng();
+    let chars: Vec<char> = "abcdefghijklmnopqrstuvwxyz234567".chars().collect();
+    let part1: String = (0..5).map(|_| chars[rng.gen_range(0..chars.len())]).collect();
+    let part2: String = (0..5).map(|_| chars[rng.gen_range(0..chars.len())]).collect();
+    format!("{}-{}", part1, part2)
+}
+
+#[derive(Deserialize)]
+pub struct RequestPasswordResetInput {
+    pub email: String,
+}
+
+pub async fn request_password_reset(
+    State(state): State<AppState>,
+    Json(input): Json<RequestPasswordResetInput>,
+) -> Response {
+    let email = input.email.trim().to_lowercase();
+    if email.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "email is required"})),
+        )
+            .into_response();
+    }
+
+    let user = sqlx::query!(
+        "SELECT id, handle 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),
+        Ok(None) => {
+            info!("Password reset requested for unknown email: {}", email);
+            return (StatusCode::OK, Json(json!({}))).into_response();
+        }
+        Err(e) => {
+            error!("DB error in request_password_reset: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError"})),
+            )
+                .into_response();
+        }
+    };
+
+    let code = generate_reset_code();
+    let expires_at = Utc::now() + Duration::minutes(10);
+
+    let update = sqlx::query!(
+        "UPDATE users SET password_reset_code = $1, password_reset_code_expires_at = $2 WHERE id = $3",
+        code,
+        expires_at,
+        user_id
+    )
+    .execute(&state.db)
+    .await;
+
+    if let Err(e) = update {
+        error!("DB error setting reset code: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response();
+    }
+
+    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
+    {
+        warn!("Failed to enqueue password reset notification: {:?}", e);
+    }
+
+    info!("Password reset requested for user {}", user_id);
+
+    (StatusCode::OK, Json(json!({}))).into_response()
+}
+
+#[derive(Deserialize)]
+pub struct ResetPasswordInput {
+    pub token: String,
+    pub password: String,
+}
+
+pub async fn reset_password(
+    State(state): State<AppState>,
+    Json(input): Json<ResetPasswordInput>,
+) -> Response {
+    let token = input.token.trim();
+    let password = &input.password;
+
+    if token.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidToken", "message": "token is required"})),
+        )
+            .into_response();
+    }
+
+    if password.is_empty() {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidRequest", "message": "password is required"})),
+        )
+            .into_response();
+    }
+
+    let user = sqlx::query!(
+        "SELECT id, password_reset_code, password_reset_code_expires_at FROM users WHERE password_reset_code = $1",
+        token
+    )
+    .fetch_optional(&state.db)
+    .await;
+
+    let (user_id, expires_at) = match user {
+        Ok(Some(row)) => {
+            let expires = row.password_reset_code_expires_at;
+            (row.id, expires)
+        }
+        Ok(None) => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "InvalidToken", "message": "Invalid or expired token"})),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            error!("DB error in reset_password: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError"})),
+            )
+                .into_response();
+        }
+    };
+
+    if let Some(exp) = expires_at {
+        if Utc::now() > exp {
+            let _ = sqlx::query!(
+                "UPDATE users SET password_reset_code = NULL, password_reset_code_expires_at = NULL WHERE id = $1",
+                user_id
+            )
+            .execute(&state.db)
+            .await;
+
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(json!({"error": "ExpiredToken", "message": "Token has expired"})),
+            )
+                .into_response();
+        }
+    } else {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({"error": "InvalidToken", "message": "Invalid or expired token"})),
+        )
+            .into_response();
+    }
+
+    let password_hash = match hash(password, DEFAULT_COST) {
+        Ok(h) => h,
+        Err(e) => {
+            error!("Failed to hash password: {:?}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(json!({"error": "InternalError"})),
+            )
+                .into_response();
+        }
+    };
+
+    let update = sqlx::query!(
+        "UPDATE users SET password_hash = $1, password_reset_code = NULL, password_reset_code_expires_at = NULL WHERE id = $2",
+        password_hash,
+        user_id
+    )
+    .execute(&state.db)
+    .await;
+
+    if let Err(e) = update {
+        error!("DB error updating password: {:?}", e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(json!({"error": "InternalError"})),
+        )
+            .into_response();
+    }
+
+    let _ = sqlx::query!("DELETE FROM sessions WHERE did = (SELECT did FROM users WHERE id = $1)", user_id)
+        .execute(&state.db)
+        .await;
+
+    info!("Password reset completed for user {}", user_id);
+
+    (StatusCode::OK, Json(json!({}))).into_response()
+}

src/lib.rs

@@ -155,6 +155,14 @@ pub fn app(state: AppState) -> Router {
             "/xrpc/com.atproto.server.requestAccountDelete",
             post(api::server::request_account_delete),
         )
+        .route(
+            "/xrpc/com.atproto.server.requestPasswordReset",
+            post(api::server::request_password_reset),
+        )
+        .route(
+            "/xrpc/com.atproto.server.resetPassword",
+            post(api::server::reset_password),
+        )
         .route(
             "/xrpc/com.atproto.identity.updateHandle",
             post(api::identity::update_handle),
@@ -223,6 +231,10 @@ pub fn app(state: AppState) -> Router {
             "/xrpc/com.atproto.admin.updateSubjectStatus",
             post(api::admin::update_subject_status),
         )
+        .route(
+            "/xrpc/com.atproto.admin.sendEmail",
+            post(api::admin::send_email),
+        )
         // I know I know, I'm not supposed to implement appview endpoints. Leave me be
         .route(
             "/xrpc/app.bsky.feed.getTimeline",

src/notifications/types.rs

@@ -29,6 +29,7 @@ pub enum NotificationType {
     PasswordReset,
     EmailUpdate,
     AccountDeletion,
+    AdminEmail,
 }
 
 #[derive(Debug, Clone, FromRow)]

tests/admin_email.rs

@@ -0,0 +1,188 @@
+mod common;
+
+use reqwest::StatusCode;
+use serde_json::{json, Value};
+use sqlx::PgPool;
+
+async fn get_pool() -> PgPool {
+    let conn_str = common::get_db_connection_string().await;
+    sqlx::postgres::PgPoolOptions::new()
+        .max_connections(5)
+        .connect(&conn_str)
+        .await
+        .expect("Failed to connect to test database")
+}
+
+#[tokio::test]
+async fn test_send_email_success() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+    let pool = get_pool().await;
+
+    let (access_jwt, did) = common::create_account_and_login(&client).await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
+        .bearer_auth(&access_jwt)
+        .json(&json!({
+            "recipientDid": did,
+            "senderDid": "did:plc:admin",
+            "content": "Hello, this is a test email from the admin.",
+            "subject": "Test Admin Email"
+        }))
+        .send()
+        .await
+        .expect("Failed to send email");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["sent"], true);
+
+    let user = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
+        .fetch_one(&pool)
+        .await
+        .expect("User not found");
+
+    let notification = sqlx::query!(
+        "SELECT subject, body, notification_type as \"notification_type: String\" FROM notification_queue WHERE user_id = $1 AND notification_type = 'admin_email' ORDER BY created_at DESC LIMIT 1",
+        user.id
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("Notification not found");
+
+    assert_eq!(notification.subject.as_deref(), Some("Test Admin Email"));
+    assert!(notification.body.contains("Hello, this is a test email from the admin."));
+}
+
+#[tokio::test]
+async fn test_send_email_default_subject() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+    let pool = get_pool().await;
+
+    let (access_jwt, did) = common::create_account_and_login(&client).await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
+        .bearer_auth(&access_jwt)
+        .json(&json!({
+            "recipientDid": did,
+            "senderDid": "did:plc:admin",
+            "content": "Email without subject"
+        }))
+        .send()
+        .await
+        .expect("Failed to send email");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["sent"], true);
+
+    let user = sqlx::query!("SELECT id FROM users WHERE did = $1", did)
+        .fetch_one(&pool)
+        .await
+        .expect("User not found");
+
+    let notification = sqlx::query!(
+        "SELECT subject FROM notification_queue WHERE user_id = $1 AND notification_type = 'admin_email' AND body = 'Email without subject' LIMIT 1",
+        user.id
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("Notification not found");
+
+    assert!(notification.subject.is_some());
+    assert!(notification.subject.unwrap().contains("Message from"));
+}
+
+#[tokio::test]
+async fn test_send_email_recipient_not_found() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let (access_jwt, _) = common::create_account_and_login(&client).await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
+        .bearer_auth(&access_jwt)
+        .json(&json!({
+            "recipientDid": "did:plc:nonexistent",
+            "senderDid": "did:plc:admin",
+            "content": "Test content"
+        }))
+        .send()
+        .await
+        .expect("Failed to send email");
+
+    assert_eq!(res.status(), StatusCode::NOT_FOUND);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["error"], "AccountNotFound");
+}
+
+#[tokio::test]
+async fn test_send_email_missing_content() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let (access_jwt, did) = common::create_account_and_login(&client).await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
+        .bearer_auth(&access_jwt)
+        .json(&json!({
+            "recipientDid": did,
+            "senderDid": "did:plc:admin",
+            "content": ""
+        }))
+        .send()
+        .await
+        .expect("Failed to send email");
+
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["error"], "InvalidRequest");
+}
+
+#[tokio::test]
+async fn test_send_email_missing_recipient() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let (access_jwt, _) = common::create_account_and_login(&client).await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
+        .bearer_auth(&access_jwt)
+        .json(&json!({
+            "recipientDid": "",
+            "senderDid": "did:plc:admin",
+            "content": "Test content"
+        }))
+        .send()
+        .await
+        .expect("Failed to send email");
+
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["error"], "InvalidRequest");
+}
+
+#[tokio::test]
+async fn test_send_email_requires_auth() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.admin.sendEmail", base_url))
+        .json(&json!({
+            "recipientDid": "did:plc:test",
+            "senderDid": "did:plc:admin",
+            "content": "Test content"
+        }))
+        .send()
+        .await
+        .expect("Failed to send email");
+
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+}

tests/password_reset.rs

@@ -0,0 +1,393 @@
+mod common;
+
+use reqwest::StatusCode;
+use serde_json::{json, Value};
+use sqlx::PgPool;
+
+async fn get_pool() -> PgPool {
+    let conn_str = common::get_db_connection_string().await;
+    sqlx::postgres::PgPoolOptions::new()
+        .max_connections(5)
+        .connect(&conn_str)
+        .await
+        .expect("Failed to connect to test database")
+}
+
+#[tokio::test]
+async fn test_request_password_reset_creates_code() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+    let pool = get_pool().await;
+
+    let handle = format!("pwreset_{}", uuid::Uuid::new_v4());
+    let email = format!("{}@example.com", handle);
+    let payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": "oldpassword"
+    });
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createAccount", base_url))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to create account");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": email}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let user = sqlx::query!(
+        "SELECT password_reset_code, password_reset_code_expires_at FROM users WHERE email = $1",
+        email
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("User not found");
+
+    assert!(user.password_reset_code.is_some());
+    assert!(user.password_reset_code_expires_at.is_some());
+
+    let code = user.password_reset_code.unwrap();
+    assert!(code.contains('-'));
+    assert_eq!(code.len(), 11);
+}
+
+#[tokio::test]
+async fn test_request_password_reset_unknown_email_returns_ok() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": "nonexistent@example.com"}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_reset_password_with_valid_token() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+    let pool = get_pool().await;
+
+    let handle = format!("pwreset2_{}", uuid::Uuid::new_v4());
+    let email = format!("{}@example.com", handle);
+    let old_password = "oldpassword";
+    let new_password = "newpassword123";
+
+    let payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": old_password
+    });
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createAccount", base_url))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to create account");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": email}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let user = sqlx::query!(
+        "SELECT password_reset_code FROM users WHERE email = $1",
+        email
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("User not found");
+
+    let token = user.password_reset_code.expect("No reset code");
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.resetPassword", base_url))
+        .json(&json!({
+            "token": token,
+            "password": new_password
+        }))
+        .send()
+        .await
+        .expect("Failed to reset password");
+
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let user = sqlx::query!(
+        "SELECT password_reset_code, password_reset_code_expires_at FROM users WHERE email = $1",
+        email
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("User not found");
+    assert!(user.password_reset_code.is_none());
+    assert!(user.password_reset_code_expires_at.is_none());
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createSession", base_url))
+        .json(&json!({
+            "identifier": handle,
+            "password": new_password
+        }))
+        .send()
+        .await
+        .expect("Failed to login");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createSession", base_url))
+        .json(&json!({
+            "identifier": handle,
+            "password": old_password
+        }))
+        .send()
+        .await
+        .expect("Failed to login attempt");
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+}
+
+#[tokio::test]
+async fn test_reset_password_with_invalid_token() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.resetPassword", base_url))
+        .json(&json!({
+            "token": "invalid-token",
+            "password": "newpassword"
+        }))
+        .send()
+        .await
+        .expect("Failed to reset password");
+
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["error"], "InvalidToken");
+}
+
+#[tokio::test]
+async fn test_reset_password_with_expired_token() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+    let pool = get_pool().await;
+
+    let handle = format!("pwreset3_{}", uuid::Uuid::new_v4());
+    let email = format!("{}@example.com", handle);
+
+    let payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": "oldpassword"
+    });
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createAccount", base_url))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to create account");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": email}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let user = sqlx::query!(
+        "SELECT password_reset_code FROM users WHERE email = $1",
+        email
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("User not found");
+
+    let token = user.password_reset_code.expect("No reset code");
+
+    sqlx::query!(
+        "UPDATE users SET password_reset_code_expires_at = NOW() - INTERVAL '1 hour' WHERE email = $1",
+        email
+    )
+    .execute(&pool)
+    .await
+    .expect("Failed to expire token");
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.resetPassword", base_url))
+        .json(&json!({
+            "token": token,
+            "password": "newpassword"
+        }))
+        .send()
+        .await
+        .expect("Failed to reset password");
+
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["error"], "ExpiredToken");
+}
+
+#[tokio::test]
+async fn test_reset_password_invalidates_sessions() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+    let pool = get_pool().await;
+
+    let handle = format!("pwreset4_{}", uuid::Uuid::new_v4());
+    let email = format!("{}@example.com", handle);
+
+    let payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": "oldpassword"
+    });
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createAccount", base_url))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to create account");
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    let original_token = body["accessJwt"].as_str().expect("No accessJwt").to_string();
+
+    let res = client
+        .get(format!("{}/xrpc/com.atproto.server.getSession", base_url))
+        .bearer_auth(&original_token)
+        .send()
+        .await
+        .expect("Failed to get session");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": email}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let user = sqlx::query!(
+        "SELECT password_reset_code FROM users WHERE email = $1",
+        email
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("User not found");
+
+    let token = user.password_reset_code.expect("No reset code");
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.resetPassword", base_url))
+        .json(&json!({
+            "token": token,
+            "password": "newpassword123"
+        }))
+        .send()
+        .await
+        .expect("Failed to reset password");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let res = client
+        .get(format!("{}/xrpc/com.atproto.server.getSession", base_url))
+        .bearer_auth(&original_token)
+        .send()
+        .await
+        .expect("Failed to get session");
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+}
+
+#[tokio::test]
+async fn test_request_password_reset_empty_email() {
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": ""}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert_eq!(body["error"], "InvalidRequest");
+}
+
+#[tokio::test]
+async fn test_reset_password_creates_notification() {
+    let pool = get_pool().await;
+    let client = common::client();
+    let base_url = common::base_url().await;
+
+    let handle = format!("pwreset5_{}", uuid::Uuid::new_v4());
+    let email = format!("{}@example.com", handle);
+
+    let payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": "oldpassword"
+    });
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.createAccount", base_url))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to create account");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let user = sqlx::query!("SELECT id FROM users WHERE email = $1", email)
+        .fetch_one(&pool)
+        .await
+        .expect("User not found");
+
+    let initial_count: i64 = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM notification_queue WHERE user_id = $1 AND notification_type = 'password_reset'",
+        user.id
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("Failed to count")
+    .unwrap_or(0);
+
+    let res = client
+        .post(format!("{}/xrpc/com.atproto.server.requestPasswordReset", base_url))
+        .json(&json!({"email": email}))
+        .send()
+        .await
+        .expect("Failed to request password reset");
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let final_count: i64 = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM notification_queue WHERE user_id = $1 AND notification_type = 'password_reset'",
+        user.id
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("Failed to count")
+    .unwrap_or(0);
+
+    assert_eq!(final_count - initial_count, 1);
+}